feat: add bounded cicd long polling

This commit is contained in:
Codex
2026-07-15 06:35:25 +02:00
parent b9c86f4d3f
commit 81163ab236
15 changed files with 363 additions and 59 deletions
+8
View File
@@ -33,6 +33,7 @@ bun scripts/cli.ts platform-infra pipelines-as-code status --target JD01 --consu
bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 --limit 10
bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 --id <pipelinerun>
bun scripts/cli.ts platform-infra pipelines-as-code delivery-timing --target NC01 --consumer selfmedia-nc01
bun scripts/cli.ts platform-infra pipelines-as-code closeout --target <NODE> --consumer <consumer> --wait --timeout 120s
bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 --consumer <consumer>
bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 --consumer <consumer> --id <pipelinerun>
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact check --target NC01 --consumer <consumer> --source-worktree <absolute-path>
@@ -46,6 +47,13 @@ bun scripts/cli.ts hwlab nodes control-plane legacy-cicd --help
- 单 consumer 再用 `platform-infra pipelines-as-code status --target <NODE> --consumer <id>` 下钻;
- 状态输出与 registry applicability 的长期判定见 [references/gitea-pac.md](references/gitea-pac.md)。
- CI/CD 只读等待:
- 默认使用 `closeout --wait --timeout 120s` 观察 Tekton、Argo 和 runtime 收敛;已处于成功或失败终态时立即返回。
- 未完成时由目标控制面的 Kubernetes watch 持有单次等待,不得在调用侧每 `1s``5s` 重复执行 `status`
- timeout 必须保留当前状态、`timedOut=true`、耗时、稳定 identity/cursor 与 `next.wait`;零变化也必须有输出。
- `--timeout` 只接受严格 `<正整数>ms|s|m`,测试使用短 timeout 或虚拟时钟,不真实等待 `120s`
- watch 断连或后端错误必须显式失败,不能折叠为 timeout;该入口固定 `mutation=false`,不触发 PipelineRun、Argo refresh、GitOps 或 rollout。
- SELFMEDIA 最新一次端到端耗时:
- 唯一推荐入口是 `platform-infra pipelines-as-code delivery-timing --target NC01 --consumer selfmedia-nc01`
- 一次调用直接披露 GitHub PR `mergedAt`、trigger wait、PipelineRun/TaskRun、端到端耗时、Argo、runtime health 和 `evidenceGaps`
@@ -143,6 +143,6 @@
解决 [UniDesk #1656](https://github.com/pikasTech/unidesk/issues/1656):由 Artificer 在独立 `.worktree/cicd-status-visibility``fix/cicd-status-visibility` 分支修复 `cicd status --node` 成功零输出,并统一 PaC consumer 的 registry applicabilityGitOps-only/publisher-only 应显示 N/A 且由 PipelineRun、GitOps/Argo 与 runtime 证据闭环,只有 owning YAML 明确声明 image/registry 的 consumer 才把 registry missing 作为 blocker。保持默认输出有界、text/JSON/YAML 同构、缺失可选 YAML 事实降级为 warning,不新增严格合同或手工交付入口;补针对性测试和 `unidesk-cicd` 说明,提交独立 PR 且不自行合并,完成任务后将详细报告写入[任务报告](./details/cli-output-progressive-disclosure/R10_Task_Report.md)。
## R11 [in_progress]
## R11 [completed]
解决 [UniDesk #2104](https://github.com/pikasTech/unidesk/issues/2104):把 CLI-SPEC 的只读观测等待规则和 `unidesk-cicd` 受控观测入口统一为默认 `120s` 长轮询,目标完成或状态满足时立即返回,未完成等待至超时并明确返回当前状态、timedOut 与稳定继续入口;禁止 CLI 本地高频短轮询、忙轮询和无输出,mutation 与交付 authority 保持 fire-and-forget,完成任务后将详细报告写入[任务报告](./details/cli-output-progressive-disclosure/R11_Task_Report.md)。
@@ -0,0 +1,36 @@
# R11 任务报告
## 任务上下文
- MDTODOR11。
- GitHub issue`pikasTech/unidesk#2104`
- TargetNC01。
- targetWorkspace`/root/unidesk/.worktree/cicd-long-poll-observe`
- repository`pikasTech/unidesk`
- ref`feat/cicd-long-poll-observe`
## 完成内容
- 新增仓库权威 `cli-spec` skill,明确只读观测默认 `120s` 长轮询、严格 duration、终态立即返回和 timeout 可见性。
- 新增通用 `read-only-long-poll` 状态机,支持稳定 cursor、虚拟时钟、错误透传和取消语义。
- 将 PaC `closeout --wait` 从 CLI 本地每 `3s` 重复全量 status 改为目标控制面 Kubernetes watch 驱动。
- PipelineRun 与 PaC 共用同一 watch 判定:空成功为正常 timeout、`141` 为首事件后的 SIGPIPE,其他非零保留认证、网络或 API 错误。
- 新增 `--timeout 120s`timeout 输出当前状态、`timedOut``elapsed`、identity/cursor 和 `next.wait`
- capture 使用显式调用级 transport timeout,并保留用户 `60m` timeout 所需的固定余量,不依赖全局 env。
- 更新 `unidesk-cicd`、CLI help 和 `docs/reference/cli.md`,明确禁止调用侧 `1s``5s` 高频轮询。
- 权威 `cli-spec` 更新位于 `pikasTech/agent_skills#12`UniDesk 仓内未保留重复 skill。
- 未修改 AgentRun events 或 `unidesk-code-queue`,未改变 CI/CD、GitOps、Tekton 或 Argo authority。
## 验证
- `bun test scripts/src/read-only-long-poll.test.ts`
- `bun test scripts/src/kubernetes-watch.test.ts scripts/src/read-only-long-poll.test.ts scripts/src/ssh.test.ts`,共 38 项通过。
- `bun test scripts/src/cicd-pr-merge-auto-delivery-doc-contract.test.ts scripts/src/cicd-node-status.test.ts`
- `bun scripts/cli.ts platform-infra pipelines-as-code help compatibility-diagnostics`
- 非法 `--timeout 1.5s` 被严格拒绝。
## 证据边界
- primary workspace 只用于读取 resource bundle skills 与执行 `trans` wrapper。
- 源码读取、修改、测试和 Git 操作全部在 Target route `NC01:/root/unidesk/.worktree/cicd-long-poll-observe` 完成。
- 本任务不部署、不触发 PipelineRun、不执行 Argo refresh,也不手工触发 CI/CD。
+10
View File
@@ -1,5 +1,15 @@
# UniDesk CLI Reference
## 只读观测长轮询
- mutation、构建、发布和部署触发保持 fire-and-forget;长轮询只属于只读观测,不改变执行 authority。
- 只读等待默认 timeout 为 `120s`,显式覆盖只接受 `<正整数>ms|s|m`,最大 `60m`
- 后端 API、Kubernetes watch、事件流或原生通知负责持有等待;CLI 调用侧禁止 `1s``5s` 高频短轮询、忙轮询和重复全量扫描。
- 成功、失败、取消或满足条件时立即返回;未完成时保持单次等待到 timeout。
- timeout 和零变化必须返回当前状态、`timedOut=true``elapsed`、稳定 identity/cursor 与继续等待命令。
- watch 断连和后端错误必须可见并保持原错误语义,不能伪装成 timeout、空变化或成功。
- CI/CD 通用入口是 `bun scripts/cli.ts platform-infra pipelines-as-code closeout --target <NODE> --consumer <consumer> --wait --timeout 120s`
UniDesk 的统一 CLI 实现入口是根目录 `scripts/cli.ts`,运行方式固定为 `bun scripts/cli.ts <command>`;普通根 CLI 子命令仍使用该入口。`trans <route> ...` 是 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 默认输出与帮助输出是直接阅读入口:
+14 -4
View File
@@ -20,6 +20,7 @@ import {
} from "../artifact-registry";
import { k3sGuardShellLines, defaultNativeKubeconfig } from "../k3s-target-guard";
import { runSshCommandCapture } from "../ssh";
import { kubernetesWatchOneEventShellFunction } from "../kubernetes-watch";
import type { DispatchResult, PipelineRunCondition } from "./types";
import { status } from "./cleanup";
@@ -48,10 +49,11 @@ export async function waitForPipelineRun(name: string, waitMs: number, target =
const command = [
"set -euo pipefail",
...ciTargetGuardShellLines(target),
`printf 'waiting_pipelinerun=%s\\n' ${shellQuote(name)}`,
kubernetesWatchOneEventShellFunction(),
`deadline=$((SECONDS + ${Math.ceil(waitMs / 1000)}))`,
"while [ \"$SECONDS\" -lt \"$deadline\" ]; do",
` condition="$(kubectl get pipelinerun/${shellQuote(name)} -n unidesk-ci -o jsonpath='{range .status.conditions[?(@.type==\"Succeeded\")]}{.status}{\"\\t\"}{.reason}{\"\\t\"}{.message}{end}' 2>/dev/null || true)"`,
`printf 'waiting_pipelinerun=%s\\n' ${shellQuote(name)}`,
`condition="$(kubectl get pipelinerun/${shellQuote(name)} -n unidesk-ci -o jsonpath='{range .status.conditions[?(@.type==\"Succeeded\")]}{.status}{\"\\t\"}{.reason}{\"\\t\"}{.message}{end}' 2>/dev/null || true)"`,
"while :; do",
" case \"$condition\" in",
" True*)",
" printf 'condition=%s\\n' \"$condition\"",
@@ -64,7 +66,15 @@ export async function waitForPipelineRun(name: string, waitMs: number, target =
" exit 1",
" ;;",
" esac",
" sleep 2",
" remaining=$((deadline - SECONDS))",
" [ \"$remaining\" -gt 0 ] || break",
" set +e",
` event="$(unidesk_watch_one_event kubectl get pipelinerun/${shellQuote(name)} -n unidesk-ci --watch-only --request-timeout="\${remaining}s" -o name)"`,
" watch_status=$?",
" set -e",
" [ \"$watch_status\" -eq 0 ] || exit \"$watch_status\"",
" [ -n \"$event\" ] || break",
` condition="$(kubectl get pipelinerun/${shellQuote(name)} -n unidesk-ci -o jsonpath='{range .status.conditions[?(@.type==\"Succeeded\")]}{.status}{\"\\t\"}{.reason}{\"\\t\"}{.message}{end}' 2>/dev/null || true)"`,
"done",
`echo "Timed out waiting for pipelinerun/${name}" >&2`,
`kubectl get pipelinerun/${shellQuote(name)} -n unidesk-ci -o json`,
+29
View File
@@ -0,0 +1,29 @@
import { describe, expect, test } from "bun:test";
import { kubernetesWatchOneEventShellFunction } from "./kubernetes-watch";
function runWatch(command: string): { exitCode: number; stdout: string; stderr: string } {
const script = `set -euo pipefail\n${kubernetesWatchOneEventShellFunction()}\nunidesk_watch_one_event bash -c ${JSON.stringify(command)}`;
const result = Bun.spawnSync(["bash", "-c", script], { stdout: "pipe", stderr: "pipe" });
return { exitCode: result.exitCode, stdout: result.stdout.toString(), stderr: result.stderr.toString() };
}
describe("kubernetes watch one event shell helper", () => {
test("returns the first change", () => {
expect(runWatch("printf 'pipelinerun/run-1\\n'")).toEqual({ exitCode: 0, stdout: "pipelinerun/run-1\n", stderr: "" });
});
test("treats an empty successful request-timeout as no change", () => {
expect(runWatch("exit 0")).toEqual({ exitCode: 0, stdout: "", stderr: "" });
});
test("accepts producer SIGPIPE after the first event", () => {
const result = runWatch("trap 'exit 141' PIPE; while :; do printf 'event\\n' || exit 141; done");
expect(result.exitCode).toBe(0);
expect(result.stdout).toBe("event\n");
});
test("preserves authentication and API failures", () => {
const result = runWatch("echo 'Unauthorized' >&2; exit 7");
expect(result).toEqual({ exitCode: 7, stdout: "", stderr: "Unauthorized\n" });
});
});
+17
View File
@@ -0,0 +1,17 @@
export function kubernetesWatchOneEventShellFunction(): string {
return [
"unidesk_watch_one_event() {",
" event_file=$(mktemp)",
" error_file=$(mktemp)",
" set +e",
" \"$@\" 2>\"$error_file\" | head -n 1 >\"$event_file\"",
" statuses=(\"${PIPESTATUS[@]}\")",
" set -e",
" watch_status=${statuses[0]:-255}",
" case \"$watch_status\" in",
" 0|141) cat \"$event_file\"; rm -f \"$event_file\" \"$error_file\"; return 0 ;;",
" *) cat \"$error_file\" >&2; rm -f \"$event_file\" \"$error_file\"; return \"$watch_status\" ;;",
" esac",
"}",
].join("\n");
}
+2 -2
View File
@@ -25,8 +25,8 @@ export interface OpsCommandOptionSpec {
flagOptions?: string[];
}
export async function capture(config: UniDeskConfig, route: string, args: string[], stdin: string): Promise<SshCaptureResult> {
return await runSshCommandCapture(config, route, args, stdin);
export async function capture(config: UniDeskConfig, route: string, args: string[], stdin: string, options: { runtimeTimeoutMs?: number } = {}): Promise<SshCaptureResult> {
return await runSshCommandCapture(config, route, args, stdin, options);
}
export function parseJsonOutput(stdout: string): Record<string, unknown> | null {
+100 -42
View File
@@ -14,7 +14,9 @@ import {
sha256Fingerprint,
} from "./platform-infra-ops-library";
import { materializeYamlComposition } from "./yaml-composition";
import { kubernetesWatchOneEventShellFunction } from "./kubernetes-watch";
import { pacNodeReadOnlyNext, pacReadOnlyNext, resolveCicdDeliveryAuthority } from "./cicd-delivery-authority";
import { DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS, parseStrictDuration, readOnlyLongPollTransportTimeoutMs, runReadOnlyLongPoll } from "./read-only-long-poll";
import {
canonicalizePacPipelineSpec,
pacSourceArtifactSafeError,
@@ -250,6 +252,7 @@ export interface PacHistoryConsumerSelection {
interface CloseoutOptions extends CommonOptions {
sourceCommit: string | null;
wait: boolean;
timeoutMs: number;
}
interface ApplyOptions extends CommonOptions {
@@ -291,7 +294,20 @@ export async function runPlatformInfraPipelinesAsCodeCommand(config: UniDeskConf
return options.full || options.raw ? result : options.json ? compactStatusJson(result) : renderStatus(result);
}
if (action === "closeout") {
const options = parseCloseoutOptions(args.slice(1));
let options: CloseoutOptions;
try {
options = parseCloseoutOptions(args.slice(1));
} catch (error) {
return {
ok: false,
action: "platform-infra-pipelines-as-code-closeout-validation",
mutation: false,
code: "validation-failed",
message: error instanceof Error ? error.message : String(error),
usage: "bun scripts/cli.ts platform-infra pipelines-as-code closeout --target <NODE> --consumer <consumer> --wait [--timeout 120s]",
valuesPrinted: false,
};
}
const result = await closeout(config, options);
return options.full || options.raw ? result : options.json ? compactCloseoutJson(result) : renderCloseout(result);
}
@@ -421,7 +437,7 @@ function help(scope: string | null): Record<string, unknown> {
scope: "explicit-readonly-or-connectivity-diagnostics",
configTruth: configLabel,
usage: [
"bun scripts/cli.ts platform-infra pipelines-as-code closeout --target <NODE> --consumer <consumer> --source-commit <sha> --wait",
"bun scripts/cli.ts platform-infra pipelines-as-code closeout --target <NODE> --consumer <consumer> --source-commit <sha> --wait [--timeout 120s]",
"bun scripts/cli.ts platform-infra pipelines-as-code status --target <NODE> --consumer <consumer> --full",
"bun scripts/cli.ts platform-infra pipelines-as-code history --target <NODE> --consumer <consumer> --limit 10 --full",
"bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target <NODE> --consumer <consumer> --json",
@@ -1398,22 +1414,25 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
const target = resolveTarget(pac, options.targetId);
const consumer = resolveConsumer(pac, options.consumerId);
const startedAt = Date.now();
const waitMs = Math.max(1, pac.release.waitTimeoutSeconds) * 1000;
let ciAttempts = 0;
let runtimeAttempts = 0;
let current = await observeCloseoutStatus(config, options, target, consumer, "ci", 1, startedAt);
ciAttempts += 1;
while (options.wait && !closeoutCiReady(current, options.sourceCommit) && Date.now() - startedAt < waitMs) {
await sleep(3000);
current = await observeCloseoutStatus(config, options, target, consumer, "ci", ciAttempts + 1, startedAt);
ciAttempts += 1;
}
let current: Record<string, unknown> = {};
let phase: "ci" | "runtime" = "ci";
const waitResult = await runReadOnlyLongPoll({
timeoutMs: options.wait ? options.timeoutMs : 1,
observe: async () => {
current = await observeCloseoutStatus(config, options, target, consumer, phase, 1, startedAt);
phase = closeoutCiReady(current, options.sourceCommit) ? "runtime" : "ci";
return current;
},
completed: (value) => !options.wait || closeoutWaitComplete(value, options.sourceCommit),
waitForChange: async (remainingMs) => await waitForCloseoutChange(config, target, consumer, phase, remainingMs),
cursor: closeoutCursor,
});
current = waitResult.value;
let summary = record(current.summary);
let latest = record(summary.latestPipelineRun);
let diagnostics = record(summary.diagnostics);
let observedSourceCommit = observedCloseoutSourceCommit(latest, diagnostics);
let sourceMatched = options.sourceCommit === null || observedSourceCommit === options.sourceCommit;
const ciReady = closeoutCiReady(current, options.sourceCommit);
let gitOpsMirrorFlush: Record<string, unknown> = {
ok: true,
enabled: false,
@@ -1426,16 +1445,6 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
configSource: configLabel,
valuesPrinted: false,
};
const runtimeStartedAt = Date.now();
if (ciReady) {
current = await observeCloseoutStatus(config, options, target, consumer, "runtime", 1, startedAt);
runtimeAttempts += 1;
while (options.wait && !closeoutReady(current, options.sourceCommit) && Date.now() - runtimeStartedAt < waitMs) {
await sleep(3000);
current = await observeCloseoutStatus(config, options, target, consumer, "runtime", runtimeAttempts + 1, startedAt);
runtimeAttempts += 1;
}
}
summary = record(current.summary);
latest = record(summary.latestPipelineRun);
diagnostics = record(summary.diagnostics);
@@ -1452,8 +1461,8 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
ready,
sourceMatched,
diagnosticCode: diagnostics.code,
ciAttempts,
runtimeAttempts,
timedOut: waitResult.timedOut,
observations: waitResult.observations,
});
return {
ok: ready,
@@ -1465,26 +1474,66 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
observedSourceCommit,
sourceMatched,
ready,
timedOut: waitResult.timedOut,
elapsedMs: waitResult.elapsedMs,
cursor: waitResult.cursor,
wait: {
requested: options.wait,
attempts: ciAttempts + runtimeAttempts,
ciAttempts,
runtimeAttempts,
elapsedMs: Date.now() - startedAt,
budgetSeconds: pac.release.waitTimeoutSeconds,
budgetScope: "per-phase",
source: `${configLabel}.release.waitTimeoutSeconds`,
completed: waitResult.completed,
timedOut: waitResult.timedOut,
cancelled: waitResult.cancelled,
observations: waitResult.observations,
elapsedMs: waitResult.elapsedMs,
budgetSeconds: options.timeoutMs / 1000,
budgetScope: "whole-observation",
source: options.wait ? "cli --timeout or default 120s" : "single observation",
identity: { target: target.id, consumer: consumer.id, pipelineRun: latest.name ?? null, sourceCommit: options.sourceCommit, valuesPrinted: false },
cursor: waitResult.cursor,
valuesPrinted: false,
},
summary,
status: current,
gitOpsMirrorFlush,
blocker: ready ? null : closeoutBlocker(current, options.sourceCommit, observedSourceCommit, sourceMatched, gitOpsMirrorFlush),
next: nextCommands(target.id, consumer.id, pac.defaults.consumerId),
next: {
...nextCommands(target.id, consumer.id, pac.defaults.consumerId),
wait: `bun scripts/cli.ts platform-infra pipelines-as-code closeout --target ${target.id} --consumer ${consumer.id}${options.sourceCommit === null ? "" : ` --source-commit ${options.sourceCommit}`} --wait --timeout 120s`,
},
valuesPrinted: false,
};
}
async function waitForCloseoutChange(config: UniDeskConfig, target: PacTarget, consumer: PacConsumer, phase: "ci" | "runtime", remainingMs: number): Promise<void> {
const timeoutSeconds = Math.max(1, Math.ceil(remainingMs / 1000));
const args = phase === "ci"
? ["kubectl", "-n", consumer.namespace, "get", "pipelinerun", "-l", `tekton.dev/pipeline=${consumer.pipeline}`, "--watch-only", `--request-timeout=${timeoutSeconds}s`, "-o", "name"]
: ["kubectl", "-n", consumer.argoNamespace, "get", "application", consumer.argoApplication, "--watch-only", `--request-timeout=${timeoutSeconds}s`, "-o", "name"];
const script = [
"set -euo pipefail",
kubernetesWatchOneEventShellFunction(),
`unidesk_watch_one_event ${args.map(shQuote).join(" ")}`,
].join("\n");
const result = await capture(config, target.route, ["bash"], script, { runtimeTimeoutMs: readOnlyLongPollTransportTimeoutMs(remainingMs) });
if (result.exitCode !== 0) throw new Error(`CI/CD ${phase} watch disconnected: exit=${result.exitCode} stderr=${stringValue(compactCapture(result, { full: true }).stderrTail)}`);
}
function closeoutWaitComplete(value: Record<string, unknown>, sourceCommit: string | null): boolean {
if (closeoutReady(value, sourceCommit)) return true;
const latest = record(record(value.summary).latestPipelineRun);
const terminal = stringValue(latest.status, stringValue(latest.reason, "")).toLowerCase();
return ["failed", "failure", "false", "cancelled", "canceled"].some((status) => terminal.includes(status));
}
function closeoutCursor(value: Record<string, unknown>): string {
const summary = record(value.summary);
const latest = record(summary.latestPipelineRun);
const argo = record(summary.argo);
const runtime = record(summary.runtime);
return [latest.name, latest.status, latest.reason, argo.revision, argo.sync, argo.health, runtime.digest, runtime.readyReplicas]
.map((item) => stringValue(item, "-"))
.join(":");
}
async function observeCloseoutStatus(config: UniDeskConfig, options: CloseoutOptions, target: PacTarget, consumer: PacConsumer, phase: "ci" | "runtime", attempt: number, startedAt: number): Promise<Record<string, unknown>> {
printCloseoutProgress(target, consumer, `${phase}-status`, "started", startedAt, { attempt });
const current = await status(config, options);
@@ -2324,6 +2373,9 @@ function compactCloseoutJson(result: Record<string, unknown>): Record<string, un
observedSourceCommit: result.observedSourceCommit,
sourceMatched: result.sourceMatched === true,
ready: result.ready === true,
timedOut: result.timedOut === true,
elapsedMs: result.elapsedMs,
cursor: result.cursor,
wait: result.wait,
summary: compactStatusSummary(record(result.summary)),
gitOpsMirrorFlush: {
@@ -2655,15 +2707,17 @@ function renderCloseout(result: Record<string, unknown>): RenderedCliResult {
...table(["TARGET", "CONSUMER", "READY", "SOURCE_MATCHED", "MUTATION"], [[stringValue(target.id), stringValue(consumer.id), boolText(result.ready), boolText(result.sourceMatched), boolText(result.mutation)]]),
"",
"WAIT",
...table(["REQUESTED", "CI_ATTEMPTS", "RUNTIME_ATTEMPTS", "ELAPSED_MS", "BUDGET_S", "SCOPE", "SOURCE"], [[
...table(["REQUESTED", "COMPLETED", "TIMED_OUT", "CANCELLED", "OBSERVATIONS", "ELAPSED_MS", "BUDGET_S"], [[
stringValue(wait.requested),
stringValue(wait.ciAttempts),
stringValue(wait.runtimeAttempts),
boolText(wait.completed),
boolText(wait.timedOut),
boolText(wait.cancelled),
stringValue(wait.observations),
stringValue(wait.elapsedMs),
stringValue(wait.budgetSeconds),
stringValue(wait.budgetScope),
stringValue(wait.source),
]]),
` identity: ${compactLine(JSON.stringify(wait.identity ?? {}))}`,
` cursor: ${stringValue(wait.cursor)}`,
"",
"SOURCE / PIPELINERUN",
...table(["EXPECTED", "OBSERVED", "PIPELINERUN", "STATUS", "DURATION_S"], [[short(stringValue(result.sourceCommit), 16), short(stringValue(result.observedSourceCommit), 16), stringValue(latest.name), stringValue(latest.reason ?? latest.status), stringValue(latest.durationSeconds)]]),
@@ -2701,6 +2755,7 @@ function renderCloseout(result: Record<string, unknown>): RenderedCliResult {
"NEXT",
` status: ${stringValue(record(result.next).status)}`,
` history: ${stringValue(record(result.next).history)} --limit 10`,
` wait: ${stringValue(record(result.next).wait)}`,
latest.name === undefined ? " pipeline-run: -" : ` pipeline-run: ${stringValue(record(result.next).history)} --id ${stringValue(latest.name)}`,
];
return rendered(result, "platform-infra pipelines-as-code closeout", lines);
@@ -2904,6 +2959,7 @@ function parseCloseoutOptions(args: string[]): CloseoutOptions {
const commonArgs: string[] = [];
let sourceCommit: string | null = null;
let wait = false;
let timeoutMs = DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--source-commit" || arg === "--commit") {
@@ -2914,6 +2970,11 @@ function parseCloseoutOptions(args: string[]): CloseoutOptions {
index += 1;
} else if (arg === "--wait") {
wait = true;
} else if (arg === "--timeout") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--timeout requires a value");
timeoutMs = parseStrictDuration(value, "--timeout");
index += 1;
} else {
commonArgs.push(arg);
if (arg === "--target" || arg === "--node" || arg === "--consumer") {
@@ -2922,7 +2983,7 @@ function parseCloseoutOptions(args: string[]): CloseoutOptions {
}
}
}
return { ...parseCommonOptions(commonArgs), sourceCommit, wait };
return { ...parseCommonOptions(commonArgs), sourceCommit, wait, timeoutMs };
}
function parseHistoryOptions(args: string[]): HistoryOptions {
@@ -3065,9 +3126,6 @@ function closeoutBlocker(value: Record<string, unknown>, expected: string | null
};
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function pipelineRunGate(latest: Record<string, unknown>): Record<string, unknown> {
+53
View File
@@ -0,0 +1,53 @@
import { describe, expect, test } from "bun:test";
import { parseStrictDuration, readOnlyLongPollTransportTimeoutMs, runReadOnlyLongPoll } from "./read-only-long-poll";
describe("read-only long poll", () => {
test("returns an immediate terminal observation", async () => {
const result = await runReadOnlyLongPoll({ observe: async () => "done", completed: (value) => value === "done", waitForChange: async () => { throw new Error("unexpected wait"); }, cursor: (value) => value });
expect(result).toMatchObject({ completed: true, timedOut: false, observations: 1, cursor: "done" });
});
test("returns after a backend-held change", async () => {
let now = 0;
let state = "running";
const result = await runReadOnlyLongPoll({
observe: async () => state,
completed: (value) => value === "done",
waitForChange: async () => { now += 25; state = "done"; },
cursor: (value) => value,
timeoutMs: 100,
clock: { now: () => now },
});
expect(result).toMatchObject({ completed: true, timedOut: false, elapsedMs: 25, observations: 2 });
});
test("returns the current state after an empty timeout", async () => {
let now = 0;
const result = await runReadOnlyLongPoll({
observe: async () => ({ state: "running", id: "run-1" }),
completed: () => false,
waitForChange: async (remainingMs) => { now += remainingMs; },
cursor: (value) => value.id,
timeoutMs: 100,
clock: { now: () => now },
});
expect(result).toMatchObject({ completed: false, timedOut: true, cancelled: false, elapsedMs: 100, cursor: "run-1" });
});
test("preserves backend errors and cancellation", async () => {
await expect(runReadOnlyLongPoll({ observe: async () => "running", completed: () => false, waitForChange: async () => { throw new Error("watch disconnected"); }, cursor: (value) => value, timeoutMs: 10 })).rejects.toThrow("watch disconnected");
const controller = new AbortController();
controller.abort();
const cancelled = await runReadOnlyLongPoll({ observe: async () => "running", completed: () => false, waitForChange: async () => {}, cursor: (value) => value, signal: controller.signal });
expect(cancelled).toMatchObject({ completed: false, timedOut: false, cancelled: true });
});
test("validates strict duration syntax", () => {
expect(parseStrictDuration("120s")).toBe(120_000);
expect(parseStrictDuration("250ms")).toBe(250);
expect(() => parseStrictDuration("1.5s")).toThrow();
expect(() => parseStrictDuration("120")).toThrow();
expect(() => parseStrictDuration("0s")).toThrow();
expect(readOnlyLongPollTransportTimeoutMs(60 * 60_000)).toBe(60 * 60_000 + 15_000);
});
});
+63
View File
@@ -0,0 +1,63 @@
export const DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS = 120_000;
export const READ_ONLY_LONG_POLL_TRANSPORT_MARGIN_MS = 15_000;
export interface ReadOnlyLongPollClock {
now(): number;
}
export interface ReadOnlyLongPollResult<T> {
value: T;
completed: boolean;
timedOut: boolean;
cancelled: boolean;
elapsedMs: number;
observations: number;
cursor: string;
}
export function parseStrictDuration(value: string, label = "duration"): number {
const match = /^(\d+)(ms|s|m)$/u.exec(value);
if (match === null) throw new Error(`${label} must match <integer>ms|s|m`);
const amount = Number.parseInt(match[1]!, 10);
if (!Number.isSafeInteger(amount) || amount <= 0) throw new Error(`${label} must be greater than zero`);
const multiplier = match[2] === "ms" ? 1 : match[2] === "s" ? 1_000 : 60_000;
const durationMs = amount * multiplier;
if (!Number.isSafeInteger(durationMs) || durationMs > 3_600_000) throw new Error(`${label} must not exceed 60m`);
return durationMs;
}
export function readOnlyLongPollTransportTimeoutMs(timeoutMs: number): number {
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0 || timeoutMs > 3_600_000) throw new Error("long-poll timeoutMs must be between 1ms and 60m");
return timeoutMs + READ_ONLY_LONG_POLL_TRANSPORT_MARGIN_MS;
}
export async function runReadOnlyLongPoll<T>(options: {
observe: () => Promise<T>;
completed: (value: T) => boolean;
waitForChange: (remainingMs: number, signal?: AbortSignal) => Promise<void>;
cursor: (value: T) => string;
timeoutMs?: number;
clock?: ReadOnlyLongPollClock;
signal?: AbortSignal;
}): Promise<ReadOnlyLongPollResult<T>> {
const timeoutMs = options.timeoutMs ?? DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS;
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) throw new Error("timeoutMs must be a positive integer");
const clock = options.clock ?? { now: () => Date.now() };
const startedAt = clock.now();
let observations = 0;
let value = await options.observe();
observations += 1;
while (!options.completed(value)) {
if (options.signal?.aborted === true) {
return { value, completed: false, timedOut: false, cancelled: true, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
}
const remainingMs = timeoutMs - (clock.now() - startedAt);
if (remainingMs <= 0) {
return { value, completed: false, timedOut: true, cancelled: false, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
}
await options.waitForChange(remainingMs, options.signal);
value = await options.observe();
observations += 1;
}
return { value, completed: true, timedOut: false, cancelled: false, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
}
+5 -2
View File
@@ -18,6 +18,7 @@ import {
parseSshInvocation,
remoteCommandForRoute,
sshFailureHint,
sshCaptureRuntimeTimeoutMs,
sshRoutePayloadCwd,
sshRouteSeparatorCompatibilityHint,
sshRuntimeTimeoutHint,
@@ -538,6 +539,7 @@ async function runRemoteSshWebSocketCaptureRemoteCommand(
invocation: ReturnType<typeof parseSshInvocation>,
remoteCommand: string,
input?: string,
options: { runtimeTimeoutMs?: number } = {},
): Promise<SshCaptureResult> {
const captureInvocation = {
...invocation,
@@ -548,7 +550,7 @@ async function runRemoteSshWebSocketCaptureRemoteCommand(
cols: Number(process.stdout.columns) > 0 ? Number(process.stdout.columns) : 100,
rows: Number(process.stdout.rows) > 0 ? Number(process.stdout.rows) : 30,
};
const runtimeTimeoutMs = sshRuntimeTimeoutMs();
const runtimeTimeoutMs = sshCaptureRuntimeTimeoutMs(options.runtimeTimeoutMs);
const payload = {
providerId: invocation.providerId,
command: wrapSshRemoteCommand(remoteCommand),
@@ -1016,6 +1018,7 @@ export async function runRemoteSshCommandCapture(
args: string[],
input?: string,
env: NodeJS.ProcessEnv = process.env,
options: { runtimeTimeoutMs?: number } = {},
): Promise<SshCaptureResult> {
const token = sshClientTokenFromEnv(env);
const session = token === null
@@ -1028,5 +1031,5 @@ export async function runRemoteSshCommandCapture(
const stdin = parsed.stdinPrefix !== undefined || parsed.stdinSuffix !== undefined
? `${parsed.stdinPrefix ?? ""}${input ?? ""}${parsed.stdinSuffix ?? ""}`
: input;
return await runRemoteSshWebSocketCaptureRemoteCommand(session, invocation, parsed.remoteCommand, stdin);
return await runRemoteSshWebSocketCaptureRemoteCommand(session, invocation, parsed.remoteCommand, stdin, options);
}
+15 -8
View File
@@ -56,6 +56,7 @@ import {
const defaultSshSlowWarningMs = 10_000;
const defaultSshRuntimeTimeoutMs = 60_000;
const maxSshRuntimeTimeoutMs = 60_000;
const maxExplicitSshCaptureRuntimeTimeoutMs = 61 * 60_000;
const defaultSshBackendCoreDetectTimeoutMs = 15_000;
const maxSshBackendCoreDetectTimeoutMs = 60_000;
const minSshStdoutStreamMaxBytes = 4 * 1024;
@@ -183,6 +184,12 @@ export function sshRuntimeTimeoutMs(env: NodeJS.ProcessEnv = process.env): numbe
return Math.min(maxSshRuntimeTimeoutMs, Math.max(1000, Math.trunc(parsed)));
}
export function sshCaptureRuntimeTimeoutMs(explicitTimeoutMs?: number, env: NodeJS.ProcessEnv = process.env): number {
if (explicitTimeoutMs === undefined) return sshRuntimeTimeoutMs(env);
if (!Number.isSafeInteger(explicitTimeoutMs) || explicitTimeoutMs <= 0) throw new Error("SSH capture runtimeTimeoutMs must be a positive integer");
return Math.min(maxExplicitSshCaptureRuntimeTimeoutMs, explicitTimeoutMs);
}
export function sshRuntimeTimeoutHint(options: {
invocation: ParsedSshInvocation;
transport: SshRuntimeTimeoutHint["transport"];
@@ -982,10 +989,10 @@ async function runSshCaptureCommand(config: UniDeskConfig, invocation: ParsedSsh
return await runSshCaptureRemoteCommand(config, invocation, remoteCommand, input);
}
async function runSshCaptureRemoteCommand(config: UniDeskConfig, invocation: ParsedSshInvocation, remoteCommand: string, input?: string): Promise<SshCaptureResult> {
async function runSshCaptureRemoteCommand(config: UniDeskConfig, invocation: ParsedSshInvocation, remoteCommand: string, input?: string, options: { runtimeTimeoutMs?: number } = {}): Promise<SshCaptureResult> {
const startedAtMs = Date.now();
const size = terminalSize();
const runtimeTimeoutMs = sshRuntimeTimeoutMs();
const runtimeTimeoutMs = sshCaptureRuntimeTimeoutMs(options.runtimeTimeoutMs);
const payload = {
providerId: invocation.providerId,
command: wrapSshRemoteCommand(remoteCommand),
@@ -1256,7 +1263,7 @@ async function runRemoteSsh(config: UniDeskConfig, host: string, providerId: str
}, config);
}
export async function runSshCommandCapture(config: UniDeskConfig, target: string, args: string[], input?: string): Promise<SshCaptureResult> {
export async function runSshCommandCapture(config: UniDeskConfig, target: string, args: string[], input?: string, options: { runtimeTimeoutMs?: number } = {}): Promise<SshCaptureResult> {
const invocation = parseSshInvocation(target, args);
const parsed = invocation.parsed;
if (parsed.remoteCommand === null) throw new Error(`ssh ${target} capture requires a non-interactive operation`);
@@ -1265,19 +1272,19 @@ export async function runSshCommandCapture(config: UniDeskConfig, target: string
: input;
const plan = sshCaptureBackendPlan(config, process.env);
if (plan.backend === "remote-frontend-websocket" && plan.remoteHost !== null) {
return await runRemoteSshCapture(config, plan.remoteHost, target, args, stdin);
return await runRemoteSshCapture(config, plan.remoteHost, target, args, stdin, options);
}
const local = await runSshCaptureRemoteCommand(config, invocation, parsed.remoteCommand, stdin);
const local = await runSshCaptureRemoteCommand(config, invocation, parsed.remoteCommand, stdin, options);
const fallbackHost = sshCaptureRemoteHost(config, process.env);
if (local.exitCode !== 0 && fallbackHost !== null && isLocalSshCaptureBackendUnavailable(local)) {
return await runRemoteSshCapture(config, fallbackHost, target, args, stdin);
return await runRemoteSshCapture(config, fallbackHost, target, args, stdin, options);
}
return local;
}
async function runRemoteSshCapture(config: UniDeskConfig, host: string, target: string, args: string[], input?: string): Promise<SshCaptureResult> {
async function runRemoteSshCapture(config: UniDeskConfig, host: string, target: string, args: string[], input?: string, options: { runtimeTimeoutMs?: number } = {}): Promise<SshCaptureResult> {
const { runRemoteSshCommandCapture } = await import("./remote-ssh");
return await runRemoteSshCommandCapture(config, host, target, args, input);
return await runRemoteSshCommandCapture(config, host, target, args, input, process.env, options);
}
export function sshCaptureBackendPlan(config: UniDeskConfig, env: NodeJS.ProcessEnv = process.env): SshCaptureBackendPlan {
+9
View File
@@ -14,6 +14,7 @@ import {
remoteCommandForRoute,
sshTruncationCompletionSummary,
sshCaptureBackendPlan,
sshCaptureRuntimeTimeoutMs,
sshStderrStreamMaxBytes,
sshStreamDumpExplicitlyRequested,
sshStdoutStreamMaxBytes,
@@ -25,6 +26,14 @@ import {
} from "./ssh";
import { readCliOutputPolicy } from "./output";
describe("ssh capture runtime timeout", () => {
test("keeps explicit long-poll transport margin beyond the 60m user timeout", () => {
expect(sshCaptureRuntimeTimeoutMs(60 * 60_000 + 15_000)).toBe(60 * 60_000 + 15_000);
expect(sshCaptureRuntimeTimeoutMs(62 * 60_000)).toBe(61 * 60_000);
expect(() => sshCaptureRuntimeTimeoutMs(0)).toThrow("positive integer");
});
});
function sha256Hex(text: string): string {
return createHash("sha256").update(text, "utf8").digest("hex");
}
+1
View File
@@ -2111,6 +2111,7 @@ export {
runSsh,
runSshCommandCapture,
sshCaptureBackendPlan,
sshCaptureRuntimeTimeoutMs,
sshFailureHint,
sshRuntimeTimeoutHint,
sshRuntimeTimeoutMs,