Merge remote-tracking branch 'origin/master' into feat/2010-hwlab-release-production-lane

scripts/src/ci/pipelinerun-runtime.ts

@@ -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`,

scripts/src/kubernetes-watch.test.ts

@@ -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" });
+  });
+});

scripts/src/kubernetes-watch.ts

@@ -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");
+}

scripts/src/platform-infra-ops-library.ts

@@ -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 {

scripts/src/platform-infra-pipelines-as-code.ts

@@ -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> {

scripts/src/read-only-long-poll.test.ts

@@ -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);
+  });
+});

scripts/src/read-only-long-poll.ts

@@ -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) };
+}

scripts/src/remote-ssh.ts

@@ -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);
 }

scripts/src/ssh-runtime.ts

@@ -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 {

scripts/src/ssh.test.ts

@@ -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");
 }

scripts/src/ssh.ts

@@ -2111,6 +2111,7 @@ export {
   runSsh,
   runSshCommandCapture,
   sshCaptureBackendPlan,
+  sshCaptureRuntimeTimeoutMs,
   sshFailureHint,
   sshRuntimeTimeoutHint,
   sshRuntimeTimeoutMs,