Merge remote-tracking branch 'origin/master'

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(

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<SshHelpScope, SshOperationHelp> = {
+  argv: {
+    group: "process",
+    description: "以原始 argv 边界运行一个远端进程；需要管道、变量或重定向时改用 sh/bash。",
+    usage: ["trans <route> argv <command> [args...]", "trans D601:/workspace argv git status --short --branch"],
+  },
+  exec: {
+    group: "process",
+    description: "在 k3s workload route 中运行进程，并可显式透传 stdin 或指定容器内 cwd。",
+    usage: ["trans <provider>:k3s:<namespace>:<workload>[:<container>] exec [--cwd /path] [--stdin] -- <command> [args...]"],
+  },
+  git: {
+    group: "process",
+    description: "在 host/workspace 或 Windows workspace route 中运行 Git convenience wrapper。",
+    usage: ["trans <provider>:/workspace git status --short --branch", "trans <provider>:win/<drive>/<path> git diff --check"],
+  },
+  node: {
+    group: "process",
+    description: "在目标 route 中直接运行 Node.js argv。",
+    usage: ["trans <route> node -e 'console.log(process.version)'"],
+  },
+  kubectl: {
+    group: "process",
+    description: "在 k3s control-plane route 上执行有界 kubectl 诊断。",
+    usage: ["trans <provider>:k3s kubectl get pods -n <namespace>"],
+  },
+  logs: {
+    group: "process",
+    description: "从 k3s control-plane 或 workload route 读取有界日志。",
+    usage: ["trans <provider>:k3s:<namespace>:<workload> logs --tail 80", "trans <provider>:k3s logs -n <namespace> -l <selector> --tail 120"],
+  },
+  sh: {
+    group: "shell",
+    description: "从 stdin 或单个命令字符串执行远端 POSIX /bin/sh。",
+    usage: ["trans <route> sh <<'SH'", "trans <route> sh -- '<command && command>'"],
+    notes: ["sh -- 只接受一个 shell command string；单进程多 argv 使用 argv。"],
+  },
+  bash: {
+    group: "shell",
+    description: "仅在需要 pipefail、数组、[[ ]] 等 Bash 语法时执行远端 Bash。",
+    usage: ["trans <route> bash <<'BASH'", "trans <route> bash -- '<bash command>'"],
+  },
+  py: {
+    group: "shell",
+    description: "从本地 stdin 执行远端 Python 源码，并保留 script argv。",
+    usage: ["trans <route> py [script-args...] < script.py"],
+  },
+  ps: {
+    group: "shell",
+    description: "在 Windows route 中执行 PowerShell 源码或 argv；route 已包含 win。",
+    usage: ["trans <provider>:win[/<drive>/<path>] ps <<'PS'", "trans <provider>:win ps '<PowerShell source>'"],
+  },
+  cmd: {
+    group: "shell",
+    description: "在 Windows route 中以 UTF-8 环境执行 cmd.exe/batch。",
+    usage: ["trans <provider>:win[/<drive>/<path>] cmd <<'CMD'", "trans <provider>:win cmd ver"],
+  },
+  pwd: { group: "filesystem", description: "读取 Windows workspace 当前目录。", usage: ["trans <provider>:win/<drive>/<path> pwd"] },
+  ls: { group: "filesystem", description: "有界列出 Windows workspace 文件。", usage: ["trans <provider>:win/<drive>/<path> ls --limit 50"] },
+  cat: { group: "filesystem", description: "读取远端文本文件；二进制文件使用 download。", usage: ["trans <route> cat <path>"] },
+  head: { group: "filesystem", description: "按行读取远端文本文件开头。", usage: ["trans <route> head -n 40 <path>"] },
+  tail: { group: "filesystem", description: "按行读取远端文本文件结尾。", usage: ["trans <route> tail -n 40 <path>"] },
+  stat: { group: "filesystem", description: "读取远端文件元数据。", usage: ["trans <route> stat <path>"] },
+  wc: { group: "filesystem", description: "统计远端文本的行、词、字符或字节。", usage: ["trans <route> wc [-l|-w|-m|-c] <path>"] },
+  rg: { group: "filesystem", description: "执行有界远端 ripgrep；Windows route 使用受控 UTF-8 子集。", usage: ["trans <route> rg [options] <pattern> [path...]"] },
+  find: { group: "filesystem", description: "不依赖 shell quoting 的结构化远端 find。", usage: ["trans <route> find <path...> [--contains TEXT] [--limit N]"] },
+  glob: { group: "filesystem", description: "通过远端 helper 执行有界 glob 匹配。", usage: ["trans <route> glob [--root DIR] [--pattern PATTERN] [--limit N]"] },
+  skills: { group: "filesystem", description: "发现 WSL/Linux 及可选 Windows skill 目录。", usage: ["trans <provider> skills [--scope all|wsl|windows] [--limit N]"] },
+  "apply-patch": {
+    group: "patch",
+    description: "默认远端文本修改入口；本地 v2 engine 计算变更，route 仅负责读写。",
+    usage: ["trans <route> 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 <route> apply-patch-v1 [--allow-loose] < patch.diff"],
+  },
+  upload: {
+    group: "transfer",
+    description: "上传必要的二进制文件或生成物并自动核对字节数与 SHA-256。",
+    usage: ["trans <route> upload <local-file> <remote-file>"],
+    notes: ["远端文本读取使用 cat/rg，修改使用 apply-patch。"],
+  },
+  download: {
+    group: "transfer",
+    description: "下载必要的二进制文件或生成物并自动核对字节数与 SHA-256。",
+    usage: ["trans <route> download <remote-file> <local-file>"],
+  },
+};
+
+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<SshOperationHelp["group"], readonly SshHelpScope[]>;
+
+const SSH_HELP_SCOPES = Object.keys(SSH_OPERATION_HELP) as SshHelpScope[];
+const SSH_HELP_SCOPE_SET = new Set<string>(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 <route>",
-      "trans <providerId> argv <command> [args...]",
-      "trans <providerId>:/absolute/workspace apply-patch < patch.diff",
-      "trans <route> upload <local-file> <remote-file>",
-      "trans <route> download <remote-file> <local-file>",
-      "trans <providerId> apply-patch-v1 [--allow-loose] < patch.diff",
-      "trans <providerId> py [script-args...] < script.py",
-      "trans <providerId> sh [arg...] <<'SH'",
-      "trans <providerId> bash [arg...] <<'BASH'",
-      "trans <providerId> skills [--scope all|wsl|windows] [--limit N]",
-      "trans <providerId> find <path...> [--contains TEXT] [--limit N]",
-      "trans <providerId> 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=<run> --tail 120",
-    ],
-    notes: [
-      "trans --help and trans <route> --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 <provider>: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 <provider>:win/<drive>/<path> 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 <route> argv ls -la` is valid, but `trans <route> 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 -- '<command && command>'` for POSIX syntax or `bash -- '<command && command>'` 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 -- <command> ...`; 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/<drive>` 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 `<provider>:/absolute/workspace`; WSL providers can use `<provider>:win ps` for Windows PowerShell and `<provider>:win cmd` for Windows cmd.exe, with `<provider>:win/c/test ...` mapping the Windows cwd to `C:\\test`; native k3s providers such as D601 and G14 use `<provider>:k3s` for the control plane and `<provider>:k3s:<namespace>:<workload>[:<container>]` 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`.",
-      "`<provider>: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} <route> <operation> [operation-args...]`,
+      hostWorkspace: "<provider>:/absolute/workspace",
+      k3sControlPlane: "<provider>:k3s",
+      k3sWorkload: "<provider>:k3s:<namespace>:<workload>[:<container>]",
+      windowsWorkspace: "<provider>:win[/<drive>/<path>]",
+      githubContent: "gh:/<owner>/<repo>[/issue|/pr]/<number>",
+    },
+    operationGroups,
+    scopedHelp: {
+      command: `${entrypoint} --help <operation>`,
+      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<SshHelpScope, "download">): 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} <route> ${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。",

scripts/src/hwlab-node-lanes.ts

@@ -206,6 +206,10 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
     agentrun: readonly string[];
     hwlab: readonly string[];
   }>;
