diff --git a/README.md b/README.md
index fbcfd4a1f3..f9ace34b34 100644
--- a/README.md
+++ b/README.md
@@ -1184,6 +1184,14 @@ overwriting them with an unevaluated fallback — so the Settings UI shows
 whether `lark-cli` will actually be present in the sandbox at chat time, rather
 than a green status hiding a later `command not found`.
 
+In Lark broker mode, AIO's fresh Bash runs close inherited pipe stdin. Persistent
+terminal commands keep their input; the shim ignores terminal stdin. Explicit
+pipelines, heredocs, and file redirections still supply input normally, and the
+shim forwards that input only after EOF. A stdin idle timeout aborts without
+executing the command, and broker execution logs omit argument values. See the
+[broker image guide](docker/lark-cli-broker/README.md) for timeout settings and
+image rebuild requirements.
+
 If a trusted operator manages the configured skills directory through an external mount such as MinIO, NFS, or CSI, an administrator can call `POST /api/skills/reload` after changing files. This invalidates skill prompt caches for the current Gateway process and waits up to the bounded refresh timeout so subsequent runs rescan the latest files; running tasks are unchanged. A loader-level filesystem failure returns a generic server error and preserves the last successfully loaded process cache rather than publishing an empty catalog. Uvicorn workers and Kubernetes Pods must each be targeted separately. Direct mount writes bypass the validation, SkillScan, and history applied by DeerFlow's install/edit APIs, so only operator-controlled systems should have write access.
 
 Skill installs and agent-managed skill edits run through **SkillScan**, a native deterministic safety scanner before the LLM-based skill scanner. Phase 1 runs offline with no Semgrep/OpenGrep dependency, blocks high-confidence `CRITICAL` findings such as private keys or shell execution, and passes warning findings to the LLM scanner for contextual review. Code files (anything under `scripts/`, a script suffix such as `.py`, `.sh`, or `.js`, or an extensionless file starting with `#!`) that are not NUL-free UTF-8 text raise a warning and are still analyzed over a lossy decode, so a single stray byte cannot hide them from `CRITICAL` checks. The moderation adapter normalizes both plain-text model responses and LangChain Responses API text blocks before parsing the required JSON decision. Python instance-client exfiltration checks follow a minimal same-scope evidence chain: a simple name bound to a known client constructor, optional name-to-name aliases, and an actual outbound method or context-manager use supported by that constructor. Constructor roots must be proven imports; bare canonical-looking names are not inferred as modules. Nested scopes do not inherit client handles and inherit only constructor import aliases that are never rebound in the enclosing scope. Comprehensions, walrus-bearing statements, annotations, complex binding targets, unsupported operations, and ambiguous branch flows produce no finding from this signal; skipped constructs conservatively invalidate every name they may bind so stale client state cannot create a finding. A deterministic work budget or recursion limit reached by this best-effort analysis does not discard findings already collected for the file. Set `skill_scan.enabled: false` in `config.yaml` to disable only the deterministic analyzers; safe archive extraction and the LLM scanner still run.
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md b/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md
new file mode 100644
index 0000000000..14ee56cf74
--- /dev/null
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md
@@ -0,0 +1,21 @@
+# AIO Sandbox
+
+## Stdin contracts per transport
+
+AIO has two distinct stdin contracts: the persistent `/v1/shell` transport is a
+PTY, while `/v1/bash` uses a subprocess pipe that stays open for writes. Keep
+persistent shell commands unchanged: the Lark broker shim already ignores TTY
+input. Only fresh `_run_bash_exec` calls prefix the original command with
+`exec < /dev/null` so default stdin gets immediate EOF; explicit
+pipes/heredocs/files still override fd0. Those sessions are created and
+released per invocation, so the prefix never closes a reusable terminal or
+alters its state. Avoid brace-group/eval wrappers: they shift top-level
+parsing (alias/extglob timing), disturb `PIPESTATUS` across calls, can fire an
+`ERR` trap an extra time, or append a delimiter a trailing backslash consumes.
+
+The broker shim forwards pipe/file input only after EOF: an idle pipe exits
+124 without a broker request, and read errors fail rather than becoming empty
+input. Window overrides must be finite positive seconds, at most 600. Broker
+INFO logs contain argc/exit/elapsed only, never argument values.
+
+Regressions: `tests/test_aio_sandbox.py`, `tests/test_lark_broker.py`.
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py
index 453155cde1..d47aec302c 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py
@@ -399,6 +399,8 @@ class AioSandbox(Sandbox):
         timeout: float,
     ) -> tuple[str, int | None, str | None]:
         kwargs = {
+            # /v1/shell is a persistent PTY. Keep its command and terminal stdin
+            # intact; the broker shim already treats a TTY as non-payload input.
             "command": command,
             "no_change_timeout": self._effective_no_change_timeout(timeout),
             "hard_timeout": timeout,
@@ -916,7 +918,14 @@ class AioSandbox(Sandbox):
                 try:
                     session_id = self._create_bash_session(self._client)
                     result = self._client.bash.exec(
-                        command=command,
+                        # /v1/bash keeps a subprocess stdin pipe open for writes.
+                        # This fresh, released session is non-interactive, so close
+                        # its default input before running the original script.
+                        # Explicit pipes/heredocs/files still override fd0. A plain
+                        # prefix keeps top-level parsing (aliases/extglob) and never
+                        # appends a delimiter that a trailing backslash can consume.
+                        # Do not apply exec to the persistent PTY transport above.
+                        command=f"exec < /dev/null\n{command}",
                         session_id=session_id,
                         env=env,
                         hard_timeout=timeout,
diff --git a/backend/packages/harness/deerflow/integrations/lark_broker.py b/backend/packages/harness/deerflow/integrations/lark_broker.py
index 35dfa17c76..1eb40d73c6 100644
--- a/backend/packages/harness/deerflow/integrations/lark_broker.py
+++ b/backend/packages/harness/deerflow/integrations/lark_broker.py
@@ -25,6 +25,7 @@ import shutil
 import subprocess
 import sys
 import threading
+import time
 from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any
@@ -120,6 +121,8 @@ def render_launcher_script(shim_path: str) -> str:
 # 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.
+# It runs as a standalone script inside the sandbox, so it carries its own
+# imports (select/time included) that deliberately duplicate this module's.
 LARK_CLI_BROKER_SHIM_SCRIPT = r'''#!/usr/bin/env python3
 """DeerFlow lark-cli broker shim (Pattern B). Forwards argv/stdin to the broker.
 
@@ -129,24 +132,71 @@ that read/write files relative to the sandbox cwd are unsupported in broker mode
 """
 import base64
 import json
+import math
 import os
+import select
 import sys
+import time
 import urllib.error
 import urllib.request
 
 BROKER_URL = os.environ.get("DEERFLOW_LARK_BROKER_URL", "http://127.0.0.1:8788")
 
+# Non-interactive executors must close inherited session stdin, while explicit
+# pipes/files are drained to EOF. These idle windows are safety budgets, never
+# evidence that input is complete: timeout must abort before calling the broker.
+_STDIN_GRACE_DEFAULT = 2.0
+_STDIN_TAIL_DEFAULT = 2.0
+STDIN_CHUNK_BYTES = 65536
+
+
+def _stdin_window_seconds(name, default):
+    """Read an operator override for a stdin window; fall back on junk values."""
+    try:
+        value = float(os.environ.get(name, ""))
+    except (TypeError, ValueError):
+        return default
+    return value if math.isfinite(value) and 0 < value <= 600 else default
+
+
+STDIN_GRACE_SECONDS = _stdin_window_seconds("DEERFLOW_LARK_BROKER_STDIN_GRACE_SECONDS", _STDIN_GRACE_DEFAULT)
+STDIN_TAIL_SECONDS = _stdin_window_seconds("DEERFLOW_LARK_BROKER_STDIN_TAIL_SECONDS", _STDIN_TAIL_DEFAULT)
+
 
 def _fail(message, code=127):
     sys.stderr.write("lark-cli: " + message + "\n")
     sys.exit(code)
 
 
+def _read_stdin():
+    """Read complete input, or abort without executing a partial command."""
+    if sys.stdin is None or sys.stdin.isatty():
+        return b""
+    if os.name == "nt":
+        # select() cannot wait on Windows pipes; deployed sandbox images are POSIX.
+        return sys.stdin.buffer.read()
+    fd = sys.stdin.fileno()
+    chunks = []
+    idle_deadline = time.monotonic() + STDIN_GRACE_SECONDS
+    while True:
+        remaining = idle_deadline - time.monotonic()
+        if remaining <= 0:
+            _fail("stdin timed out before EOF; command was not executed", 124)
+        ready, _, _ = select.select([fd], [], [], remaining)
+        if not ready:
+            _fail("stdin timed out before EOF; command was not executed", 124)
+        chunk = os.read(fd, STDIN_CHUNK_BYTES)
+        if not chunk:
+            return b"".join(chunks)
+        chunks.append(chunk)
+        idle_deadline = time.monotonic() + STDIN_TAIL_SECONDS
+
+
 def main():
     try:
-        stdin_bytes = b"" if sys.stdin is None or sys.stdin.isatty() else sys.stdin.buffer.read()
+        stdin_bytes = _read_stdin()
     except Exception:
-        stdin_bytes = b""
+        _fail("cannot read stdin; command was not executed")
     payload = json.dumps(
         {
             "args": sys.argv[1:],
@@ -353,8 +403,10 @@ def make_handler(config: BrokerConfig) -> type[BaseHTTPRequestHandler]:
                 return
 
             if not semaphore.acquire(blocking=False):
+                logger.warning("lark-cli exec rejected: broker busy (concurrency cap %d)", LARK_BROKER_MAX_CONCURRENCY)
                 self._send_json(503, {"error": "broker busy"})
                 return
+            started = time.monotonic()
             try:
                 result = run_lark_cli(config, args, stdin)
             except Exception:  # noqa: BLE001 - keep the wire contract uniform
@@ -369,6 +421,13 @@ def make_handler(config: BrokerConfig) -> type[BaseHTTPRequestHandler]:
             finally:
                 semaphore.release()
 
+            logger.info(
+                "lark-cli exec argc=%d exit=%d in %.2fs%s",
+                len(args),
+                result.exit_code,
+                time.monotonic() - started,
+                " (output truncated)" if result.truncated else "",
+            )
             self._send_json(
                 200,
                 {
diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py
index f1ab584659..7f41635012 100644
--- a/backend/tests/test_agent_guidance_check.py
+++ b/backend/tests/test_agent_guidance_check.py
@@ -22,6 +22,7 @@ EXPECTED_GUIDANCE_PATHS = {
     "backend/packages/harness/deerflow/community/ragflow/AGENTS.md",
     "backend/packages/harness/deerflow/community/tavily/AGENTS.md",
     "backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md",
+    "backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md",
     "backend/packages/harness/deerflow/config/AGENTS.md",
     "backend/packages/harness/deerflow/extensions/AGENTS.md",
     "backend/packages/harness/deerflow/runtime/AGENTS.md",
diff --git a/backend/tests/test_aio_sandbox.py b/backend/tests/test_aio_sandbox.py
index feefb39821..e56cdc5a3e 100644
--- a/backend/tests/test_aio_sandbox.py
+++ b/backend/tests/test_aio_sandbox.py
@@ -1,5 +1,9 @@
 """Tests for AioSandbox concurrent command serialization (#1433)."""
 
+import os
+import shlex
+import subprocess
+import sys
 import threading
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -120,6 +124,204 @@ def test_bash_exec_appends_exit_marker_when_failure_has_output(sandbox):
     assert sandbox.execute_command("make test", env={"A": "1"}) == "5 passed, 1 error\n\nExit Code: 1"
 
 
+def _execute_with_open_session_stdin(command, env=None, *, shell="/bin/sh", terminal=False, **_kwargs):
+    terminal_fds = None
+    if terminal:
+        import pty
+
+        terminal_fds = pty.openpty()
+    proc = subprocess.Popen(
+        [shell, "-c", command],
+        stdin=terminal_fds[1] if terminal_fds else subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env={**os.environ, **(env or {})},
+    )
+    try:
+        # communicate() would close the inherited session stdin and mask the bug.
+        proc.wait(timeout=5)
+        stdout = proc.stdout.read().decode()
+        return SimpleNamespace(data=SimpleNamespace(output=stdout, stdout=stdout, stderr=proc.stderr.read().decode(), exit_code=proc.returncode))
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+        proc.communicate()
+        if terminal_fds:
+            for fd in terminal_fds:
+                os.close(fd)
+
+
+@pytest.mark.skipif(os.name == "nt", reason="executes the POSIX sandbox shell transport")
+@pytest.mark.parametrize("input_kind", ["session", "pipeline", "heredoc", "file"])
+@pytest.mark.parametrize("execution_path", ["shell", "scope", "env"])
+def test_command_preserves_transport_stdin_and_explicit_input(sandbox, tmp_path, input_kind, execution_path):
+    """Model the actual AIO transports: shell PTY versus bash.exec's open pipe."""
+    from deerflow.integrations.lark_broker import LARK_BROKER_URL_ENV
+
+    # The shim ignores terminal input, while explicit input is drained to EOF.
+    reader = f"{shlex.quote(sys.executable)} -c 'import sys; print(\"tty\" if sys.stdin.isatty() else repr(sys.stdin.read()))'"
+    expected = "''\n" if execution_path == "env" else "tty\n"
+    if input_kind == "pipeline":
+        command = f"printf 'pipeline input' | {reader}"
+        expected = "'pipeline input'\n"
+    elif input_kind == "heredoc":
+        command = f"{reader} <<'INPUT'\nheredoc input\nINPUT"
+        expected = "'heredoc input\\n'\n"
+    elif input_kind == "file":
+        input_file = tmp_path / "input with spaces.txt"
+        input_file.write_text("file input", encoding="utf-8")
+        command = f"{reader} < {shlex.quote(str(input_file))}"
+        expected = "'file input'\n"
+    else:
+        command = reader
+
+    sandbox._client.bash.exec = _execute_with_open_session_stdin
+
+    def execute_on_terminal(command, **kwargs):
+        return _execute_with_open_session_stdin(command, terminal=True, **kwargs)
+
+    sandbox._client.shell.exec_command = execute_on_terminal
+    if execution_path == "env":
+        result = sandbox.execute_command(command, env={LARK_BROKER_URL_ENV: "http://127.0.0.1:8788"})
+        sandbox._client.bash.close_session.assert_called_once()
+    elif execution_path == "scope":
+        result = sandbox.execute_command_in_scope(command, scope_id="subagent")
+    else:
+        result = sandbox.execute_command(command)
+    assert result == expected
+
+
+@pytest.mark.skipif(os.name == "nt", reason="executes the Bash sandbox shell transport")
+@pytest.mark.parametrize("shell", ["/bin/sh", "/bin/bash"])
+@pytest.mark.parametrize("execution_path", ["shell", "scope", "env"])
+@pytest.mark.parametrize(
+    ("command", "expected"),
+    [
+        pytest.param("printf '%s' " + "\\" * 2, "\\", id="literal_backslash"),
+        pytest.param("echo a \\\nb", "a b\n", id="internal_continuation"),
+        pytest.param("printf '%s' '中文 \"quoted\" $HOME $(echo literal)'", '中文 "quoted" $HOME $(echo literal)', id="quoted_shell_characters"),
+    ],
+)
+def test_command_preserves_line_continuations(sandbox, shell, execution_path, command, expected):
+    """The transport delimiter must not become part of the original command."""
+
+    def execute_in_shell(command, **kwargs):
+        return _execute_with_open_session_stdin(command, shell=shell, **kwargs)
+
+    sandbox._client.bash.exec = execute_in_shell
+    sandbox._client.shell.exec_command = execute_in_shell
+    if execution_path == "env":
+        result = sandbox.execute_command(command, env={"REQUEST_VALUE": "present"})
+    elif execution_path == "scope":
+        result = sandbox.execute_command_in_scope(command, scope_id="subagent")
+    else:
+        result = sandbox.execute_command(command)
+    assert result == expected
+
+
+@pytest.mark.parametrize("execution_path", ["shell", "scope", "env"])
+@pytest.mark.parametrize("command", ["echo a " + "\\", "printf '%s' " + "\\" * 3])
+def test_command_delivers_trailing_backslash_unmodified(sandbox, execution_path, command):
+    """A lone backslash at EOF is unspecified (bash drops it, dash keeps it),
+    so assert the transport property instead: the command arrives byte-for-byte
+    and nothing follows it that the backslash could consume."""
+    delivered = []
+
+    def capture(command, **kwargs):
+        delivered.append(command)
+        return SimpleNamespace(data=SimpleNamespace(output="ok", exit_code=0))
+
+    sandbox._client.bash.exec = capture
+    sandbox._client.shell.exec_command = capture
+    if execution_path == "env":
+        sandbox.execute_command(command, env={"REQUEST_VALUE": "present"})
+        assert delivered == [f"exec < /dev/null\n{command}"]
+    elif execution_path == "scope":
+        sandbox.execute_command_in_scope(command, scope_id="subagent")
+        assert delivered == [command]
+    else:
+        sandbox.execute_command(command)
+        assert delivered == [command]
+
+
+@pytest.mark.skipif(os.name == "nt", reason="executes the Bash sandbox shell transport")
+@pytest.mark.parametrize("execution_path", ["shell", "scope", "env"])
+@pytest.mark.parametrize(
+    ("command", "expected", "exit_code"),
+    [
+        pytest.param("shopt -s extglob\ncase test.py in *.@(py|sh)) printf extglob-ok;; esac", "extglob-ok", 0, id="enable_extglob"),
+        pytest.param("shopt -s expand_aliases\nalias project_status='printf alias-ok'\nproject_status", "alias-ok", 0, id="define_alias"),
+        pytest.param("false | true\nprintf '%s' \"${PIPESTATUS[*]}\"", "1 0", 0, id="pipeline_status"),
+        pytest.param("trap 'printf error-trap' ERR\nfalse", "error-trap", 1, id="error_trap"),
+    ],
+)
+def test_command_preserves_bash_parsing_and_status(sandbox, execution_path, command, expected, exit_code):
+    def execute_in_bash(command, **kwargs):
+        return _execute_with_open_session_stdin(command, shell="/bin/bash", terminal=execution_path != "env", **kwargs)
+
+    sandbox._client.bash.exec = execute_in_bash
+    sandbox._client.shell.exec_command = execute_in_bash
+    if execution_path == "env":
+        result = sandbox.execute_command(command, env={"REQUEST_VALUE": "present"})
+    elif execution_path == "scope":
+        result = sandbox.execute_command_in_scope(command, scope_id="subagent")
+    else:
+        result = sandbox.execute_command(command)
+    assert result == expected + (f"\nExit Code: {exit_code}" if exit_code else "")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="executes the POSIX sandbox shell transport")
+def test_persistent_shell_preserves_session_state(sandbox, tmp_path):
+    """PTY commands must not be rewritten or permanently close session fd0."""
+    commands = []
+
+    def record_command(command, **_kwargs):
+        commands.append(command)
+        return SimpleNamespace(data=SimpleNamespace(output="ok", exit_code=0))
+
+    sandbox._client.shell.exec_command = record_command
+    sandbox.execute_command(f"cd {shlex.quote(str(tmp_path))}; export SESSION_VALUE=preserved")
+    sandbox.execute_command('printf "%s\\n" "$PWD" "$SESSION_VALUE"; exit 7')
+    # Run the commands in one shell, as the persistent AIO session does.
+    completed = subprocess.run(["/bin/sh", "-c", "\n".join(commands)], input=b"", capture_output=True, timeout=5)
+    assert completed.returncode == 7
+    assert completed.stdout.decode().splitlines() == [str(tmp_path), "preserved"]
+
+    commands.clear()
+    sandbox.execute_command("false")
+    sandbox.execute_command('printf "%s" "$?"')
+    completed = subprocess.run(["/bin/sh", "-c", "\n".join(commands)], input=b"", capture_output=True, timeout=5)
+    assert completed.returncode == 0
+    assert completed.stdout == b"1"
+
+    commands.clear()
+    sandbox.execute_command("false | true")
+    sandbox.execute_command('printf "%s" "${PIPESTATUS[*]}"')
+    completed = subprocess.run(["/bin/bash", "-c", "\n".join(commands)], input=b"", capture_output=True, timeout=5)
+    assert completed.returncode == 0
+    assert completed.stdout == b"1 0"
+
+    commands.clear()
+    sandbox.execute_command("trap 'printf error-trap' ERR")
+    sandbox.execute_command("false")
+    completed = subprocess.run(["/bin/bash", "-c", "\n".join(commands)], input=b"", capture_output=True, timeout=5)
+    assert completed.returncode == 1
+    assert completed.stdout == b"error-trap"
+
+    commands.clear()
+    sandbox.execute_command("true")
+    completed = subprocess.run(["/bin/sh", "-c", commands[0] + '\nread session_input; printf "%s" "$session_input"'], input=b"session still open\n", capture_output=True, timeout=5)
+    assert completed.returncode == 0
+    assert completed.stdout == b"session still open"
+
+
+@pytest.mark.parametrize("command", ["", " \t\n", "# comment\n \t# another comment"])
+def test_empty_or_comment_only_command_keeps_session_status(sandbox, command):
+    sandbox._client.shell.exec_command = MagicMock(return_value=SimpleNamespace(data=SimpleNamespace(output="ok", exit_code=0)))
+    sandbox.execute_command(command)
+    assert sandbox._client.shell.exec_command.call_args.kwargs["command"] == command
+
+
 class TestExecuteCommandSerialization:
     """Verify that concurrent exec_command calls are serialized."""
 
diff --git a/backend/tests/test_github_token_plumbing.py b/backend/tests/test_github_token_plumbing.py
index 70839f8a00..e10fbe6bc8 100644
--- a/backend/tests/test_github_token_plumbing.py
+++ b/backend/tests/test_github_token_plumbing.py
@@ -144,7 +144,8 @@ def test_aio_sandbox_env_routes_through_bash_exec() -> None:
     out = sbx.execute_command("gh pr create", env={"GH_TOKEN": "tok-123"})
 
     assert out == "ok"
-    assert captured["command"] == "gh pr create"
+    assert captured["command"] == "exec < /dev/null\ngh pr create"
+    assert "tok-123" not in captured["command"]
     assert captured["env"] == {"GH_TOKEN": "tok-123"}
     assert captured["created_session"] == captured["exec_session"] == captured["closed_session"]
     assert captured["create_options"] == {
@@ -175,7 +176,7 @@ def test_aio_sandbox_no_env_leaves_command_unchanged() -> None:
     sbx._recovery_session_id = None
     sbx._default_shell_corrupted = False
 
-    sbx.execute_command("echo hello")
+    assert sbx.execute_command("echo hello") == "ok"
 
     assert captured["command"] == "echo hello"
 
diff --git a/backend/tests/test_lark_broker.py b/backend/tests/test_lark_broker.py
index 67a2c0a4d3..48fbab4bad 100644
--- a/backend/tests/test_lark_broker.py
+++ b/backend/tests/test_lark_broker.py
@@ -4,10 +4,13 @@ from __future__ import annotations
 
 import base64
 import json
+import logging
 import os
+import re
 import subprocess
 import sys
 import textwrap
+import time
 import urllib.request
 from http.client import HTTPConnection
 from pathlib import Path
@@ -144,6 +147,41 @@ def test_exec_endpoint_round_trips(broker_server) -> None:
     assert payload["stdin"] == "hi"
 
 
+def test_exec_endpoint_logs_each_call(broker_server, caplog, monkeypatch) -> None:
+    """The HTTP log contract does not depend on a POSIX CLI executable."""
+    monkeypatch.setattr(lark_broker, "run_lark_cli", lambda *_args: lark_broker.ExecResult(0, b"", b"", False))
+    host, port = broker_server
+    with caplog.at_level(logging.INFO, logger=lark_broker.__name__):
+        status, _ = _post_exec(host, port, {"args": ["ping"]})
+    assert status == 200
+    messages = [record.getMessage() for record in caplog.records]
+    assert any(re.search(r"lark-cli exec argc=1 exit=0 in \d+\.\d+s", message) for message in messages)
+
+
+def test_exec_endpoint_logs_never_include_argument_values(broker_server, caplog, monkeypatch) -> None:
+    """Neither short secrets nor document bodies may enter server-side logs."""
+    received = []
+
+    def run_cli(_config, args, stdin):
+        received.append((args, stdin))
+        return lark_broker.ExecResult(0, b"", b"", False)
+
+    monkeypatch.setattr(lark_broker, "run_lark_cli", run_cli)
+    host, port = broker_server
+    long_arg = "x" * 500
+    args = ["im", "+messages-send", "--content", "synthetic-private-body", "--token=SYNTHETIC_SECRET", long_arg]
+    with caplog.at_level(logging.INFO, logger=lark_broker.__name__):
+        status, _ = _post_exec(host, port, {"args": args})
+    assert status == 200
+    assert received == [(args, b"")]
+    messages = [record.getMessage() for record in caplog.records if "lark-cli exec" in record.getMessage()]
+    assert messages
+    assert "argc=6" in messages[0]
+    assert "synthetic-private-body" not in messages[0]
+    assert "SYNTHETIC_SECRET" not in messages[0]
+    assert "x" * 200 not in messages[0]
+
+
 @_skip_windows
 def test_exec_endpoint_ignores_client_supplied_credential_paths(broker_server) -> None:
     host, port = broker_server
@@ -220,6 +258,228 @@ def test_shim_fails_loudly_when_broker_unreachable(tmp_path: Path) -> None:
     assert b"broker unreachable" in completed.stderr
 
 
+@_skip_windows
+def test_shim_terminal_stdin_needs_no_command_wrapper(broker_server, tmp_path: Path) -> None:
+    """The legacy AIO shell is a PTY, which the shim already treats as no input."""
+    import pty
+
+    host, port = broker_server
+    shim = tmp_path / "lark-cli-shim.py"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    master, slave = pty.openpty()
+    try:
+        # Keep the terminal open throughout execution: there is no EOF to read.
+        completed = subprocess.run(
+            [sys.executable, str(shim), "auth", "status"],
+            stdin=slave,
+            capture_output=True,
+            env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}"},
+            timeout=5,
+        )
+    finally:
+        os.close(slave)
+        os.close(master)
+    assert completed.returncode == 0, completed.stderr
+    assert json.loads(completed.stdout)["stdin"] == ""
+
+
+@_skip_windows
+def test_shim_open_pipe_stdin_fails_without_executing(broker_server, tmp_path: Path, monkeypatch) -> None:
+    """An open pipe could be a delayed producer; never execute with guessed input."""
+    executions = []
+    monkeypatch.setattr(lark_broker, "run_lark_cli", lambda *args: executions.append(args) or lark_broker.ExecResult(0, b"", b"", False))
+    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)
+
+    proc = subprocess.Popen(
+        [sys.executable, str(shim), "auth", "status"],
+        stdin=subprocess.PIPE,  # held open by the parent, never written: no EOF ever arrives
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env={
+            **os.environ,
+            lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}",
+            "DEERFLOW_LARK_BROKER_STDIN_GRACE_SECONDS": "0.2",
+        },
+    )
+    try:
+        # Do not use communicate(): it would close stdin and mask the bug.
+        proc.wait(timeout=30)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+        pytest.fail("shim blocked waiting for EOF on a session-held stdin pipe")
+    stdout = proc.stdout.read()
+    stderr = proc.stderr.read()
+    proc.stdin.close()
+    assert proc.returncode == 124, stderr
+    assert stdout == b""
+    assert b"stdin" in stderr and b"EOF" in stderr
+    assert executions == []
+
+
+@_skip_windows
+def test_shim_piped_stdin_still_forwarded(broker_server, tmp_path: Path) -> None:
+    """`echo x | lark-cli ...` must still pass stdin through to the broker."""
+    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), "docs", "+fetch"],
+        input=b"streamed-input",
+        capture_output=True,
+        env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}"},
+        timeout=30,
+    )
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout.decode())
+    assert payload["stdin"] == "streamed-input"
+
+
+@_skip_windows
+def test_shim_drip_fed_stdin_round_trips(broker_server, tmp_path: Path) -> None:
+    """A slow producer writing in chunks below the idle window keeps its input."""
+    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)
+
+    proc = subprocess.Popen(
+        [sys.executable, str(shim), "docs", "+fetch"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}"},
+    )
+    try:
+        for piece in (b"streamed", b"-", b"input"):
+            proc.stdin.write(piece)
+            proc.stdin.flush()
+            time.sleep(0.3)  # each pause stays below the default tail window
+        proc.stdin.close()
+        proc.wait(timeout=30)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+        pytest.fail("shim stalled on drip-fed stdin")
+    stdout = proc.stdout.read()
+    stderr = proc.stderr.read()
+    assert proc.returncode == 0, stderr
+    payload = json.loads(stdout.decode())
+    assert payload["stdin"] == "streamed-input"
+    assert b"warning" not in stderr
+
+
+@_skip_windows
+def test_shim_idle_stdin_fails_without_executing_partial_input(broker_server, tmp_path: Path, monkeypatch) -> None:
+    """Never submit a partial document when a producer pauses before EOF."""
+    executions = []
+    monkeypatch.setattr(lark_broker, "run_lark_cli", lambda *args: executions.append(args) or lark_broker.ExecResult(0, b"", b"", False))
+    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)
+
+    proc = subprocess.Popen(
+        [sys.executable, str(shim), "docs", "+fetch"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env={
+            **os.environ,
+            lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}",
+            "DEERFLOW_LARK_BROKER_STDIN_TAIL_SECONDS": "0.3",
+        },
+    )
+    proc.stdin.write(b"first-chunk")
+    proc.stdin.flush()
+    try:
+        # The producer never writes again and never closes: the tail window fires.
+        proc.wait(timeout=30)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+        pytest.fail("shim stalled on an idle stdin pipe")
+    stdout = proc.stdout.read()
+    stderr = proc.stderr.read()
+    proc.stdin.close()
+    assert proc.returncode == 124, stderr
+    assert stdout == b""
+    assert b"stdin" in stderr and b"EOF" in stderr
+    assert executions == []
+
+
+@_skip_windows
+@pytest.mark.parametrize("window", ["DEERFLOW_LARK_BROKER_STDIN_GRACE_SECONDS", "DEERFLOW_LARK_BROKER_STDIN_TAIL_SECONDS"])
+@pytest.mark.parametrize("value", ["inf", "1e309", "nan", "-1", "junk"])
+def test_shim_invalid_stdin_window_preserves_input(broker_server, tmp_path: Path, window: str, value: str) -> None:
+    """Invalid operator settings fall back rather than discarding valid input."""
+    host, port = broker_server
+    shim = tmp_path / "shim.py"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    completed = subprocess.run(
+        [sys.executable, str(shim), "docs", "+fetch"],
+        input=b"complete-input",
+        capture_output=True,
+        env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}", window: value},
+        timeout=30,
+    )
+    assert completed.returncode == 0, completed.stderr
+    assert json.loads(completed.stdout)["stdin"] == "complete-input"
+
+
+@_skip_windows
+def test_shim_delayed_first_byte_within_window_is_preserved(broker_server, tmp_path: Path) -> None:
+    """An explicit larger grace window supports producers with startup latency."""
+    host, port = broker_server
+    shim = tmp_path / "shim.py"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    proc = subprocess.Popen(
+        [sys.executable, str(shim), "docs", "+fetch"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env={
+            **os.environ,
+            lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}",
+            "DEERFLOW_LARK_BROKER_STDIN_GRACE_SECONDS": "5",
+        },
+    )
+    try:
+        time.sleep(2.4)
+        stdout, stderr = proc.communicate(input=b"delayed-complete-input", timeout=30)
+        assert proc.returncode == 0, stderr
+        assert json.loads(stdout)["stdin"] == "delayed-complete-input"
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+        proc.communicate()
+
+
+@_skip_windows
+def test_shim_large_file_input_is_preserved(broker_server, tmp_path: Path) -> None:
+    host, port = broker_server
+    shim = tmp_path / "shim.py"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    contents = "文档内容\n" * 16000
+    input_file = tmp_path / "input.txt"
+    input_file.write_text(contents, encoding="utf-8")
+    with input_file.open("rb") as stdin:
+        completed = subprocess.run(
+            [sys.executable, str(shim), "docs", "+fetch"],
+            stdin=stdin,
+            capture_output=True,
+            env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}"},
+            timeout=30,
+        )
+    assert completed.returncode == 0, completed.stderr
+    assert json.loads(completed.stdout)["stdin"] == contents
+
+
 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.
diff --git a/docker/lark-cli-broker/README.md b/docker/lark-cli-broker/README.md
index 3af3ddb15c..950102acb5 100644
--- a/docker/lark-cli-broker/README.md
+++ b/docker/lark-cli-broker/README.md
@@ -94,6 +94,27 @@ and pointing the provisioner at it:
 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.
 
+### Stdin and execution logs
+
+AIO's fresh non-interactive Bash runs close inherited pipe stdin, so commands
+without input receive immediate EOF. Persistent Shell terminal commands remain
+unchanged; the shim ignores their TTY stdin. Explicit pipelines, heredocs, and
+file redirections supply input normally. The shim forwards pipe/file stdin only
+after EOF; a pipe that exceeds its idle budget exits `124` before contacting
+the broker, so incomplete input cannot execute a partial write.
+
+The sandbox process can override the first-byte budget with
+`DEERFLOW_LARK_BROKER_STDIN_GRACE_SECONDS` and the between-chunk idle budget with
+`DEERFLOW_LARK_BROKER_STDIN_TAIL_SECONDS`. Both default to 2 seconds and accept
+finite values in `(0, 600]`; invalid values fall back to the default. Increase the
+appropriate budget when an explicit input producer needs longer startup or pauses.
+These settings belong on the sandbox invocation, not the broker sidecar.
+
+Execution logs record argument count, exit code, elapsed time, and output
+truncation, but never argument values or stdin. Updating the Gateway alone does
+not update an installed shim: rebuild and publish this broker image with a new
+immutable tag or digest, update `LARK_CLI_BROKER_IMAGE`, and recreate sandbox Pods.
+
 ### No file I/O relative to the sandbox cwd
 
 The broker runs `lark-cli` in the **sidecar's** working directory and cannot see