From bfb825b0a0bcbb23fb21ce4aa5b013964a2b18c6 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 20:33:39 +0200 Subject: [PATCH 1/7] =?UTF-8?q?fix:=20=E6=94=B6=E6=95=9B=20trans=20?= =?UTF-8?q?=E9=A1=B6=E5=B1=82=E5=B8=AE=E5=8A=A9=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/src/help.ts | 255 +++++++++++++++++++++++------------ scripts/src/ssh-help.test.ts | 129 ++++++++++++------ 2 files changed, 252 insertions(+), 132 deletions(-) diff --git a/scripts/src/help.ts b/scripts/src/help.ts index 8f8bda0f..0a927ba2 100644 --- a/scripts/src/help.ts +++ b/scripts/src/help.ts @@ -169,111 +169,188 @@ export function serverHelp(action: string | undefined = undefined): unknown { }; } -export type SshHelpScope = "download"; +export type SshHelpScope = + | "argv" | "exec" | "git" | "node" | "kubectl" | "logs" + | "sh" | "bash" | "py" | "ps" | "cmd" + | "pwd" | "ls" | "cat" | "head" | "tail" | "stat" | "wc" | "rg" | "find" | "glob" | "skills" + | "apply-patch" | "apply-patch-v1" | "upload" | "download"; + +type SshOperationHelp = { + group: "process" | "shell" | "filesystem" | "patch" | "transfer"; + description: string; + usage: string[]; + notes?: string[]; +}; + +const SSH_OPERATION_HELP: Record = { + argv: { + group: "process", + description: "以原始 argv 边界运行一个远端进程；需要管道、变量或重定向时改用 sh/bash。", + usage: ["trans argv [args...]", "trans D601:/workspace argv git status --short --branch"], + }, + exec: { + group: "process", + description: "在 k3s workload route 中运行进程，并可显式透传 stdin 或指定容器内 cwd。", + usage: ["trans :k3s::[:] exec [--cwd /path] [--stdin] -- [args...]"], + }, + git: { + group: "process", + description: "在 host/workspace 或 Windows workspace route 中运行 Git convenience wrapper。", + usage: ["trans :/workspace git status --short --branch", "trans :win// git diff --check"], + }, + node: { + group: "process", + description: "在目标 route 中直接运行 Node.js argv。", + usage: ["trans node -e 'console.log(process.version)'"], + }, + kubectl: { + group: "process", + description: "在 k3s control-plane route 上执行有界 kubectl 诊断。", + usage: ["trans :k3s kubectl get pods -n "], + }, + logs: { + group: "process", + description: "从 k3s control-plane 或 workload route 读取有界日志。", + usage: ["trans :k3s:: logs --tail 80", "trans :k3s logs -n -l --tail 120"], + }, + sh: { + group: "shell", + description: "从 stdin 或单个命令字符串执行远端 POSIX /bin/sh。", + usage: ["trans sh <<'SH'", "trans sh -- ''"], + notes: ["sh -- 只接受一个 shell command string；单进程多 argv 使用 argv。"], + }, + bash: { + group: "shell", + description: "仅在需要 pipefail、数组、[[ ]] 等 Bash 语法时执行远端 Bash。", + usage: ["trans bash <<'BASH'", "trans bash -- ''"], + }, + py: { + group: "shell", + description: "从本地 stdin 执行远端 Python 源码，并保留 script argv。", + usage: ["trans py [script-args...] < script.py"], + }, + ps: { + group: "shell", + description: "在 Windows route 中执行 PowerShell 源码或 argv；route 已包含 win。", + usage: ["trans :win[//] ps <<'PS'", "trans :win ps ''"], + }, + cmd: { + group: "shell", + description: "在 Windows route 中以 UTF-8 环境执行 cmd.exe/batch。", + usage: ["trans :win[//] cmd <<'CMD'", "trans :win cmd ver"], + }, + pwd: { group: "filesystem", description: "读取 Windows workspace 当前目录。", usage: ["trans :win// pwd"] }, + ls: { group: "filesystem", description: "有界列出 Windows workspace 文件。", usage: ["trans :win// ls --limit 50"] }, + cat: { group: "filesystem", description: "读取远端文本文件；二进制文件使用 download。", usage: ["trans cat "] }, + head: { group: "filesystem", description: "按行读取远端文本文件开头。", usage: ["trans head -n 40 "] }, + tail: { group: "filesystem", description: "按行读取远端文本文件结尾。", usage: ["trans tail -n 40 "] }, + stat: { group: "filesystem", description: "读取远端文件元数据。", usage: ["trans stat "] }, + wc: { group: "filesystem", description: "统计远端文本的行、词、字符或字节。", usage: ["trans wc [-l|-w|-m|-c] "] }, + rg: { group: "filesystem", description: "执行有界远端 ripgrep；Windows route 使用受控 UTF-8 子集。", usage: ["trans rg [options] [path...]"] }, + find: { group: "filesystem", description: "不依赖 shell quoting 的结构化远端 find。", usage: ["trans find [--contains TEXT] [--limit N]"] }, + glob: { group: "filesystem", description: "通过远端 helper 执行有界 glob 匹配。", usage: ["trans glob [--root DIR] [--pattern PATTERN] [--limit N]"] }, + skills: { group: "filesystem", description: "发现 WSL/Linux 及可选 Windows skill 目录。", usage: ["trans skills [--scope all|wsl|windows] [--limit N]"] }, + "apply-patch": { + group: "patch", + description: "默认远端文本修改入口；本地 v2 engine 计算变更，route 仅负责读写。", + usage: ["trans apply-patch [--cwd /path] < patch.diff"], + notes: ["失败后重读当前目标块并缩小 hunk；不得自动回退 apply-patch-v1 或整文件 upload。"], + }, + "apply-patch-v1": { + group: "patch", + description: "显式调用 legacy remote apply_patch helper；只作为人工确认后的兼容入口。", + usage: ["trans apply-patch-v1 [--allow-loose] < patch.diff"], + }, + upload: { + group: "transfer", + description: "上传必要的二进制文件或生成物并自动核对字节数与 SHA-256。", + usage: ["trans upload "], + notes: ["远端文本读取使用 cat/rg，修改使用 apply-patch。"], + }, + download: { + group: "transfer", + description: "下载必要的二进制文件或生成物并自动核对字节数与 SHA-256。", + usage: ["trans download "], + }, +}; + +const SSH_OPERATION_GROUPS = { + process: ["argv", "exec", "git", "node", "kubectl", "logs"], + shell: ["sh", "bash", "py", "ps", "cmd"], + filesystem: ["pwd", "ls", "cat", "head", "tail", "stat", "wc", "rg", "find", "glob", "skills"], + patch: ["apply-patch", "apply-patch-v1"], + transfer: ["upload", "download"], +} as const satisfies Record; + +const SSH_HELP_SCOPES = Object.keys(SSH_OPERATION_HELP) as SshHelpScope[]; +const SSH_HELP_SCOPE_SET = new Set(SSH_HELP_SCOPES); + +function sshEntrypoint(): "trans" | "tran" | "ssh" { + const rawEntrypoint = process.env.UNIDESK_SSH_ENTRYPOINT?.trim(); + return rawEntrypoint === "trans" || rawEntrypoint === "tran" || rawEntrypoint === "ssh" ? rawEntrypoint : "ssh"; +} export function sshHelpScope(args: string[]): SshHelpScope | undefined { const [top, helpToken, scope] = args; - if (top !== "ssh" || args.length !== 3 || !isHelpToken(helpToken)) return undefined; - return scope === "download" ? scope : undefined; + if (top !== "ssh" || args.length !== 3 || !isHelpToken(helpToken) || !SSH_HELP_SCOPE_SET.has(scope ?? "")) return undefined; + return scope as SshHelpScope; } export function sshHelp(scope: SshHelpScope | undefined = undefined): unknown { if (scope === "download") return sshDownloadHelp(); + if (scope !== undefined) return sshOperationHelp(scope); + const entrypoint = sshEntrypoint(); + const operationGroups = Object.fromEntries( + Object.entries(SSH_OPERATION_GROUPS).map(([group, operations]) => [group, operations.join(" | ")]), + ); return { - command: "ssh", + command: `${entrypoint} --help`, output: "json", - description: "Open a Host SSH / WSL SSH maintenance session through the provider-gateway bridge.", - usage: [ - "trans ", - "trans argv [args...]", - "trans :/absolute/workspace apply-patch < patch.diff", - "trans upload ", - "trans download ", - "trans apply-patch-v1 [--allow-loose] < patch.diff", - "trans py [script-args...] < script.py", - "trans sh [arg...] <<'SH'", - "trans bash [arg...] <<'BASH'", - "trans skills [--scope all|wsl|windows] [--limit N]", - "trans find [--contains TEXT] [--limit N]", - "trans glob [--root DIR] [--pattern PATTERN]", - "trans NC01:/root/hwlab-v03 git status --short --branch", - "trans D601:win ps <<'PS'", - "trans D601:win/c/test ps <<'PS'", - "trans D601:win/c/test cmd <<'CMD'", - "trans D601:win cmd ver", - "trans D601:win/c/test cmd cd", - "trans D601:win/c/test pwd", - "trans D601:win/c/test ls --limit 50", - "trans D601:win/c/test cat README.md", - "trans D601:win/c/test head -n 40 README.md", - "trans D601:win/c/test tail -n 40 README.md", - "trans D601:win/c/test stat README.md", - "trans D601:win/c/test wc README.md", - "trans D601:win/c/test rg -i needle .", - "trans D601:win/c/test git status --short --branch", - "trans D601:win/c/test git diff --check", - "trans D601:win/c/test git commit -m 'fix: update docs'", - "trans D601:win/c/test git exact-commit status", - "trans D601:win/c/test git exact-commit plan --path src/main.c", - "trans D601:win/c/test git exact-commit run --path src/main.c --message 'fix: exact change' --confirm", - "trans D601:win skills [--scope agents|codex|all] [--limit N]", - "trans NC01:k3s", - "trans NC01:k3s kubectl get pods -n hwlab-dev", - "trans NC01:k3s", - "trans NC01:k3s kubectl get pipelineruns -n hwlab-ci", - "trans NC01:k3s sh <<'SH'", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api:hwlab-cloud-api exec --cwd /app -- pwd", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api:hwlab-cloud-api apply-patch --cwd /app <<'PATCH'", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api apply-patch-v1 <<'PATCH'", - "tar -C /path/to/files -cf - . | trans NC01:k3s:unidesk:code-queue exec --cwd /root/unidesk --stdin -- tar -xf - -C /root/unidesk", - "trans D601:win/c/test apply-patch <<'PATCH'", - "trans D601:win upload ./tool.mjs C:\\Temp\\tool.mjs", - "trans D601:win download C:\\Temp\\tool.mjs ./tool.mjs", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api node -e 'console.log(process.version)'", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api sh <<'SH'", - "trans NC01:k3s:hwlab-dev:hwlab-cloud-api logs --tail 80", - "trans NC01:k3s logs -n agentrun-ci -l tekton.dev/pipelineRun= --tail 120", - ], - notes: [ - "trans --help and trans --help print this JSON help and never open an interactive session; the underlying ssh subcommand keeps the same help behavior.", - "For non-interactive remote commands, prefer argv for a single process and explicit sh/bash stdin for shell logic.", - "Windows routes have explicit Windows operations, not POSIX shell aliases: `ps` runs Windows PowerShell from stdin or one inline command, `cmd` runs cmd.exe/batch from stdin or one command line, and `skills` discovers Windows skill directories.", - "On Windows routes, the route already contains `win`; write `trans D601:win/c/test ps`, not `trans D601:win/c/test win ps`. Repository commands can use the git convenience wrapper, including `git status`, `git diff`, and non-interactive `git commit -m ...`; commands that need shell review should use `ps` or `cmd`.", - "Windows routes include read-only filesystem convenience operations `pwd`, `ls`, `cat`, `head`, `tail`, `stat`, `wc`, and a bounded UTF-8 `rg` subset. These are implemented through a Windows fs backend with UTF-8/binary checks and bounded output; they do not imply POSIX `sh`/`bash` availability.", - "For Windows PowerShell, use `trans :win ps <<'PS'`; the PowerShell body is written to a temporary .ps1 with UTF-8 settings and executed by powershell.exe. Do not use POSIX `sh` or `bash` for Windows PowerShell.", - "For Windows cmd.exe, use `trans :win// cmd <<'CMD'`; `cmd` with no command-line arguments reads the UTF-8 batch body from stdin, injects UTF-8/Python encoding defaults, runs it from a temp .cmd file, and deletes the temp file.", - "`argv` executes direct argv tokens only: `trans argv ls -la` is valid, but `trans argv 'ls -la'` is rejected because the single string would be treated as an executable path; use `sh -- 'ls -la'` or `bash -- 'ls -la'` for one-line shell logic.", - "For one-line remote shell logic without a heredoc, use `sh -- ''` for POSIX syntax or `bash -- ''` for Bash syntax. Outer shell operators written outside trans, such as `trans NC01:/repo sed ... && sed ...`, are parsed by the local shell before UniDesk starts and therefore cannot be redirected by the CLI.", - "`sh --` and `bash --` accept exactly one shell command string. For direct argv commands with multiple tokens, use `argv`, `exec`, or a known direct subcommand such as `git`, `rg`, `sed`, or `cat`.", - "The removed `script` and `shell` operations intentionally fail. The operation name must declare the shell dialect: `sh` maps to target `/bin/sh`, while `bash` maps to target `bash`.", - "sh and bash helper modes inject a tiny POSIX-compatible printf wrapper before user shell text, so portable printf headings such as `printf \"--- section ---\\n\"` work consistently under dash/sh and bash. Direct argv commands are unchanged.", - "For arbitrary stdin streams into a workload command, use a workload route plus `exec --stdin -- ...`; this keeps the route as location-only and avoids heredoc/base64/tar shell wrapping.", - "`apply-patch` is the default remote text patch entry and uses the v2 local line-based patch engine with remote read/write operations, including Windows routes such as `D601:win/c/test`, so long Unicode/Chinese lines and pure insertion hunks avoid the legacy remote shell hunk parser. Plain multi-file Update File patches on POSIX host/k3s and Windows workspace routes use bulk read/write operations to avoid per-file SSH round trips. Its stdout follows Codex apply_patch text output rather than UniDesk JSON output; stderr keeps Codex-style failure text and appends one `UNIDESK_APPLY_PATCH_TIMING` JSON summary with durationMs, patchBytes, fileCount, hunkCount, changedCount, remoteOperationCount, remoteOperationCounts and remoteElapsedMs so slow patch runs can be attributed without changing success stdout.", - "`upload` 和 `download` 只用于必要的二进制文件或生成物；远端文本应避免反复传输，读取使用 `cat`/`rg`，修改使用 `apply-patch`。成功传输的 JSON 会在 `hint` 中重复该提示，通过远端临时文件写入，自动核对两端字节数和 SHA-256，并返回 `verification.automatic=true`、`verification.verified=true` 和 `verification.match.{bytes,sha256}=true`；该 JSON 已是完整性证据，无需额外执行 `sha256sum`。下载通过 `host.ssh.tcp-pool` 流式传输并输出进度 JSON；受控异步产物任务可传入 `--inactivity-timeout-ms`，只要数据持续流动就不受普通短命令预算影响。Windows route 下载（例如 `trans D601:win download C:\\Temp\\tool.mjs ./tool.mjs`）会自动把盘符路径映射到同一 provider 的 WSL `/mnt/` host route，同时仍返回字节数和 SHA-256 校验证据。", - "`apply-patch-v1` is the only legacy fallback entry: it rejects low-context update hunks by default, reports the matched file:line for each hunk on stderr, and only accepts --allow-loose when the caller has manually reviewed an intentionally ambiguous insertion.", - "`sh` inherits provider proxy variables such as HTTP_PROXY/HTTPS_PROXY/ALL_PROXY/NO_PROXY and runs target `/bin/sh`; use `bash` only for Bash syntax such as `pipefail`, arrays, substring expansion, or `[[ ... ]]`, not as a proxy workaround.", - "Route syntax is `{provider}:{plane}[:{scope...}] {operation} [operation-args...]`: the first argv token locates a distributed target only, and every following token belongs to the operation parser. Host workspace routes use `:/absolute/workspace`; WSL providers can use `:win ps` for Windows PowerShell and `:win cmd` for Windows cmd.exe, with `:win/c/test ...` mapping the Windows cwd to `C:\\test`; native k3s providers such as D601 and G14 use `:k3s` for the control plane and `:k3s::[:]` for a workload/container. In k3s routes, `:` is the distributed route separator; `/...` is only an in-container filesystem cwd and never selects a container. Prefer operation `--cwd /path` when a container is also specified.", - "Use `win`, not `win32`; the win route is a Windows operation plane. `ps` and `cmd` both set UTF-8/Python encoding defaults, while `cmd` also sets `chcp 65001`.", - "`:win skills` discovers the current Windows user's `%USERPROFILE%\\.agents\\skills` by default; use `--scope all` to include `%USERPROFILE%\\.codex\\skills`.", - "Do not put operation names in any colon route segment, including nested k3s namespace/workload/container segments.", - "Do not use post-provider shorthand such as `trans NC01 k3s ...`; write `trans NC01:k3s ...` so location and operation stay separated.", - "If an ssh-like remote command fails with timeout/kex/exit-255 friction, stderr includes one low-noise UNIDESK_SSH_HINT JSON line with the argv retry command.", - "Ordinary non-interactive ssh/trans/tran remote commands have a hard top-level runtime timeout capped at 60s. Timeout writes UNIDESK_SSH_RUNTIME_TIMEOUT or UNIDESK_TRAN_TIMEOUT_HINT and disconnects the broker; long CI/CD, trace, logs, build, or hardware work must use submit-and-poll / short query loops instead of keeping trans open. Whole-file `download` is the narrow exception: controlled async callers can pass `--inactivity-timeout-ms` for a verified progress-emitting tcp-pool transfer.", - "Only slow ssh/trans/tran runtime writes UNIDESK_SSH_TIMING JSON to stderr; operations over 10s are marked level=warning even when they succeed, because slow successful calls are a distributed performance monitoring signal. Check provider latency, remote command cost, helper bootstrap, or remote patch optimization before repeating high-frequency work. Routine short calls do not emit timing noise.", - "The local trans/tran wrapper must not add provider/plane directory locks; rely on k8s/Tekton/Argo/Lease or server-side TTL queues for coordination.", - "Use -- before a remote command that intentionally starts with a dash.", + description: "通过 provider-gateway bridge 访问 Host SSH、WSL、Windows 或 k3s 目标。", + routeSyntax: { + general: `${entrypoint} [operation-args...]`, + hostWorkspace: ":/absolute/workspace", + k3sControlPlane: ":k3s", + k3sWorkload: ":k3s::[:]", + windowsWorkspace: ":win[//]", + githubContent: "gh://[/issue|/pr]/", + }, + operationGroups, + scopedHelp: { + command: `${entrypoint} --help `, + availableOperationCount: SSH_HELP_SCOPES.length, + examples: [`${entrypoint} --help apply-patch`, `${entrypoint} --help download`, `${entrypoint} --help ps`], + }, + boundaries: [ + "route 只定位目标，route 后的 token 全部交给 operation parser。", + "单进程优先 argv；shell 逻辑显式使用 sh/bash，Windows 使用 ps/cmd。", + "远端文本读取优先 cat/rg，修改优先 apply-patch；upload/download 仅用于必要的二进制或生成物。", + "普通 trans 短连接上限保持 60s；长流程使用受控 submit-and-poll 入口。", ], }; } +function sshOperationHelp(scope: Exclude): unknown { + const entrypoint = sshEntrypoint(); + const detail = SSH_OPERATION_HELP[scope]; + return { + command: `${entrypoint} --help ${scope}`, + output: "json", + scope, + group: detail.group, + description: detail.description, + routeSyntax: `${entrypoint} ${scope} [operation-args...]`, + usage: detail.usage.map((value) => value.replace(/^trans\b/u, entrypoint)), + notes: detail.notes ?? [], + next: [`${entrypoint} --help`, `${entrypoint} --help ${scope}`], + }; +} + function sshDownloadHelp(): unknown { const configuredRepoRoot = process.env.UNIDESK_TRANS_REPO_ROOT?.trim() || null; - const rawEntrypoint = process.env.UNIDESK_SSH_ENTRYPOINT?.trim(); - const entrypoint = rawEntrypoint === "trans" || rawEntrypoint === "tran" || rawEntrypoint === "ssh" - ? rawEntrypoint - : "ssh"; + const entrypoint = sshEntrypoint(); return { - command: `${entrypoint} download`, + command: `${entrypoint} --help download`, output: "json", scope: "download", description: "下载一个必要的二进制文件或生成物，并自动核对远端/本地字节数与 SHA-256；远端文本改用 cat/rg/apply-patch。", diff --git a/scripts/src/ssh-help.test.ts b/scripts/src/ssh-help.test.ts index d0b8cbd1..7c7000f3 100644 --- a/scripts/src/ssh-help.test.ts +++ b/scripts/src/ssh-help.test.ts @@ -3,15 +3,74 @@ import { describe, expect, test } from "bun:test"; import { repoRoot } from "./config"; import { sshHelp, sshHelpScope } from "./help"; -describe("ssh download scoped help", () => { - test("selects only the exact help-first download scope", () => { +type RenderedHelp = { + ok: boolean; + command: string; + data: { + command?: string; + scope?: string; + runtime?: { repoRoot: string }; + routeSyntax?: Record | string; + operationGroups?: Record; + scopedHelp?: { availableOperationCount: number }; + usage?: string[]; + outputTruncated?: boolean; + dump?: unknown; + }; +}; + +function runTransHelp(...args: string[]): { stdout: string; rendered: RenderedHelp } { + const result = spawnSync("bun", ["scripts/ssh-cli.ts", ...args], { + cwd: repoRoot, + env: { + ...process.env, + UNIDESK_SSH_ENTRYPOINT: "trans", + UNIDESK_TRANS_REPO_ROOT: repoRoot, + }, + encoding: "utf8", + }); + expect(result.status).toBe(0); + expect(result.stderr).toBe(""); + return { stdout: result.stdout, rendered: JSON.parse(result.stdout) as RenderedHelp }; +} + +describe("ssh bounded progressive help", () => { + test("selects exact help-first operation scopes", () => { expect(sshHelpScope(["ssh", "--help", "download"])).toBe("download"); - expect(sshHelpScope(["ssh", "help", "download"])).toBe("download"); + expect(sshHelpScope(["ssh", "help", "apply-patch"])).toBe("apply-patch"); + expect(sshHelpScope(["ssh", "--help", "ps"])).toBe("ps"); expect(sshHelpScope(["ssh", "--help"])).toBeUndefined(); + expect(sshHelpScope(["ssh", "--help", "unknown"])).toBeUndefined(); expect(sshHelpScope(["ssh", "NC01:k3s", "--help"])).toBeUndefined(); expect(sshHelpScope(["ssh", "--help", "download", "extra"])).toBeUndefined(); }); + test("keeps the top-level model compact while indexing routes and every scoped operation", () => { + const top = sshHelp() as { + routeSyntax: Record; + operationGroups: Record; + scopedHelp: { command: string; availableOperationCount: number }; + outputTruncated?: boolean; + }; + const serialized = JSON.stringify(top); + const grouped = Object.values(top.operationGroups).flatMap((value) => value.split(" | ")); + + expect(Buffer.byteLength(serialized, "utf8")).toBeLessThan(6_144); + expect(top.outputTruncated).toBeUndefined(); + expect(top.routeSyntax.general).toContain(" "); + expect(top.routeSyntax.k3sWorkload).toContain("::"); + expect(top.routeSyntax.windowsWorkspace).toContain(":win"); + expect(top.operationGroups.patch).toBe("apply-patch | apply-patch-v1"); + expect(top.operationGroups.transfer).toBe("upload | download"); + expect(new Set(grouped).size).toBe(top.scopedHelp.availableOperationCount); + expect(top.scopedHelp.command).toContain("--help "); + for (const operation of grouped) { + const scope = sshHelpScope(["ssh", "--help", operation]); + expect(scope).toBe(operation); + expect(Buffer.byteLength(JSON.stringify(sshHelp(scope)), "utf8")).toBeLessThan(6_144); + } + }); + test("keeps download help bounded and discloses the executing source root", () => { const scoped = sshHelp("download") as { scope: string; @@ -20,13 +79,11 @@ describe("ssh download scoped help", () => { examples: Record; contract: { pathOrder: string[]; transport: string; verification: { automatic: boolean } }; }; - const full = sshHelp() as { usage: string[] }; const serialized = JSON.stringify(scoped); expect(scoped.scope).toBe("download"); expect(scoped.usage.length).toBeLessThanOrEqual(4); - expect(scoped.usage.length).toBeLessThan(full.usage.length); - expect(Buffer.byteLength(serialized, "utf8")).toBeLessThan(10_240); + expect(Buffer.byteLength(serialized, "utf8")).toBeLessThan(6_144); expect(serialized).not.toContain("git exact-commit"); expect(scoped.runtime.repoRoot).toBe(repoRoot); expect(scoped.runtime.cliPath).toBe(`${repoRoot}/scripts/ssh-cli.ts`); @@ -40,46 +97,32 @@ describe("ssh download scoped help", () => { "trans NC01:k3s:hwlab-v03:hwlab-cloud-api:hwlab-cloud-api download /tmp/trace.json ./trace.json", ); expect(scoped.examples.windowsDrive).toContain("D:\\tmp\\trace.json"); - expect(scoped.examples.explicitWorktree).toBe( - `UNIDESK_TRANS_REPO_ROOT=${repoRoot} trans NC01:k3s:hwlab-v03:hwlab-cloud-api:hwlab-cloud-api download /tmp/trace.json ./trace.json`, - ); }); - test("renders the original trans command without stdout dump fallback", () => { - const result = spawnSync("bun", ["scripts/ssh-cli.ts", "--help", "download"], { - cwd: repoRoot, - env: { - ...process.env, - UNIDESK_SSH_ENTRYPOINT: "trans", - UNIDESK_TRANS_REPO_ROOT: repoRoot, - }, - encoding: "utf8", - }); + test("renders compact top-level and scoped help without dump fallback", () => { + const top = runTransHelp("--help"); + expect(Buffer.byteLength(top.stdout, "utf8")).toBeLessThan(6_144); + expect(top.rendered.ok).toBe(true); + expect(top.rendered.command).toBe("trans --help"); + expect(top.rendered.data.outputTruncated).toBeUndefined(); + expect(top.rendered.data.dump).toBeUndefined(); + expect(top.rendered.data.operationGroups?.transfer).toContain("download"); - expect(result.status).toBe(0); - expect(result.stderr).toBe(""); - expect(Buffer.byteLength(result.stdout, "utf8")).toBeLessThan(10_240); - const rendered = JSON.parse(result.stdout) as { - ok: boolean; - command: string; - data: { scope: string; runtime: { repoRoot: string }; usage: string[]; outputTruncated?: boolean }; - }; - expect(rendered.ok).toBe(true); - expect(rendered.command).toBe("trans --help download"); - expect(rendered.data.scope).toBe("download"); - expect(rendered.data.runtime.repoRoot).toBe(repoRoot); - expect(rendered.data.usage.length).toBeLessThanOrEqual(4); - expect(rendered.data.outputTruncated).toBeUndefined(); + const routeTop = runTransHelp("NC01:k3s", "--help"); + expect(routeTop.rendered.command).toBe("trans NC01:k3s --help"); + expect(routeTop.rendered.data.outputTruncated).toBeUndefined(); + expect(routeTop.rendered.data.operationGroups).toEqual(top.rendered.data.operationGroups); - const unified = spawnSync("bun", ["scripts/cli.ts", "ssh", "--help", "download"], { - cwd: repoRoot, - env: process.env, - encoding: "utf8", - }); - expect(unified.status).toBe(0); - expect(unified.stderr).toBe(""); - const unifiedRendered = JSON.parse(unified.stdout) as { data: { scope: string; runtime: { repoRoot: string } } }; - expect(unifiedRendered.data.scope).toBe("download"); - expect(unifiedRendered.data.runtime.repoRoot).toBe(repoRoot); + const applyPatch = runTransHelp("--help", "apply-patch"); + expect(applyPatch.rendered.command).toBe("trans --help apply-patch"); + expect(applyPatch.rendered.data.scope).toBe("apply-patch"); + expect(applyPatch.rendered.data.usage?.[0]).toContain("apply-patch"); + expect(applyPatch.rendered.data.outputTruncated).toBeUndefined(); + + const download = runTransHelp("--help", "download"); + expect(download.rendered.command).toBe("trans --help download"); + expect(download.rendered.data.scope).toBe("download"); + expect(download.rendered.data.runtime?.repoRoot).toBe(repoRoot); + expect(download.rendered.data.outputTruncated).toBeUndefined(); }); }); From 8d2a7f2e7cca983c9e6c807df839ed5b66ca252f Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 20:34:20 +0200 Subject: [PATCH 2/7] =?UTF-8?q?docs:=20=E5=9B=BA=E5=8C=96=E9=AB=98?= =?UTF-8?q?=E9=A2=91=E5=B8=AE=E5=8A=A9=E8=BE=93=E5=87=BA=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .agents/skills/unidesk-trans/SKILL.md | 5 +++++ docs/reference/cli.md | 8 +++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/.agents/skills/unidesk-trans/SKILL.md b/.agents/skills/unidesk-trans/SKILL.md index 7ac9b06b..800e3117 100644 --- a/.agents/skills/unidesk-trans/SKILL.md +++ b/.agents/skills/unidesk-trans/SKILL.md @@ -31,6 +31,11 @@ Host workspace、k3s、Windows、GitHub issue/PR route 见 [references/routes.md ## P0 边界 +- `trans --help` 是高频直接阅读入口： + - 顶层只展示 route 语法、operation 分组和 `trans --help ` 下钻索引。 + - 顶层与 scoped help 都必须有界且直接返回；出现 `outputTruncated` 或 dump 即视为 CLI 缺陷。 + - 修复帮助过长时必须收敛信息结构，禁止提高 stdout 阈值。 + - dump 只允许用于显式 `--full`、`--raw` 或确实无法预先界定的一次性异常证据。 - 远端文本修改优先 `trans apply-patch`；不要 download/upload/sed 拼临时 diff 代替 patch。 - `upload` / `download` 仅用于必要的二进制文件或生成物： - 远端文本读取优先使用 `cat` / `rg`； diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 32ae3309..902e2a8b 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -2,6 +2,12 @@ UniDesk 的统一 CLI 实现入口是根目录 `scripts/cli.ts`，运行方式固定为 `bun scripts/cli.ts `；普通根 CLI 子命令仍使用该入口。`trans ...` 是 SSH/WSL/k3s 透传专用入口，wrapper 委托轻量 `scripts/ssh-cli.ts` 启动链路，避免被无关根 CLI 子命令模块的解析或语法错误拖垮；长期参考文档、AGENTS 索引、CLI help 和人工远端操作示例都必须优先写 `trans ...`，不得再把 `bun scripts/cli.ts ssh ...` 作为默认透传入口。面向人工的默认输出采用 Kubernetes 风格的简洁表格、短摘要和 drill-down 命令；JSON 只用于 `--json`、`--raw`、`--full`、`-o json` 或其他明确机器消费模式。所有成功和失败路径仍必须给出可判定状态，避免无输出造成不可观测。 +- 高频 CLI 默认输出与帮助输出是直接阅读入口： + - 必须保持有界、低噪声并直接展示，不得触发 `outputTruncated` 或 `/tmp/unidesk-cli-output` dump。 + - 顶层帮助只展示分组索引和可执行的下钻命令，详细合同通过 ` --help ` 或等价 scoped help 获取。 + - 默认输出或帮助输出触发 dump 视为 CLI 缺陷，必须收敛信息结构，禁止用提高 stdout 阈值掩盖。 + - dump 只允许用于用户显式请求的 `--full`、`--raw`、机器完整输出，或确实无法预先界定的一次性异常证据。 + 主 server 必须在 PATH 上提供 `/root/.local/bin/trans` 可执行 wrapper，内容委托 repo 内版本化 `scripts/trans` 并执行 `bun scripts/ssh-cli.ts ssh "$@"`；交互 shell 可额外提供 alias，但非交互 Codex `exec` 和脚本不能依赖 alias 展开。 主 server 固定提供 PATH 可见的 `/usr/local/bin/unidesk`，委托版本化 `scripts/unidesk` 执行根 CLI；`/root/.local/bin/unidesk` 可以保留为用户级兼容链接，但不能作为非交互入口的唯一安装位置。高频人工入口优先写 `unidesk `；仓库内验证和未安装 wrapper 的 checkout 使用等价的 `bun scripts/cli.ts `。 @@ -274,7 +280,7 @@ GitHub issue/PR 正文局部修补必须优先使用 `trans gh:/owner/repo/issue `trans [ssh-like args...]` 是面向人的终端透传入口，不包装 JSON 输出，默认由 `scripts/ssh-cli.ts` 只加载 SSH/route/远程前端转发相关模块。主 server 本地执行时会在宿主机启动 `docker exec -i unidesk-backend-core backend-core --ssh-broker ...`，broker 只连接 backend-core 的 Docker 内网 `/ws/ssh`；core 使用 provider WebSocket 下发 open/dispatch 控制消息，但 stdin/stdout/stderr 数据面必须走 provider 主动连接 main server 的 `host.ssh.tcp-pool` TCP warm pool，provider-gateway 最终执行维护用 SSH 连接宿主或 WSL sshd。TTY 策略固定为交互登录 shell 使用 `ssh -tt`，带远端命令的会话使用 `ssh -T`；`apply-patch`、脚本 stdin、`py` 和旧 `apply-patch-v1` fallback 这类命令模式不得被伪终端回显或注入控制字符。该入口不暴露 database，也不改变 frontend/dev frontend/provider ingress 之外的业务边界；provider data TCP port 是 provider 主动连入的数据面端口，不是计算节点入站要求。 -`trans --help` 和 `trans --help` 是本地 JSON 帮助命令，必须快速返回；不能把 `--help` 解析成 Provider ID，不能打开交互 shell，也不能等待 provider 会话。 +`trans --help` 和 `trans --help` 是本地 JSON 帮助命令，必须快速返回；不能把 `--help` 解析成 Provider ID，不能打开交互 shell，也不能等待 provider 会话。顶层只展示 route 语法、operation 分组和 scoped help 索引；operation 细节通过 `trans --help ` 获取。任一默认帮助触发 `outputTruncated` 或 dump 都是缺陷，必须缩减顶层内容，禁止提高 stdout 阈值。 主 server 固定提供 `trans` 缩写，等价于 `bun scripts/ssh-cli.ts ssh "$@"` 的受控 UniDesk SSH 透传入口。这里必须同时保留两层入口：交互式 shell 可额外配置 alias；Codex `exec`、脚本和其他非交互 shell 不会自动展开 alias，所以还必须有 `/root/.local/bin/trans` 可执行 wrapper，内容固定为委托 repo 内版本化脚本： From 98e4cde51f0dbbb4b6ece8505eedf5be61804d7b Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 20:56:01 +0200 Subject: [PATCH 3/7] =?UTF-8?q?fix:=20=E6=A0=A1=E6=AD=A3=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=8E=A2=E9=92=88=E5=8F=AF=E9=80=89=E5=B7=A5=E5=85=B7=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=90=88=E5=90=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/hwlab-node-lanes.yaml | 11 ++- scripts/src/hwlab-node-lanes.ts | 23 +++++ ...web-observe-runner-realtime-source.test.ts | 86 +++++++++++++++++++ ...node-web-observe-runner-realtime-source.ts | 73 ++++++++++++++-- .../web-probe-observe-actions.test.ts | 8 ++ .../web-probe-observe-options.test.ts | 10 ++- 6 files changed, 196 insertions(+), 15 deletions(-) diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index c474c40f..7f3aea15 100644 --- a/config/hwlab-node-lanes.yaml +++ b/config/hwlab-node-lanes.yaml @@ -602,16 +602,19 @@ templates: requiredEventTypes: agentrun: - user_message - - tool_call - - command_output - assistant_message - terminal_status hwlab: - user - - tool - - status - assistant - terminal + conditionalEventTypePairs: + tool-call: + agentrun: tool_call + hwlab: tool + command-output: + agentrun: command_output + hwlab: status workbenchKafkaDebugReplay: enabled: true topic: hwlab.event.debug.v1 diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index b67d9713..8f02280b 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -206,6 +206,10 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec { agentrun: readonly string[]; hwlab: readonly string[]; }>; + readonly conditionalEventTypePairs: Readonly>>; } export interface HwlabRuntimeWebProbeAuthLoginSpec { @@ -1514,6 +1518,24 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla const hwlabEventTypes = nonEmptyStringArrayField(requiredEventTypes, "hwlab", `${path}.requiredEventTypes`); if (new Set(agentrunEventTypes).size !== agentrunEventTypes.length) throw new Error(`${path}.requiredEventTypes.agentrun must not contain duplicates`); if (new Set(hwlabEventTypes).size !== hwlabEventTypes.length) throw new Error(`${path}.requiredEventTypes.hwlab must not contain duplicates`); + const conditionalEventTypePairsRaw = asRecord(raw.conditionalEventTypePairs, `${path}.conditionalEventTypePairs`); + if (Object.keys(conditionalEventTypePairsRaw).length === 0) throw new Error(`${path}.conditionalEventTypePairs must contain at least one pair`); + const conditionalEventTypePairs = Object.fromEntries(Object.entries(conditionalEventTypePairsRaw).map(([id, value]) => { + if (!/^[a-z][a-z0-9-]*$/u.test(id)) throw new Error(`${path}.conditionalEventTypePairs.${id} must use a stable lowercase id`); + const pairPath = `${path}.conditionalEventTypePairs.${id}`; + const pair = asRecord(value, pairPath); + const unknownFields = Object.keys(pair).filter((field) => field !== "agentrun" && field !== "hwlab"); + if (unknownFields.length > 0) throw new Error(`${pairPath}.${unknownFields[0]} is not allowed`); + const agentrun = stringField(pair, "agentrun", pairPath); + const hwlab = stringField(pair, "hwlab", pairPath); + if (agentrunEventTypes.includes(agentrun)) throw new Error(`${pairPath}.agentrun duplicates a required event type`); + if (hwlabEventTypes.includes(hwlab)) throw new Error(`${pairPath}.hwlab duplicates a required event type`); + return [id, { agentrun, hwlab }]; + })); + const conditionalAgentrunTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.agentrun); + const conditionalHwlabTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.hwlab); + if (new Set(conditionalAgentrunTypes).size !== conditionalAgentrunTypes.length) throw new Error(`${path}.conditionalEventTypePairs agentrun types must not contain duplicates`); + if (new Set(conditionalHwlabTypes).size !== conditionalHwlabTypes.length) throw new Error(`${path}.conditionalEventTypePairs hwlab types must not contain duplicates`); return { subscriberCount: 2, debugStreams: debugStreams as ("stdio" | "agentrun" | "hwlab")[], @@ -1535,6 +1557,7 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla agentrun: agentrunEventTypes, hwlab: hwlabEventTypes, }, + conditionalEventTypePairs, }; } diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts index f1082eff..cd22896b 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts @@ -36,6 +36,7 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent" }, forbiddenRequestPaths: [], requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] }, + conditionalEventTypePairs: { "tool-call": { agentrun: "tool_call", hwlab: "tool" } }, expectedKafka: { topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" }, groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" }, @@ -46,6 +47,91 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent" assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000); assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000); assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000); + assert.deepEqual(effectiveProfile(profile, null).conditionalEventTypePairs, profile.conditionalEventTypePairs); +}); + +test("realtime fanout terminal completion does not require absent optional tool families", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const build = new Function(`${source}\nreturn realtimeDebugTurnCompletion;`) as () => ( + snapshot: Record, traceId: string, profile: Record, + ) => Record; + const completion = build(); + const traceId = "trc_direct_answer"; + const profile = { + debugStreams: ["stdio", "agentrun", "hwlab"], + requiredEventTypes: { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }, + conditionalEventTypePairs: { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }, + }; + const snapshot = { + records: { + stdio: [{ traceId, eventType: "codex.stdio.stdout" }], + agentrun: ["user_message", "backend_status", "assistant_message", "terminal_status"].map((eventType) => ({ traceId, eventType })), + hwlab: ["user", "backend", "assistant", "terminal"].map((eventType) => ({ traceId, eventType })), + }, + }; + + const directAnswer = completion(snapshot, traceId, profile); + assert.equal(directAnswer.complete, true); + assert.deepEqual(directAnswer.missingEventTypes, { agentrun: [], hwlab: [] }); + + const missingTerminal = completion({ + records: { + ...snapshot.records, + agentrun: snapshot.records.agentrun.filter((row: Record) => row.eventType !== "terminal_status"), + hwlab: snapshot.records.hwlab.filter((row: Record) => row.eventType !== "terminal"), + }, + }, traceId, profile); + assert.equal(missingTerminal.complete, false); + assert.deepEqual(missingTerminal.missingEventTypes, { agentrun: ["terminal_status"], hwlab: ["terminal"] }); + assert.ok(missingTerminal.observedEventTypes.agentrun.includes("assistant_message")); +}); + +test("realtime fanout validates optional tool events only when they occur", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const build = new Function(`${source}\nreturn realtimeValidateConditionalEventTypePairs;`) as () => ( + input: Record, + ) => void; + const validate = build(); + const traceId = "trc_optional_tool"; + const base = { + turn: 1, + traceId, + profile: { + conditionalEventTypePairs: { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }, + }, + }; + + assert.doesNotThrow(() => validate({ ...base, debug: { records: { agentrun: [], hwlab: [] } } })); + assert.doesNotThrow(() => validate({ + ...base, + debug: { + records: { + agentrun: [{ traceId, eventType: "tool_call", eventId: "evt_tool" }], + hwlab: [{ traceId, eventType: "tool", sourceEventId: "evt_tool" }], + }, + }, + })); + assert.throws( + () => validate({ + ...base, + debug: { + records: { + agentrun: [{ traceId, eventType: "command_output", eventId: "evt_output" }], + hwlab: [], + }, + }, + }), + /conditional event pair command-output count differs/u, + ); }); test("realtime fanout retries a YAML-bounded transient health navigation", async () => { diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts index ffd82a98..cfd7b8c0 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts @@ -747,6 +747,7 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) { reconnectQuietMs: Number(profile.reconnectQuietMs), forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [], requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {}, + conditionalEventTypePairs: profile.conditionalEventTypePairs && typeof profile.conditionalEventTypePairs === "object" ? sanitize(profile.conditionalEventTypePairs) : {}, expectedProductSse: sanitize(expectedProductSse), expectedKafka: sanitize(expectedKafka), }; @@ -1043,15 +1044,44 @@ async function realtimeWaitTurnIds(debug, key, traceId, timeoutMs) { } async function realtimeWaitDebugTurnComplete(debug, key, traceId, profile) { - return realtimePoll(async () => { - const snapshot = await realtimeDebugSnapshot(debug, key); - if (!profile.debugStreams.every((stream) => (snapshot.records[stream] || []).some((item) => item.traceId === traceId))) return null; - const agentrunTypes = new Set((snapshot.records.agentrun || []).filter((item) => item.traceId === traceId).map((item) => item.eventType)); - const hwlabTypes = new Set((snapshot.records.hwlab || []).filter((item) => item.traceId === traceId).map((item) => item.eventType)); - if (!(profile.requiredEventTypes.agentrun || []).every((type) => agentrunTypes.has(type))) return null; - if (!(profile.requiredEventTypes.hwlab || []).every((type) => hwlabTypes.has(type))) return null; - return snapshot; - }, profile.terminalTimeoutMs, "realtime debug terminal event families"); + let latest = null; + try { + return await realtimePoll(async () => { + const snapshot = await realtimeDebugSnapshot(debug, key); + latest = realtimeDebugTurnCompletion(snapshot, traceId, profile); + return latest.complete ? snapshot : null; + }, profile.terminalTimeoutMs, "realtime debug terminal event families"); + } catch (error) { + const wrapped = error instanceof Error ? error : new Error(String(error)); + const missing = latest?.missingEventTypes || { agentrun: [], hwlab: [] }; + wrapped.message = "realtime debug terminal event families missing agentrun=[" + missing.agentrun.join(",") + "] hwlab=[" + missing.hwlab.join(",") + "] after " + profile.terminalTimeoutMs + "ms"; + wrapped.details = { ...(wrapped.details || {}), traceId, completion: latest, valuesRedacted: true }; + throw wrapped; + } +} + +function realtimeDebugTurnCompletion(snapshot, traceId, profile) { + const observedEventTypes = {}; + const recordCounts = {}; + for (const stream of profile.debugStreams || []) { + const rows = (snapshot.records?.[stream] || []).filter((item) => item.traceId === traceId); + observedEventTypes[stream] = [...new Set(rows.map((item) => item.eventType).filter(Boolean))]; + recordCounts[stream] = rows.length; + } + const missingEventTypes = { + agentrun: (profile.requiredEventTypes?.agentrun || []).filter((type) => !observedEventTypes.agentrun?.includes(type)), + hwlab: (profile.requiredEventTypes?.hwlab || []).filter((type) => !observedEventTypes.hwlab?.includes(type)), + }; + const missingStreams = (profile.debugStreams || []).filter((stream) => !recordCounts[stream]); + return { + complete: missingStreams.length === 0 && missingEventTypes.agentrun.length === 0 && missingEventTypes.hwlab.length === 0, + traceId, + observedEventTypes, + missingEventTypes, + missingStreams, + recordCounts, + valuesRedacted: true, + }; } async function realtimeWaitTurnStable(debug, key, traceId, subscribers, profile) { @@ -1241,6 +1271,7 @@ function realtimeValidateTurnEvidence(input) { const hwlabTypes = new Set(input.debug.records.hwlab.filter((item) => item.traceId === input.traceId).map((item) => item.eventType)); for (const type of profile.requiredEventTypes.agentrun || []) if (!agentrunTypes.has(type)) throw new Error("turn " + input.turn + " AgentRun lacks " + type); for (const type of profile.requiredEventTypes.hwlab || []) if (!hwlabTypes.has(type)) throw new Error("turn " + input.turn + " HWLAB lacks " + type); + realtimeValidateConditionalEventTypePairs(input); const identityRows = Object.values(input.debug.records).flat().filter((item) => item.traceId === input.traceId); if (!input.runId || new Set(identityRows.map((item) => item.runId)).size !== 1) throw new Error("turn " + input.turn + " runId missing or mismatched across streams"); const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean)); @@ -1267,6 +1298,30 @@ function realtimeValidateTurnEvidence(input) { } } +function realtimeValidateConditionalEventTypePairs(input) { + const pairs = input.profile.conditionalEventTypePairs || {}; + const agentrunRows = (input.debug.records.agentrun || []).filter((item) => item.traceId === input.traceId); + const hwlabRows = (input.debug.records.hwlab || []).filter((item) => item.traceId === input.traceId); + for (const [id, pair] of Object.entries(pairs)) { + const sourceRows = agentrunRows.filter((item) => item.eventType === pair.agentrun); + const projectedRows = hwlabRows.filter((item) => item.eventType === pair.hwlab); + if (sourceRows.length === 0 && projectedRows.length === 0) continue; + if (sourceRows.length !== projectedRows.length) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " count differs AgentRun=" + sourceRows.length + " HWLAB=" + projectedRows.length); + } + for (const source of sourceRows) { + if (!source.eventId || !projectedRows.some((projected) => projected.sourceEventId === source.eventId)) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " lacks HWLAB sourceEventId lineage"); + } + } + for (const projected of projectedRows) { + if (!projected.sourceEventId || !sourceRows.some((source) => source.eventId === projected.sourceEventId)) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " has an orphan HWLAB event"); + } + } + } +} + function realtimeExpectedStreamSessionId(stream, hwlabSessionId) { if (stream !== "stdio") return hwlabSessionId; const base = String(hwlabSessionId || "") diff --git a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts index 466b9cf1..149facfc 100644 --- a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts @@ -28,6 +28,14 @@ test("realtime fanout runner contract derives topics, groups, and independent ca projectionOutboxRelay: false, projectionRealtime: false, }); + assert.deepEqual(profiles["pure-kafka-live"]?.requiredEventTypes, { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }); + assert.deepEqual(profiles["pure-kafka-live"]?.conditionalEventTypePairs, { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }); }); test("Workbench Kafka debug replay runner contract is derived from owning YAML", () => { diff --git a/scripts/src/hwlab-node/web-probe-observe-options.test.ts b/scripts/src/hwlab-node/web-probe-observe-options.test.ts index 18d9f737..c7d11dcb 100644 --- a/scripts/src/hwlab-node/web-probe-observe-options.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-options.test.ts @@ -41,8 +41,14 @@ test("validateRealtimeFanout parses the YAML profile and two independent prompts refreshHandoff: true, replayPriorTraceOnReconnect: true, }); - assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.agentrun.includes("user_message")); - assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.hwlab.includes("user")); + assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes, { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }); + assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.conditionalEventTypePairs, { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }); }); test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => { From 28e2cff32c1cbf36dbb251e99ce623e19d8103e9 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 20:59:37 +0200 Subject: [PATCH 4/7] docs(webdev): clarify conditional realtime events --- .agents/skills/unidesk-webdev/SKILL.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.agents/skills/unidesk-webdev/SKILL.md b/.agents/skills/unidesk-webdev/SKILL.md index 98e38116..856872fc 100644 --- a/.agents/skills/unidesk-webdev/SKILL.md +++ b/.agents/skills/unidesk-webdev/SKILL.md @@ -42,6 +42,9 @@ description: UniDesk Web 开发与浏览器验证技能。用户处理 UniDesk/H - Workbench Kafka 实时 fanout 的重复验收： - 使用 `observe command --type validateRealtimeFanout --profile --provider --text --second-text `； - connected、refresh/reconnect 和 capability 期望来自 owning YAML，可独立组合，禁止在命令中硬编码 live-only 或 refresh mode； + - 每轮只把 `user`、`assistant` 和 `terminal` 对应的 AgentRun/HWLAB 事件族作为必需核心事件； + - `tool_call`/`command_output` 与 HWLAB `tool`/`status` 是条件事件对，仅在实际出现时校验数量和 lineage； + - 直接答复不得为了通过探针而合成工具事件，超时证据必须披露已经观测和仍然缺失的事件族； - refresh 启用时必须证明 retention 到 live handoff、正式 `user` event、`hwlab.event.v1` 同 envelope，以及主页面单一 EventSource 和同一 UI reducer 终态； - observer 后台维持两个独立 product subscriber 和三条 debug SSE； - 命令提交后按 `status --command-id ` 轮询，不得回退为长连接 `web-probe script`。 From ba0a5a07794f7a582e5653f7065aa0ce96925aba Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 21:51:00 +0200 Subject: [PATCH 5/7] =?UTF-8?q?fix:=20=E6=A0=A1=E6=AD=A3=20WebProbe=20?= =?UTF-8?q?=E9=87=8D=E8=BF=9E=E5=9B=9E=E6=94=BE=E6=BA=90=E9=9B=86=E5=90=88?= =?UTF-8?q?=E6=97=B6=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...web-observe-runner-realtime-source.test.ts | 171 ++++++++++++++ ...node-web-observe-runner-realtime-source.ts | 211 +++++++++++++++--- 2 files changed, 357 insertions(+), 25 deletions(-) diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts index cd22896b..11dc173c 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts @@ -407,3 +407,174 @@ test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka /product subscriber lacks pre-terminal or terminal event|product SSE lacks the formal user event/u, ); }); + +function replayEnvelope(eventId: string, sourceSeq: number, terminal = false) { + return { + envelopeFingerprint: `fingerprint:${eventId}`, + eventId, + sourceEventId: `source:${eventId}`, + sourceSeq, + eventType: terminal ? "terminal" : "assistant", + traceId: "trc_joint_quiet", + hwlabSessionId: "ses_joint_quiet", + runId: "run_joint_quiet", + commandId: "cmd_joint_quiet", + userMessageId: null, + terminal, + }; +} + +test("realtime replay joint quiet waits for a late formal Kafka envelope before accepting 32/32", async () => { + const source = nodeWebObserveRunnerRealtimeSource(); + let now = 0; + let sleepCount = 0; + const first = replayEnvelope("evt_first", 1); + const late = replayEnvelope("evt_late", 2, true); + let kafkaRows = [first]; + const productRows = [first, late]; + const build = new Function( + "sleep", + "Date", + `${source}\nreturn realtimeWaitReplayedTraceJointQuiet;`, + ) as ( + sleep: () => Promise, + date: { now: () => number }, + ) => ( + debug: Record, + key: string, + subscriber: Record, + traceId: string, + label: string, + profile: Record, + ) => Promise>; + const waitForJointQuiet = build( + async () => { + sleepCount += 1; + now += 250; + if (sleepCount === 1) kafkaRows = [first, late]; + }, + { now: () => now }, + ); + const debug = { + page: { + async evaluate() { + return { + connected: {}, connectedEventCount: { hwlab: 1 }, openCount: { hwlab: 1 }, errors: { hwlab: [] }, + records: { stdio: [], agentrun: [], hwlab: structuredClone(kafkaRows) }, + }; + }, + }, + }; + const subscriber = { + page: { + async evaluate() { + return { connected: {}, connectedEventCount: 1, openCount: 1, errors: [], records: structuredClone(productRows) }; + }, + }, + }; + + const result = await waitForJointQuiet( + debug, + "turn-1-trc_joint_quiet", + subscriber, + "trc_joint_quiet", + "reconnected", + { barrierTimeoutMs: 2_000, terminalQuietMs: 500 }, + ); + + assert.equal(result.comparison.equal, true); + assert.equal(result.comparison.sourceCount, 2); + assert.equal(result.comparison.productCount, 2); + assert.equal(result.quietForMs, 500); + assert.ok(sleepCount >= 3); +}); + +test("realtime replay joint quiet rejects a permanent 31/32 split with bounded identity diagnostics", async () => { + const source = nodeWebObserveRunnerRealtimeSource(); + let now = 0; + const first = replayEnvelope("evt_first", 1); + const late = replayEnvelope("evt_late", 2, true); + const build = new Function( + "sleep", + "Date", + `${source}\nreturn realtimeWaitReplayedTraceJointQuiet;`, + ) as ( + sleep: () => Promise, + date: { now: () => number }, + ) => ( + debug: Record, + key: string, + subscriber: Record, + traceId: string, + label: string, + profile: Record, + ) => Promise>; + const waitForJointQuiet = build(async () => { now += 250; }, { now: () => now }); + const debug = { + page: { + async evaluate() { + return { connected: {}, connectedEventCount: { hwlab: 1 }, openCount: { hwlab: 1 }, errors: { hwlab: [] }, records: { stdio: [], agentrun: [], hwlab: [first] } }; + }, + }, + }; + const subscriber = { + page: { + async evaluate() { + return { connected: {}, connectedEventCount: 1, openCount: 1, errors: [], records: [first, late] }; + }, + }, + }; + + await assert.rejects( + () => waitForJointQuiet( + debug, + "turn-1-trc_joint_quiet", + subscriber, + "trc_joint_quiet", + "reconnected", + { barrierTimeoutMs: 750, terminalQuietMs: 250 }, + ), + (error: Error & { details?: Record }) => { + assert.match(error.message, /did not converge before the joint quiet timeout/u); + assert.equal(error.details?.sourceCount, 1); + assert.equal(error.details?.productCount, 2); + assert.equal(error.details?.productOnly?.[0]?.identity?.eventId, "evt_late"); + assert.equal(error.details?.productOnly?.[0]?.count, 1); + return true; + }, + ); +}); + +test("realtime envelope equality is an exact multiset even when counts match", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const build = new Function(`${source}\nreturn { compare: realtimeEnvelopeMultisetComparison, assertEqual: realtimeAssertEnvelopePassthrough };`) as () => { + compare: (kafkaRows: Record[], productRows: Record[]) => Record; + assertEqual: (kafkaRows: Record[], productRows: Record[], requireAll: boolean, label: string) => void; + }; + const helpers = build(); + const first = replayEnvelope("evt_first", 1); + const sourceTerminal = replayEnvelope("evt_terminal", 2, true); + const changedTerminal = { ...sourceTerminal, envelopeFingerprint: "fingerprint:changed", sourceEventId: "source:changed" }; + const comparison = helpers.compare([first, sourceTerminal], [first, changedTerminal]); + + assert.equal(comparison.equal, false); + assert.equal(comparison.sourceCount, 2); + assert.equal(comparison.productCount, 2); + assert.equal(comparison.sourceOnly[0].identity.eventId, "evt_terminal"); + assert.equal(comparison.productOnly[0].identity.sourceEventId, "source:changed"); + assert.throws( + () => helpers.assertEqual([first, sourceTerminal], [first, changedTerminal], true, "reconnected"), + /product SSE envelope absent from hwlab\.event\.v1/u, + ); +}); + +test("realtime fanout closes the first debug stream only after the final joint quiet gate", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const afterSecond = source.indexOf("const firstReplayAfterSecond = await realtimeWaitReplayedTraceJointQuiet"); + const finalGate = source.indexOf("const firstReplayFinal = await realtimeWaitReplayedTraceJointQuiet"); + const close = source.indexOf("await realtimeCloseDebugStreams(debug, first.debugKey);"); + + assert.ok(afterSecond > 0); + assert.ok(finalGate > afterSecond); + assert.ok(close > finalGate); +}); diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts index cfd7b8c0..fe13655f 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts @@ -112,7 +112,7 @@ async function validateRealtimeFanout(command) { const firstTerminal = await realtimeWaitTraceEvent(subscriberA, first.traceId, true, effective.terminalTimeoutMs); await realtimeWaitUiTerminal(first.traceId, effective.terminalTimeoutMs); const firstStable = await realtimeWaitTurnStable(debug, first.debugKey, first.traceId, [subscriberA], effective); - const firstDebug = firstStable.debug; + let firstDebug = firstStable.debug; realtimeEnrichTurnExecutionIds(first, firstDebug); const firstProductA = firstStable.products[0]; realtimeAssertStableProductTransport(firstProductA, "subscriber-a-first-terminal"); @@ -127,13 +127,25 @@ async function validateRealtimeFanout(command) { preTerminalSnapshots: [bBeforeDisconnect], profile: effective, }); - await realtimeCloseDebugStreams(debug, first.debugKey); - subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective); const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs); realtimeAssertConnectedContract(bReconnected, sessionId, "subscriber-b-reconnected", effective, { requireRetainedEvents: effective.expectedProductSse.replayPriorTraceOnReconnect }); await sleep(effective.reconnectQuietMs); - const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2); + let reconnectBeforeSecond; + if (effective.expectedProductSse.replayPriorTraceOnReconnect) { + const firstReplayBeforeSecond = await realtimeWaitReplayedTraceJointQuiet( + debug, + first.debugKey, + subscriberB2, + first.traceId, + "subscriber-b-reconnected-before-second-turn", + effective, + ); + firstDebug = firstReplayBeforeSecond.debug; + reconnectBeforeSecond = firstReplayBeforeSecond.product; + } else { + reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2); + } realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn"); if (effective.expectedProductSse.replayPriorTraceOnReconnect) { realtimeAssertReplayedTrace(firstDebug, reconnectBeforeSecond, first.traceId, "subscriber-b-reconnected-before-second-turn"); @@ -161,7 +173,17 @@ async function validateRealtimeFanout(command) { realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal"); realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal"); if (effective.expectedProductSse.replayPriorTraceOnReconnect) { - realtimeAssertReplayedTrace(firstDebug, secondProductB, first.traceId, "subscriber-b-reconnected-after-second-turn"); + const firstReplayAfterSecond = await realtimeWaitReplayedTraceJointQuiet( + debug, + first.debugKey, + subscriberB2, + first.traceId, + "subscriber-b-reconnected-after-second-turn", + effective, + ); + firstDebug = firstReplayAfterSecond.debug; + realtimeAssertStableProductTransport(firstReplayAfterSecond.product, "subscriber-b-reconnected-after-second-turn"); + realtimeAssertReplayedTrace(firstDebug, firstReplayAfterSecond.product, first.traceId, "subscriber-b-reconnected-after-second-turn"); } else if (secondProductB.records.some((item) => item.traceId === first.traceId)) { throw new Error("reconnected subscriber received first-turn events contrary to the owning YAML"); } @@ -179,12 +201,25 @@ async function validateRealtimeFanout(command) { await realtimeCloseDebugStreams(debug, second.debugKey); const productA = await realtimeProductSnapshot(subscriberA); - const productB2 = await realtimeProductSnapshot(subscriberB2); + let productB2 = await realtimeProductSnapshot(subscriberB2); + if (effective.expectedProductSse.replayPriorTraceOnReconnect) { + const firstReplayFinal = await realtimeWaitReplayedTraceJointQuiet( + debug, + first.debugKey, + subscriberB2, + first.traceId, + "subscriber-b-reconnected-final", + effective, + ); + firstDebug = firstReplayFinal.debug; + productB2 = firstReplayFinal.product; + } realtimeAssertStableProductTransport(productA, "subscriber-a-final"); realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final"); if (effective.expectedProductSse.replayPriorTraceOnReconnect) { realtimeAssertReplayedTrace(firstDebug, productB2, first.traceId, "subscriber-b-reconnected-final"); } + await realtimeCloseDebugStreams(debug, first.debugKey); const mainStreams = network.filter((item) => item.path === effective.productEventsPath && item.sessionId === sessionId); const forbidden = network.filter((item) => realtimeForbiddenRequest(item, effective.forbiddenRequestPaths)); if (mainStreams.length !== 1) throw new Error("main Workbench session SSE connection count is " + mainStreams.length + ", expected 1"); @@ -1214,6 +1249,61 @@ function realtimeAssertReplayedTrace(debug, product, traceId, label) { realtimeAssertEnvelopePassthrough(kafkaRows, productRows, true, label + " retained trace"); } +async function realtimeWaitReplayedTraceJointQuiet(debug, key, subscriber, traceId, label, profile) { + const timeoutMs = profile.barrierTimeoutMs; + const deadline = Date.now() + timeoutMs; + let previousSignature = null; + let quietSince = null; + let latest = null; + while (Date.now() < deadline) { + const debugSnapshot = await realtimeDebugSnapshot(debug, key); + const productSnapshot = await realtimeProductSnapshot(subscriber); + const kafkaRows = (debugSnapshot.records?.hwlab || []).filter((item) => item.traceId === traceId); + const productRows = (productSnapshot.records || []).filter((item) => item.traceId === traceId); + const comparison = realtimeEnvelopeMultisetComparison(kafkaRows, productRows); + const signature = JSON.stringify({ + kafka: realtimeEnvelopeMultisetSignature(kafkaRows), + product: realtimeEnvelopeMultisetSignature(productRows), + }); + const now = Date.now(); + if (signature !== previousSignature) { + previousSignature = signature; + quietSince = now; + } + const quietForMs = quietSince === null ? 0 : now - quietSince; + const terminalSeen = productRows.some((item) => item.terminal === true); + const sourceTransport = { + openCount: debugSnapshot.openCount?.hwlab || 0, + connectedEventCount: debugSnapshot.connectedEventCount?.hwlab || 0, + errorCount: (debugSnapshot.errors?.hwlab || []).length, + }; + const sourceTransportStable = sourceTransport.openCount === 1 && sourceTransport.connectedEventCount === 1 && sourceTransport.errorCount === 0; + latest = { comparison, terminalSeen, quietForMs, sourceTransport, sourceTransportStable }; + if (comparison.equal && terminalSeen && sourceTransportStable && quietForMs >= profile.terminalQuietMs) { + return { debug: debugSnapshot, product: productSnapshot, comparison, quietForMs, valuesRedacted: true }; + } + await sleep(250); + } + const error = new Error(label + " Kafka source and product replay envelope multisets did not converge before the joint quiet timeout"); + error.details = { + traceId, + timeoutMs, + terminalQuietMs: profile.terminalQuietMs, + terminalSeen: latest?.terminalSeen === true, + sourceTransport: latest?.sourceTransport || null, + sourceTransportStable: latest?.sourceTransportStable === true, + quietForMs: latest?.quietForMs || 0, + sourceCount: latest?.comparison?.sourceCount || 0, + productCount: latest?.comparison?.productCount || 0, + sourceOnly: latest?.comparison?.sourceOnly || [], + productOnly: latest?.comparison?.productOnly || [], + invalidSource: latest?.comparison?.invalidSource || [], + invalidProduct: latest?.comparison?.invalidProduct || [], + valuesRedacted: true, + }; + throw error; +} + function realtimeConnectedSummary(connected) { return { realtimeSource: connected?.realtimeSource || null, @@ -1331,26 +1421,97 @@ function realtimeExpectedStreamSessionId(stream, hwlabSessionId) { return base ? "ses_agentrun_" + base : null; } -function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) { - const sameEnvelope = (left, right) => left.envelopeFingerprint && left.envelopeFingerprint === right.envelopeFingerprint - && left.eventId === right.eventId - && left.sourceEventId === right.sourceEventId - && left.sourceSeq === right.sourceSeq - && left.eventType === right.eventType - && left.traceId === right.traceId - && left.hwlabSessionId === right.hwlabSessionId - && left.runId === right.runId - && left.commandId === right.commandId - && left.userMessageId === right.userMessageId - && left.terminal === right.terminal; - for (const productRow of productRows) { - if (!kafkaRows.some((kafkaRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " received a product SSE envelope absent from hwlab.event.v1"); - } - if (requireAllKafkaRows) { - if (productRows.length !== kafkaRows.length) throw new Error(label + " envelope count differs from hwlab.event.v1"); - for (const kafkaRow of kafkaRows) { - if (!productRows.some((productRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " did not receive an unchanged hwlab.event.v1 envelope"); +function realtimeEnvelopeIdentity(row) { + return { + envelopeFingerprint: row?.envelopeFingerprint || null, + eventId: row?.eventId || null, + sourceEventId: row?.sourceEventId || null, + sourceSeq: Number.isFinite(row?.sourceSeq) ? row.sourceSeq : null, + eventType: row?.eventType || null, + traceId: row?.traceId || null, + hwlabSessionId: row?.hwlabSessionId || null, + runId: row?.runId || null, + commandId: row?.commandId || null, + userMessageId: row?.userMessageId || null, + terminal: row?.terminal === true, + }; +} + +function realtimeEnvelopeIdentityKey(row) { + const identity = realtimeEnvelopeIdentity(row); + return identity.envelopeFingerprint ? JSON.stringify(identity) : null; +} + +function realtimeEnvelopeMultiset(rows) { + const counts = new Map(); + const invalid = []; + for (const row of rows) { + const identity = realtimeEnvelopeIdentity(row); + const key = realtimeEnvelopeIdentityKey(row); + if (!key) { + if (invalid.length < 8) invalid.push(identity); + continue; } + const current = counts.get(key); + if (current) current.count += 1; + else counts.set(key, { count: 1, identity }); + } + return { counts, invalid }; +} + +function realtimeEnvelopeMultisetSignature(rows) { + const multiset = realtimeEnvelopeMultiset(rows); + const counts = [...multiset.counts.entries()].map(([key, entry]) => [key, entry.count]).sort((left, right) => left[0].localeCompare(right[0])); + return JSON.stringify({ counts, invalid: multiset.invalid }); +} + +function realtimeEnvelopeMultisetComparison(kafkaRows, productRows) { + const source = realtimeEnvelopeMultiset(kafkaRows); + const product = realtimeEnvelopeMultiset(productRows); + const sourceOnly = []; + const productOnly = []; + let differs = source.invalid.length > 0 || product.invalid.length > 0; + const keys = new Set([...source.counts.keys(), ...product.counts.keys()]); + for (const key of keys) { + const sourceEntry = source.counts.get(key); + const productEntry = product.counts.get(key); + const sourceCount = sourceEntry?.count || 0; + const productCount = productEntry?.count || 0; + if (sourceCount > productCount) { + differs = true; + if (sourceOnly.length < 8) sourceOnly.push({ identity: sourceEntry.identity, count: sourceCount - productCount }); + } + if (productCount > sourceCount) { + differs = true; + if (productOnly.length < 8) productOnly.push({ identity: productEntry.identity, count: productCount - sourceCount }); + } + } + return { + equal: !differs && kafkaRows.length === productRows.length, + sourceCount: kafkaRows.length, + productCount: productRows.length, + sourceOnly, + productOnly, + invalidSource: source.invalid, + invalidProduct: product.invalid, + valuesRedacted: true, + }; +} + +function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) { + const comparison = realtimeEnvelopeMultisetComparison(kafkaRows, productRows); + if (comparison.productOnly.length > 0 || comparison.invalidProduct.length > 0) { + const error = new Error(label + " received a product SSE envelope absent from hwlab.event.v1"); + error.details = comparison; + throw error; + } + if (requireAllKafkaRows && !comparison.equal) { + const message = comparison.sourceCount !== comparison.productCount + ? label + " envelope count differs from hwlab.event.v1" + : label + " did not receive an unchanged hwlab.event.v1 envelope"; + const error = new Error(message); + error.details = comparison; + throw error; } } From 8fcad60cc93f3cbdb4a3abc70fb28069b8bfed99 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 21:55:32 +0200 Subject: [PATCH 6/7] =?UTF-8?q?fix:=20=E6=81=A2=E5=A4=8D=20Decision=20CLI?= =?UTF-8?q?=20=E6=88=AA=E6=96=AD=20JSON?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/src/decision-center.ts | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/scripts/src/decision-center.ts b/scripts/src/decision-center.ts index 5054fde8..d5766550 100644 --- a/scripts/src/decision-center.ts +++ b/scripts/src/decision-center.ts @@ -3,6 +3,7 @@ import { createHash } from "node:crypto"; import { resolve } from "node:path"; import { type UniDeskConfig, repoRoot } from "./config"; import { runCommand } from "./command"; +import { resolveCliChildJsonCommandResult } from "./cli-child-json-recovery"; import { coreInternalFetch } from "./microservices"; import { capture, compactCapture, parseJsonOutput, shQuote } from "./platform-infra-ops-library"; @@ -659,11 +660,24 @@ console.log(output); stderrTail: result.stderr.slice(-1000), }; } - const jsonLine = result.stdout.trim().split("\n").reverse().find((line) => line.trim().startsWith("{")); - if (jsonLine === undefined) { - return { ok: false, status: 502, error: "NC01 decision-center k8s proxy returned no JSON", stdoutTail: result.stdout.slice(-1000), stderrTail: result.stderr.slice(-1000) }; - } - return JSON.parse(jsonLine) as unknown; + const resolved = resolveCliChildJsonCommandResult({ + result, + requestedStdoutType: "NC01 decision-center k8s proxy JSON", + acceptParsed: (value) => typeof value.ok === "boolean" && typeof value.status === "number", + }); + if (resolved.parsed !== null) return resolved.parsed; + return { + ok: false, + status: 502, + error: "NC01 decision-center k8s proxy returned unparseable JSON", + transport: { + exitCode: result.exitCode, + timedOut: result.timedOut, + stdoutBytes: Buffer.byteLength(result.stdout), + stderrBytes: Buffer.byteLength(result.stderr), + }, + diagnostics: resolved.diagnostics, + }; } async function decisionProxyAsync( From 6a119a5364d5ad4b8d985dfdf19d0c26ac38cb7d Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 22:11:52 +0200 Subject: [PATCH 7/7] =?UTF-8?q?docs:=20=E5=9B=BA=E5=8C=96=20HWLAB=20Kafka?= =?UTF-8?q?=20=E5=88=B7=E6=96=B0=E5=AE=9E=E6=97=B6=E4=BA=A4=E6=8E=A5?= =?UTF-8?q?=E5=90=88=E5=90=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/reference/hwlab.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/docs/reference/hwlab.md b/docs/reference/hwlab.md index 4ac45189..2656db39 100644 --- a/docs/reference/hwlab.md +++ b/docs/reference/hwlab.md @@ -226,6 +226,30 @@ HWLAB 与 AgentRun 协同修复必须按 `docs/reference/agentrun.md` 的职责 HWLAB v0.3 订阅 AgentRun Kafka event 的权威输入是 `agentrun.event.v1`，内存 direct mapper 的权威输出是 `hwlab.event.v1`；三 topic 的 ownership 统一见 `docs/reference/platform-infra.md#kafka-event-bus-boundary`。`directPublish` 使用固定独立 group 消费 AgentRun event、调用生产 mapper 并直接发布完整 HWLAB envelope；`liveKafkaSse` 使用另一个固定 group 的进程级 shared consumer，再向当前 SSE subscriber fanout，同一 Kafka record 不为每个浏览器重复消费。两者不读取数据库、projection inbox/checkpoint/outbox 或 snapshot，不得回写 `agentrun.event.v1`，也不得把 `codex-stdio.raw.v1` 混入 HWLAB product schema。transactional projector、projection outbox relay 和 projection realtime 是三个可独立组合的 capability，关闭时必须保持不启动、不就绪门禁、不参与产品链路。 +### 纯 Kafka 页面刷新与实时交接合同 + +- Workbench 的产品权威链固定为： + - `codex-stdio.raw.v1 -> agentrun.event.v1 -> HWLAB direct mapper -> hwlab.event.v1 retention -> bounded refresh bootstrap -> shared live fanout -> SSE -> Web ingress/reducer`； + - 用户输入必须先成为 `agentrun.event.v1` 的正式 `user_message`，再一对一映射为 `hwlab.event.v1` 的正式 user event； + - 用户消息、assistant、tool、terminal 和 final 不得另建数据库、HTTP history 或浏览器本地事实源。 +- 页面刷新只允许从 retained `hwlab.event.v1` 恢复已授权的 session/trace： + - 服务端先建立 shared live handoff 保护，再读取目标分区 end offset 作为 barrier； + - barrier 内由有界 retention bootstrap 输出，barrier 后由 shared live fanout 输出； + - overlap 只按 Kafka topic/partition/offset 与稳定 event identity 收敛； + - messageId 必须由正式 trace/item identity 派生，禁止按正文、时间戳、数组位置或 DOM 内容去重与重排。 +- refresh bootstrap 是短生命周期读取器： + - 到达 barrier 后必须退出，不得把每个浏览器连接变成新的长期 Kafka consumer； + - 实时阶段继续复用进程级 shared consumer/fanout； + - 同一 Workbench session 只建立一个产品 EventSource，刷新与重连不得复制业务 subscriber。 +- `kafkaRefreshReplay` 与 `liveKafkaSse` 是可组合、可独立开关的 capability： + - topic、group、扫描预算、等待预算、live buffer 上限和失败策略只从选中 node/lane 的 owning YAML/runtime config 读取； + - replay 与 live frame 必须进入同一 decode、queue、reducer、row model 和 card； + - barrier 不可获得、扫描或 buffer 超限、授权 scope 不完整时必须 typed fail-closed，不得回退 projector/read model/snapshot/sync/gap-fill/finalizer、polling 或 HTTP history。 +- 关闭此类问题时先用 repo-owned CLI 复用生产 bootstrap/codec/reducer，披露 scanned、matched、applied、barrier、buffered、deduplicated、terminal 与 stable identity： + - CLI 通过后，使用 semantic internal WebProbe 在同一已有 session 上执行只刷新、不提交新输入的验收； + - 刷新前后 user/agent identity 和可见顺序必须一致，retention 与产品 SSE envelope 多重集必须精确相等； + - OTel 应只出现 direct publish、shared fanout receive 和 product SSE write，projector/projection write span 必须为 0。 + 纯 Kafka 调试分成三个可复用入口，不以临时脚本作为长期路径。AgentRun 使用 `./scripts/agentrun kafka regenerate agentrun --session-id --trace-id `，从 stdio Kafka/JSONL 调用生产 reducer，只输出 `agentrun.event.debug.v1` 并明示 partial reconstruction；HWLAB 使用 `hwlab-cli kafka regenerate hwlab --from kafka|jsonl --session-id --trace-id ` 调用生产 mapper，只输出 `hwlab.event.debug.v1`。原 Workbench 页面的管理员 YAML 调试开关只在显式点击后按当前 traceId 建立唯一 debug group，清空隔离面板并复用生产 reducer/card；正式浏览器验收使用 `web-probe observe command --type validateWorkbenchKafkaDebugReplay`。typed command 继承 observer 的 YAML `internal|public` origin，不接受 URL/IP 来选择内外网。 产品实时链路的最小关闭证据必须来自一个真实 Workbench turn，并同时证明：AgentRun→HWLAB direct publish 数与 HWLAB shared fanout receive 数一致；产品 SSE 收到 terminal/final response；OTel 覆盖 `hwlab-cloud-api`、`agentrun-manager`、`agentrun-runner` 且 error=0；projection write/projector span 为 0。单步 debug 关闭证据另需证明 stdio frame→AgentRun debug→HWLAB debug→Workbench applied 的逐级 count、顺序、lineage 和 terminal，不得把 debug topic/group 冒充产品 topic/group。KafkaJS 2.2.4 `TimeoutNegativeWarning` 已由 `pikasTech/unidesk#1704` 跟踪空 pending queue 的 1ms 自循环；它不是数据丢失证据，也不得通过 suppress warning、全局 timer patch 或退回产品 polling 路径掩盖。