+  readonly conditionalEventTypePairs: Readonly<Record<string, Readonly<{
+    agentrun: string;
+    hwlab: string;
+  }>>>;
 }
 
 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,
   };
 }
 

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<string, any>, traceId: string, profile: Record<string, any>,
+  ) => Record<string, any>;
+  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<string, string>) => row.eventType !== "terminal_status"),
+      hwlab: snapshot.records.hwlab.filter((row: Record<string, string>) => 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<string, any>,
+  ) => 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 () => {
@@ -321,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<void>,
+    date: { now: () => number },
+  ) => (
+    debug: Record<string, any>,
+    key: string,
+    subscriber: Record<string, any>,
+    traceId: string,
+    label: string,
+    profile: Record<string, number>,
+  ) => Promise<Record<string, any>>;
+  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<void>,
+    date: { now: () => number },
+  ) => (
+    debug: Record<string, any>,
+    key: string,
+    subscriber: Record<string, any>,
+    traceId: string,
+    label: string,
+    profile: Record<string, number>,
+  ) => Promise<Record<string, any>>;
+  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<string, any> }) => {
+      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<string, any>[], productRows: Record<string, any>[]) => Record<string, any>;
+    assertEqual: (kafkaRows: Record<string, any>[], productRows: Record<string, any>[], 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);
+});

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");
@@ -747,6 +782,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 +1079,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) {
@@ -1184,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,
@@ -1241,6 +1361,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 +1388,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 || "")
@@ -1276,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;
   }
 }
 

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", () => {

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", () => {

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, string> | string;
+    operationGroups?: Record<string, string>;
+    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<string, string>;
+      operationGroups: Record<string, string>;
+      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("<route> <operation>");
+    expect(top.routeSyntax.k3sWorkload).toContain(":<namespace>:<workload>");
+    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 <operation>");
+    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<string, string>;
       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();
   });
 });