Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8be4adb97b |
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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",)
|
||||
|
||||
@@ -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={
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user