ci: run external e2e in main ci

.github/workflows/main-ci.yml

@@ -99,6 +99,7 @@ jobs:
               - '.nvmrc'
               - 'docker/docker-compose.middleware.yaml'
               - 'docker/envs/middleware.env.example'
+              - '.github/workflows/main-ci.yml'
               - '.github/workflows/web-e2e.yml'
               - '.github/actions/setup-web/**'
             vdb:
@@ -326,6 +327,8 @@ jobs:
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
     uses: ./.github/workflows/web-e2e.yml
+    with:
+      run-external-runtime: true
     secrets: inherit
 
   web-e2e-skip:

.github/workflows/web-e2e.yml

@@ -96,6 +96,15 @@ jobs:
           vp run e2e:external:prepare
           vp run e2e:external
 
+      - name: Print E2E log tails
+        if: ${{ failure() && inputs.run-external-runtime }}
+        run: |
+          while IFS= read -r log_file; do
+            echo "::group::${log_file}"
+            tail -n 200 "${log_file}"
+            echo "::endgroup::"
+          done < <(find e2e/.logs -type f | sort)
+
       - name: Upload Cucumber report
         if: ${{ !cancelled() }}
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
@@ -111,5 +120,5 @@ jobs:
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: e2e-logs
-          path: e2e/.logs
+          path: e2e/.logs/**
           retention-days: 7

dify-agent/src/dify_agent/layers/shell/layer.py

@@ -60,6 +60,7 @@ class _HasErrorCode(Protocol):
 
 DEFAULT_TIMEOUT_SECONDS = 30.0
 DEFAULT_TERMINATE_GRACE_SECONDS = 10.0
+_INTERNAL_COMMAND_TIMEOUT_SECONDS = 90.0
 _WORKSPACE_ROOT = "~/workspace"
 _WORKSPACE_DIR_NAME = "workspace"
 _WORKSPACE_COLLISION_EXIT_CODE = 17
@@ -559,7 +560,7 @@ class DifyShellLayer(PydanticAILayer[DifyShellLayerDeps, object, DifyShellLayerC
             script,
             cwd=cwd,
             env=self._build_shell_command_env(include_agent_stub_env=False),
-            timeout=DEFAULT_TIMEOUT_SECONDS,
+            timeout=_INTERNAL_COMMAND_TIMEOUT_SECONDS,
             max_output_bytes=_REMOTE_COMPLETE_OUTPUT_MAX_BYTES,
         )
 

dify-agent/src/shellctl/server/config.py

@@ -65,6 +65,8 @@ class ShellctlConfig:
     poll_interval_seconds: float = 0.05
     pipe_monitor_interval_seconds: float = 1.0
     pipe_ready_timeout_seconds: float = 10.0
+    tmux_command_timeout_seconds: float = 15.0
+    tmux_session_start_timeout_seconds: float = 90.0
     sqlite_busy_timeout_ms: int = DEFAULT_SQLITE_BUSY_TIMEOUT_MS
     sanitize_pty_command: tuple[str, ...] = ("shellctl-sanitize-pty",)
     runner_exit_command: tuple[str, ...] = ("shellctl-runner-exit",)

dify-agent/src/shellctl/server/service.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import logging
 import shlex
 import shutil
 import stat
@@ -74,6 +75,9 @@ from shellctl.shared.schemas import (
 )
 
 
+logger = logging.getLogger(__name__)
+
+
 class ShellctlService:
     """SQLite-backed shellctl job service used by HTTP and local server entrypoints.
 
@@ -273,6 +277,7 @@ class ShellctlService:
                 require_exit_code_null=True,
             )
         except Exception as exc:
+            logger.warning("shellctl job %s failed during startup: %s", job_id, exc, exc_info=True)
             await self._transition_status(
                 job_id,
                 allowed_from={

dify-agent/src/shellctl/server/tmux.py

@@ -124,20 +124,31 @@ class TmuxController:
                 str(cwd),
             ]
         )
-        result = await self._run_tmux(
-            "-f",
-            "/dev/null",
-            "new-session",
-            "-d",
-            "-s",
-            job_session_name(job_id),
-            "-x",
-            str(terminal.cols),
-            "-y",
-            str(terminal.rows),
-            runner_command,
-            check=False,
-        )
+        session_name = job_session_name(job_id)
+        try:
+            result = await self._run_tmux(
+                "-f",
+                "/dev/null",
+                "new-session",
+                "-d",
+                "-s",
+                session_name,
+                "-x",
+                str(terminal.cols),
+                "-y",
+                str(terminal.rows),
+                runner_command,
+                check=False,
+                timeout_seconds=self._config.tmux_session_start_timeout_seconds,
+            )
+        except ShellctlServerError as exc:
+            if exc.code == "tmux_timeout":
+                session_probe = await self._session_exists_after_start_timeout(session_name)
+                if session_probe is True:
+                    return
+                message = f"{exc.message}; session_probe={self._format_optional_probe(session_probe)}"
+                raise ShellctlServerError(exc.status_code, exc.code, message) from exc
+            raise
         if result.returncode != 0:
             raise ShellctlServerError(
                 500,
@@ -299,19 +310,36 @@ class TmuxController:
             check=False,
         )
 
-    async def _run_tmux(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[bytes]:
-        env = dict(os.environ)
-        env.pop("TMUX", None)
+    async def _run_tmux(
+        self,
+        *args: str,
+        check: bool = True,
+        timeout_seconds: float | None = None,
+    ) -> subprocess.CompletedProcess[bytes]:
+        command = ["tmux", "-S", str(self._config.tmux_socket), *args]
+        effective_timeout = (
+            timeout_seconds if timeout_seconds is not None else self._config.tmux_command_timeout_seconds
+        )
         try:
-            result = await anyio.run_process(
-                ["tmux", "-S", str(self._config.tmux_socket), *args],
-                env=env,
-                check=False,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-            )
+            with anyio.fail_after(effective_timeout):
+                result = await anyio.run_process(
+                    command,
+                    env=self._tmux_env(),
+                    check=False,
+                    stdin=subprocess.DEVNULL,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    start_new_session=True,
+                )
         except FileNotFoundError as exc:
             raise ShellctlServerError(500, "tmux_not_installed", "tmux executable was not found") from exc
+        except TimeoutError as exc:
+            message = f"tmux command timed out after {effective_timeout:.3f}s: {self._shell_join(command)}"
+            raise ShellctlServerError(
+                504,
+                "tmux_timeout",
+                message,
+            ) from exc
         if check and result.returncode != 0:
             raise ShellctlServerError(
                 500,
@@ -320,10 +348,48 @@ class TmuxController:
             )
         return result
 
+    async def _session_exists_after_start_timeout(self, session_name: str) -> bool | None:
+        try:
+            with anyio.fail_after(min(5.0, self._config.tmux_command_timeout_seconds)):
+                result = await anyio.run_process(
+                    [
+                        "tmux",
+                        "-S",
+                        str(self._config.tmux_socket),
+                        "list-sessions",
+                        "-F",
+                        "#{session_name}",
+                    ],
+                    env=self._tmux_env(),
+                    check=False,
+                    stdin=subprocess.DEVNULL,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    start_new_session=True,
+                )
+        except (FileNotFoundError, TimeoutError):
+            return None
+        if result.returncode != 0:
+            return None
+        output = result.stdout.decode("utf-8", errors="replace")
+        return session_name in {line.strip() for line in output.splitlines() if line.strip()}
+
+    @staticmethod
+    def _format_optional_probe(probe: bool | None) -> str:
+        if probe is None:
+            return "unavailable"
+        return "found" if probe else "missing"
+
     @staticmethod
     def _shell_join(parts: tuple[str, ...] | list[str]) -> str:
         return " ".join(shlex.quote(part) for part in parts)
 
+    @staticmethod
+    def _tmux_env() -> dict[str, str]:
+        env = dict(os.environ)
+        env.pop("TMUX", None)
+        return env
+
 
 def _tmux_target_missing(stderr: str) -> bool:
     normalized = stderr.lower()

dify-agent/tests/local/dify_agent/layers/shell/test_layer.py

@@ -319,6 +319,12 @@ def test_shell_layer_create_allocates_workspace_and_bootstraps(monkeypatch: pyte
     assert layer.runtime_state.session_id == "abc12ff"
     assert layer.runtime_state.workspace_cwd == "~/workspace/abc12ff"
     assert [call.job_id for call in provider.resource.commands.delete_calls] == ["mkdir-job", "bootstrap-job"]
+    assert [call.timeout for call in provider.resource.commands.run_calls] == pytest.approx(
+        [
+            shell_layer_module._INTERNAL_COMMAND_TIMEOUT_SECONDS,
+            shell_layer_module._INTERNAL_COMMAND_TIMEOUT_SECONDS,
+        ]
+    )
 
 
 def test_shell_layer_uses_agent_specific_home_and_workspace_cwd(

dify-agent/tests/local/shellctl/test_shellctl_service.py

@@ -1583,10 +1583,136 @@ def test_shellctl_config_defaults_to_lightweight_sanitize_entrypoint(
     config = ShellctlConfig(state_dir=tmp_path / "state", runtime_dir=tmp_path / "run")
 
     assert config.pipe_ready_timeout_seconds == 10.0
+    assert config.tmux_command_timeout_seconds == 15.0
+    assert config.tmux_session_start_timeout_seconds == 90.0
     assert config.sanitize_pty_command == ("shellctl-sanitize-pty",)
     assert config.runner_exit_command == ("shellctl-runner-exit",)
 
 
+@pytest.mark.anyio
+async def test_tmux_command_timeout_returns_structured_error(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def slow_run_process(*args: object, **kwargs: object) -> subprocess.CompletedProcess[bytes]:
+        del args, kwargs
+        await anyio.sleep(1)
+        return subprocess.CompletedProcess(["tmux"], 0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr(anyio, "run_process", slow_run_process)
+    controller = TmuxController(
+        ShellctlConfig(
+            state_dir=tmp_path / "state",
+            runtime_dir=tmp_path / "run",
+            tmux_command_timeout_seconds=0.01,
+        )
+    )
+
+    with pytest.raises(ShellctlServerError) as exc_info:
+        await controller.list_sessions()
+
+    assert exc_info.value.status_code == 504
+    assert exc_info.value.code == "tmux_timeout"
+    assert "tmux command timed out after 0.010s" in exc_info.value.message
+    assert "list-sessions" in exc_info.value.message
+
+
+@pytest.mark.anyio
+async def test_tmux_commands_do_not_inherit_process_stdin_or_session(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured_kwargs: dict[str, object] = {}
+
+    async def fake_run_process(*args: object, **kwargs: object) -> subprocess.CompletedProcess[bytes]:
+        del args
+        captured_kwargs.update(kwargs)
+        return subprocess.CompletedProcess(["tmux"], 0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr(anyio, "run_process", fake_run_process)
+    controller = TmuxController(
+        ShellctlConfig(
+            state_dir=tmp_path / "state",
+            runtime_dir=tmp_path / "run",
+        )
+    )
+
+    await controller.start_server()
+
+    assert captured_kwargs["stdin"] == subprocess.DEVNULL
+    assert captured_kwargs["start_new_session"] is True
+
+
+@pytest.mark.anyio
+async def test_tmux_session_start_uses_dedicated_timeout(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def slow_run_process(*args: object, **kwargs: object) -> subprocess.CompletedProcess[bytes]:
+        del args, kwargs
+        await anyio.sleep(1)
+        return subprocess.CompletedProcess(["tmux"], 0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr(anyio, "run_process", slow_run_process)
+    controller = TmuxController(
+        ShellctlConfig(
+            state_dir=tmp_path / "state",
+            runtime_dir=tmp_path / "run",
+            tmux_command_timeout_seconds=30.0,
+            tmux_session_start_timeout_seconds=0.01,
+        )
+    )
+
+    with pytest.raises(ShellctlServerError) as exc_info:
+        await controller.create_job_session(
+            job_id="job-timeout",
+            job_dir=tmp_path / "job-timeout",
+            cwd=tmp_path,
+            terminal=TerminalSize(cols=120, rows=80),
+        )
+
+    assert exc_info.value.status_code == 504
+    assert exc_info.value.code == "tmux_timeout"
+    assert "tmux command timed out after 0.010s" in exc_info.value.message
+    assert "new-session" in exc_info.value.message
+
+
+@pytest.mark.anyio
+async def test_tmux_session_start_timeout_succeeds_when_session_was_created(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session_name = job_session_name("job-timeout")
+
+    async def fake_run_process(
+        command: list[str],
+        **kwargs: object,
+    ) -> subprocess.CompletedProcess[bytes]:
+        del kwargs
+        if "new-session" in command:
+            await anyio.sleep(1)
+            return subprocess.CompletedProcess(command, 0, stdout=b"", stderr=b"")
+        if "list-sessions" in command:
+            return subprocess.CompletedProcess(command, 0, stdout=f"{session_name}\n".encode(), stderr=b"")
+        return subprocess.CompletedProcess(command, 1, stdout=b"", stderr=b"unexpected command")
+
+    monkeypatch.setattr(anyio, "run_process", fake_run_process)
+    controller = TmuxController(
+        ShellctlConfig(
+            state_dir=tmp_path / "state",
+            runtime_dir=tmp_path / "run",
+            tmux_session_start_timeout_seconds=0.01,
+        )
+    )
+
+    await controller.create_job_session(
+        job_id="job-timeout",
+        job_dir=tmp_path / "job-timeout",
+        cwd=tmp_path,
+        terminal=TerminalSize(cols=120, rows=80),
+    )
+
+
 def test_pipe_command_finalizer_commits_runner_exit_after_drain(tmp_path: Path) -> None:
     controller = TmuxController(
         ShellctlConfig(

e2e/features/agent-v2/support/access-point.ts

@@ -6,11 +6,11 @@ import type {
   ChatRequestPayloadWithUser,
   PostChatMessagesResponse,
 } from '@dify/contracts/api/service/types.gen'
-import type { APIResponse } from '@playwright/test'
-import { request } from '@playwright/test'
 import { createApiContext, expectApiResponseOK, setAppSiteEnabled } from '../../../support/api'
 import { getTestAgent } from './agent'
 
+const SERVICE_API_STREAM_TIMEOUT_MS = 120_000
+
 export type AgentServiceApiChatResult = {
   body: PostChatMessagesResponse | unknown
   ok: boolean
@@ -22,12 +22,13 @@ type ServiceApiSseEvent = {
   event?: string
 }
 
-async function parseServiceApiChatResponse(response: APIResponse) {
-  const contentType = response.headers()['content-type'] ?? ''
-  const text = await response.text().catch(() => '')
+async function parseServiceApiChatResponse(response: Response) {
+  const contentType = response.headers.get('content-type') ?? ''
 
   if (contentType.includes('text/event-stream'))
-    return parseServiceApiSseText(text)
+    return parseServiceApiSseStream(response)
+
+  const text = await response.text().catch(() => '')
 
   if (contentType.includes('application/json')) {
     try {
@@ -46,6 +47,36 @@ async function parseServiceApiChatResponse(response: APIResponse) {
   }
 }
 
+async function parseServiceApiSseStream(response: Response) {
+  if (!response.body)
+    return parseServiceApiSseText(await response.text().catch(() => ''))
+
+  const reader = response.body.getReader()
+  const decoder = new TextDecoder()
+  let raw = ''
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done)
+        break
+
+      raw += decoder.decode(value, { stream: true })
+      const parsed = parseServiceApiSseText(raw)
+      if (parsed.events.some(isTerminalServiceApiEvent)) {
+        await reader.cancel().catch(() => {})
+        return parsed
+      }
+    }
+  }
+  finally {
+    reader.releaseLock()
+  }
+
+  raw += decoder.decode()
+  return parseServiceApiSseText(raw)
+}
+
 function parseServiceApiSseText(text: string) {
   const events: ServiceApiSseEvent[] = []
   const answers: string[] = []
@@ -95,6 +126,24 @@ function parseServiceApiSseText(text: string) {
   }
 }
 
+function isTerminalServiceApiEvent(event: ServiceApiSseEvent) {
+  if (event.event === 'message_end' || event.event === 'workflow_finished' || event.event === 'error')
+    return true
+
+  const data = event.data
+  return Boolean(
+    data
+    && typeof data === 'object'
+    && !Array.isArray(data)
+    && 'event' in data
+    && (
+      data.event === 'message_end'
+      || data.event === 'workflow_finished'
+      || data.event === 'error'
+    ),
+  )
+}
+
 export async function setAgentSiteAccessAndGetURL(
   agentId: string,
   enabled: boolean,
@@ -152,7 +201,6 @@ export async function sendAgentServiceApiChatMessage({
   query?: string
   serviceApiBaseURL: string
 }): Promise<AgentServiceApiChatResult> {
-  const ctx = await request.newContext()
   const body = {
     inputs: {},
     query,
@@ -160,22 +208,28 @@ export async function sendAgentServiceApiChatMessage({
     user: 'e2e-agent-access-point',
   } satisfies ChatRequestPayloadWithUser
 
+  const controller = new AbortController()
+  const timeout = setTimeout(() => controller.abort(), SERVICE_API_STREAM_TIMEOUT_MS)
+
   try {
-    const response = await ctx.post(`${serviceApiBaseURL.replace(/\/$/, '')}/chat-messages`, {
-      data: body,
+    const response = await fetch(`${serviceApiBaseURL.replace(/\/$/, '')}/chat-messages`, {
+      body: JSON.stringify(body),
       headers: {
-        Authorization: `Bearer ${apiKey}`,
+        'Authorization': `Bearer ${apiKey}`,
+        'Content-Type': 'application/json',
       },
+      method: 'POST',
+      signal: controller.signal,
     })
     const responseBody = await parseServiceApiChatResponse(response)
 
     return {
       body: responseBody as PostChatMessagesResponse | unknown,
-      ok: response.ok(),
-      status: response.status(),
+      ok: response.ok,
+      status: response.status,
     }
   }
   finally {
-    await ctx.dispose()
+    clearTimeout(timeout)
   }
 }

e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts

@@ -9,6 +9,8 @@ import {
 import { agentBuilderExpectedTokens, agentBuilderFixedInputs } from '../../agent-v2/support/agent-builder-resources'
 import { getCurrentAgentId, getServiceApiCard } from './access-point-helpers'
 
+const SERVICE_API_RUNTIME_STEP_TIMEOUT_MS = 130_000
+
 async function enableAgentApiAccessWithKey(world: DifyWorld) {
   const agentId = getCurrentAgentId(world)
   const apiAccess = await setAgentApiAccess(agentId, true)
@@ -172,34 +174,42 @@ Then('the Agent v2 API Reference should open in a new tab', async function (this
   this.agentBuilder.accessPoint.apiReferencePage = undefined
 })
 
-When('I send the Agent v2 Backend service API minimal request', async function (this: DifyWorld) {
-  const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
-  const apiKey = this.agentBuilder.accessPoint.generatedApiKey
-  if (!serviceApiBaseURL)
-    throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
-  if (!apiKey)
-    throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
+When(
+  'I send the Agent v2 Backend service API minimal request',
+  { timeout: SERVICE_API_RUNTIME_STEP_TIMEOUT_MS },
+  async function (this: DifyWorld) {
+    const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
+    const apiKey = this.agentBuilder.accessPoint.generatedApiKey
+    if (!serviceApiBaseURL)
+      throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
+    if (!apiKey)
+      throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
 
-  this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
-    apiKey,
-    serviceApiBaseURL,
-  })
-})
+    this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
+      apiKey,
+      serviceApiBaseURL,
+    })
+  },
+)
 
-When('I send the Agent v2 Backend service API knowledge request', async function (this: DifyWorld) {
-  const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
-  const apiKey = this.agentBuilder.accessPoint.generatedApiKey
-  if (!serviceApiBaseURL)
-    throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
-  if (!apiKey)
-    throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
+When(
+  'I send the Agent v2 Backend service API knowledge request',
+  { timeout: SERVICE_API_RUNTIME_STEP_TIMEOUT_MS },
+  async function (this: DifyWorld) {
+    const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
+    const apiKey = this.agentBuilder.accessPoint.generatedApiKey
+    if (!serviceApiBaseURL)
+      throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
+    if (!apiKey)
+      throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
 
-  this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
-    apiKey,
-    query: agentBuilderFixedInputs.knowledgeRuntimeQuery,
-    serviceApiBaseURL,
-  })
-})
+    this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
+      apiKey,
+      query: agentBuilderFixedInputs.knowledgeRuntimeQuery,
+      serviceApiBaseURL,
+    })
+  },
+)
 
 const stringifyServiceApiBody = (body: unknown) => {
   try {

e2e/features/step-definitions/agent-v2/build-draft.steps.ts

@@ -34,6 +34,7 @@ import {
 } from './configure-helpers'
 
 const BUILD_DRAFT_RUNTIME_STEP_TIMEOUT_MS = 180_000
+const BUILD_DRAFT_CHAT_RESPONSE_TIMEOUT_MS = 150_000
 const BUILD_DRAFT_NOTE_SYNC_TIMEOUT_MS = 30_000
 const BUILD_NOTE_FILE_NAME = 'build_note.md'
 const BUILD_NOTE_MARKER = 'E2E_BUILD_DRAFT_PASS'
@@ -161,7 +162,7 @@ When(
     const chatResponsePromise = page.waitForResponse(response => (
       response.request().method() === 'POST'
       && new URL(response.url()).pathname.endsWith(`/console/api/agent/${agentId}/chat-messages`)
-    ))
+    ), { timeout: BUILD_DRAFT_CHAT_RESPONSE_TIMEOUT_MS })
 
     await page.getByRole('button', { name: 'Start build' }).click()
     expect((await checkoutResponsePromise).ok()).toBe(true)

e2e/features/step-definitions/agent-v2/tools.steps.ts

@@ -12,6 +12,8 @@ import { hasToolEntry } from '../../agent-v2/support/preflight/tools'
 import { getPreseededToolContract } from '../../agent-v2/support/tools'
 import { expectProviderToolActionVisible, getCurrentAgentId } from './configure-helpers'
 
+const TOOL_RUNTIME_STEP_TIMEOUT_MS = 180_000
+
 const getToolsSection = (world: DifyWorld) =>
   world.getPage().getByRole('region', { name: 'Tools' })
 
@@ -271,24 +273,28 @@ Then(
   },
 )
 
-When('I send the Agent v2 Backend service API JSON Replace request', async function (this: DifyWorld) {
-  const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
-  const apiKey = this.agentBuilder.accessPoint.generatedApiKey
-  if (!serviceApiBaseURL)
-    throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
-  if (!apiKey)
-    throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
+When(
+  'I send the Agent v2 Backend service API JSON Replace request',
+  { timeout: TOOL_RUNTIME_STEP_TIMEOUT_MS },
+  async function (this: DifyWorld) {
+    const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL
+    const apiKey = this.agentBuilder.accessPoint.generatedApiKey
+    if (!serviceApiBaseURL)
+      throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.')
+    if (!apiKey)
+      throw new Error('No Agent v2 API key found. Create a Backend service API key first.')
 
-  this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
-    apiKey,
-    query: [
-      'Verify JSON Replace now.',
-      `Replace ${agentBuilderExpectedTokens.jsonToolBefore} with ${agentBuilderExpectedTokens.jsonToolAfter}.`,
-      `Return the JSON result and include ${agentBuilderExpectedTokens.jsonToolAfter}.`,
-    ].join(' '),
-    serviceApiBaseURL,
-  })
-})
+    this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({
+      apiKey,
+      query: [
+        'Verify JSON Replace now.',
+        `Replace ${agentBuilderExpectedTokens.jsonToolBefore} with ${agentBuilderExpectedTokens.jsonToolAfter}.`,
+        `Return the JSON result and include ${agentBuilderExpectedTokens.jsonToolAfter}.`,
+      ].join(' '),
+      serviceApiBaseURL,
+    })
+  },
+)
 
 Then(
   'the Agent v2 Backend service API response should include the JSON Replace E2E marker',

e2e/scripts/run-cucumber.ts

@@ -68,6 +68,87 @@ const readLogTail = async (logFilePath: string) => {
     .join('\n')
 }
 
+const getShellctlAuthHeaders = () => {
+  const token = process.env.E2E_SHELLCTL_AUTH_TOKEN || process.env.DIFY_AGENT_SHELLCTL_AUTH_TOKEN
+
+  return token ? { Authorization: `Bearer ${token}` } : undefined
+}
+
+type ShellctlJobResult = {
+  done?: boolean
+  exit_code?: number | null
+  job_id?: string
+  offset?: number
+  output?: string
+  status?: string
+}
+
+const prewarmShellctlSandbox = async (shellctlURL: string) => {
+  const marker = 'shellctl-e2e-ready'
+  const controller = new AbortController()
+  const timeout = setTimeout(() => controller.abort(), 210_000)
+  const deadline = Date.now() + 180_000
+  let output = ''
+  let latestBody: ShellctlJobResult | undefined
+
+  try {
+    const response = await fetch(`${shellctlURL}/v1/jobs/run`, {
+      body: JSON.stringify({
+        idle_flush_seconds: 0.2,
+        output_limit: 4096,
+        script: `printf '${marker}'`,
+        timeout: 180,
+      }),
+      headers: {
+        ...getShellctlAuthHeaders(),
+        'Content-Type': 'application/json',
+      },
+      method: 'POST',
+      signal: controller.signal,
+    })
+    latestBody = await response.json().catch(() => undefined) as ShellctlJobResult | undefined
+    output += latestBody?.output ?? ''
+
+    if (!response.ok)
+      throw new Error(`Shellctl sandbox prewarm failed: ${response.status} ${JSON.stringify(latestBody)}`)
+
+    while (latestBody?.job_id && !latestBody.done && Date.now() < deadline) {
+      const waitResponse = await fetch(`${shellctlURL}/v1/jobs/${latestBody.job_id}/wait`, {
+        body: JSON.stringify({
+          idle_flush_seconds: 0.2,
+          offset: latestBody.offset ?? output.length,
+          output_limit: 4096,
+          timeout: Math.min(30, Math.max(1, Math.ceil((deadline - Date.now()) / 1000))),
+        }),
+        headers: {
+          ...getShellctlAuthHeaders(),
+          'Content-Type': 'application/json',
+        },
+        method: 'POST',
+        signal: controller.signal,
+      })
+      latestBody = await waitResponse.json().catch(() => undefined) as ShellctlJobResult | undefined
+      output += latestBody?.output ?? ''
+
+      if (!waitResponse.ok)
+        throw new Error(`Shellctl sandbox prewarm wait failed: ${waitResponse.status} ${JSON.stringify(latestBody)}`)
+    }
+
+    if (
+      !latestBody?.done
+      || latestBody.exit_code !== 0
+      || !output.includes(marker)
+    ) {
+      throw new Error(
+        `Shellctl sandbox prewarm failed: ${JSON.stringify(latestBody)} output=${JSON.stringify(output)}`,
+      )
+    }
+  }
+  finally {
+    clearTimeout(timeout)
+  }
+}
+
 const waitForUnexpectedProcessExit = async (
   managedProcess: ManagedProcess,
   shouldIgnoreExit: () => boolean,
@@ -196,17 +277,20 @@ const main = async () => {
       let waitingForShellctl = true
       try {
         const shellctlPort = process.env.E2E_SHELLCTL_PORT || '5004'
+        const shellctlURL = `http://127.0.0.1:${shellctlPort}`
         await Promise.race([
-          waitForUrl(`http://127.0.0.1:${shellctlPort}/openapi.json`, 180_000, 1_000),
+          waitForUrl(`${shellctlURL}/openapi.json`, 180_000, 1_000),
           waitForUnexpectedProcessExit(shellctlProcess, () => !waitingForShellctl),
         ])
+        await prewarmShellctlSandbox(shellctlURL)
       }
       catch (error) {
         if (error instanceof Error && error.message.includes('exited before becoming ready'))
           throw error
 
+        const detail = error instanceof Error ? error.message : String(error)
         throw new Error(
-          `Shellctl sandbox did not become ready. See ${shellctlProcess.logFilePath}.`,
+          `Shellctl sandbox did not become ready or prewarm successfully: ${detail}. See ${shellctlProcess.logFilePath}.`,
         )
       }
       finally {