From a59bed9a2a60f98866597e1345d024320f7f012f Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 25 May 2026 03:16:55 +0000 Subject: [PATCH] fix gateway shell head-of-line blocking --- AGENTS.md | 4 +- cmd/hwlab-gateway/main.mjs | 107 +++++++++++++++++++- docs/reference/code-agent-chat-readiness.md | 4 + docs/reference/commander-collaboration.md | 4 +- docs/reference/deployment-publish.md | 10 +- docs/reference/gateway-outbound-demo.md | 12 +++ internal/cloud/codex-stdio-session.mjs | 1 + internal/cloud/gateway-demo-registry.mjs | 25 +++++ internal/cloud/json-rpc.mjs | 25 +++-- internal/cloud/json-rpc.test.mjs | 2 + scripts/gateway-outbound-demo-smoke.mjs | 58 ++++++++++- tools/hwlab-gateway-shell.mjs | 3 + web/hwlab-cloud-web/scripts/check.mjs | 13 +++ 13 files changed, 245 insertions(+), 23 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a3078e1f..1ea71eb1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,7 +18,7 @@ HWLAB 是硬件实验室运行面和控制面项目。本文是 agent、指挥 - Runner 和指挥常用工作区是 `/workspace/hwlab`；进入仓库先检查分支与工作树状态，详见 [docs/reference/commander-collaboration.md](docs/reference/commander-collaboration.md)。 - D601 发布/构建工作区是 `/home/ubuntu/workspace/hwlab`；不要把 runner 临时目录当作发布真相，详见 [docs/reference/deployment-publish.md](docs/reference/deployment-publish.md)。 - Master server 只做控制、Git、issue/PR 和短轮询；HWLAB `check`、Playwright、本地构建、发布预检和 CI/CD 验证必须放到 D601/runner/CI/CD，详见 [docs/reference/deployment-publish.md](docs/reference/deployment-publish.md)。 -- 当前一律走 PR 工作流；不要直推 `main`、不要合并自己的 PR、不要改 PROD、不要重启服务。 +- 当前一律走 PR 工作流；不要直推 `main`，默认不要合并自己的 PR；用户或指挥官明确授权且满足门禁时可按长期参考自合并，不要改 PROD、不要重启服务。 - `DC-DCSN-P0-2026-003` / [pikasTech/HWLAB#78](https://github.com/pikasTech/HWLAB/issues/78) 是当前 M3 虚拟硬件可信闭环的上位约束；其他任务不得把 SOURCE、LOCAL、DRY-RUN、fixture 或前端状态误报为 M3 DEV-LIVE。 - 仓库禁止创建或提交 repo report 目录；验收、进展和结论只承载在 #7、专题 issue、每日简报或 PR/issue 评论。临时 JSON 只能写入 `/tmp`、`.state` 或 CI artifact，不能进入源码仓库。 @@ -50,7 +50,7 @@ HWLAB 是硬件实验室运行面和控制面项目。本文是 agent、指挥 - 中文优先：issue、PR 正文、长期参考文档和用户可见说明默认用中文；英文术语只在命令、协议、接口、ID、路径和标准名需要保真时保留，详见 [docs/reference/chinese-first-documentation.md](docs/reference/chinese-first-documentation.md)。 - 用户反馈优先：用户和参谋提出的问题默认按高优先级用户反馈处理，blocker 状态不能替代反馈分流，必须挂到 [pikasTech/HWLAB#7](https://github.com/pikasTech/HWLAB/issues/7) 醒目位置，详见 [docs/reference/user-feedback-triage.md](docs/reference/user-feedback-triage.md)。 - docs-spec 本地权威优先：涉及 `AGENTS.md`、`docs/reference/*.md` 或过程文档蒸馏时，先按 [docs/reference/documentation-governance.md](docs/reference/documentation-governance.md) 执行，不另建同级规则副本。 -- PR 工作流优先：从最新 `origin/main` 创建短分支，提交 PR 到 `pikasTech/HWLAB:main`，runner 不直推 `main`、不合并 PR、不改 PROD、不重启服务，详见 [docs/reference/commander-collaboration.md](docs/reference/commander-collaboration.md)。 +- PR 工作流优先：从最新 `origin/main` 创建短分支，提交 PR 到 `pikasTech/HWLAB:main`，runner 不直推 `main`，默认不自合并；显式授权且满足门禁时按协作规则收口，不改 PROD、不重启服务，详见 [docs/reference/commander-collaboration.md](docs/reference/commander-collaboration.md)。 ## 常用轻量命令 diff --git a/cmd/hwlab-gateway/main.mjs b/cmd/hwlab-gateway/main.mjs index 88a39ade..aa408252 100644 --- a/cmd/hwlab-gateway/main.mjs +++ b/cmd/hwlab-gateway/main.mjs @@ -14,9 +14,11 @@ const cloudUrl = trimUrl(process.env.HWLAB_GATEWAY_CLOUD_URL); const pollIntervalMs = parsePositiveInteger(process.env.HWLAB_GATEWAY_POLL_INTERVAL_MS, 500); const commandTimeoutMs = parsePositiveInteger(process.env.HWLAB_GATEWAY_CMD_TIMEOUT_MS, 120000); const outputLimitBytes = parsePositiveInteger(process.env.HWLAB_GATEWAY_CMD_OUTPUT_LIMIT_BYTES, 65536); +const maxInflightRequests = parsePositiveInteger(process.env.HWLAB_GATEWAY_MAX_INFLIGHT, 3); const resourceId = process.env.HWLAB_GATEWAY_RESOURCE_ID ?? "res_windows_host"; const boxId = process.env.HWLAB_GATEWAY_BOX_ID ?? "box_windows_host"; const commandExecutionEnabled = truthy(process.env.HWLAB_GATEWAY_CMD_EXEC_ENABLED) || truthy(process.env.HWLAB_GATEWAY_DEMO_OPEN); +const inflightRequests = new Map(); const state = createGatewayState({ gatewayId, serviceId: "hwlab-gateway", @@ -37,9 +39,14 @@ state.outbound = { cloudUrl, pollIntervalMs, commandExecutionEnabled, + maxInflightRequests, + inflightCount: 0, + inflightRequests: [], lastPollAt: null, lastPollError: null, - lastResultAt: null + lastResultAt: null, + lastResultError: null, + lastBusyAt: null }; const routes = new Map(); @@ -90,13 +97,50 @@ async function pollCloudOnce() { state.outbound.lastPollError = null; state.session.lastSeenAt = state.outbound.lastPollAt; if (response.body?.request) { - await handleCloudRequest(response.body.request); + startCloudRequest(response.body.request); } } catch (error) { state.outbound.lastPollError = error instanceof Error ? error.message : String(error); } } +function startCloudRequest(request) { + const requestId = String(request?.id ?? `req_gateway_${Date.now()}`); + if (inflightRequests.size >= maxInflightRequests) { + state.outbound.lastBusyAt = new Date().toISOString(); + postGatewayBusyResult({ + ...request, + id: requestId + }).catch((error) => { + state.outbound.lastResultError = error instanceof Error ? error.message : String(error); + }); + return; + } + + const startedAt = new Date().toISOString(); + inflightRequests.set(requestId, { + requestId, + method: request?.method ?? null, + operationId: request?.params?.operationId ?? null, + traceId: request?.meta?.traceId ?? null, + startedAt + }); + refreshInflightState(); + + Promise.resolve() + .then(() => handleCloudRequest({ + ...request, + id: requestId + })) + .catch((error) => { + state.outbound.lastResultError = error instanceof Error ? error.message : String(error); + }) + .finally(() => { + inflightRequests.delete(requestId); + refreshInflightState(); + }); +} + async function handleCloudRequest(request) { const context = operationContext({ operationId: request.params?.operationId, @@ -136,7 +180,31 @@ async function handleCloudRequest(request) { }; } - await fetchJson(new URL("/v1/gateway/result", ensureTrailingSlash(cloudUrl)), { + await postGatewayResult(response); +} + +async function postGatewayBusyResult(request) { + const traceId = request?.meta?.traceId ?? `trc_gateway_busy_${Date.now()}`; + const body = createRpcErrorBody({ + code: ERROR_CODES.operationRejected, + message: "Gateway is busy; retry after an in-flight command finishes or increase HWLAB_GATEWAY_MAX_INFLIGHT.", + data: { + reason: "gateway_busy", + operationId: request?.params?.operationId ?? null, + inflightCount: inflightRequests.size, + maxInflightRequests + } + }); + await postGatewayResult({ + jsonrpc: JSON_RPC_VERSION, + id: request?.id ?? null, + error: body.error, + meta: responseMeta(traceId) + }); +} + +async function postGatewayResult(response) { + const result = await fetchJson(new URL("/v1/gateway/result", ensureTrailingSlash(cloudUrl)), { method: "POST", body: { gatewayId, @@ -144,7 +212,11 @@ async function handleCloudRequest(request) { response } }); + if (!result.ok) { + throw new Error(`cloud result failed: HTTP ${result.status}`); + } state.outbound.lastResultAt = new Date().toISOString(); + state.outbound.lastResultError = null; } async function invokeShell(params, context) { @@ -365,6 +437,7 @@ function createLimitedCollector(limitBytes) { } function registrationPayload() { + refreshInflightState(); return { serviceId: "hwlab-gateway", projectId: process.env.HWLAB_PROJECT_ID ?? "prj_mvp_topology", @@ -374,6 +447,13 @@ function registrationPayload() { resourceId, boxId, capabilities: shellCapabilities(), + outbound: { + pollIntervalMs, + commandExecutionEnabled, + maxInflightRequests, + inflightCount: inflightRequests.size, + inflightRequests: snapshotInflightRequests() + }, system: { platform: process.platform, arch: process.arch, @@ -395,13 +475,32 @@ function shellCapabilities() { valueType: "object", constraints: { gatewayOutboundOnly: true, - demoOpen: truthy(process.env.HWLAB_GATEWAY_DEMO_OPEN) + demoOpen: truthy(process.env.HWLAB_GATEWAY_DEMO_OPEN), + maxInflightRequests }, mutatesState: true } ]; } +function refreshInflightState() { + state.outbound.inflightCount = inflightRequests.size; + state.outbound.inflightRequests = snapshotInflightRequests(); +} + +function snapshotInflightRequests() { + const now = Date.now(); + return [...inflightRequests.values()].map((request) => ({ + ...request, + durationMs: durationSince(request.startedAt, now) + })); +} + +function durationSince(value, now = Date.now()) { + const started = Date.parse(value ?? ""); + return Number.isFinite(started) ? Math.max(0, now - started) : 0; +} + function normalizeShellInput(params = {}) { const input = params.input && typeof params.input === "object" && !Array.isArray(params.input) ? params.input : {}; const command = params.command ?? input.command ?? input.shell?.command; diff --git a/docs/reference/code-agent-chat-readiness.md b/docs/reference/code-agent-chat-readiness.md index f2a8028c..54d0f484 100644 --- a/docs/reference/code-agent-chat-readiness.md +++ b/docs/reference/code-agent-chat-readiness.md @@ -100,6 +100,8 @@ PowerShell 默认使用 wrapper 的 `-EncodedCommand` 路径；不要手写 `cmd Workbench 会把“Gateway 命令超时”控件的毫秒值随 `/v1/agent/chat` 传入 `gatewayShellTimeoutMs`。Codex prompt 必须把该值落实到 wrapper 的 `--timeout-ms`，cloud-api `hardware.invoke.shell` dispatch timeout 必须取环境配置、请求 `input.timeoutMs` 和 120s 默认值中的较大值并加 grace；不得让 20s/30s 的旧默认提前返回 `dispatchStatus=timed_out`。wrapper 自身 HTTP request timeout 要比 shell timeout 稍长，确保用户看到的是 gateway/cloud 返回的结构化 `status/operationId/dispatch`，不是 wrapper 先超时丢失结果。 +调大 timeout 不能替代正确的长任务控制语义。Gateway poll loop 必须支持后台 in-flight 执行，长 Keil/UV4 命令运行期间仍能处理短 `job-status`、state/log 读取和健康探测；如果 trace 出现 `shellExecuted=false` 的 dispatch timeout，优先检查 gateway 是否队头阻塞或离线，而不是把所有 wrapper 调用改成长等待。 + Workbench trace 对已知 JSON-RPC gateway 响应应按普通 tool call 展示：首行展示 `tool hardware.invoke.shell status= op= exit= s=`，正文展示 request、gateway/resource/capability、dispatch、command、audit/evidence 以及有界 stdout/stderr。不要把整段 JSON 原样刷屏；复制/下载完整 trace 仍保留原始 JSON。 ## 短连接 result 轮询 @@ -135,6 +137,8 @@ py -3 keil-cli.py job-status 多 probe、烧录和 reset-run 的具体参数以 Windows 侧 `C:\Users\liang\.agents\skills\keil\SKILL.md` 为准；Code Agent 只负责通过 repo wrapper 调用该 skill CLI 并返回 trace、operation/evidence 和 bounded stdout/stderr 摘要。 +对 build/download 这类长任务，Code Agent 应优先使用 skill 自带的异步 job 语义：启动命令用短 wrapper timeout 拿到 job id 或明确的启动失败，再用短 `job-status`、state 文件和日志读取轮询进展。除非用户明确要求同步等待并设置了足够大的 Gateway 命令超时，不要通过 gateway 执行 `--wait` 长轮询；同步等待会占用一个 in-flight 槽位，旧 gateway 还会造成队头阻塞。 + ## Smoke Gate 本地合同检查： diff --git a/docs/reference/commander-collaboration.md b/docs/reference/commander-collaboration.md index 4447d4d3..1f4eeda6 100644 --- a/docs/reference/commander-collaboration.md +++ b/docs/reference/commander-collaboration.md @@ -26,7 +26,9 @@ - 不要修改 PROD。 - 除非任务明确授权，不要重启服务。 - 向 `pikasTech/HWLAB:main` 创建 PR。 -- runner 不合并自己的 PR。 +- runner 默认不合并自己的 PR；用户或指挥官可以对单个 PR 明确授权 runner 自合并。 +- 自合并前必须同时满足：PR 为 `MERGEABLE/CLEAN` 或等价无冲突状态；required checks 没有失败；D601/CI 或指定运行态验证证据已贴到 PR/issue；变更不涉及 PROD、Secret、权限提升、数据迁移或未授权重启；runner 在最终评论中列出提交 SHA、验证命令和回滚边界。 +- 当前 GitHub 写入仍优先走 UniDesk CLI 或 repo-owned GitHub 路径；若当前 CLI 不支持 merge，必须使用可审计的授权路径，不能用无记录的本地绕行来规避审计。 - PR 冲突由指挥官审阅并处理；runner 不做大范围冲突手术，除非被明确分配。 - 指挥官合并 PR 时必须同时 review，确认方向没有偏离 `#7`、`#78`、`#99` 和当前用户反馈。 diff --git a/docs/reference/deployment-publish.md b/docs/reference/deployment-publish.md index 157d625e..bc6f1b23 100644 --- a/docs/reference/deployment-publish.md +++ b/docs/reference/deployment-publish.md @@ -353,13 +353,13 @@ Deployment/CD 工作按以下边界分工： | 角色 | 可以执行 | 没有明确 rollout 授权时不得执行 | | --- | --- | --- | -| Code Queue runner | 准备分支和 PR；在任务授权且检查通过后更新或自合并 docs/code/artifact PR；通过 repo-owned 路径 build/publish artifact；在被分配时刷新 artifact report 或 desired-state 文件；运行只读 `status`/`dry-run`/preflight 检查。 | 竞争 DEV CD Lease、执行 DEV apply、rollout、声称 `16666/16667` live verification、收口 `#7` 或 `#242`、修改 PROD、打印 Secret value、手工修 live resource。 | +| Code Queue runner | 准备分支和 PR；在任务明确授权且检查通过后更新或自合并 docs/code/artifact PR；通过 repo-owned 路径 build/publish artifact；在被分配时刷新 artifact report 或 desired-state 文件；运行只读 `status`/`dry-run`/preflight 检查。 | 未授权自合并、竞争 DEV CD Lease、执行 DEV apply、rollout、声称 `16666/16667` live verification、收口 `#7` 或 `#242`、修改 PROD、打印 Secret value、手工修 live resource。 | | Host commander | 在需要指挥官 review 时审阅并合并最终 PR；执行单一 DEV deploy apply 事务；执行 rollout 和 `16666/16667` live verification；收口 `#7`/`#242` 状态。 | 绕过 repo-owned CD 入口、使用第二控制面、用 UniDesk runtime 替代 HWLAB runtime、只凭报告而不做一手复验就当作 live evidence。 | -Runner 自合并权限只在任务边界内生效。它不授予 DEV apply、rollout、 -live health verification 或看板/status 收口权限。Runner 发布 artifact 时， -输出必须包含 tag、digest 和 report 证据，并把 desired-state convergence 与 -live rollout evidence 分开。 +Runner 自合并权限只在单个任务和单个 PR 的显式授权边界内生效。它不授予 +DEV apply、rollout、live health verification 或看板/status 收口权限。Runner +发布 artifact 时，输出必须包含 tag、digest 和 report 证据，并把 +desired-state convergence 与 live rollout evidence 分开。 ## CI/CD 职责拆分 diff --git a/docs/reference/gateway-outbound-demo.md b/docs/reference/gateway-outbound-demo.md index 1bbb2acc..6106f994 100644 --- a/docs/reference/gateway-outbound-demo.md +++ b/docs/reference/gateway-outbound-demo.md @@ -16,6 +16,17 @@ - `POST /v1/gateway/result`：gateway 主动回传 JSON-RPC response，cloud-api 用 request id 完成等待中的 `hardware.invoke.shell`。 - `POST /json-rpc` 或 `POST /v1/rpc/hardware.invoke.shell`：用户、agent 或 smoke 仍只调用 cloud-api，不直连 gateway。 +## Poll Loop 和长命令并发 + +Gateway 的 poll loop 不能等待单条 shell 命令结束后才继续 poll。Keil build/download、烧录或 Windows skill 可能运行数分钟；这些长命令必须在 gateway 内作为 in-flight 请求后台执行，poll loop 继续注册心跳并领取短状态读取、日志读取和取消/诊断请求。 + +判断问题类型时区分两种 timeout： + +- `shellExecuted=true` 且有 `operationId`：命令已经到达 gateway，本次 shell 超过了命令超时；下一步应读取 job state、日志或产物时间戳，而不是盲目把超时继续调大。 +- `shellExecuted=false`、`dispatchStatus=timed_out` 且旧链路可能 `operationId=null`：cloud-api 等不到 gateway 领取或回传，通常是 gateway poll loop 被前一个长命令队头阻塞，或 gateway 离线；修复重点是非阻塞 poll 和 in-flight 可观测性，不是只加大 dispatch timeout。 + +Gateway 必须在 registration payload 和 `/v1/gateway/sessions` 中暴露 `inflightCount`、`maxInflightRequests` 和当前 in-flight 摘要。默认允许少量并发，使一个长 Keil 操作不会阻塞后续只读 `job-status`、state/log 读取或健康探测；超过并发上限时应返回结构化 `gateway_busy`，不能让请求静默排队到 cloud dispatch timeout。 + ## Gateway 环境变量 | 变量 | 作用 | @@ -26,6 +37,7 @@ | `HWLAB_GATEWAY_CMD_EXEC_ENABLED=1` | 允许执行 shell 命令；未设置时 gateway 拒绝 `hardware.invoke.shell`。 | | `HWLAB_GATEWAY_DEMO_OPEN=1` | demo 明确打开标记；本地 smoke 会设置。 | | `HWLAB_GATEWAY_POLL_INTERVAL_MS` | poll 间隔，默认 500ms。 | +| `HWLAB_GATEWAY_MAX_INFLIGHT` | 同一 gateway 同时执行的 cloud 请求数，默认 3；用于避免长 Keil/UV4 命令阻塞短状态查询。 | | `HWLAB_GATEWAY_CMD_TIMEOUT_MS` | 单条命令超时，默认 120000ms；Workbench 发起 Code Agent 请求时可通过“Gateway 命令超时”控件把本轮 wrapper `--timeout-ms` 调到 1/2/3/5/10 分钟。 | | `HWLAB_GATEWAY_CMD_OUTPUT_LIMIT_BYTES` | stdout/stderr 单路输出上限，默认 65536 bytes。 | diff --git a/internal/cloud/codex-stdio-session.mjs b/internal/cloud/codex-stdio-session.mjs index 744de1d0..b0ebef01 100644 --- a/internal/cloud/codex-stdio-session.mjs +++ b/internal/cloud/codex-stdio-session.mjs @@ -101,6 +101,7 @@ const CODEX_STDIO_BOUNDARY_INSTRUCTIONS = [ "For Windows filesystem inventory, start with one small bounded PowerShell stdin script: use -LiteralPath, Select-Object -First, ConvertTo-HwlabJson, and keep stdout under about 12 KB. Do not dump full directory JSON and then retry.", "For Windows-side skills under C:\\Users\\liang\\.agents\\skills\\, read the skill manifest when needed and call its CLI from a PowerShell stdin script using explicit paths or argument arrays. Do not reimplement skill logic, and do not use shell working-directory tricks like cd &&.", "For F:\\work or F:\\work\\ConStart discovery, first list top-level project markers and *.uvprojx candidates with bounded output. For Keil build/program requests, use the Windows skill CLI at C:\\Users\\liang\\.agents\\skills\\keil with py -3 keil-cli.py from the generic PowerShell stdin wrapper path; do not reimplement Keil build logic.", + "For long Keil build/program/download jobs, prefer the skill CLI async job flow: start the job with a short bounded wrapper call, then poll job-status, state files, or logs with short wrapper calls. Do not use gateway --wait long polling unless the user explicitly asks for synchronous waiting and sets a sufficient gateway timeout.", "If a Windows gateway command reaches the gateway but fails due script syntax or output size, simplify once and report the failed operationId plus the corrected bounded command evidence. Do not spend multiple turns on exploratory rewrites.", "Use the Windows-side skill CLIs under C:\\Users\\liang\\.agents\\skills\\ from that cmd command when they exist; do not reimplement those tools in the prompt.", "Do not satisfy PC gateway requests by string matching, assistant-only claims, internal shortcut dispatch, or direct gateway URLs; the Codex turn must run the wrapper and report command evidence.", diff --git a/internal/cloud/gateway-demo-registry.mjs b/internal/cloud/gateway-demo-registry.mjs index a10a6b55..28e485ed 100644 --- a/internal/cloud/gateway-demo-registry.mjs +++ b/internal/cloud/gateway-demo-registry.mjs @@ -31,6 +31,7 @@ export function createGatewayDemoRegistry({ resourceId: input.resourceId ?? previous?.resourceId ?? "res_windows_host", boxId: input.boxId ?? previous?.boxId ?? "box_windows_host", capabilities: Array.isArray(input.capabilities) ? input.capabilities : previous?.capabilities ?? [], + outbound: normalizeGatewayOutbound(input.outbound ?? previous?.outbound), system: input.system ?? previous?.system ?? {}, firstSeenAt: previous?.firstSeenAt ?? observedAt, lastSeenAt: observedAt, @@ -134,6 +135,9 @@ export function createGatewayDemoRegistry({ boxId: session.boxId, capabilityCount: session.capabilities.length, queueDepth: session.queue.length, + inflightCount: session.outbound.inflightCount, + maxInflightRequests: session.outbound.maxInflightRequests, + inflightRequests: session.outbound.inflightRequests, firstSeenAt: session.firstSeenAt, lastSeenAt: session.lastSeenAt, system: session.system @@ -153,6 +157,27 @@ export function createGatewayDemoRegistry({ }; } +function normalizeGatewayOutbound(input = {}) { + const source = input && typeof input === "object" && !Array.isArray(input) ? input : {}; + const inflightRequests = Array.isArray(source.inflightRequests) + ? source.inflightRequests.map((request) => ({ + requestId: request?.requestId ?? null, + method: request?.method ?? null, + operationId: request?.operationId ?? null, + traceId: request?.traceId ?? null, + startedAt: request?.startedAt ?? null, + durationMs: Number.isInteger(request?.durationMs) ? request.durationMs : null + })) + : []; + return { + pollIntervalMs: Number.isInteger(source.pollIntervalMs) ? source.pollIntervalMs : null, + commandExecutionEnabled: source.commandExecutionEnabled === true, + maxInflightRequests: Number.isInteger(source.maxInflightRequests) ? source.maxInflightRequests : null, + inflightCount: Number.isInteger(source.inflightCount) ? source.inflightCount : inflightRequests.length, + inflightRequests + }; +} + export function createGatewayShellRequest({ id, params = {}, meta = {} }) { return { jsonrpc: JSON_RPC_VERSION, diff --git a/internal/cloud/json-rpc.mjs b/internal/cloud/json-rpc.mjs index 06814194..a8deb1d8 100644 --- a/internal/cloud/json-rpc.mjs +++ b/internal/cloud/json-rpc.mjs @@ -274,29 +274,34 @@ function getRuntimeStore(context = {}) { async function handleHardwareInvokeShell(params, envelope, context) { const registry = context.gatewayRegistry; - const gatewaySessionId = params.gatewaySessionId; + const operationId = params.operationId ?? `op_gateway_shell_${randomUUID()}`; + const dispatchParams = { + ...params, + operationId + }; + const gatewaySessionId = dispatchParams.gatewaySessionId; if (!registry?.isOnline?.(gatewaySessionId)) { - return getRuntimeStore(context).invokeHardwareShell(params, envelope.meta); + return getRuntimeStore(context).invokeHardwareShell(dispatchParams, envelope.meta); } const dispatch = await registry.enqueue({ gatewaySessionId, request: createGatewayShellRequest({ id: envelope.id, - params, + params: dispatchParams, meta: envelope.meta }), - timeoutMs: parseDispatchTimeout(context.env ?? process.env, params) + timeoutMs: parseDispatchTimeout(context.env ?? process.env, dispatchParams) }); if (!dispatch.ok) { return { accepted: true, status: dispatch.status === "timed_out" ? "timed_out" : "accepted", - operationId: params.operationId ?? null, + operationId, gatewaySessionId, - resourceId: params.resourceId ?? null, - capabilityId: params.capabilityId ?? null, + resourceId: dispatchParams.resourceId ?? null, + capabilityId: dispatchParams.capabilityId ?? null, dispatch: { shellExecuted: false, dispatchStatus: dispatch.status, @@ -310,10 +315,10 @@ async function handleHardwareInvokeShell(params, envelope, context) { return { accepted: gatewayResult.accepted !== false, status: gatewayResult.status ?? statusFromShellResult(gatewayResult), - operationId: gatewayResult.operationId ?? params.operationId ?? null, + operationId: gatewayResult.operationId ?? operationId, gatewaySessionId: gatewayResult.gatewaySessionId ?? gatewaySessionId, - resourceId: params.resourceId ?? null, - capabilityId: params.capabilityId ?? null, + resourceId: dispatchParams.resourceId ?? null, + capabilityId: dispatchParams.capabilityId ?? null, dispatch: { shellExecuted: gatewayResult.shellExecuted === true, dispatchStatus: gatewayResult.dispatchStatus ?? statusFromShellResult(gatewayResult), diff --git a/internal/cloud/json-rpc.test.mjs b/internal/cloud/json-rpc.test.mjs index b2a40bdc..827b8dde 100644 --- a/internal/cloud/json-rpc.test.mjs +++ b/internal/cloud/json-rpc.test.mjs @@ -103,6 +103,7 @@ test("JSON-RPC invalid meta reports unknown serviceId reason without leaking sec method: "hardware.invoke.shell", params: { projectId: "prj_mvp_topology", + operationId: "op_gateway_timeout_requested", gatewaySessionId: "gws_DESKTOP-1MHOD9I", resourceId: "res_windows_host", capabilityId: "cap_windows_cmd_exec", @@ -169,6 +170,7 @@ test("hardware.invoke.shell dispatch timeout honors requested gateway shell time validateResponse(response); assert.equal(observedTimeoutMs, 185000); assert.equal(response.result.status, "timed_out"); + assert.equal(response.result.operationId, "op_gateway_timeout_requested"); assert.equal(response.result.dispatch.dispatchStatus, "timed_out"); assert.equal(response.result.dispatch.shellExecuted, false); }); diff --git a/scripts/gateway-outbound-demo-smoke.mjs b/scripts/gateway-outbound-demo-smoke.mjs index 8bd78452..699d98c0 100644 --- a/scripts/gateway-outbound-demo-smoke.mjs +++ b/scripts/gateway-outbound-demo-smoke.mjs @@ -86,13 +86,44 @@ try { assert.equal(rpc.result?.dispatch?.shellExecuted, true); assert.equal(rpc.result?.dispatch?.exitCode, 0); assert.match(rpc.result?.dispatch?.stdout ?? "", /hwlab-demo/u); + + const slow = postJson(`${apiBaseUrl}/json-rpc`, shellRpc({ + id: "req_gateway_outbound_demo_slow", + traceId: "trc_gateway_outbound_demo_slow", + command: process.platform === "win32" ? "ping -n 4 127.0.0.1 >NUL" : "sleep 3", + cwd, + timeoutMs: 5000 + })); + await delay(250); + const quickStartedAt = Date.now(); + const quick = await postJson(`${apiBaseUrl}/json-rpc`, shellRpc({ + id: "req_gateway_outbound_demo_quick", + traceId: "trc_gateway_outbound_demo_quick", + command: "echo hwlab-quick", + cwd, + timeoutMs: 5000 + })); + const quickDurationMs = Date.now() - quickStartedAt; + assert.equal(quick.result?.dispatch?.shellExecuted, true); + assert.equal(quick.result?.dispatch?.exitCode, 0); + assert.match(quick.result?.dispatch?.stdout ?? "", /hwlab-quick/u); + assert.ok(quickDurationMs < 2500, `quick request was blocked by slow request for ${quickDurationMs}ms`); + + const sessionDuringSlow = await waitForGatewaySession(apiBaseUrl, "gws_gtw_local_demo"); + assert.equal(Number.isInteger(sessionDuringSlow.inflightCount), true); + const slowResult = await slow; + assert.equal(slowResult.result?.dispatch?.shellExecuted, true); + assert.equal(slowResult.result?.dispatch?.exitCode, 0); + console.log(JSON.stringify({ status: "pass", mode: useEdgeProxy ? "edge-proxy" : "direct-cloud-api", cloudUrl: `http://127.0.0.1:${cloudPort}`, apiBaseUrl, gatewaySessionId: "gws_gtw_local_demo", - stdout: rpc.result.dispatch.stdout.trim() + stdout: rpc.result.dispatch.stdout.trim(), + nonBlockingQuickMs: quickDurationMs, + inflightCountObserved: sessionDuringSlow.inflightCount }, null, 2)); } finally { for (const child of children.reverse()) { @@ -174,6 +205,31 @@ async function postJson(url, body) { return text ? JSON.parse(text) : null; } +function shellRpc({ id, traceId, command, cwd, timeoutMs }) { + return { + jsonrpc: "2.0", + id, + method: "hardware.invoke.shell", + params: { + projectId: "prj_mvp_topology", + gatewaySessionId: "gws_gtw_local_demo", + resourceId: "res_windows_host", + capabilityId: "cap_windows_cmd_exec", + input: { + command, + cwd, + timeoutMs + } + }, + meta: { + traceId, + actorId: "usr_gateway_outbound_demo", + serviceId: "hwlab-cloud-api", + environment: "dev" + } + }; +} + function freePort() { return new Promise((resolve, reject) => { const server = createServer(); diff --git a/tools/hwlab-gateway-shell.mjs b/tools/hwlab-gateway-shell.mjs index f1e6b07f..84a66d54 100644 --- a/tools/hwlab-gateway-shell.mjs +++ b/tools/hwlab-gateway-shell.mjs @@ -36,6 +36,7 @@ async function main() { const apiBaseUrl = stripTrailingSlash(args.apiBaseUrl || process.env.HWLAB_GATEWAY_SHELL_API_BASE_URL || DEFAULT_API_BASE_URL); const traceId = args.traceId || `trc_gateway_shell_cli_${Date.now()}`; const requestId = args.requestId || `req_gateway_shell_cli_${randomUUID()}`; + const operationId = args.operationId || `op_gateway_shell_cli_${randomUUID()}`; const timeoutMs = boundedPositiveInteger(args.timeoutMs ?? process.env.HWLAB_GATEWAY_SHELL_TIMEOUT_MS, DEFAULT_GATEWAY_SHELL_TIMEOUT_MS, { min: 1000, max: MAX_GATEWAY_SHELL_TIMEOUT_MS @@ -56,6 +57,7 @@ async function main() { }, body: { projectId: args.projectId || DEFAULT_PROJECT_ID, + operationId, gatewaySessionId: args.gatewaySessionId || DEFAULT_GATEWAY_SESSION_ID, resourceId: args.resourceId || DEFAULT_RESOURCE_ID, capabilityId: args.capabilityId || DEFAULT_CAPABILITY_ID, @@ -123,6 +125,7 @@ function printHelp() { " --cwd WINDOWS_PATH Gateway-side working directory; prefer this over shell-level cd &&", " --timeout-ms N Gateway command timeout, default 120000ms; can also use HWLAB_GATEWAY_SHELL_TIMEOUT_MS", " --request-timeout-ms N HTTP wait timeout; defaults to command timeout plus a small response grace", + " --operation-id ID Stable operationId to preserve correlation if dispatch times out", " --project-id ID Default prj_mvp_topology", " --gateway-session-id ID Default gws_DESKTOP-1MHOD9I", " --resource-id ID Default res_windows_host", diff --git a/web/hwlab-cloud-web/scripts/check.mjs b/web/hwlab-cloud-web/scripts/check.mjs index d68c8aa3..4e167790 100644 --- a/web/hwlab-cloud-web/scripts/check.mjs +++ b/web/hwlab-cloud-web/scripts/check.mjs @@ -69,6 +69,8 @@ const cloudWebProxy = fs.readFileSync(path.resolve(repoRoot, "internal/dev-entry const cloudApiServer = fs.readFileSync(path.resolve(repoRoot, "internal/cloud/server.mjs"), "utf8"); const cloudJsonRpc = fs.readFileSync(path.resolve(repoRoot, "internal/cloud/json-rpc.mjs"), "utf8"); const gatewayDemoRegistry = fs.readFileSync(path.resolve(repoRoot, "internal/cloud/gateway-demo-registry.mjs"), "utf8"); +const gatewayMain = fs.readFileSync(path.resolve(repoRoot, "cmd/hwlab-gateway/main.mjs"), "utf8"); +const gatewayDemoSmoke = fs.readFileSync(path.resolve(repoRoot, "scripts/gateway-outbound-demo-smoke.mjs"), "utf8"); const codexStdioSession = fs.readFileSync(path.resolve(repoRoot, "internal/cloud/codex-stdio-session.mjs"), "utf8"); const gatewayShellTool = fs.readFileSync(path.resolve(repoRoot, "tools/hwlab-gateway-shell.mjs"), "utf8"); const frontendSource = `${html}\n${auth}\n${app}\n${artifactPublisher}`; @@ -1138,9 +1140,12 @@ assert.match(codexStdioSession, /Windows filesystem inventory/); assert.match(codexStdioSession, /Select-Object -First/); assert.match(codexStdioSession, /ConvertTo-Json -Compress/); assert.match(codexStdioSession, /C:\\\\Users\\\\liang\\\\\.agents\\\\skills\\\\keil/); +assert.match(codexStdioSession, /async job flow/); assert.match(gatewayShellTool, /function powershellEncodedCommand/); assert.match(gatewayShellTool, /DEFAULT_GATEWAY_SHELL_TIMEOUT_MS\s*=\s*120000/); assert.match(gatewayShellTool, /REQUEST_TIMEOUT_GRACE_MS/); +assert.match(gatewayShellTool, /op_gateway_shell_cli_\$\{randomUUID\(\)\}/); +assert.match(gatewayShellTool, /operationId,/); assert.match(gatewayShellTool, /Read-HwlabText/); assert.match(gatewayShellTool, /Buffer\.from\(`\$\{prologue\}\$\{source\}`,\s*"utf16le"\)\.toString\("base64"\)/); assert.match(gatewayShellTool, /--powershell-stdin/); @@ -1148,7 +1153,15 @@ assert.match(gatewayShellTool, /--cwd WINDOWS_PATH/); assert.match(gatewayShellTool, /powershell\.exe -NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -EncodedCommand/); assert.doesNotMatch(gatewayShellTool, /keil-find|keil-build|keil-job-status/); assert.match(cloudJsonRpc, /Math\.max\(configured \?\? 0,\s*requested \?\? 0,\s*120000\)/); +assert.match(cloudJsonRpc, /op_gateway_shell_\$\{randomUUID\(\)\}/); assert.match(gatewayDemoRegistry, /DEFAULT_DISPATCH_TIMEOUT_MS\s*=\s*120000/); +assert.match(gatewayDemoRegistry, /inflightCount/); +assert.match(gatewayDemoRegistry, /maxInflightRequests/); +assert.match(gatewayMain, /HWLAB_GATEWAY_MAX_INFLIGHT/); +assert.match(gatewayMain, /function startCloudRequest/); +assert.match(gatewayMain, /Promise\.resolve\(\)\s*\n\s*\.then\(\(\) => handleCloudRequest/); +assert.doesNotMatch(gatewayMain, /await handleCloudRequest\(response\.body\.request\)/); +assert.match(gatewayDemoSmoke, /quick request was blocked by slow request/); assert.match(cloudApiServer, /HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS,\s*120000/); assert.match(codexStdioSession, /detached:\s*childDetached/); assert.match(codexStdioSession, /function terminateChildProcess/);