From 3c613afdb569625693fba20363793eceec4e621b Mon Sep 17 00:00:00 2001 From: Relayflow Date: Mon, 21 Sep 2026 22:28:17 +0000 Subject: [PATCH 1/5] flows: watch a hosted run's status and follow its runner log `flows status --cloud` and `flows logs` were built for a run that had already finished. This adds the two live forms and the live step rows the watched page needs. - `flows status --cloud` renders `running`, `backoff`, `waiting` and `needs_human` rows in the local view's grammar, with the number of the attempt now running and the time since its start. Only what Cloud's step snapshot establishes is printed: no maximum-attempt denominator, no wait id, no attempt for a row that was never dispatched, and no advancing of a finished step whose end timestamp is missing. A running run with no rows says so rather than claiming the run has no steps. - `flows status --cloud --watch` redraws every two seconds until the run is terminal and leaves the final page up. Each frame is rendered, checked against the abort, then written in one call, so an interrupt cannot leave a torn page and a failed poll leaves the previous page alone. - `flows logs --follow` appends new runner output until the run ends and Cloud marks the log complete. Lines are buffered across polls and redacted whole, so a secret split on a chunk boundary cannot escape. - Both exit with the run's outcome, not the read's, through the validator `flows run --cloud --wait` already blocks on (extracted to `cloud-run-record.ts` and shared): 0 only on an attested completed/success, 1 on an attested failure or cancellation, and `cloud_invalid_response` on a terminal record that attests neither. Ctrl-C is `observation_aborted`, exit 1, and does not cancel the run. The runner log is re-read whole on each poll and only the new suffix is printed: the route's `offset` is a byte count and the content is a string, and mixing them loses text on the first non-ASCII character. A snapshot that no longer extends what was printed refuses with `cloud_log_rewritten` rather than guessing. `--follow --step` is refused before any request; a step transcript is not an append-only stream. `cli/cloud-read.ts` was over the 500-line smell threshold, so the shared formatting, refusals and status view moved to dependency-only modules that the one-shot and live entry points both import. Co-Authored-By: Claude Opus 5 --- docs/CLOUD.md | 79 +- docs/SURFACE.md | 9 +- packages/sdk/src/cli-commands.ts | 2 + packages/sdk/src/cli.ts | 34 +- packages/sdk/src/cli/cloud-format.ts | 82 ++ packages/sdk/src/cli/cloud-live.ts | 310 ++++++++ packages/sdk/src/cli/cloud-read.ts | 295 +------- packages/sdk/src/cli/cloud-refusal.ts | 113 +++ packages/sdk/src/cli/cloud-status-view.ts | 208 ++++++ packages/sdk/src/cli/status.ts | 15 + packages/sdk/src/cloud-read.ts | 26 +- packages/sdk/src/cloud-run-record.ts | 64 ++ packages/sdk/src/cloud-run.ts | 29 +- packages/sdk/tests/cloud-live.test.ts | 743 +++++++++++++++++++ packages/sdk/tests/cloud-read.test.ts | 11 +- packages/sdk/tests/relay-cli-surface.test.ts | 4 + 16 files changed, 1702 insertions(+), 322 deletions(-) create mode 100644 packages/sdk/src/cli/cloud-format.ts create mode 100644 packages/sdk/src/cli/cloud-live.ts create mode 100644 packages/sdk/src/cli/cloud-refusal.ts create mode 100644 packages/sdk/src/cli/cloud-status-view.ts create mode 100644 packages/sdk/src/cloud-run-record.ts create mode 100644 packages/sdk/tests/cloud-live.test.ts diff --git a/docs/CLOUD.md b/docs/CLOUD.md index 1fc6bc839..152cf7eb1 100644 --- a/docs/CLOUD.md +++ b/docs/CLOUD.md @@ -140,7 +140,9 @@ agent holding its user's own Cloud credential does not have to hand-roll HTTP: ```sh flows runs [--limit ] [--json] # recent runs flows logs [--step ] [--raw] [--json] # runner log, or a step's transcript +flows logs --follow [--json] # ...and keep reading it until the run ends flows status --cloud [--json] # the run's steps, as `flows status` renders a local one +flows status --cloud --watch [--json] # ...and redraw it until the run ends ``` They resolve their credential exactly the way every other hosted verb does @@ -222,9 +224,82 @@ authority surface 2.0.22 · artifact 9c361a2cbb0a · commit b4dd665eb433 ✓ complete-3 deterministic completed 1 attempt 0.0s success gate: exit_code pass ``` +A step that is still going renders in the same grammar as a finished one, in +the local view's vocabulary: `↻` for `running` and `backoff`, `⏸` for +`waiting` and `needs_human`, the number of the attempt now running, and the +time since its `startTime`. + +```text + ↻ agent-5 agent running attempt 1 6m50s +``` + +Only what the snapshot establishes is printed. Cloud's step rows carry no +maximum-attempt budget, no wait id and no backoff deadline, so — unlike the +local view — no `attempt 1/3`, no `awaiting human: ...` and no +`backoff until ...` appears; those cells arrive if and when the step route +carries the fields. A row with no `startTime` has not been dispatched, so it +shows no attempt number rather than `attempt 1`, and a step that has ended +keeps the duration Cloud reported instead of being advanced to now. A running +run whose snapshot has no rows prints `steps 0` followed by +`No step snapshot available yet.` — a fact about the snapshot, where a bare +`steps 0` would be a claim about the run. + +### Following a run that is still going + +`flows status --cloud --watch` redraws the page every two seconds until the +run reaches a terminal status, then leaves the final page up. `flows logs + --follow` appends new runner output on the same cadence until the run +ends and Cloud marks the log complete, then prints the run's outcome: + +```text +LOG 20d04c99-3fa8-48c9-9286-92d364a5bc2e runner following 1,204 bytes so far +[relayflow] ▶ agent-5 (agent) started +[relayflow] ✓ agent-1 … done in 5m16s · claude-opus-5 · 37 turns · $2.26 · wrote plan.md +COMPLETED 20d04c99-3fa8-48c9-9286-92d364a5bc2e completionReason: success +``` + +Both differ from their one-shot forms in one visible way: **the exit code is +the run's, not the read's.** They exit 0 only on a run Cloud attests as +`completed` with `completionReason: success`, 1 on an attested failure or +cancellation, and 1 with `cloud_invalid_response` on a terminal record that +attests neither — the same validation `flows run --cloud --wait` blocks on, so +the two cannot disagree. A plain `flows logs` on a failed run still exits 0, +because there the exit code describes the read. Ctrl-C exits 1 with +`observation_aborted`; the hosted run is **not** cancelled by it. + +Under `--json` both poll silently and print exactly one document at the end — +the one their one-shot form would have printed, with `--follow` carrying the +whole redacted log. That is deliberately unlike `flows check --watch`, which +emits one JSON report per check: these two have a single result, and a script +that wants it wants to block and then parse once. In that document `ok: true` +means the read succeeded; the run's outcome is the exit code. + +A watched page is drawn in one write after the cancellation check, so an +interrupt never leaves half a frame, and a failed poll leaves the previous page +alone rather than clearing the screen to report it. Transient failures and +HTTP 408/429/500/502/503/504 are retried with the same doubling delay, capped +at 30 seconds, that the hosted waiter uses; every other failure refuses with +the codes below. The page is two reads against two projections (the run record +then the step rows), so a step row can lag the header above it by a poll; it is +not an atomic snapshot and does not claim to be. + +`--follow` does not take `--step`. A step transcript is not an append-only +stream: a retry replaces it, and the rendered form is built from the whole +JSONL. The combination is refused with `invalid_invocation` before any request, +naming both alternatives. Following the runner log re-reads it whole on every +poll and prints only the part that is new — the route's `offset` is a byte +count and the content is a string, and mixing the two silently loses text the +moment a log contains a non-ASCII character, which the runner's own transition +lines do. The cost is a full read per poll and the log held in memory; the +benefit is that no line can be duplicated or skipped. If what was already +printed is no longer a prefix of what Cloud serves, `--follow` refuses with +`cloud_log_rewritten` rather than guessing which bytes are new. + `--cloud` takes neither `--data-dir` nor `--tail`: both name things on this filesystem, which a hosted run has none of, so pairing them is refused as an -invocation rather than quietly ignored. The spend total is summed from the +invocation rather than quietly ignored. `--watch` is refused without `--cloud` +for the same reason: the local `flows status` reads one journal file and +returns, so there is no loop for it to hang in. The spend total is summed from the step rows because the run record carries no total, and Cloud stores each step's cost as a float — unlike the local view, which adds the journal's decimal strings exactly (`run-state.ts`). @@ -239,6 +314,8 @@ Every refusal is one `REFUSED [code] message` line naming what to do next: | `cloud_forbidden` | 403: authenticated, but not allowed to read that run or log | | `cloud_run_not_found` | 404: no such run for this credential; points at `flows runs` | | `cloud_step_no_transcript` | `--step` named a step with no transcript, or no such step; names the ones that have one | +| `cloud_log_rewritten` | `--follow` found the log no longer starts with what it printed; following it would skip or repeat output | +| `observation_aborted` | Ctrl-C (or a caller's abort) during `--watch`/`--follow`; the hosted run continues | | `invalid_invocation` | the run id is not a run id (wrong characters, too long); refused before any request | | `cloud_invalid_response` | Cloud answered something this client cannot trust — a record for a different run, a list that is not a list, a row with no id, a pagination cursor that does not advance | | `cloud_unreachable` / `cloud_transport_failed` | the request never completed | diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 8b98dad6f..3f23962fb 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -734,10 +734,15 @@ journal. They are documented in [CLOUD.md](CLOUD.md#reading-a-hosted-run): ```text flows runs [--limit ] [--json] -flows logs [--step ] [--raw] [--json] -flows status --cloud [--json] +flows logs [--step ] [--raw] [--json] [--follow] +flows status --cloud [--json] [--watch] ``` +`--watch` and `--follow` keep reading until the hosted run is terminal and +exit with *its* outcome rather than the read's, using `flows run --cloud +--wait`'s mapping; Ctrl-C ends the observation, not the run. `--watch` needs +`--cloud`, and `--follow` does not take `--step`. + ### Agent sidechannel (initial byte-stream slice) Local agent workers (`flows run --local-agent`) open diff --git a/packages/sdk/src/cli-commands.ts b/packages/sdk/src/cli-commands.ts index 0bf944748..be0649907 100644 --- a/packages/sdk/src/cli-commands.ts +++ b/packages/sdk/src/cli-commands.ts @@ -190,6 +190,7 @@ export const CLI_VERBS = [ options: [ { flags: '--step ', description: 'Show that agent step’s transcript instead of the runner log' }, { flags: '--raw', description: 'Print the transcript JSONL unrendered (still redacted)' }, + { flags: '--follow', description: 'Append new runner output until the run ends; exits with the run’s outcome' }, JSON_OPTION, ], variants: ['logs'], @@ -325,6 +326,7 @@ export const CLI_VERBS = [ DATA_DIR_OPTION, { flags: '--tail ', description: 'Lines of each agent attempt’s transcript tail to show' }, { flags: '--cloud', description: 'Read the run from Cloud instead of a local journal; needs the run id' }, + { flags: '--watch', description: 'With --cloud: redraw until the run ends, then exit with its outcome' }, ], variants: ['status'], }, diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index 202a69a4a..61e2351e0 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -30,6 +30,7 @@ import { parseLogsArgs, parseRunsArgs, runCloudLogsCli, runCloudRunsCli, runCloudStatusCli, type LogsArgs, type RunsArgs, } from './cli/cloud-read.js'; +import { runCloudLogsFollow, runCloudStatusWatch } from './cli/cloud-live.js'; import { transcriptTailSource } from './transcript-tail.js'; import { checkTypeScriptFlow } from './cli/check-typescript.js'; import { runCloudCli } from './cli/cloud-run.js'; @@ -55,6 +56,16 @@ export type { CheckInputDiagnostic, CheckReport } from './cli/check.js'; export interface CliIo { stdout(line: string): void; stderr(line: string): void; + /** + * True when stdout is an interactive terminal. + * + * Only the live views read it, and only to decide whether a redraw may use + * ANSI control sequences: `flows status --cloud --watch` clears the screen + * for a terminal and appends whole pages when its output is redirected or + * mounted in a host that renders text. Absent means "not a terminal", so an + * embedder that says nothing gets the safe form. + */ + tty?: boolean; } type CliExitCode = 0 | 1 | 2 | 3; @@ -121,9 +132,9 @@ const USAGE = [ 'flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] ', 'flows replay [--allow-human-influenced] [--json] [--data-dir ] [--at ]', 'flows status [--json] [--data-dir ] [--tail ] []', - 'flows status --cloud [--json] ', + 'flows status --cloud [--json] [--watch] ', 'flows runs [--limit ] [--json]', - 'flows logs [--step ] [--raw] [--json] ', + 'flows logs [--step ] [--raw] [--json] [--follow] ', 'flows observer [--data-dir ]', 'flows hn-monitor start [--data-dir ] [--poll-interval-ms ] ', ].join('\n'); @@ -141,6 +152,7 @@ function spawnAllowedByEnv(env: NodeJS.ProcessEnv = process.env): boolean { const PROCESS_IO: CliIo = { stdout: (line) => process.stdout.write(`${line}\n`), stderr: (line) => process.stderr.write(`${line}\n`), + tty: process.stdout.isTTY === true, }; /** Optional knobs for an embedded caller. `bin/flows.js` passes none. */ @@ -220,13 +232,21 @@ export async function runCli( // works inside a step of a run whose daemon is gone (kernel/DAEMON-LIFECYCLE.md §4). if (parsed.command === 'status') { // One verb, two sources. `--cloud` never reaches `runStatus`, so the - // offline reader stays offline (cli/status.ts). - return parsed.cloud === true - ? runCloudStatusCli(parsed, io) - : runStatus(parsed, io, { tails: transcriptTailSource() }); + // offline reader stays offline (cli/status.ts). Only `--watch` blocks, so + // only `--watch` takes a signal: a one-shot read keeps installing none. + if (parsed.cloud !== true) return runStatus(parsed, io, { tails: transcriptTailSource() }); + const cloudStatus = parsed; + return cloudStatus.watch === true + ? withInterrupt(options.signal, (signal) => runCloudStatusWatch(cloudStatus, io, { signal })) + : runCloudStatusCli(cloudStatus, io); } if (parsed.command === 'runs') return runCloudRunsCli(parsed, io); - if (parsed.command === 'logs') return runCloudLogsCli(parsed, io); + if (parsed.command === 'logs') { + const logs = parsed; + return logs.follow + ? withInterrupt(options.signal, (signal) => runCloudLogsFollow(logs, io, { signal })) + : runCloudLogsCli(logs, io); + } if (parsed.command === 'answer') { const execution = await answerFlow(parsed.runId, parsed.waitId, parsed.answer, parsed.dataDir, { ...(parsed.note === undefined ? {} : { note: parsed.note }), diff --git a/packages/sdk/src/cli/cloud-format.ts b/packages/sdk/src/cli/cloud-format.ts new file mode 100644 index 000000000..2adae2f42 --- /dev/null +++ b/packages/sdk/src/cli/cloud-format.ts @@ -0,0 +1,82 @@ +// The cells every hosted-run page is built from. +// +// Moved out of `cli/cloud-read.ts` verbatim when the live views (`--watch`, +// `--follow`) needed the same formatting: the run list, the status page and +// the watched page must render a count, a cost and an instant identically, or +// a reader would have to learn which verb they were looking at first. +// +// Depends on nothing in `cli/` but `formatDuration`, so every other Cloud CLI +// module can import it without an import cycle. + +import { formatDuration } from './status.js'; + +export function thousands(value: number): string { + return String(value).replace(/\B(?=(\d{3})+(?!\d))/gu, ','); +} + +export function dollars(value: number): string { + return `$${value.toFixed(6).replace(/0+$/u, '').replace(/\.$/u, '')}`; +} + +/** Agent- and flow-authored names cannot inject terminal control sequences. */ +export function safe(text: string): string { + return text.replace(/[\u0000-\u001F\u007F-\u009F]/gu, '?'); +} + +/** + * Render a run's `error` as readable lines rather than one control-char smear. + * + * Cloud stores the runner's terminal output in this field, newlines and all, so + * passing it through `safe()` alone turns a 200-line tail into a single line of + * `?` separators. Long runs are dominated by lease renewals — one line every + * 10s for the life of every agent step, differing only in the deadline — which + * are worth counting, not reading. + * + * Collapsing is deliberately narrow: two adjacent lines merge only when they + * are character-for-character identical once a trailing integer is masked. A + * shared prefix is NOT line identity — runner lines put the step name, reason + * or message after a long fixed prefix, so collapsing on a prefix would hide + * distinct diagnostics behind a similarity count. The final line never + * collapses into an earlier one, because that is where the failure is. + */ +const TRAILING_NUMBER = /\d+(?=\D{0,2}$)/u; + +export function errorLines(text: string, indent: string): string[] { + const raw = text.split(/\r\n|\r|\n/u).map((line) => line.trimEnd()).filter((line) => line !== ''); + if (raw.length === 0) return []; + + const key = (line: string): string => line.replace(TRAILING_NUMBER, '#'); + const collapsed: { line: string; count: number }[] = []; + raw.forEach((line, index) => { + const previous = collapsed.at(-1); + const isLast = index === raw.length - 1; + if (previous !== undefined && !isLast && key(previous.line) === key(line)) previous.count += 1; + else collapsed.push({ line, count: 1 }); + }); + + const rendered = collapsed.map(({ line, count }) => + count === 1 ? safe(line) : `${safe(line)} (${count} times, differing only in a number)`); + + const HEAD = 2; + const TAIL = 12; + if (rendered.length <= HEAD + TAIL + 1) return rendered.map((line) => `${indent}${line}`); + const elided = rendered.length - HEAD - TAIL; + return [ + ...rendered.slice(0, HEAD), + `… ${elided} more line${elided === 1 ? '' : 's'} (full text: --json)`, + ...rendered.slice(-TAIL), + ].map((line) => `${indent}${line}`); +} + +/** ISO-8601 to the second: a list column, not a timestamp to do arithmetic on. */ +export function instant(value: string | null): string { + if (value === null) return 'unknown'; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? new Date(parsed).toISOString().replace(/\.\d{3}Z$/u, 'Z') : 'unknown'; +} + +export function ago(value: string | null, now: number): string { + if (value === null) return 'unknown'; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? `${formatDuration(now - parsed)} ago` : 'unknown'; +} diff --git a/packages/sdk/src/cli/cloud-live.ts b/packages/sdk/src/cli/cloud-live.ts new file mode 100644 index 000000000..04d05d97f --- /dev/null +++ b/packages/sdk/src/cli/cloud-live.ts @@ -0,0 +1,310 @@ +// `flows status --cloud --watch` and `flows logs --follow`: the two +// hosted reads that keep reading. +// +// The one-shot verbs (cli/cloud-read.ts) answer "what did that run do". These +// two answer "what is it doing", by asking again until the run is over. Three +// properties separate them from a loop around the one-shot reads: +// +// 1. **The exit code is the run's.** A finished page is not evidence that the +// run succeeded; `getCloudRunDetailLive` validates the same record +// `flows run --wait` blocks on, so `--watch` and `--follow` exit 0 only on +// an attested `completed`/`success`, 1 on an attested failure or a +// cancellation, and refuse a terminal record that attests neither. +// 2. **A frame is whole or absent.** The page is rendered, then the abort is +// checked, then the whole page leaves in one `io.stdout` call. A Ctrl-C +// mid-poll ends the command with a refusal and no half-drawn screen. +// 3. **Nothing is invented while waiting.** A failed poll leaves the previous +// page and the consumed log exactly where they were; it never clears the +// screen to show an error, and never re-prints a line it has shown. +// +// Cancellation is the caller's: `runCli` hands these functions the signal it +// owns for the verb's duration, and nothing here installs a process handler. +// Aborting stops the observation, not the hosted run — the refusal says so. + +import { canonicalize } from '../canonical.js'; +import { + getCloudRunDetailLive, getCloudRunLog, getCloudRunSteps, + type CloudRunDetail, type CloudRunLog, type CloudStep, +} from '../cloud-read.js'; +import { isCloudRunActive, type CloudRunState } from '../cloud-run-record.js'; +import { redact } from '../redact.js'; +import { thousands } from './cloud-format.js'; +import { fail, isTransientRead, refusalFor, RUN_ID_REQUIRED, type CloudReadOptions } from './cloud-refusal.js'; +import { renderCloudStatus, scrubRun, scrubSteps } from './cloud-status-view.js'; +import { sleepInterruptible } from './interruptible-sleep.js'; +import type { CliIo } from '../cli.js'; + +/** The cadence `waitForCloudFlowRun` polls at; not a user flag on either verb. */ +const DEFAULT_POLL_INTERVAL_MS = 2_000; +const MAX_BACKOFF_MS = 30_000; + +export interface CloudLiveOptions extends CloudReadOptions { + /** Milliseconds between polls. Injected by tests; 1..60000, as the waiter accepts. */ + pollIntervalMs?: number; + /** Injected by tests so no suite sleeps; production uses the abortable sleep. */ + sleep?: (ms: number, signal?: AbortSignal) => Promise; +} + +/** Clears the screen only for a terminal; a redirected page must stay text. */ +function clearSequence(io: CliIo): string { + return io.tty === true ? '\u001b[2J\u001b[H' : ''; +} + +/** The same delay schedule as `waitForCloudFlowRun`: ×2 per failure, capped. */ +function delayFor(failures: number, interval: number): number { + return failures === 0 ? interval : Math.min(interval * 2 ** (failures - 1), MAX_BACKOFF_MS); +} + +/** The hosted `flows run --wait` mapping: only an attested success is 0. */ +function exitFor(state: CloudRunState): 0 | 1 { + return state.status === 'completed' ? 0 : 1; +} + +function aborted(verb: 'watching' | 'following', runId: string, json: boolean, io: CliIo): 1 { + fail({ + code: 'observation_aborted', exit: 1, + message: `Stopped ${verb} run ${runId}; the hosted run has not been cancelled.`, + }, json, io); + return 1; +} + +/** + * Has the caller cancelled? + * + * A function rather than `signal?.aborted` at each site on purpose: `aborted` + * is a readonly property, so an inline check narrows the signal's type for the + * rest of the block and the compiler then calls every later check dead — which + * is exactly backwards for a value that changes under an await. + */ +function isAborted(signal: AbortSignal | undefined): boolean { + return signal !== undefined && signal.aborted; +} + +function invalidInterval(interval: number): boolean { + return !Number.isSafeInteger(interval) || interval < 1 || interval > 60_000; +} + +// -------------------------------------------------- flows status --cloud --watch + +/** + * Redraw the hosted status page until the run is terminal, then leave the + * final page up and exit with the run's outcome. + * + * The detail and the step rows are two requests against two projections, so + * the page is a pair of reads and not an atomic snapshot: a step row can lag + * the header it is drawn under by one poll. That is a property of the API, and + * this view reports what each read said rather than reconciling them. + */ +export async function runCloudStatusWatch( + args: { runId?: string; json: boolean }, + io: CliIo, + options: CloudLiveOptions = {}, +): Promise<0 | 1 | 2> { + const env = options.env ?? process.env; + const clock = options.now ?? Date.now; + const signal = options.signal; + const interval = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + const sleep = options.sleep ?? sleepInterruptible; + if (args.runId === undefined || args.runId.length === 0) return fail(RUN_ID_REQUIRED, args.json, io); + const runId = args.runId; + if (invalidInterval(interval)) { + return fail({ code: 'cloud_configuration', exit: 2, message: 'pollIntervalMs must be an integer from 1 to 60000.' }, + args.json, io); + } + + let failures = 0; + for (;;) { + if (isAborted(signal)) return aborted('watching', runId, args.json, io); + let live: { detail: CloudRunDetail; state: CloudRunState }; + let steps: CloudStep[]; + try { + live = await getCloudRunDetailLive(runId, options); + if (isAborted(signal)) return aborted('watching', runId, args.json, io); + steps = await getCloudRunSteps(runId, options); + } catch (error) { + if (isAborted(signal)) return aborted('watching', runId, args.json, io); + if (!isTransientRead(error)) return fail(refusalFor(error, `run ${runId}`, env), args.json, io); + // The previous page stays exactly as it was: a poll that did not + // complete is not news about the run. + failures = Math.min(failures + 1, 16); + await sleep(delayFor(failures, interval), signal); + continue; + } + failures = 0; + + const run = scrubRun(live.detail, env); + const scrubbed = scrubSteps(steps, env); + const terminal = !isCloudRunActive(live.state.status); + // Rendered before the abort check and emitted after it, in one call: an + // interrupt between two lines of a frame would leave a torn page. + const page = args.json ? '' : `${clearSequence(io)}${renderCloudStatus(run, scrubbed, clock()).join('\n')}`; + if (isAborted(signal)) return aborted('watching', runId, args.json, io); + if (!args.json) io.stdout(page); + if (terminal) { + if (args.json) io.stdout(canonicalize({ v: 1, ok: true, run, steps: scrubbed, now_ms: clock() })); + return exitFor(live.state); + } + await sleep(interval, signal); + } +} + +// ------------------------------------------------------ flows logs --follow + +/** + * What has been shown of the runner log, and what is still half a line. + * + * Byte and line bookkeeping is kept on the log exactly as Cloud served it, + * before redaction: a secret split across two polls is reassembled here and + * redacted as one whole line, so no chunk boundary can let half a token past + * the redactor. + */ +interface LogFollowState { + /** The whole log as last served, held to prove the next read extends it. */ + consumed: string; + /** A trailing fragment with no newline yet; never printed until it ends. */ + pending: string; + emitted: boolean; +} + +/** + * Emit whatever whole lines the new content completed. + * + * Splitting is on `\n` only, and one trailing `\r` per line is dropped: a + * CRLF log printed with its carriage returns intact makes a terminal overwrite + * each line it just drew. Blank lines and repeated identical lines are + * preserved — they are the log. + */ +function emitLines(fresh: string, state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): void { + state.pending += fresh; + let newline: number; + while ((newline = state.pending.indexOf('\n')) >= 0) { + const line = state.pending.slice(0, newline).replace(/\r$/u, ''); + state.pending = state.pending.slice(newline + 1); + io.stdout(redact(line, env)); + state.emitted = true; + } +} + +/** + * Has Cloud finished publishing this log? + * + * Three facts, all required. The run being over does not mean the last bytes + * have been written; `done` alone does not mean the run is over (the route can + * report a completed *upload* of a log the run is still adding to); and a + * `done` envelope that carries fewer bytes than it advertises has not served + * the tail. Stopping on any one of them alone would truncate the output, which + * is the one thing a follow must not do. + */ +function logDrained(log: CloudRunLog, terminal: boolean, content: string): boolean { + return terminal && log.done && Buffer.byteLength(content, 'utf8') >= log.total_size; +} + +/** + * Append the hosted runner log until the run is terminal and its log is drained. + * + * The route serves the log from an offset, but this client does not page it: + * the offsets are byte counts and the content is a JavaScript string, and + * subtracting one from the other silently loses text the moment a log contains + * a non-ASCII character — which the runner's own transition lines do. Until + * the byte-range contract can be checked against the server, each poll reads + * the whole log and prints the part that is new, refusing outright if what it + * already showed is no longer a prefix of what Cloud serves. That costs a full + * read per poll and holds the log in memory; it cannot duplicate or drop a line. + */ +export async function runCloudLogsFollow( + args: { runId: string; step: string | undefined; json: boolean }, + io: CliIo, + options: CloudLiveOptions = {}, +): Promise<0 | 1 | 2> { + const env = options.env ?? process.env; + const signal = options.signal; + const interval = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + const sleep = options.sleep ?? sleepInterruptible; + const runId = args.runId; + // Refused before any request: a step transcript is not an append-only stream + // this verb can follow. A retry replaces it, and the rendered form is built + // from the whole JSONL — attempt markers, session header, result footer — + // rather than a line at a time. + if (args.step !== undefined) { + return fail({ + code: 'invalid_invocation', exit: 2, + message: '`flows logs --follow` follows the runner log, which is append-only; it cannot follow a step ' + + 'transcript. Follow the run with `flows logs --follow`, or read the step transcript once ' + + 'with `flows logs --step `.', + }, args.json, io); + } + if (invalidInterval(interval)) { + return fail({ code: 'cloud_configuration', exit: 2, message: 'pollIntervalMs must be an integer from 1 to 60000.' }, + args.json, io); + } + + const state: LogFollowState = { consumed: '', pending: '', emitted: false }; + let header = false; + let failures = 0; + for (;;) { + if (isAborted(signal)) return aborted('following', runId, args.json, io); + let outcome: CloudRunState; + let log: CloudRunLog; + try { + // The run record first: a log read that follows a terminal status can + // only be missing bytes Cloud had not published yet, never bytes written + // after the check. + const live = await getCloudRunDetailLive(runId, options); + if (isAborted(signal)) return aborted('following', runId, args.json, io); + log = await getCloudRunLog(runId, undefined, options); + outcome = live.state; + } catch (error) { + if (isAborted(signal)) return aborted('following', runId, args.json, io); + if (!isTransientRead(error)) return fail(refusalFor(error, `run ${runId}`, env), args.json, io); + failures = Math.min(failures + 1, 16); + await sleep(delayFor(failures, interval), signal); + continue; + } + failures = 0; + if (isAborted(signal)) return aborted('following', runId, args.json, io); + + if (!log.content.startsWith(state.consumed)) { + return fail({ + code: 'cloud_log_rewritten', exit: 1, + message: `Cloud's runner log for run ${runId} no longer begins with the ${thousands(state.consumed.length)} ` + + 'characters already shown, so following it would skip or repeat output. Read it whole with ' + + `\`flows logs ${runId}\`.`, + }, args.json, io); + } + const fresh = log.content.slice(state.consumed.length); + state.consumed = log.content; + const terminal = !isCloudRunActive(outcome.status); + const drained = logDrained(log, terminal, log.content); + + if (!args.json) { + if (!header) { + io.stdout(`LOG ${runId} runner following ${thousands(log.total_size)} bytes so far`); + header = true; + } + emitLines(fresh, state, io, env); + } + if (!drained) { + await sleep(interval, signal); + continue; + } + // Drained: the last line may have no newline, and this is the only moment + // at which printing it cannot cut a line in half. + if (args.json) { + io.stdout(canonicalize({ + v: 1, ok: true, run_id: runId, step: null, + bytes: log.offset, total_bytes: log.total_size, done: log.done, + content: redact(log.content, env), + })); + return exitFor(outcome); + } + if (state.pending.length > 0) { + io.stdout(redact(state.pending.replace(/\r$/u, ''), env)); + state.pending = ''; + state.emitted = true; + } + if (!state.emitted) io.stdout(' (empty)'); + io.stdout(`${outcome.status.toUpperCase()} ${runId} completionReason: ` + + `${'completionReason' in outcome ? outcome.completionReason : 'unavailable'}`); + return exitFor(outcome); + } +} diff --git a/packages/sdk/src/cli/cloud-read.ts b/packages/sdk/src/cli/cloud-read.ts index 5e946208a..2776434d3 100644 --- a/packages/sdk/src/cli/cloud-read.ts +++ b/packages/sdk/src/cli/cloud-read.ts @@ -13,14 +13,15 @@ // facts beneath it) so a reader who knows one knows the other. import { canonicalize } from '../canonical.js'; -import { CloudFlowError, type CloudConnectionOptions } from '../cloud-http.js'; import { getCloudRunDetail, getCloudRunLog, getCloudRunSteps, listCloudRuns, type CloudRunDetail, type CloudStep, } from '../cloud-read.js'; import { parseAgentTranscript, renderAgentTranscript } from '../cloud-transcript.js'; import { redact } from '../redact.js'; -import { formatDuration } from './status.js'; +import { errorLines, instant, safe, thousands } from './cloud-format.js'; +import { fail, refusalFor, RUN_ID_REQUIRED, type CloudReadOptions } from './cloud-refusal.js'; +import { renderCloudStatus, scrubRun, scrubSteps } from './cloud-status-view.js'; import type { CliIo } from '../cli.js'; export interface RunsArgs { @@ -35,17 +36,13 @@ export interface LogsArgs { step: string | undefined; raw: boolean; json: boolean; + /** `--follow`: append new runner output until the run is terminal (cli/cloud-live.ts). */ + follow: boolean; } /** Enough runs to recognise the one you mean; `--limit` raises it. */ export const DEFAULT_RUN_LIMIT = 20; -/** Injected by tests; production takes the ambient clock and environment. */ -export interface CloudReadOptions extends CloudConnectionOptions { - env?: NodeJS.ProcessEnv; - now?: () => number; -} - export function parseRunsArgs(args: readonly string[]): RunsArgs | undefined { let limit: number | undefined; let json = false; @@ -69,6 +66,7 @@ export function parseLogsArgs(args: readonly string[]): LogsArgs | undefined { let step: string | undefined; let raw = false; let json = false; + let follow = false; const positionals: string[] = []; for (let index = 0; index < args.length; index += 1) { const argument = args[index]!; @@ -78,6 +76,9 @@ export function parseLogsArgs(args: readonly string[]): LogsArgs | undefined { } else if (argument === '--raw') { if (raw) return undefined; raw = true; + } else if (argument === '--follow') { + if (follow) return undefined; + follow = true; } else if (argument === '--step') { const value = args[++index]; if (step !== undefined || value === undefined || value.length === 0 || value.startsWith('-')) return undefined; @@ -89,147 +90,7 @@ export function parseLogsArgs(args: readonly string[]): LogsArgs | undefined { } } if (positionals.length !== 1) return undefined; - return { command: 'logs', runId: positionals[0]!, step, raw, json }; -} - -/** - * Every way these verbs can refuse, as a code and a sentence. - * - * Each one names the thing to do next. A missing credential names - * `agent-relay cloud login` because that is the command that fixes it; a 404 - * names `flows runs` because the usual cause is a run id from another - * workspace, and the list is how you find yours. - */ -interface Refusal { code: string; message: string; exit: 1 | 2 } - -function refusalFor(error: unknown, subject: string, env: NodeJS.ProcessEnv): Refusal { - const clean = (message: string): string => redact(message, env); - if (error instanceof CloudFlowError) { - if (error.code === 'configuration') { - switch (error.reason) { - case 'auth_missing': - return { - code: 'cloud_auth_missing', exit: 2, - message: 'No Cloud credential. Sign in with `agent-relay cloud login`, or set FLOWS_CLOUD_TOKEN to a ' - + 'Cloud API token with workflow:runs:read (and workflow:logs:read for `flows logs`).', - }; - case 'auth_expired': - return { code: 'cloud_auth_expired', exit: 2, message: clean(error.message) }; - case 'url_mismatch': - return { code: 'cloud_url_mismatch', exit: 2, message: clean(error.message) }; - default: - return { code: 'cloud_configuration', exit: 2, message: clean(error.message) }; - } - } - if (error.code === 'http_error') { - if (error.status === 401) { - return { - code: 'cloud_auth_rejected', exit: 2, - message: 'Cloud rejected the credential (HTTP 401). The token is unknown or revoked; ' - + 'sign in again with `agent-relay cloud login`.', - }; - } - if (error.status === 403) { - return { - code: 'cloud_forbidden', exit: 2, - message: `The credential is valid but not allowed to read ${subject} (HTTP 403). ` - + 'A run-scoped sandbox token may only read its own run; a workspace token needs ' - + 'workflow:runs:read, and workflow:logs:read or workflow:invoke:read for logs.', - }; - } - if (error.status === 404) { - return { - code: 'cloud_run_not_found', exit: 2, - message: `Cloud has no ${subject} visible to this credential (HTTP 404). ` - + 'It may belong to another workspace; `flows runs` lists the ones this credential can read.', - }; - } - return { code: 'cloud_http_error', exit: 1, message: clean(error.message) }; - } - if (error.code === 'invalid_input') return { code: 'invalid_invocation', exit: 2, message: clean(error.message) }; - if (error.code === 'transient_error' || error.code === 'transport_error') { - return { code: error.code === 'transient_error' ? 'cloud_unreachable' : 'cloud_transport_failed', exit: 1, message: clean(error.message) }; - } - return { code: 'cloud_invalid_response', exit: 1, message: clean(error.message) }; - } - return { code: 'cloud_read_failed', exit: 1, message: clean(error instanceof Error ? error.message : String(error)) }; -} - -function fail(refusal: Refusal, json: boolean, io: CliIo): 1 | 2 { - if (json) io.stdout(canonicalize({ v: 1, ok: false, code: refusal.code, message: refusal.message })); - else io.stderr(`REFUSED [${refusal.code}] ${refusal.message}`); - return refusal.exit; -} - -function thousands(value: number): string { - return String(value).replace(/\B(?=(\d{3})+(?!\d))/gu, ','); -} - -function dollars(value: number): string { - return `$${value.toFixed(6).replace(/0+$/u, '').replace(/\.$/u, '')}`; -} - -/** Agent- and flow-authored names cannot inject terminal control sequences. */ -function safe(text: string): string { - return text.replace(/[\u0000-\u001F\u007F-\u009F]/gu, '?'); -} - -/** - * Render a run's `error` as readable lines rather than one control-char smear. - * - * Cloud stores the runner's terminal output in this field, newlines and all, so - * passing it through `safe()` alone turns a 200-line tail into a single line of - * `?` separators. Long runs are dominated by lease renewals — one line every - * 10s for the life of every agent step, differing only in the deadline — which - * are worth counting, not reading. - * - * Collapsing is deliberately narrow: two adjacent lines merge only when they - * are character-for-character identical once a trailing integer is masked. A - * shared prefix is NOT line identity — runner lines put the step name, reason - * or message after a long fixed prefix, so collapsing on a prefix would hide - * distinct diagnostics behind a similarity count. The final line never - * collapses into an earlier one, because that is where the failure is. - */ -const TRAILING_NUMBER = /\d+(?=\D{0,2}$)/u; - -export function errorLines(text: string, indent: string): string[] { - const raw = text.split(/\r\n|\r|\n/u).map((line) => line.trimEnd()).filter((line) => line !== ''); - if (raw.length === 0) return []; - - const key = (line: string): string => line.replace(TRAILING_NUMBER, '#'); - const collapsed: { line: string; count: number }[] = []; - raw.forEach((line, index) => { - const previous = collapsed.at(-1); - const isLast = index === raw.length - 1; - if (previous !== undefined && !isLast && key(previous.line) === key(line)) previous.count += 1; - else collapsed.push({ line, count: 1 }); - }); - - const rendered = collapsed.map(({ line, count }) => - count === 1 ? safe(line) : `${safe(line)} (${count} times, differing only in a number)`); - - const HEAD = 2; - const TAIL = 12; - if (rendered.length <= HEAD + TAIL + 1) return rendered.map((line) => `${indent}${line}`); - const elided = rendered.length - HEAD - TAIL; - return [ - ...rendered.slice(0, HEAD), - `… ${elided} more line${elided === 1 ? '' : 's'} (full text: --json)`, - ...rendered.slice(-TAIL), - ].map((line) => `${indent}${line}`); -} - -/** ISO-8601 to the second: a list column, not a timestamp to do arithmetic on. */ -function instant(value: string | null): string { - if (value === null) return 'unknown'; - const parsed = Date.parse(value); - return Number.isFinite(parsed) ? new Date(parsed).toISOString().replace(/\.\d{3}Z$/u, 'Z') : 'unknown'; -} - -function ago(value: string | null, now: number): string { - if (value === null) return 'unknown'; - const parsed = Date.parse(value); - return Number.isFinite(parsed) ? `${formatDuration(now - parsed)} ago` : 'unknown'; + return { command: 'logs', runId: positionals[0]!, step, raw, json, follow }; } // ---------------------------------------------------------------- flows runs @@ -361,113 +222,6 @@ export async function runCloudLogsCli( // -------------------------------------------------------- flows status --cloud -const GLYPH: Record = { - completed: '✓', succeeded: '✓', failed: '✗', cancelled: '✗', running: '↻', pending: '○', queued: '○', -}; - -function glyph(step: CloudStep): string { - if (step.status === 'completed' && step.completion_reason !== null && step.completion_reason !== 'success') return '✗'; - return GLYPH[step.status] ?? '○'; -} - -/** The `steps N: 3 done · 1 running` line, counted off whatever Cloud called each status. */ -function stepCounts(steps: readonly CloudStep[]): string { - const counts = new Map(); - for (const step of steps) counts.set(step.status, (counts.get(step.status) ?? 0) + 1); - const parts = [...counts].map(([status, count]) => `${count} ${status}`); - return `steps ${steps.length}${parts.length === 0 ? '' : `: ${parts.join(' · ')}`}`; -} - -function renderStep(step: CloudStep, runId: string, idWidth: number): string[] { - const cells = [ - safe(step.step_name).slice(0, 24).padEnd(idWidth), - step.step_type.padEnd(13), - step.status.padEnd(11), - ]; - const attempts = step.attempts.length; - if (attempts > 0) cells.push(`${attempts} attempt${attempts === 1 ? '' : 's'}`); - if (step.duration_ms !== null) cells.push(formatDuration(step.duration_ms)); - if (step.completion_reason !== null) cells.push(safe(step.completion_reason)); - if (step.gate !== null) cells.push(`gate: ${safe(step.gate.gate)} ${safe(step.gate.verdict)}`); - const lines = [` ${glyph(step)} ${cells.join(' ')}`.trimEnd()]; - - if (step.gate !== null && step.gate.verdict !== 'pass' && step.gate.detail.length > 0) { - lines.push(` gate: ${safe(step.gate.gate)} ${step.gate.verdict.toUpperCase()} — ${JSON.stringify(safe(step.gate.detail))}`); - } - const transcript = step.transcript; - if (transcript !== null) { - const facts = [ - transcript.model === null ? null : safe(transcript.model), - transcript.num_turns === null ? null : `${transcript.num_turns} turn${transcript.num_turns === 1 ? '' : 's'}`, - transcript.total_calls === null ? null : `${transcript.total_calls} tool call${transcript.total_calls === 1 ? '' : 's'}`, - transcript.total_cost_usd === null ? null : dollars(transcript.total_cost_usd), - ].filter((fact): fact is string => fact !== null); - lines.push(` transcript (attempt ${transcript.attempt ?? step.attempts.length})${facts.length === 0 ? '' : `: ${facts.join(' · ')}`}`); - if (transcript.frames_total !== null || transcript.bytes_total !== null) { - const frames = transcript.frames_total === null ? '' - : `${thousands(transcript.frames_kept ?? transcript.frames_total)} of ${thousands(transcript.frames_total)} frames`; - const bytes = transcript.bytes_total === null ? '' : `${thousands(transcript.bytes_total)} bytes`; - const cut = transcript.truncated ? ', truncated' : ''; - lines.push(` ${[frames, bytes].filter((part) => part.length > 0).join(', ')}${cut}`); - } - if (transcript.tokens_in !== null || transcript.tokens_out !== null) { - const cache = [ - transcript.cache_read === null ? null : `${thousands(transcript.cache_read)} cache read`, - transcript.cache_creation === null ? null : `${thousands(transcript.cache_creation)} cache write`, - ].filter((part): part is string => part !== null); - lines.push(` ${thousands(transcript.tokens_in ?? 0)} in / ${thousands(transcript.tokens_out ?? 0)} out` - + `${cache.length === 0 ? '' : ` · ${cache.join(' · ')}`}`); - } - if (transcript.tools.length > 0) { - lines.push(` tools: ${transcript.tools.map((tool) => - `${safe(tool.name)} ×${tool.calls ?? 0}${tool.errors ? ` (${tool.errors} error${tool.errors === 1 ? '' : 's'})` : ''}`).join(', ')}`); - } - for (const call of transcript.last_calls) { - const excerpt = call.input_excerpt === null ? '' : ` ${safe(call.input_excerpt).slice(0, 120)}`; - lines.push(` call ${call.seq ?? '?'} ${safe(call.name)}${excerpt} → ${thousands(call.result_bytes ?? 0)} bytes`); - } - lines.push(` artifacts: ${transcript.artifacts.length === 0 - ? 'none' : transcript.artifacts.slice(0, 10).map(safe).join(', ')}`); - } - if (step.sandbox_id.length > 0) lines.push(` logs: flows logs ${runId} --step ${safe(step.sandbox_id)}`); - return lines; -} - -function renderCloudStatus(run: CloudRunDetail, steps: readonly CloudStep[], now: number): string[] { - let tokensIn = 0; - let tokensOut = 0; - let cost = 0; - for (const step of steps) { - tokensIn += step.tokens_in ?? 0; - tokensOut += step.tokens_out ?? 0; - cost += step.cost_usd ?? 0; - } - // Summed from the step rows, because the run record carries no total. Cloud - // stores `costUsd` as a float, so this is a float sum — not the exact decimal - // addition the local journal fold does (run-state.ts `addDollars`). Shown to - // six places and labelled `spend`, the same word the local view uses. - const spend = `${thousands(tokensIn)} in / ${thousands(tokensOut)} out / ${dollars(cost)}`; - const finished = run.completion_reason === null ? '' : ` finished ${safe(run.completion_reason)}`; - const lines = [ - `RUN ${run.run_id} ${safe(run.name)} ${run.status} started ${ago(run.created_at, now)}${finished} spend ${spend}`, - stepCounts(steps), - ]; - if (run.authority !== null) { - const facts = [ - run.authority.surface_version === null ? null : `surface ${run.authority.surface_version}`, - run.authority.artifact_sha256 === null ? null : `artifact ${run.authority.artifact_sha256.slice(0, 12)}`, - run.authority.source_commit === null ? null : `commit ${run.authority.source_commit.slice(0, 12)}`, - ].filter((fact): fact is string => fact !== null); - if (facts.length > 0) lines.push(`authority ${facts.join(' · ')}`); - } - if (run.pull_request_url !== null) lines.push(`pr ${run.pull_request_url}`); - if (run.error !== null) lines.push('error', ...errorLines(run.error, ' ')); - lines.push(''); - const idWidth = Math.max(4, ...steps.map((step) => Math.min(24, step.step_name.length))); - for (const step of steps) lines.push(...renderStep(step, run.run_id, idWidth)); - return lines; -} - export async function runCloudStatusCli( args: { runId?: string; json: boolean }, io: CliIo, @@ -475,13 +229,7 @@ export async function runCloudStatusCli( ): Promise<0 | 1 | 2> { const env = options.env ?? process.env; const now = (options.now ?? Date.now)(); - if (args.runId === undefined || args.runId.length === 0) { - return fail({ - code: 'run_unknown', exit: 2, - message: '`flows status --cloud` needs the hosted run id; `flows runs` lists them. ' - + '(Without --cloud the run id defaults to RELAYFLOW_RUN_ID inside a step.)', - }, args.json, io); - } + if (args.runId === undefined || args.runId.length === 0) return fail(RUN_ID_REQUIRED, args.json, io); let run: CloudRunDetail; let steps: CloudStep[]; try { @@ -490,25 +238,8 @@ export async function runCloudStatusCli( } catch (error) { return fail(refusalFor(error, `run ${args.runId}`, env), args.json, io); } - const scrubbed: CloudRunDetail = { - ...run, - name: redact(run.name, env), - error: run.error === null ? null : redact(run.error, env), - }; - const scrubbedSteps = steps.map((step) => ({ - ...step, - output_summary: step.output_summary === null ? null : redact(step.output_summary, env), - gate: step.gate === null ? null : { ...step.gate, detail: redact(step.gate.detail, env) }, - transcript: step.transcript === null ? null : { - ...step.transcript, - model: step.transcript.model === null ? null : redact(step.transcript.model, env), - last_calls: step.transcript.last_calls.map((call) => ({ - ...call, - input_excerpt: call.input_excerpt === null ? null : redact(call.input_excerpt, env), - })), - artifacts: step.transcript.artifacts.map((path) => redact(path, env)), - }, - })); + const scrubbed = scrubRun(run, env); + const scrubbedSteps = scrubSteps(steps, env); if (args.json) { io.stdout(canonicalize({ v: 1, ok: true, run: scrubbed, steps: scrubbedSteps, now_ms: now })); return 0; diff --git a/packages/sdk/src/cli/cloud-refusal.ts b/packages/sdk/src/cli/cloud-refusal.ts new file mode 100644 index 000000000..8e06169dd --- /dev/null +++ b/packages/sdk/src/cli/cloud-refusal.ts @@ -0,0 +1,113 @@ +// Every way a hosted read can refuse, as a code, a sentence and an exit. +// +// Moved out of `cli/cloud-read.ts` when the live views needed to refuse in +// exactly the same words: `flows status --cloud --watch` that meets a 404 must +// say what `flows status --cloud` says, or a reader would have to learn two +// vocabularies for one failure. A dependency-only module — it imports no other +// Cloud CLI module, so the one-shot and live entry points can both use it with +// no import cycle between them. + +import { canonicalize } from '../canonical.js'; +import { CloudFlowError, type CloudConnectionOptions } from '../cloud-http.js'; +import { redact } from '../redact.js'; +import type { CliIo } from '../cli.js'; + +/** Injected by tests; production takes the ambient clock and environment. */ +export interface CloudReadOptions extends CloudConnectionOptions { + env?: NodeJS.ProcessEnv; + now?: () => number; +} + +/** + * Every way these verbs can refuse, as a code and a sentence. + * + * Each one names the thing to do next. A missing credential names + * `agent-relay cloud login` because that is the command that fixes it; a 404 + * names `flows runs` because the usual cause is a run id from another + * workspace, and the list is how you find yours. + */ +export interface Refusal { code: string; message: string; exit: 1 | 2 } + +export function refusalFor(error: unknown, subject: string, env: NodeJS.ProcessEnv): Refusal { + const clean = (message: string): string => redact(message, env); + if (error instanceof CloudFlowError) { + if (error.code === 'configuration') { + switch (error.reason) { + case 'auth_missing': + return { + code: 'cloud_auth_missing', exit: 2, + message: 'No Cloud credential. Sign in with `agent-relay cloud login`, or set FLOWS_CLOUD_TOKEN to a ' + + 'Cloud API token with workflow:runs:read (and workflow:logs:read for `flows logs`).', + }; + case 'auth_expired': + return { code: 'cloud_auth_expired', exit: 2, message: clean(error.message) }; + case 'url_mismatch': + return { code: 'cloud_url_mismatch', exit: 2, message: clean(error.message) }; + default: + return { code: 'cloud_configuration', exit: 2, message: clean(error.message) }; + } + } + if (error.code === 'http_error') { + if (error.status === 401) { + return { + code: 'cloud_auth_rejected', exit: 2, + message: 'Cloud rejected the credential (HTTP 401). The token is unknown or revoked; ' + + 'sign in again with `agent-relay cloud login`.', + }; + } + if (error.status === 403) { + return { + code: 'cloud_forbidden', exit: 2, + message: `The credential is valid but not allowed to read ${subject} (HTTP 403). ` + + 'A run-scoped sandbox token may only read its own run; a workspace token needs ' + + 'workflow:runs:read, and workflow:logs:read or workflow:invoke:read for logs.', + }; + } + if (error.status === 404) { + return { + code: 'cloud_run_not_found', exit: 2, + message: `Cloud has no ${subject} visible to this credential (HTTP 404). ` + + 'It may belong to another workspace; `flows runs` lists the ones this credential can read.', + }; + } + return { code: 'cloud_http_error', exit: 1, message: clean(error.message) }; + } + if (error.code === 'invalid_input') return { code: 'invalid_invocation', exit: 2, message: clean(error.message) }; + if (error.code === 'transient_error' || error.code === 'transport_error') { + return { code: error.code === 'transient_error' ? 'cloud_unreachable' : 'cloud_transport_failed', exit: 1, message: clean(error.message) }; + } + return { code: 'cloud_invalid_response', exit: 1, message: clean(error.message) }; + } + return { code: 'cloud_read_failed', exit: 1, message: clean(error instanceof Error ? error.message : String(error)) }; +} + +export function fail(refusal: Refusal, json: boolean, io: CliIo): 1 | 2 { + if (json) io.stdout(canonicalize({ v: 1, ok: false, code: refusal.code, message: refusal.message })); + else io.stderr(`REFUSED [${refusal.code}] ${refusal.message}`); + return refusal.exit; +} + +/** + * `flows status --cloud` with no run id. + * + * Shared by the one-shot and watched forms so both name the same remedy: a + * hosted run has no ambient id the way a step inside a local run does. + */ +export const RUN_ID_REQUIRED: Refusal = { + code: 'run_unknown', exit: 2, + message: '`flows status --cloud` needs the hosted run id; `flows runs` lists them. ' + + '(Without --cloud the run id defaults to RELAYFLOW_RUN_ID inside a step.)', +}; + +/** + * A read that a poll can retry rather than end on. + * + * The same set `waitForCloudFlowRun` retries: the failure says the request did + * not complete, not that the request was wrong. Anything else — a 404, a + * rejected credential, a response this client cannot trust — ends the command + * through `refusalFor`, exactly as it ends a one-shot read. + */ +export function isTransientRead(error: unknown): boolean { + return error instanceof CloudFlowError && (error.code === 'transient_error' + || (error.code === 'http_error' && [408, 429, 500, 502, 503, 504].includes(error.status ?? 0))); +} diff --git a/packages/sdk/src/cli/cloud-status-view.ts b/packages/sdk/src/cli/cloud-status-view.ts new file mode 100644 index 000000000..843ad1703 --- /dev/null +++ b/packages/sdk/src/cli/cloud-status-view.ts @@ -0,0 +1,208 @@ +// The `flows status --cloud` page: one run record and its step rows, rendered +// in the grammar `cli/status.ts` already uses for a local run. +// +// Moved out of `cli/cloud-read.ts` so the one-shot read and the watched read +// (cli/cloud-live.ts) draw the identical page through the identical scrubber. +// A watched frame that redacted differently to the page it replaces would be +// a leak nobody could see from either file alone. +// +// Live rows are the part this view has that the local one does not have to +// think about: Cloud's step snapshot is the only evidence, and where it is +// silent the row stays silent. An elapsed time is derived only for a step that +// is demonstrably running, an attempt number is printed only where the +// snapshot establishes one, and neither a maximum-attempt denominator nor a +// wait id is invented — Cloud's step rows carry neither. + +import { isCloudRunActive } from '../cloud-run-record.js'; +import { redact } from '../redact.js'; +import type { CloudRunDetail, CloudStep } from '../cloud-read.js'; +import { ago, dollars, errorLines, safe, thousands } from './cloud-format.js'; +import { formatDuration } from './status.js'; + +const GLYPH: Record = { + completed: '✓', succeeded: '✓', failed: '✗', cancelled: '✗', + running: '↻', backoff: '↻', waiting: '⏸', needs_human: '⏸', + pending: '○', queued: '○', +}; + +/** + * Step statuses that mean the step is in flight, in the local view's + * vocabulary (`cli/status.ts`). These are the rows with no `durationMs` and no + * `endTime`: their timing has to be derived, and their attempt number is the + * one being attempted rather than a count of the ones that finished. + */ +const ACTIVE_STEP_STATUSES: readonly string[] = ['running', 'backoff', 'waiting', 'needs_human']; + +function glyph(step: CloudStep): string { + if (step.status === 'completed' && step.completion_reason !== null && step.completion_reason !== 'success') return '✗'; + return GLYPH[step.status] ?? '○'; +} + +/** The `steps N: 3 done · 1 running` line, counted off whatever Cloud called each status. */ +function stepCounts(steps: readonly CloudStep[]): string { + const counts = new Map(); + for (const step of steps) counts.set(step.status, (counts.get(step.status) ?? 0) + 1); + const parts = [...counts].map(([status, count]) => `${count} ${status}`); + return `steps ${steps.length}${parts.length === 0 ? '' : `: ${parts.join(' · ')}`}`; +} + +/** + * How long an in-flight step has been in flight. + * + * Only from a start timestamp the row actually carries, and only while the row + * says the step has not ended: a finished step whose `endTime` is missing keeps + * whatever duration Cloud reported rather than being advanced to now, which + * would make a step that ended an hour ago look like it is still burning. A + * start in the future is clock skew between Cloud and here, not negative time, + * so it clamps to zero. + */ +function liveElapsed(step: CloudStep, now: number): number | null { + if (!ACTIVE_STEP_STATUSES.includes(step.status) || step.ended_at !== null || step.started_at === null) return null; + const started = Date.parse(step.started_at); + return Number.isFinite(started) ? Math.max(0, now - started) : null; +} + +/** + * The attempt cell. + * + * A finished step counts the attempt records Cloud kept. An in-flight step + * wants the number of the attempt *now running*, and `attempts` cannot supply + * it: those entries are completion records, so during a second attempt the + * array still holds one entry and would print `attempt 1`. `retryCount` plus + * one is that number — but only where the row also shows the step was + * dispatched, because a row with no start timestamp has not attempted + * anything and `attempt 1` would be a claim about work that has not begun. + * Where neither establishes it, the cell is omitted: the local view prints no + * attempt at zero either. No `/max` denominator — Cloud's step rows carry no + * maximum, and printing one would invent a budget. + */ +function attemptCell(step: CloudStep): string | null { + if (!ACTIVE_STEP_STATUSES.includes(step.status)) { + const attempts = step.attempts.length; + return attempts > 0 ? `${attempts} attempt${attempts === 1 ? '' : 's'}` : null; + } + if (step.retry_count === null || step.started_at === null) return null; + return `attempt ${Math.max(0, Math.trunc(step.retry_count)) + 1}`; +} + +function renderStep(step: CloudStep, runId: string, idWidth: number, now: number): string[] { + const cells = [ + safe(step.step_name).slice(0, 24).padEnd(idWidth), + step.step_type.padEnd(13), + step.status.padEnd(11), + ]; + const attempt = attemptCell(step); + if (attempt !== null) cells.push(attempt); + const elapsed = step.duration_ms ?? liveElapsed(step, now); + if (elapsed !== null) cells.push(formatDuration(elapsed)); + if (step.completion_reason !== null) cells.push(safe(step.completion_reason)); + if (step.gate !== null) cells.push(`gate: ${safe(step.gate.gate)} ${safe(step.gate.verdict)}`); + const lines = [` ${glyph(step)} ${cells.join(' ')}`.trimEnd()]; + + if (step.gate !== null && step.gate.verdict !== 'pass' && step.gate.detail.length > 0) { + lines.push(` gate: ${safe(step.gate.gate)} ${step.gate.verdict.toUpperCase()} — ${JSON.stringify(safe(step.gate.detail))}`); + } + const transcript = step.transcript; + if (transcript !== null) { + const facts = [ + transcript.model === null ? null : safe(transcript.model), + transcript.num_turns === null ? null : `${transcript.num_turns} turn${transcript.num_turns === 1 ? '' : 's'}`, + transcript.total_calls === null ? null : `${transcript.total_calls} tool call${transcript.total_calls === 1 ? '' : 's'}`, + transcript.total_cost_usd === null ? null : dollars(transcript.total_cost_usd), + ].filter((fact): fact is string => fact !== null); + lines.push(` transcript (attempt ${transcript.attempt ?? step.attempts.length})${facts.length === 0 ? '' : `: ${facts.join(' · ')}`}`); + if (transcript.frames_total !== null || transcript.bytes_total !== null) { + const frames = transcript.frames_total === null ? '' + : `${thousands(transcript.frames_kept ?? transcript.frames_total)} of ${thousands(transcript.frames_total)} frames`; + const bytes = transcript.bytes_total === null ? '' : `${thousands(transcript.bytes_total)} bytes`; + const cut = transcript.truncated ? ', truncated' : ''; + lines.push(` ${[frames, bytes].filter((part) => part.length > 0).join(', ')}${cut}`); + } + if (transcript.tokens_in !== null || transcript.tokens_out !== null) { + const cache = [ + transcript.cache_read === null ? null : `${thousands(transcript.cache_read)} cache read`, + transcript.cache_creation === null ? null : `${thousands(transcript.cache_creation)} cache write`, + ].filter((part): part is string => part !== null); + lines.push(` ${thousands(transcript.tokens_in ?? 0)} in / ${thousands(transcript.tokens_out ?? 0)} out` + + `${cache.length === 0 ? '' : ` · ${cache.join(' · ')}`}`); + } + if (transcript.tools.length > 0) { + lines.push(` tools: ${transcript.tools.map((tool) => + `${safe(tool.name)} ×${tool.calls ?? 0}${tool.errors ? ` (${tool.errors} error${tool.errors === 1 ? '' : 's'})` : ''}`).join(', ')}`); + } + for (const call of transcript.last_calls) { + const excerpt = call.input_excerpt === null ? '' : ` ${safe(call.input_excerpt).slice(0, 120)}`; + lines.push(` call ${call.seq ?? '?'} ${safe(call.name)}${excerpt} → ${thousands(call.result_bytes ?? 0)} bytes`); + } + lines.push(` artifacts: ${transcript.artifacts.length === 0 + ? 'none' : transcript.artifacts.slice(0, 10).map(safe).join(', ')}`); + } + if (step.sandbox_id.length > 0) lines.push(` logs: flows logs ${runId} --step ${safe(step.sandbox_id)}`); + return lines; +} + +export function renderCloudStatus(run: CloudRunDetail, steps: readonly CloudStep[], now: number): string[] { + let tokensIn = 0; + let tokensOut = 0; + let cost = 0; + for (const step of steps) { + tokensIn += step.tokens_in ?? 0; + tokensOut += step.tokens_out ?? 0; + cost += step.cost_usd ?? 0; + } + // Summed from the step rows, because the run record carries no total. Cloud + // stores `costUsd` as a float, so this is a float sum — not the exact decimal + // addition the local journal fold does (run-state.ts `addDollars`). Shown to + // six places and labelled `spend`, the same word the local view uses. + const spend = `${thousands(tokensIn)} in / ${thousands(tokensOut)} out / ${dollars(cost)}`; + const finished = run.completion_reason === null ? '' : ` finished ${safe(run.completion_reason)}`; + const lines = [ + `RUN ${run.run_id} ${safe(run.name)} ${run.status} started ${ago(run.created_at, now)}${finished} spend ${spend}`, + stepCounts(steps), + ]; + // `steps 0` alone reads as "this run has no steps", which is a claim about + // the run. For a run that is still going it is a fact about the snapshot: + // Cloud served no rows, this time. + if (steps.length === 0 && isCloudRunActive(run.status)) lines.push('No step snapshot available yet.'); + if (run.authority !== null) { + const facts = [ + run.authority.surface_version === null ? null : `surface ${run.authority.surface_version}`, + run.authority.artifact_sha256 === null ? null : `artifact ${run.authority.artifact_sha256.slice(0, 12)}`, + run.authority.source_commit === null ? null : `commit ${run.authority.source_commit.slice(0, 12)}`, + ].filter((fact): fact is string => fact !== null); + if (facts.length > 0) lines.push(`authority ${facts.join(' · ')}`); + } + if (run.pull_request_url !== null) lines.push(`pr ${run.pull_request_url}`); + if (run.error !== null) lines.push('error', ...errorLines(run.error, ' ')); + lines.push(''); + const idWidth = Math.max(4, ...steps.map((step) => Math.min(24, step.step_name.length))); + for (const step of steps) lines.push(...renderStep(step, run.run_id, idWidth, now)); + return lines; +} + +/** Every free-text field of the run record, through the status page's redactor. */ +export function scrubRun(run: CloudRunDetail, env: NodeJS.ProcessEnv): CloudRunDetail { + return { + ...run, + name: redact(run.name, env), + error: run.error === null ? null : redact(run.error, env), + }; +} + +/** The same, for the step rows — including every string inside the digest. */ +export function scrubSteps(steps: readonly CloudStep[], env: NodeJS.ProcessEnv): CloudStep[] { + return steps.map((step) => ({ + ...step, + output_summary: step.output_summary === null ? null : redact(step.output_summary, env), + gate: step.gate === null ? null : { ...step.gate, detail: redact(step.gate.detail, env) }, + transcript: step.transcript === null ? null : { + ...step.transcript, + model: step.transcript.model === null ? null : redact(step.transcript.model, env), + last_calls: step.transcript.last_calls.map((call) => ({ + ...call, + input_excerpt: call.input_excerpt === null ? null : redact(call.input_excerpt, env), + })), + artifacts: step.transcript.artifacts.map((path) => redact(path, env)), + }, + })); +} diff --git a/packages/sdk/src/cli/status.ts b/packages/sdk/src/cli/status.ts index 5567b3dee..4458e616f 100644 --- a/packages/sdk/src/cli/status.ts +++ b/packages/sdk/src/cli/status.ts @@ -31,6 +31,15 @@ export interface StatusArgs { * module keeps its property of opening one file and no socket. */ cloud?: true; + /** + * `--watch`: redraw the hosted page until the run is terminal. + * + * Only with `--cloud`. A local `flows status` reads one journal file and + * returns; there is no loop for it to hang in, so `--watch` without + * `--cloud` is a contradiction rather than a narrower request, and is + * refused with the rest of the invocation. Handled in `cli/cloud-live.ts`. + */ + watch?: true; } export const DEFAULT_TAIL_LINES = 20; @@ -63,6 +72,7 @@ export interface StatusOptions { export function parseStatusArgs(args: readonly string[]): StatusArgs | undefined { let json = false; let cloud = false; + let watch = false; let dataDir: string | undefined; let tail: number | undefined; const positionals: string[] = []; @@ -74,6 +84,9 @@ export function parseStatusArgs(args: readonly string[]): StatusArgs | undefined } else if (argument === '--cloud') { if (cloud) return undefined; cloud = true; + } else if (argument === '--watch') { + if (watch) return undefined; + watch = true; } else if (argument === '--data-dir' || argument === '--tail') { const value = args[++index]; if (value === undefined || value.length === 0 || value.startsWith('-')) return undefined; @@ -97,9 +110,11 @@ export function parseStatusArgs(args: readonly string[]): StatusArgs | undefined // but a contradiction, and is refused as an invocation rather than silently // ignored. A hosted run id is mandatory: there is no ambient one. if (cloud && (dataDir !== undefined || tail !== undefined || positionals.length === 0)) return undefined; + if (watch && !cloud) return undefined; return { command: 'status', json, ...(cloud ? { cloud: true as const } : {}), + ...(watch ? { watch: true as const } : {}), ...(dataDir === undefined ? {} : { dataDir }), ...(tail === undefined ? {} : { tail }), ...(positionals[0] === undefined ? {} : { runId: positionals[0] }), diff --git a/packages/sdk/src/cloud-read.ts b/packages/sdk/src/cloud-read.ts index 4a21ab20e..fa8f6220b 100644 --- a/packages/sdk/src/cloud-read.ts +++ b/packages/sdk/src/cloud-read.ts @@ -15,6 +15,7 @@ // key is ignored, a route that drops one yields `null`, never a throw. import { CloudFlowError, cloudFetch, cloudRunId, isCloudRecord, type CloudConnectionOptions } from './cloud-http.js'; +import { cloudRunState, type CloudRunState } from './cloud-run-record.js'; /** One row of `GET /api/v1/workflows/runs`. */ export interface CloudRunSummary { @@ -334,8 +335,31 @@ export async function getCloudRunDetail( options: CloudConnectionOptions = {}, ): Promise { const asked = callerRunId(runId); - const body = record(await cloudFetch(`/api/v1/workflows/runs/${asked}`, options, + return runDetail(asked, await cloudFetch(`/api/v1/workflows/runs/${asked}`, options, { method: 'GET', detail: true })); +} + +/** + * The run record twice over: what to render, and what it attests. + * + * `--watch` and `--follow` need both, from the same bytes. The rendered view + * is the permissive mapping below, which shows whatever Cloud said; the exit + * code comes from `cloudRunState`, the same validator `flows run --wait` + * blocks on, so a live read cannot exit 0 on a record the waiter refuses. + * One request, because a second GET per poll could observe a different run + * than the one that was drawn. + */ +export async function getCloudRunDetailLive( + runId: string, + options: CloudConnectionOptions = {}, +): Promise<{ detail: CloudRunDetail; state: CloudRunState }> { + const asked = callerRunId(runId); + const answered = await cloudFetch(`/api/v1/workflows/runs/${asked}`, options, { method: 'GET', detail: true }); + return { detail: runDetail(asked, answered), state: cloudRunState(answered, asked) }; +} + +function runDetail(asked: string, answeredBody: unknown): CloudRunDetail { + const body = record(answeredBody); if (body === null) throw new CloudFlowError('invalid_response', 'Cloud returned no run record.'); // The record has to be the record that was asked for, and it has to say what // the run is doing. `getCloudFlowRun` (cloud-run.ts) already refuses a diff --git a/packages/sdk/src/cloud-run-record.ts b/packages/sdk/src/cloud-run-record.ts new file mode 100644 index 000000000..bc9196258 --- /dev/null +++ b/packages/sdk/src/cloud-run-record.ts @@ -0,0 +1,64 @@ +// What a Cloud run record has to say before this client will act on it. +// +// One validator, two callers. `waitForCloudFlowRun` (cloud-run.ts) blocks on +// it to decide when a hosted `flows run --wait` is over, and the live read +// verbs (cli/cloud-live.ts, through `getCloudRunDetailLive`) block on it to +// decide when `--watch` and `--follow` stop and with which exit code. Both +// have to mean the same thing by "this run is over, and it ended well", or a +// watch would exit 0 on a record the waiter refuses. +// +// Deliberately strict and deliberately separate from the *permissive* +// `getCloudRunDetail` mapping, which exists to show a reader whatever Cloud +// said. Attesting an outcome is a different job to rendering one: a +// `completed` header with no valid `completionReason` attests nothing, and +// must not become exit 0. + +import { RUN_COMPLETION_REASONS } from '@relayflows/surface'; +import { CloudFlowError, isCloudRecord } from './cloud-http.js'; +import type { RunCompletionReason } from './protocol.js'; + +export type CloudRunState = + | { runId: string; status: 'pending' | 'launching' | 'running' } + | { runId: string; status: 'completed' | 'failed' | 'cancelled'; completionReason: RunCompletionReason }; + +/** The statuses that mean the hosted run has not finished. */ +const ACTIVE_RUN_STATUSES: readonly string[] = ['pending', 'launching', 'running']; + +/** + * Is this run still going? + * + * The one vocabulary for the question, so `run --cloud --wait`, + * `status --cloud --watch` and `logs --follow` cannot disagree about what + * `launching` means. Note this is about the *run*: a `needs_human` step is + * not a run status and never reaches here. + */ +export function isCloudRunActive(status: string): boolean { + return ACTIVE_RUN_STATUSES.includes(status); +} + +/** + * Validate a raw run record into an execution outcome, or refuse. + * + * Takes the body as Cloud answered it so a caller that also wants the + * renderable projection can make one request and validate the same bytes it + * rendered, rather than issuing a second GET per poll. + */ +export function cloudRunState(body: unknown, runId: string): CloudRunState { + if (!isCloudRecord(body) || body.runId !== runId || body.relayflowVersion !== 'v2' + || typeof body.status !== 'string' + || !['pending', 'launching', 'running', 'completed', 'failed', 'cancelled'].includes(body.status)) { + throw new CloudFlowError('invalid_response', 'Cloud returned an invalid v2 run record.'); + } + if (isCloudRunActive(body.status)) { + return { runId, status: body.status as 'pending' | 'launching' | 'running' }; + } + const report = body.result; + const reason = isCloudRecord(report) ? report.completionReason : undefined; + if (typeof reason !== 'string' || !(RUN_COMPLETION_REASONS as readonly string[]).includes(reason) + || (body.status === 'completed' && (reason !== 'success' || !isCloudRecord(report) || report.ok !== true || report.status !== 'completed')) + || (body.status === 'failed' && !['step_failed', 'budget_exceeded'].includes(reason)) + || (body.status === 'cancelled' && reason !== 'canceled')) { + throw new CloudFlowError('invalid_response', 'Cloud terminal record lacks a valid, consistent run completionReason; no execution outcome is attested.'); + } + return { runId, status: body.status as 'completed' | 'failed' | 'cancelled', completionReason: reason as RunCompletionReason }; +} diff --git a/packages/sdk/src/cloud-run.ts b/packages/sdk/src/cloud-run.ts index fb92081e9..ef4c29307 100644 --- a/packages/sdk/src/cloud-run.ts +++ b/packages/sdk/src/cloud-run.ts @@ -1,5 +1,4 @@ -import { RUN_COMPLETION_REASONS, type ScheduleTriggerSource } from '@relayflows/surface'; -import type { RunCompletionReason } from './protocol.js'; +import { type ScheduleTriggerSource } from '@relayflows/surface'; import { readFile } from 'node:fs/promises'; import { createHash } from 'node:crypto'; import { setTimeout } from 'node:timers/promises'; @@ -14,6 +13,7 @@ import { CloudFlowError, cloudConnection, cloudFetch, cloudRequest, cloudRunId, isCloudRecord, type CloudConnectionOptions, } from './cloud-http.js'; +import { cloudRunState, isCloudRunActive, type CloudRunState } from './cloud-run-record.js'; import { packWorkingTree, prepareCloudSync, uploadCloudCode } from './cloud-sync.js'; export type CloudFlowSource = FlowSpec | { path: string }; @@ -66,9 +66,7 @@ export function cloudSubmissionBody(submission: CloudSubmission): Record { cloudRunId(runId); - const result = await cloudRequest(`/api/v1/workflows/runs/${runId}`, options); - if (!isCloudRecord(result) || result.runId !== runId || result.relayflowVersion !== 'v2' - || typeof result.status !== 'string' - || !['pending', 'launching', 'running', 'completed', 'failed', 'cancelled'].includes(result.status)) { - throw new CloudFlowError('invalid_response', 'Cloud returned an invalid v2 run record.'); - } - if (result.status === 'pending' || result.status === 'launching' || result.status === 'running') { - return { runId, status: result.status }; - } - const report = result.result; - const reason = isCloudRecord(report) ? report.completionReason : undefined; - if (typeof reason !== 'string' || !(RUN_COMPLETION_REASONS as readonly string[]).includes(reason) - || (result.status === 'completed' && (reason !== 'success' || !isCloudRecord(report) || report.ok !== true || report.status !== 'completed')) - || (result.status === 'failed' && !['step_failed', 'budget_exceeded'].includes(reason)) - || (result.status === 'cancelled' && reason !== 'canceled')) { - throw new CloudFlowError('invalid_response', 'Cloud terminal record lacks a valid, consistent run completionReason; no execution outcome is attested.'); - } - return { runId, status: result.status as 'completed' | 'failed' | 'cancelled', completionReason: reason as RunCompletionReason }; + return cloudRunState(await cloudRequest(`/api/v1/workflows/runs/${runId}`, options), runId); } /** Wait without a fixed execution deadline. Abort stops observation, not the hosted run. */ @@ -304,7 +285,7 @@ export async function waitForCloudFlowRun( try { const run = await getCloudFlowRun(runId, options); failures = 0; - if (run.status !== 'pending' && run.status !== 'launching' && run.status !== 'running') return run; + if (!isCloudRunActive(run.status)) return run; } catch (error) { options.signal?.throwIfAborted(); const transient = error instanceof CloudFlowError && (error.code === 'transient_error' diff --git a/packages/sdk/tests/cloud-live.test.ts b/packages/sdk/tests/cloud-live.test.ts new file mode 100644 index 000000000..e24babe03 --- /dev/null +++ b/packages/sdk/tests/cloud-live.test.ts @@ -0,0 +1,743 @@ +// `flows status --cloud --watch` and `flows logs --follow` against a +// fake Cloud that changes its mind between polls. +// +// The fixtures keep `cloud-read.test.ts`'s vocabulary — the camelCase field +// names the real routes emit — and add the shapes only a live run has: a step +// row with a `startTime` and no `endTime`, a log envelope that is not `done`, +// and a run record that answers `running` for the first polls and `completed` +// after. Every test injects the clock and the sleep, so nothing here waits. + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { runCli } from '../src/cli.js'; +import { runCloudLogsFollow, runCloudStatusWatch } from '../src/cli/cloud-live.js'; +import { runCloudStatusCli, parseLogsArgs } from '../src/cli/cloud-read.js'; +import { parseStatusArgs } from '../src/cli/status.js'; + +const RUN = '20d04c99-3fa8-48c9-9286-92d364a5bc2e'; +const CONNECTION = { apiUrl: 'https://cloud-contract.example', token: 'test-scoped-cloud-token', env: {} }; +const STARTED = '2026-09-19T20:34:58.286Z'; +/** `startTime` of the live `agent-5` row; `NOW` is 6m50s after it. */ +const STEP_STARTED = '2026-09-19T20:35:10.000Z'; +const NOW = Date.parse(STEP_STARTED) + 410_000; + +afterEach(() => { + vi.unstubAllEnvs(); + vi.restoreAllMocks(); +}); + +interface Answer { status?: number; body: unknown } +/** One poll's worth of answers: the run record, its step rows, its runner log. */ +interface Stage { run: Answer; steps?: Answer; log?: Answer } + +function io(tty = false) { + const stdout: string[] = []; + const stderr: string[] = []; + // A watched frame arrives as one `stdout` call, so `stdout[n]` is the nth + // frame and not the nth line; tests that care about lines split it. + return { + stdout, stderr, + io: { stdout: (line: string) => stdout.push(line), stderr: (line: string) => stderr.push(line), tty }, + }; +} + +/** + * A fake Cloud that advances one stage per run-record read. + * + * The run record is the first request of every tick, so consuming a stage + * there makes "the third poll" expressible; the steps and logs routes answer + * from the stage the current tick is on. The last stage repeats, so a test + * only writes the transitions it cares about. + */ +function stagedCloud(stages: readonly Stage[], onRequest?: (path: string, index: number) => void) { + const requests: { path: string; query: string }[] = []; + let index = -1; + vi.spyOn(globalThis, 'fetch').mockImplementation(async (input) => { + const url = new URL(String(input)); + requests.push({ path: url.pathname, query: url.search }); + if (url.pathname.endsWith('/steps')) onRequest?.('steps', Math.max(index, 0)); + else if (url.pathname.endsWith('/logs')) onRequest?.('logs', Math.max(index, 0)); + else { + index = Math.min(index + 1, stages.length - 1); + onRequest?.('run', index); + } + const stage = stages[Math.max(index, 0)]!; + const answer = url.pathname.endsWith('/steps') ? stage.steps ?? { body: { steps: [] } } + : url.pathname.endsWith('/logs') ? stage.log ?? { body: { content: '', offset: 0, totalSize: 0, done: false } } + : stage.run; + return new Response(JSON.stringify(answer.body), { + status: answer.status ?? 200, headers: { 'content-type': 'application/json' }, + }); + }); + return { requests }; +} + +function runRecord(overrides: Record = {}): Answer { + return { + body: { + runId: RUN, relayflowVersion: 'v2', fileType: 'ts', status: 'running', + workflow: '{"name":"insight-proof-2022"}', createdAt: STARTED, updatedAt: STARTED, + ...overrides, + }, + }; +} + +const RUNNING = runRecord(); +const COMPLETED = runRecord({ + status: 'completed', + result: { ok: true, command: 'run', completedSteps: 3, status: 'completed', completionReason: 'success' }, +}); +const FAILED = runRecord({ + status: 'failed', error: 'Relayflow v2 CLI failed with exit code 1', + result: { ok: false, command: 'run', status: 'failed', completionReason: 'step_failed' }, +}); +const CANCELLED = runRecord({ + status: 'cancelled', result: { ok: false, command: 'run', status: 'cancelled', completionReason: 'canceled' }, +}); + +function stepRow(overrides: Record = {}) { + return { + stepName: 'agent-5', stepType: 'agent', status: 'running', sandboxId: '', + startTime: STEP_STARTED, endTime: null, retryCount: 0, + detail: { attempts: [] }, + ...overrides, + }; +} + +function stepsBody(rows: readonly unknown[]): Answer { + return { body: { steps: rows } }; +} + +function logBody(content: string, overrides: { done?: boolean; totalSize?: number; offset?: number } = {}): Answer { + const bytes = Buffer.byteLength(content, 'utf8'); + return { + body: { + content, + offset: overrides.offset ?? bytes, + totalSize: overrides.totalSize ?? bytes, + done: overrides.done ?? false, + }, + }; +} + +/** Injected sleep: records what would have been waited, waits for nothing. */ +function recorder() { + const delays: number[] = []; + return { delays, sleep: async (ms: number): Promise => { delays.push(ms); } }; +} + +function live(extra: Record = {}) { + return { ...CONNECTION, pollIntervalMs: 2_000, now: () => NOW, sleep: recorder().sleep, ...extra }; +} + +describe('live step rows', () => { + it('renders running, backoff, waiting and needs_human beside a finished step', async () => { + stagedCloud([{ + run: RUNNING, + steps: stepsBody([ + stepRow({ + stepName: 'run-1', stepType: 'deterministic', status: 'completed', completionReason: 'success', + endTime: '2026-09-19T20:35:09.000Z', durationMs: 3, + detail: { attempts: [{ attempt: 1, disposition: 'step_done' }], wallclockMs: 3 }, + }), + stepRow(), + stepRow({ stepName: 'agent-6', status: 'backoff', retryCount: 1 }), + stepRow({ stepName: 'human-7', stepType: 'deterministic', status: 'waiting' }), + stepRow({ stepName: 'human-8', stepType: 'deterministic', status: 'needs_human' }), + ]), + }]); + const out = io(); + expect(await runCloudStatusCli({ runId: RUN, json: false }, out.io, { ...CONNECTION, now: () => NOW })).toBe(0); + const rendered = out.stdout.join('\n'); + expect(out.stdout[1]).toBe('steps 5: 1 completed · 1 running · 1 backoff · 1 waiting · 1 needs_human'); + expect(rendered).toContain('↻ agent-5 agent running attempt 1 6m50s'); + expect(rendered).toContain('↻ agent-6 agent backoff attempt 2 6m50s'); + expect(rendered).toContain('⏸ human-7 deterministic waiting attempt 1 6m50s'); + expect(rendered).toContain('⏸ human-8 deterministic needs_human attempt 1 6m50s'); + expect(rendered).toContain('✓ run-1 deterministic completed 1 attempt 0.0s success'); + // Cloud's step rows carry no maximum-attempt budget, so none is printed. + expect(rendered).not.toContain('attempt 1/'); + }); + + it('counts the attempt now running, not the completed attempt records', async () => { + // The array holds completion records: during a second attempt it still + // has one entry, and `attempts.length` would print `attempt 1`. + stagedCloud([{ + run: RUNNING, + steps: stepsBody([stepRow({ + retryCount: 1, + detail: { attempts: [{ attempt: 1, disposition: 'retry', completionReason: 'step_failed' }] }, + })]), + }]); + const out = io(); + await runCloudStatusCli({ runId: RUN, json: false }, out.io, { ...CONNECTION, now: () => NOW }); + expect(out.stdout.join('\n')).toContain('running attempt 2'); + }); + + it('omits the attempt and the elapsed time a snapshot does not establish', async () => { + stagedCloud([{ + run: RUNNING, + steps: stepsBody([ + // Queued: no dispatch evidence at all, so no attempt is claimed. + stepRow({ stepName: 'pending-1', status: 'pending', startTime: null, retryCount: null }), + // Waiting, but the row carries no retry count: the number is unknown. + stepRow({ stepName: 'wait-2', status: 'waiting', retryCount: null }), + // A start Cloud wrote in a shape this client cannot parse. + stepRow({ stepName: 'odd-3', startTime: 'yesterday' }), + // Clock skew: a start in the future is zero elapsed, never negative. + stepRow({ stepName: 'skew-4', startTime: new Date(NOW + 60_000).toISOString() }), + ]), + }]); + const out = io(); + await runCloudStatusCli({ runId: RUN, json: false }, out.io, { ...CONNECTION, now: () => NOW }); + const rendered = out.stdout.join('\n'); + expect(rendered).toContain('○ pending-1 agent pending'); + expect(rendered).not.toMatch(/pending-1.*attempt/u); + expect(rendered).toContain('⏸ wait-2 agent waiting'); + expect(rendered).not.toMatch(/wait-2.*attempt/u); + expect(rendered).toContain('↻ odd-3 agent running attempt 1\n'); + expect(rendered).toContain('↻ skew-4 agent running attempt 1 0.0s'); + }); + + it('does not advance a finished step whose end timestamp is missing', async () => { + stagedCloud([{ + run: COMPLETED, + steps: stepsBody([stepRow({ + status: 'completed', completionReason: 'success', endTime: null, durationMs: 9252, + detail: { attempts: [{ attempt: 1 }], wallclockMs: 9252 }, + })]), + }]); + const out = io(); + await runCloudStatusCli({ runId: RUN, json: false }, out.io, { ...CONNECTION, now: () => NOW }); + expect(out.stdout.join('\n')).toContain('completed 1 attempt 9.3s'); + expect(out.stdout.join('\n')).not.toContain('6m50s'); + }); + + it('says a running run has no snapshot rather than that it has no steps', async () => { + stagedCloud([{ run: RUNNING, steps: stepsBody([]) }]); + const running = io(); + await runCloudStatusCli({ runId: RUN, json: false }, running.io, { ...CONNECTION, now: () => NOW }); + expect(running.stdout[1]).toBe('steps 0'); + expect(running.stdout[2]).toBe('No step snapshot available yet.'); + + stagedCloud([{ run: COMPLETED, steps: stepsBody([]) }]); + const finished = io(); + await runCloudStatusCli({ runId: RUN, json: false }, finished.io, { ...CONNECTION, now: () => NOW }); + expect(finished.stdout.join('\n')).not.toContain('No step snapshot available yet.'); + }); +}); + +describe('flows status --cloud --watch', () => { + it('redraws each poll, prints the terminal page once and exits with the run', async () => { + const server = stagedCloud([ + { run: RUNNING, steps: stepsBody([stepRow()]) }, + { run: RUNNING, steps: stepsBody([stepRow()]) }, + { + run: COMPLETED, + steps: stepsBody([stepRow({ + status: 'completed', completionReason: 'success', endTime: '2026-09-19T20:42:00.000Z', durationMs: 410_000, + detail: { attempts: [{ attempt: 1 }], wallclockMs: 410_000 }, + })]), + }, + ]); + const out = io(); + const clock = recorder(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ sleep: clock.sleep }))).toBe(0); + + // One frame per poll, each a single write; the last is the terminal page. + expect(out.stdout).toHaveLength(3); + expect(out.stdout[0]).toContain('running'); + expect(out.stdout[2]).toContain('completed'); + expect(out.stdout[2]).toContain('finished success'); + expect(out.stdout.filter((frame) => frame.includes('finished success'))).toHaveLength(1); + // Detail then steps, per tick, and nothing after the terminal page. + expect(server.requests.map((request) => request.path)).toEqual([ + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/steps`, + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/steps`, + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/steps`, + ]); + expect(clock.delays).toEqual([2_000, 2_000]); + expect(out.stderr).toEqual([]); + }); + + it('advances a running step’s elapsed time between frames', async () => { + stagedCloud([{ run: RUNNING, steps: stepsBody([stepRow()]) }, { run: COMPLETED, steps: stepsBody([stepRow()]) }]); + let tick = 0; + const out = io(); + await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ now: () => NOW + (tick++) * 60_000 })); + expect(out.stdout[0]).toContain('6m50s'); + expect(out.stdout[1]).toContain('7m50s'); + }); + + it('draws one page on a run that is already terminal, and polls once', async () => { + const server = stagedCloud([{ run: COMPLETED, steps: stepsBody([]) }]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live())).toBe(0); + expect(out.stdout).toHaveLength(1); + expect(server.requests).toHaveLength(2); + }); + + it('exits 1 on a failed run and on a cancelled one, with the failure on the page', async () => { + stagedCloud([{ run: FAILED, steps: stepsBody([]) }]); + const failed = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, failed.io, live())).toBe(1); + expect(failed.stdout[0]).toContain('failed'); + expect(failed.stdout[0]).toContain('Relayflow v2 CLI failed with exit code 1'); + + stagedCloud([{ run: CANCELLED, steps: stepsBody([]) }]); + const cancelled = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, cancelled.io, live())).toBe(1); + expect(cancelled.stdout[0]).toContain('cancelled'); + }); + + it('keeps watching a run whose step is waiting on a human', async () => { + // `needs_human` is a step state, not a run status: the run is still going. + const server = stagedCloud([ + { run: RUNNING, steps: stepsBody([stepRow({ status: 'needs_human' })]) }, + { run: COMPLETED, steps: stepsBody([stepRow({ status: 'completed', completionReason: 'success', durationMs: 1 })]) }, + ]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live())).toBe(0); + expect(out.stdout).toHaveLength(2); + expect(server.requests).toHaveLength(4); + }); + + it('clears the screen for a terminal and appends whole pages when redirected', async () => { + stagedCloud([{ run: RUNNING, steps: stepsBody([]) }, { run: COMPLETED, steps: stepsBody([]) }]); + const terminal = io(true); + await runCloudStatusWatch({ runId: RUN, json: false }, terminal.io, live()); + expect(terminal.stdout[0]!.startsWith('\u001b[2J\u001b[H')).toBe(true); + + stagedCloud([{ run: RUNNING, steps: stepsBody([]) }, { run: COMPLETED, steps: stepsBody([]) }]); + const redirected = io(false); + await runCloudStatusWatch({ runId: RUN, json: false }, redirected.io, live()); + expect(redirected.stdout.join('')).not.toContain('\u001b'); + expect(redirected.stdout[0]!.startsWith(`RUN ${RUN}`)).toBe(true); + }); + + it('redacts the run name and error in every frame', async () => { + const env = { CI_TOKEN: 'supersecrettokenvalue' }; + stagedCloud([ + { run: runRecord({ workflow: '{"name":"flow rk_live_NOTAREALSECRET1234"}' }), steps: stepsBody([]) }, + { + run: runRecord({ + status: 'failed', error: 'died holding supersecrettokenvalue', + workflow: '{"name":"flow rk_live_NOTAREALSECRET1234"}', + result: { ok: false, status: 'failed', completionReason: 'step_failed' }, + }), + steps: stepsBody([]), + }, + ]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live({ env }))).toBe(1); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('rk_live_NOTAREALSECRET1234'); + expect(rendered).not.toContain('supersecrettokenvalue'); + expect(rendered).toContain('[redacted]'); + expect(rendered).toContain('[redacted:CI_TOKEN]'); + }); +}); + +describe('outcome integrity', () => { + it('refuses a completed record that attests no outcome, rather than exiting 0', async () => { + for (const body of [ + runRecord({ status: 'completed' }), + runRecord({ status: 'completed', result: { ok: true, status: 'completed', completionReason: 'step_failed' } }), + runRecord({ status: 'failed', result: { ok: false, status: 'failed', completionReason: 'success' } }), + ]) { + stagedCloud([{ run: body, steps: stepsBody([]) }]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live())).toBe(1); + expect(out.stdout).toEqual([]); + expect(out.stderr[0]).toContain('REFUSED [cloud_invalid_response]'); + expect(out.stderr[0]).toContain('no execution outcome is attested'); + } + }); + + it('refuses a run status this client does not know rather than calling it terminal', async () => { + stagedCloud([{ run: runRecord({ status: 'cancelling' }), steps: stepsBody([]) }]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live())).toBe(1); + expect(out.stderr[0]).toContain('REFUSED [cloud_invalid_response]'); + expect(out.stdout).toEqual([]); + }); + + it('refuses a record for a different run', async () => { + stagedCloud([{ run: runRecord({ runId: 'some-other-run' }), steps: stepsBody([]) }]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, live())).toBe(1); + expect(out.stderr[0]).toContain('REFUSED [cloud_invalid_response]'); + }); +}); + +describe('flows logs --follow', () => { + const HEADER = new RegExp(`^LOG ${RUN} runner following`, 'u'); + + it('prints each line once as it arrives, then the run’s footer', async () => { + const first = '[relayflow] ▶ agent-5 (agent) started\n'; + const second = first + '[relayflow] ✓ agent-5 … done in 5m16s · claude-opus-5 · $2.26\n'; + const server = stagedCloud([ + { run: RUNNING, log: logBody(first) }, + { run: RUNNING, log: logBody(second) }, + { run: COMPLETED, log: logBody(second, { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout[0]).toMatch(HEADER); + expect(out.stdout.slice(1)).toEqual([ + '[relayflow] ▶ agent-5 (agent) started', + '[relayflow] ✓ agent-5 … done in 5m16s · claude-opus-5 · $2.26', + `COMPLETED ${RUN} completionReason: success`, + ]); + expect(out.stdout.filter((line) => line.startsWith('LOG '))).toHaveLength(1); + expect(server.requests.map((request) => request.path)).toEqual([ + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, + ]); + // The runner log is followed whole; no offset is guessed at. + expect(server.requests.map((request) => request.query)).toEqual(['', '', '', '', '', '']); + }); + + it('preserves blank lines and repeats, and prints nothing for a poll that added nothing', async () => { + const content = 'one\n\none\n'; + stagedCloud([ + { run: RUNNING, log: logBody(content) }, + { run: RUNNING, log: logBody(content) }, + { run: COMPLETED, log: logBody(content, { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1, -1)).toEqual(['one', '', 'one']); + }); + + it('holds an incomplete line until its newline arrives, and flushes the last one once', async () => { + stagedCloud([ + { run: RUNNING, log: logBody('half') }, + { run: RUNNING, log: logBody('half a line\nno newline here') }, + { run: COMPLETED, log: logBody('half a line\nno newline here', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1)).toEqual([ + 'half a line', 'no newline here', `COMPLETED ${RUN} completionReason: success`, + ]); + }); + + it('drops the carriage return of a CRLF log rather than overwriting each line', async () => { + stagedCloud([{ run: COMPLETED, log: logBody('alpha\r\nbeta\r\n', { done: true }) }]); + const out = io(); + await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live()); + expect(out.stdout.slice(1, -1)).toEqual(['alpha', 'beta']); + }); + + it('says a terminal run’s log is empty rather than printing nothing at all', async () => { + stagedCloud([{ run: COMPLETED, log: logBody('', { done: true }) }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout[1]).toBe(' (empty)'); + }); + + it('follows a log whose bytes outnumber its characters', async () => { + // `totalSize` is a byte count and `content` is a string: comparing the two + // by `content.length` would leave a drained log looking unfinished. + const content = '▶ résumé 🙂 done\n'; + expect(Buffer.byteLength(content, 'utf8')).toBeGreaterThan(content.length); + stagedCloud([{ run: COMPLETED, log: logBody(content, { done: true }) }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout[1]).toBe('▶ résumé 🙂 done'); + }); + + it('keeps following while the log claims bytes it has not served', async () => { + const tail = 'first\nsecond\n'; + const server = stagedCloud([ + { run: COMPLETED, log: logBody('first\n', { done: true, totalSize: Buffer.byteLength(tail, 'utf8') }) }, + { run: COMPLETED, log: logBody(tail, { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1, -1)).toEqual(['first', 'second']); + expect(server.requests).toHaveLength(4); + }); + + it('keeps following a done log while the run is still going, and a live log after it ends', async () => { + const server = stagedCloud([ + // `done` with an active run: the log upload finished, the run did not. + { run: RUNNING, log: logBody('a\n', { done: true }) }, + // Terminal with `done: false`: the tail is still being published. + { run: COMPLETED, log: logBody('a\n', { done: false }) }, + { run: COMPLETED, log: logBody('a\nb\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1, -1)).toEqual(['a', 'b']); + expect(server.requests).toHaveLength(6); + }); + + it('prints a finished failed run’s whole log and exits 1', async () => { + stagedCloud([{ run: FAILED, log: logBody('boom\n', { done: true }) }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(1); + expect(out.stdout.slice(1)).toEqual(['boom', `FAILED ${RUN} completionReason: step_failed`]); + }); + + it('never lets a secret split across two polls reach stdout', async () => { + const env = { CI_TOKEN: 'supersecrettokenvalue' }; + stagedCloud([ + { run: RUNNING, log: logBody('exported rk_live_NOTARE') }, + { run: RUNNING, log: logBody('exported rk_live_NOTAREALSECRET1234 and supersecret') }, + { run: COMPLETED, log: logBody('exported rk_live_NOTAREALSECRET1234 and supersecrettokenvalue\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live({ env }))).toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('rk_live_NOTAREALSECRET1234'); + expect(rendered).not.toContain('supersecrettokenvalue'); + expect(rendered).toContain('[redacted]'); + expect(rendered).toContain('[redacted:CI_TOKEN]'); + }); + + it('refuses a log that no longer begins with what was already shown', async () => { + stagedCloud([ + { run: RUNNING, log: logBody('alpha\nbeta\n') }, + { run: RUNNING, log: logBody('rewritten\n') }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(1); + expect(out.stderr[0]).toContain('REFUSED [cloud_log_rewritten]'); + expect(out.stderr[0]).toContain('would skip or repeat output'); + expect(out.stdout.slice(1)).toEqual(['alpha', 'beta']); + }); + + it('refuses `--follow --step` before it sends a request', async () => { + const server = stagedCloud([{ run: RUNNING }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: 'agent-5', json: false }, out.io, live())).toBe(2); + expect(server.requests).toEqual([]); + expect(out.stderr[0]).toContain('REFUSED [invalid_invocation]'); + expect(out.stderr[0]).toContain('cannot follow a step transcript'); + expect(out.stderr[0]).toContain(`flows logs --step `); + }); +}); + +describe('refusals and retries', () => { + it('refuses a run this credential cannot see, printing no page', async () => { + stagedCloud([{ run: { status: 404, body: { error: 'Run not found' } } }]); + const watch = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, watch.io, live())).toBe(2); + expect(watch.stdout).toEqual([]); + expect(watch.stderr[0]).toContain('REFUSED [cloud_run_not_found]'); + + stagedCloud([{ run: { status: 401, body: { error: 'Unauthorized' } } }]); + const follow = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, follow.io, live())).toBe(2); + expect(follow.stdout).toEqual([]); + expect(follow.stderr[0]).toContain('REFUSED [cloud_auth_rejected]'); + }); + + it('retries a 503 with a growing, capped delay and resets after a good tick', async () => { + const clock = recorder(); + stagedCloud([ + { run: { status: 503, body: { error: 'busy' } } }, + { run: { status: 503, body: { error: 'busy' } } }, + { run: { status: 503, body: { error: 'busy' } } }, + { run: RUNNING, steps: stepsBody([]) }, + { run: { status: 503, body: { error: 'busy' } } }, + { run: COMPLETED, steps: stepsBody([]) }, + ]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ sleep: clock.sleep, pollIntervalMs: 20_000 }))).toBe(0); + // 20s, 40s, then the 30s cap; then the interval after a good tick, and + // the first backoff again — the failure count reset with the good tick. + expect(clock.delays).toEqual([20_000, 30_000, 30_000, 20_000, 20_000]); + expect(out.stdout).toHaveLength(2); + expect(out.stderr).toEqual([]); + }); + + it('leaves the previous page and the consumed log alone when a poll fails', async () => { + stagedCloud([ + { run: RUNNING, log: logBody('alpha\n') }, + { run: { status: 500, body: { error: 'boom' } } }, + { run: COMPLETED, log: logBody('alpha\nbeta\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1)).toEqual(['alpha', 'beta', `COMPLETED ${RUN} completionReason: success`]); + expect(out.stderr).toEqual([]); + }); + + it('retries a transient failure while draining the final tail rather than dropping it', async () => { + stagedCloud([ + { run: COMPLETED, log: { status: 429, body: { error: 'slow down' } } }, + { run: COMPLETED, log: logBody('tail\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1, -1)).toEqual(['tail']); + }); +}); + +describe('cancellation', () => { + it('stops before the first request when the signal is already aborted', async () => { + const server = stagedCloud([{ run: RUNNING, steps: stepsBody([]) }]); + const controller = new AbortController(); + controller.abort(); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ signal: controller.signal }))).toBe(1); + expect(server.requests).toEqual([]); + expect(out.stdout).toEqual([]); + expect(out.stderr[0]).toBe(`REFUSED [observation_aborted] Stopped watching run ${RUN}; ` + + 'the hosted run has not been cancelled.'); + }); + + it('prints no partial frame when the abort lands between the two reads', async () => { + const controller = new AbortController(); + stagedCloud([{ run: RUNNING, steps: stepsBody([stepRow()]) }], (route) => { + if (route === 'steps') controller.abort(); + }); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ signal: controller.signal }))).toBe(1); + expect(out.stdout).toEqual([]); + expect(out.stderr[0]).toContain('observation_aborted'); + }); + + it('stops during the sleep between frames, leaving the frames already drawn', async () => { + const controller = new AbortController(); + stagedCloud([{ run: RUNNING, steps: stepsBody([stepRow()]) }]); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ signal: controller.signal, sleep: async () => { controller.abort(); } }))).toBe(1); + expect(out.stdout).toHaveLength(1); + expect(out.stderr[0]).toContain('observation_aborted'); + }); + + it('stops a follow without printing a fragment or a footer', async () => { + const controller = new AbortController(); + stagedCloud([{ run: RUNNING, log: logBody('whole\nfrag') }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, + live({ signal: controller.signal, sleep: async () => { controller.abort(); } }))).toBe(1); + expect(out.stdout.slice(1)).toEqual(['whole']); + expect(out.stdout.join('\n')).not.toContain('frag'); + expect(out.stdout.join('\n')).not.toContain('completionReason'); + expect(out.stderr[0]).toBe(`REFUSED [observation_aborted] Stopped following run ${RUN}; ` + + 'the hosted run has not been cancelled.'); + }); + + it('reports an aborted fetch as the abort it was, not as a read failure', async () => { + const controller = new AbortController(); + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + controller.abort(); + throw new DOMException('This operation was aborted', 'AbortError'); + }); + const out = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: false }, out.io, + live({ signal: controller.signal }))).toBe(1); + expect(out.stderr[0]).toContain('observation_aborted'); + }); + + it('installs no process signal handler of its own', async () => { + const controller = new AbortController(); + const before = process.listenerCount('SIGINT'); + stagedCloud([{ run: RUNNING, steps: stepsBody([]) }]); + await runCloudStatusWatch({ runId: RUN, json: false }, io().io, + live({ signal: controller.signal, sleep: async () => { controller.abort(); } })); + expect(process.listenerCount('SIGINT')).toBe(before); + }); +}); + +describe('--json', () => { + it('polls silently and emits one status document at the end', async () => { + stagedCloud([ + { run: RUNNING, steps: stepsBody([stepRow()]) }, + { run: COMPLETED, steps: stepsBody([stepRow({ status: 'completed', completionReason: 'success', durationMs: 1 })]) }, + ]); + const out = io(true); + expect(await runCloudStatusWatch({ runId: RUN, json: true }, out.io, live())).toBe(0); + expect(out.stdout).toHaveLength(1); + expect(out.stdout[0]).not.toContain('\u001b'); + const payload = JSON.parse(out.stdout[0]!) as { v: number; ok: boolean; run: { status: string }; now_ms: number }; + expect(payload).toMatchObject({ v: 1, ok: true, now_ms: NOW }); + expect(payload.run.status).toBe('completed'); + }); + + it('emits one log document carrying the whole redacted content', async () => { + const env = { CI_TOKEN: 'supersecrettokenvalue' }; + stagedCloud([ + { run: RUNNING, log: logBody('one\n') }, + { run: FAILED, log: logBody('one\ntwo supersecrettokenvalue\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: true }, out.io, live({ env }))).toBe(1); + expect(out.stdout).toHaveLength(1); + const payload = JSON.parse(out.stdout[0]!) as { ok: boolean; content: string; done: boolean; step: null }; + expect(payload.ok).toBe(true); + expect(payload.done).toBe(true); + expect(payload.step).toBeNull(); + expect(payload.content).toBe('one\ntwo [redacted:CI_TOKEN]\n'); + }); + + it('emits one refusal document and no partial output when aborted or refused', async () => { + const controller = new AbortController(); + stagedCloud([{ run: RUNNING, log: logBody('one\n') }]); + const stopped = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: true }, stopped.io, + live({ signal: controller.signal, sleep: async () => { controller.abort(); } }))).toBe(1); + expect(stopped.stdout).toHaveLength(1); + expect(JSON.parse(stopped.stdout[0]!)).toMatchObject({ v: 1, ok: false, code: 'observation_aborted' }); + expect(stopped.stderr).toEqual([]); + + stagedCloud([{ run: { status: 404, body: { error: 'Run not found' } } }]); + const refused = io(); + expect(await runCloudStatusWatch({ runId: RUN, json: true }, refused.io, live())).toBe(2); + expect(refused.stdout).toHaveLength(1); + expect(JSON.parse(refused.stdout[0]!)).toMatchObject({ v: 1, ok: false, code: 'cloud_run_not_found' }); + }); +}); + +describe('argv and wiring', () => { + it('accepts the two flags, and refuses the invocations that contradict', () => { + expect(parseStatusArgs(['--cloud', '--watch', RUN])).toMatchObject({ cloud: true, watch: true, runId: RUN }); + // Local status opens one journal file and returns; there is no loop to watch. + expect(parseStatusArgs(['--watch', RUN])).toBeUndefined(); + expect(parseStatusArgs(['--cloud', '--watch', '--watch', RUN])).toBeUndefined(); + expect(parseStatusArgs(['--cloud', '--watch'])).toBeUndefined(); + expect(parseStatusArgs(['--cloud', '--watch', '--tail', '5', RUN])).toBeUndefined(); + expect(parseStatusArgs(['--cloud', '--watch', '--data-dir', '/tmp/x', RUN])).toBeUndefined(); + + expect(parseLogsArgs([RUN, '--follow'])).toMatchObject({ runId: RUN, follow: true, step: undefined }); + expect(parseLogsArgs([RUN, '--follow', '--follow'])).toBeUndefined(); + expect(parseLogsArgs(['--follow'])).toBeUndefined(); + }); + + it('routes both live invocations through runCli, and refuses a missing run id', async () => { + vi.stubEnv('FLOWS_CLOUD_URL', 'https://cloud-contract.example'); + vi.stubEnv('FLOWS_CLOUD_TOKEN', 'test-scoped-cloud-token'); + + stagedCloud([{ run: COMPLETED, steps: stepsBody([]) }]); + const watched = io(); + expect(await runCli(['status', '--cloud', '--watch', '--json', RUN], watched.io)).toBe(0); + expect(JSON.parse(watched.stdout[0]!)).toMatchObject({ ok: true }); + + stagedCloud([{ run: FAILED, log: logBody('x\n', { done: true }) }]); + const followed = io(); + expect(await runCli(['logs', '--follow', RUN], followed.io)).toBe(1); + expect(followed.stdout.at(-1)).toBe(`FAILED ${RUN} completionReason: step_failed`); + + const refused = io(); + expect(await runCloudStatusWatch({ json: false }, refused.io, live())).toBe(2); + expect(refused.stderr[0]).toContain('REFUSED [run_unknown]'); + }); + + it('lists both flags in --help', async () => { + const help = io(); + expect(await runCli(['--help'], help.io)).toBe(0); + expect(help.stdout[0]).toContain('flows status --cloud [--json] [--watch] '); + expect(help.stdout[0]).toContain('flows logs [--step ] [--raw] [--json] [--follow] '); + }); +}); diff --git a/packages/sdk/tests/cloud-read.test.ts b/packages/sdk/tests/cloud-read.test.ts index 548772530..76cf2df36 100644 --- a/packages/sdk/tests/cloud-read.test.ts +++ b/packages/sdk/tests/cloud-read.test.ts @@ -11,8 +11,9 @@ import { join } from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { runCli } from '../src/cli.js'; import { - errorLines, parseLogsArgs, parseRunsArgs, runCloudLogsCli, runCloudRunsCli, runCloudStatusCli, + parseLogsArgs, parseRunsArgs, runCloudLogsCli, runCloudRunsCli, runCloudStatusCli, } from '../src/cli/cloud-read.js'; +import { errorLines } from '../src/cli/cloud-format.js'; import { parseStatusArgs } from '../src/cli/status.js'; const RUN = '20d04c99-3fa8-48c9-9286-92d364a5bc2e'; @@ -657,9 +658,9 @@ describe('argv', () => { expect(parseRunsArgs(['--limit', '0'])).toBeUndefined(); expect(parseRunsArgs(['--limit'])).toBeUndefined(); expect(parseRunsArgs(['extra'])).toBeUndefined(); - expect(parseLogsArgs([RUN])).toEqual({ command: 'logs', runId: RUN, step: undefined, raw: false, json: false }); + expect(parseLogsArgs([RUN])).toEqual({ command: 'logs', runId: RUN, step: undefined, raw: false, json: false, follow: false }); expect(parseLogsArgs([RUN, '--step', 'agent-2', '--raw', '--json'])) - .toEqual({ command: 'logs', runId: RUN, step: 'agent-2', raw: true, json: true }); + .toEqual({ command: 'logs', runId: RUN, step: 'agent-2', raw: true, json: true, follow: false }); expect(parseLogsArgs([])).toBeUndefined(); expect(parseLogsArgs([RUN, 'second'])).toBeUndefined(); expect(parseLogsArgs([RUN, '--step'])).toBeUndefined(); @@ -679,8 +680,8 @@ describe('argv', () => { const help = io(); expect(await runCli(['--help'], help.io)).toBe(0); expect(help.stdout[0]).toContain('flows runs [--limit ] [--json]'); - expect(help.stdout[0]).toContain('flows logs [--step ] [--raw] [--json] '); - expect(help.stdout[0]).toContain('flows status --cloud [--json] '); + expect(help.stdout[0]).toContain('flows logs [--step ] [--raw] [--json] [--follow] '); + expect(help.stdout[0]).toContain('flows status --cloud [--json] [--watch] '); }); }); diff --git a/packages/sdk/tests/relay-cli-surface.test.ts b/packages/sdk/tests/relay-cli-surface.test.ts index ef8b02b6f..c626f35cd 100644 --- a/packages/sdk/tests/relay-cli-surface.test.ts +++ b/packages/sdk/tests/relay-cli-surface.test.ts @@ -93,6 +93,8 @@ const INVOCATIONS: readonly { verb: string; argv: readonly string[]; variant: Pa // its flags are all optional, and the bare form is the runner log. { verb: 'logs', argv: ['logs', RUN_ID], variant: 'logs' }, { verb: 'logs', argv: ['logs', '--step', 'agent-2', '--raw', '--json', RUN_ID], variant: 'logs' }, + // `--follow` follows the runner log, so it carries no `--step`. + { verb: 'logs', argv: ['logs', '--follow', '--json', RUN_ID], variant: 'logs' }, { verb: 'observer', argv: ['observer'], variant: 'observer' }, { verb: 'observer', argv: ['observer', '--data-dir', '.relayflowd'], variant: 'observer' }, { verb: 'replay', argv: ['replay', RUN_ID], variant: 'replay' }, @@ -151,6 +153,8 @@ const INVOCATIONS: readonly { verb: string; argv: readonly string[]; variant: Pa // `--cloud` is the same verb against the Cloud API; it takes neither of // the two filesystem flags above, so it needs its own sample. { verb: 'status', argv: ['status', '--cloud', '--json', RUN_ID], variant: 'status' }, + // `--watch` is refused without `--cloud`, so its sample carries both. + { verb: 'status', argv: ['status', '--cloud', '--watch', RUN_ID], variant: 'status' }, { verb: 'sync', argv: ['sync', RUN_ID], variant: 'sync' }, { verb: 'sync', argv: ['sync', '--dry-run', '--json', '--dir', '.', RUN_ID], variant: 'sync' }, { From e191e3dfd2d2ee9c1087f4ab8fbcb64f11f1ce37 Mon Sep 17 00:00:00 2001 From: Relayflow Date: Mon, 21 Sep 2026 22:43:23 +0000 Subject: [PATCH 2/5] flows: redact a followed log in blocks, and let a live row's clock run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on #545. F1 (P1, review.md): `flows logs --follow` split the new content into lines before redacting each one, while the one-shot read redacts the whole content first. A secret env value spanning more than one line — a PEM private key is the usual one — matches no single line, so every line of key material reached stdout. The un-released remainder is now redacted as one block: a cut is released only where no secret value has begun and not ended (`openSecretStart`, which also covers a value whose halves arrive in two different polls) and only where redacting the block alone gives the same text as the front of the whole redacted remainder, so a cut that fell inside something the redactor would have caught moves back a line instead. A line can therefore appear one poll late; none of it can appear unredacted. Cursor Bugbot (inline, medium): `flows status --cloud` preferred a step row's `duration_ms` over the derived elapsed time. A row still in flight can carry the `wallclockMs` of the attempt that already ended, so a watched frame froze its clock at that attempt's duration. The derived time comes first now; `liveElapsed` is null for every row that is not demonstrably in flight, so a finished row still prints what Cloud reported. Co-Authored-By: Claude Opus 5 --- docs/CLOUD.md | 9 +++ packages/sdk/src/cli/cloud-live.ts | 81 ++++++++++++++++------- packages/sdk/src/cli/cloud-status-view.ts | 7 +- packages/sdk/src/redact.ts | 27 ++++++++ packages/sdk/tests/cloud-live.test.ts | 59 +++++++++++++++++ packages/sdk/tests/redact.test.ts | 30 ++++++++- 6 files changed, 187 insertions(+), 26 deletions(-) diff --git a/docs/CLOUD.md b/docs/CLOUD.md index 152cf7eb1..548e17100 100644 --- a/docs/CLOUD.md +++ b/docs/CLOUD.md @@ -295,6 +295,15 @@ benefit is that no line can be duplicated or skipped. If what was already printed is no longer a prefix of what Cloud serves, `--follow` refuses with `cloud_log_rewritten` rather than guessing which bytes are new. +What has not been printed yet is redacted as one block, never a line at a +time, and a line is held back while a secret env value has begun in it and not +ended — whether the rest of that value is further down the same response or +has not been served yet. A multi-line value, a PEM private key being the usual +one, is therefore replaced whole by `[redacted:]`: no line of it can +reach stdout on its own, which is exactly what a line-at-a-time redactor can +never prevent. The cost is that a line can appear one poll later than the byte +that completed it. + `--cloud` takes neither `--data-dir` nor `--tail`: both name things on this filesystem, which a hosted run has none of, so pairing them is refused as an invocation rather than quietly ignored. `--watch` is refused without `--cloud` diff --git a/packages/sdk/src/cli/cloud-live.ts b/packages/sdk/src/cli/cloud-live.ts index 04d05d97f..d3ed51783 100644 --- a/packages/sdk/src/cli/cloud-live.ts +++ b/packages/sdk/src/cli/cloud-live.ts @@ -27,7 +27,7 @@ import { type CloudRunDetail, type CloudRunLog, type CloudStep, } from '../cloud-read.js'; import { isCloudRunActive, type CloudRunState } from '../cloud-run-record.js'; -import { redact } from '../redact.js'; +import { openSecretStart, redact } from '../redact.js'; import { thousands } from './cloud-format.js'; import { fail, isTransientRead, refusalFor, RUN_ID_REQUIRED, type CloudReadOptions } from './cloud-refusal.js'; import { renderCloudStatus, scrubRun, scrubSteps } from './cloud-status-view.js'; @@ -151,40 +151,75 @@ export async function runCloudStatusWatch( // ------------------------------------------------------ flows logs --follow /** - * What has been shown of the runner log, and what is still half a line. + * What has been shown of the runner log, and where the rest starts. * - * Byte and line bookkeeping is kept on the log exactly as Cloud served it, - * before redaction: a secret split across two polls is reassembled here and - * redacted as one whole line, so no chunk boundary can let half a token past - * the redactor. + * Bookkeeping is kept on the log exactly as Cloud served it, before redaction: + * the un-released remainder is redacted as one block on every poll, so a + * secret is matched against the whole text it spans — several lines of a PEM, + * or halves that arrived in two different polls — rather than against one line + * at a time, which no multi-line value can ever match. */ interface LogFollowState { /** The whole log as last served, held to prove the next read extends it. */ consumed: string; - /** A trailing fragment with no newline yet; never printed until it ends. */ - pending: string; + /** How much of `consumed` has been printed; the rest is still redactable. */ + released: number; emitted: boolean; } /** - * Emit whatever whole lines the new content completed. + * Print `text` as lines. * * Splitting is on `\n` only, and one trailing `\r` per line is dropped: a * CRLF log printed with its carriage returns intact makes a terminal overwrite - * each line it just drew. Blank lines and repeated identical lines are + * each line it just drew. A final newline ends the last line rather than + * starting an empty one. Blank lines and repeated identical lines are * preserved — they are the log. */ -function emitLines(fresh: string, state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): void { - state.pending += fresh; - let newline: number; - while ((newline = state.pending.indexOf('\n')) >= 0) { - const line = state.pending.slice(0, newline).replace(/\r$/u, ''); - state.pending = state.pending.slice(newline + 1); - io.stdout(redact(line, env)); +function writeLines(text: string, state: LogFollowState, io: CliIo): void { + const lines = text.split('\n'); + if (lines.at(-1) === '') lines.pop(); + for (const line of lines) { + io.stdout(line.replace(/\r$/u, '')); state.emitted = true; } } +/** The last line boundary at or before `at`: 0, or the index after a newline. */ +function lineBoundaryAtOrBefore(text: string, at: number): number { + return at <= 0 ? 0 : text.lastIndexOf('\n', at - 1) + 1; +} + +/** + * Release every line the log has settled on, redacted as one block. + * + * Two things hold a line back. `openSecretStart` marks where a secret value + * has begun and not ended, so nothing at or past it can be shown until the + * next poll completes it. And a candidate block is released only when + * redacting it alone gives the same text as the front of the whole redacted + * remainder: if the cut fell inside something the redactor would have caught — + * the second line of a private key, a header whose value is on the next line — + * the two disagree, and the cut moves back a line and is tried again. + * + * A poll that can release nothing prints nothing; the bytes are not lost, they + * are held until a later poll or the drain flush can redact them whole. + */ +function releaseLines(state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): void { + const raw = state.consumed; + if (state.released >= raw.length) return; + const remainder = redact(raw.slice(state.released), env); + let cut = lineBoundaryAtOrBefore(raw, openSecretStart(raw, env)); + while (cut > state.released) { + const block = redact(raw.slice(state.released, cut), env); + if (remainder.startsWith(block)) { + writeLines(block, state, io); + state.released = cut; + return; + } + cut = lineBoundaryAtOrBefore(raw, cut - 1); + } +} + /** * Has Cloud finished publishing this log? * @@ -238,7 +273,7 @@ export async function runCloudLogsFollow( args.json, io); } - const state: LogFollowState = { consumed: '', pending: '', emitted: false }; + const state: LogFollowState = { consumed: '', released: 0, emitted: false }; let header = false; let failures = 0; for (;;) { @@ -271,7 +306,6 @@ export async function runCloudLogsFollow( + `\`flows logs ${runId}\`.`, }, args.json, io); } - const fresh = log.content.slice(state.consumed.length); state.consumed = log.content; const terminal = !isCloudRunActive(outcome.status); const drained = logDrained(log, terminal, log.content); @@ -281,7 +315,7 @@ export async function runCloudLogsFollow( io.stdout(`LOG ${runId} runner following ${thousands(log.total_size)} bytes so far`); header = true; } - emitLines(fresh, state, io, env); + releaseLines(state, io, env); } if (!drained) { await sleep(interval, signal); @@ -297,10 +331,9 @@ export async function runCloudLogsFollow( })); return exitFor(outcome); } - if (state.pending.length > 0) { - io.stdout(redact(state.pending.replace(/\r$/u, ''), env)); - state.pending = ''; - state.emitted = true; + if (state.released < state.consumed.length) { + writeLines(redact(state.consumed.slice(state.released), env), state, io); + state.released = state.consumed.length; } if (!state.emitted) io.stdout(' (empty)'); io.stdout(`${outcome.status.toUpperCase()} ${runId} completionReason: ` diff --git a/packages/sdk/src/cli/cloud-status-view.ts b/packages/sdk/src/cli/cloud-status-view.ts index 843ad1703..e00144417 100644 --- a/packages/sdk/src/cli/cloud-status-view.ts +++ b/packages/sdk/src/cli/cloud-status-view.ts @@ -93,7 +93,12 @@ function renderStep(step: CloudStep, runId: string, idWidth: number, now: number ]; const attempt = attemptCell(step); if (attempt !== null) cells.push(attempt); - const elapsed = step.duration_ms ?? liveElapsed(step, now); + // The derived time first, not second: a row that is still in flight can + // carry a `wallclockMs` from the attempt that already ended, and preferring + // it would freeze the cell at that attempt's duration while the step runs + // on. `liveElapsed` is null for every row that is not demonstrably in + // flight, so a finished row still prints the duration Cloud reported. + const elapsed = liveElapsed(step, now) ?? step.duration_ms; if (elapsed !== null) cells.push(formatDuration(elapsed)); if (step.completion_reason !== null) cells.push(safe(step.completion_reason)); if (step.gate !== null) cells.push(`gate: ${safe(step.gate.gate)} ${safe(step.gate.verdict)}`); diff --git a/packages/sdk/src/redact.ts b/packages/sdk/src/redact.ts index 27c48f80c..6c47252b4 100644 --- a/packages/sdk/src/redact.ts +++ b/packages/sdk/src/redact.ts @@ -104,3 +104,30 @@ export function redactRelayError(message: string, env: NodeJS.ProcessEnv = proce } return redact(message, env); } + +/** + * Where a secret value may still be half-arrived at the end of `text`. + * + * A stream is redacted in pieces, and `redact` only replaces a secret it can + * see whole: releasing text up to a point where a secret has begun but not + * ended would print the first half of it and never match the second. This + * returns the smallest index `i` for which `text.slice(i)` is a proper prefix + * of some secret env value — the point past which nothing may be released + * until more of the stream arrives — or `text.length` when no value is open. + * + * Values only, not the token and header *shapes*: those are bounded by + * whitespace, so a reader that releases whole lines cannot cut one in half. + */ +export function openSecretStart(text: string, env: NodeJS.ProcessEnv = process.env): number { + let start = text.length; + for (const [, value] of secretEnvValues(env)) { + // A proper prefix is shorter than the value, so it can only begin inside + // the last `value.length - 1` characters; the first character narrows the + // scan to the few positions worth comparing. + const first = Math.max(0, text.length - value.length + 1); + for (let at = text.indexOf(value[0]!, first); at >= 0 && at < start; at = text.indexOf(value[0]!, at + 1)) { + if (value.startsWith(text.slice(at))) { start = at; break; } + } + } + return start; +} diff --git a/packages/sdk/tests/cloud-live.test.ts b/packages/sdk/tests/cloud-live.test.ts index e24babe03..b54daefc8 100644 --- a/packages/sdk/tests/cloud-live.test.ts +++ b/packages/sdk/tests/cloud-live.test.ts @@ -130,6 +130,22 @@ function live(extra: Record = {}) { } describe('live step rows', () => { + it('advances an in-flight row that carries a finished attempt’s wallclock', async () => { + // A retried step keeps the previous attempt's `wallclockMs` in its detail. + // Printing that as the running row's elapsed time would freeze the cell at + // the attempt that already ended while the new one burns. + stagedCloud([{ + run: RUNNING, + steps: stepsBody([stepRow({ + status: 'backoff', retryCount: 1, + detail: { attempts: [{ attempt: 1, disposition: 'step_failed' }], wallclockMs: 3 }, + })]), + }]); + const out = io(); + expect(await runCloudStatusCli({ runId: RUN, json: false }, out.io, { ...CONNECTION, now: () => NOW })).toBe(0); + expect(out.stdout.join('\n')).toContain('↻ agent-5 agent backoff attempt 2 6m50s'); + }); + it('renders running, backoff, waiting and needs_human beside a finished step', async () => { stagedCloud([{ run: RUNNING, @@ -499,6 +515,49 @@ describe('flows logs --follow', () => { expect(rendered).toContain('[redacted:CI_TOKEN]'); }); + // A PEM in the environment is the shape a line-at-a-time redactor cannot + // catch: no single line of it equals the value, so every line of key + // material goes straight to stdout. The value is fake. + const PEM = '-----BEGIN PRIVATE KEY-----\nFAKE_PRIVATE_KEY_MATERIAL_1234567890\n-----END PRIVATE KEY-----'; + const PEM_ENV = { SERVICE_PRIVATE_KEY: PEM }; + + it('redacts a multiline secret that arrives whole in one response', async () => { + stagedCloud([{ run: COMPLETED, log: logBody(`credential dump:\n${PEM}\n`, { done: true }) }]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live({ env: PEM_ENV }))) + .toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('FAKE_PRIVATE_KEY_MATERIAL_1234567890'); + expect(rendered).not.toContain('BEGIN PRIVATE KEY'); + expect(out.stdout.slice(1, -1)).toEqual(['credential dump:', '[redacted:SERVICE_PRIVATE_KEY]']); + }); + + it('holds a multiline secret’s first lines back until the polls that complete it', async () => { + const head = `credential dump:\n${PEM.split('\n')[0]!}\n`; + const most = `credential dump:\n${PEM.split('\n').slice(0, 2).join('\n')}\n`; + const whole = `credential dump:\n${PEM}\ndone\n`; + // What stdout held when each poll's run record was answered, so a line + // released late is distinguishable from one released at the drain. + const seen: string[][] = []; + const out = io(); + stagedCloud([ + { run: RUNNING, log: logBody(head) }, + { run: RUNNING, log: logBody(most) }, + { run: COMPLETED, log: logBody(whole, { done: true }) }, + ], (path) => { if (path === 'run') seen.push([...out.stdout]); }); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live({ env: PEM_ENV }))) + .toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('FAKE_PRIVATE_KEY_MATERIAL_1234567890'); + expect(rendered).not.toContain('BEGIN PRIVATE KEY'); + expect(rendered).not.toContain('END PRIVATE KEY'); + expect(out.stdout.slice(1, -1)).toEqual(['credential dump:', '[redacted:SERVICE_PRIVATE_KEY]', 'done']); + // The line before the secret is not held hostage by it: it left on the + // first poll, and nothing of the key left before the third. + expect(seen[1]!.slice(1)).toEqual(['credential dump:']); + expect(seen[2]!.slice(1)).toEqual(['credential dump:']); + }); + it('refuses a log that no longer begins with what was already shown', async () => { stagedCloud([ { run: RUNNING, log: logBody('alpha\nbeta\n') }, diff --git a/packages/sdk/tests/redact.test.ts b/packages/sdk/tests/redact.test.ts index e3154fb49..8a732f7c8 100644 --- a/packages/sdk/tests/redact.test.ts +++ b/packages/sdk/tests/redact.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { redact, redactRelayError } from '../src/redact.js'; +import { openSecretStart, redact, redactRelayError } from '../src/redact.js'; const ENV: NodeJS.ProcessEnv = { GITHUB_TOKEN: 'ghp_abcdefghijklmnopqrstuvwxyz0123', @@ -128,3 +128,31 @@ describe('redact — JSON credential fields', () => { expect(redact('{"token":"[redacted]"}', {})).toBe('{"token":"[redacted]"}'); }); }); + +describe('openSecretStart', () => { + const PEM = '-----BEGIN PRIVATE KEY-----\nFAKE_KEY_MATERIAL_0123456789\n-----END PRIVATE KEY-----'; + const ENV_PEM: NodeJS.ProcessEnv = { SERVICE_PRIVATE_KEY: PEM }; + + it('answers the length when nothing is half-arrived', () => { + expect(openSecretStart('nothing to see here\n', ENV_PEM)).toBe(20); + expect(openSecretStart('', ENV_PEM)).toBe(0); + expect(openSecretStart(`already whole: ${PEM}\n`, ENV_PEM)).toBe(16 + PEM.length); + }); + + it('marks where a value has begun and not ended, across lines', () => { + const head = 'dump:\n-----BEGIN PRIVATE KEY-----\n'; + expect(openSecretStart(head, ENV_PEM)).toBe(6); + expect(openSecretStart(`dump:\n${PEM.slice(0, 3)}`, ENV_PEM)).toBe(6); + }); + + it('ignores a value no secret name exports, and one too short to be material', () => { + expect(openSecretStart('dump:\n-----BEGIN PRIVATE KEY-----\n', {})).toBe(34); + expect(openSecretStart('mode is of', { AUTH_MODE: 'off' })).toBe(10); + }); + + it('takes the earliest start when two secrets are open at once', () => { + const env = { A_TOKEN: 'abcdefgh-longer-one', B_TOKEN: 'gh-longer-one-still' }; + // `abcdefgh-` opens at 4; `gh-longer-one-still` would open at 11. + expect(openSecretStart('tailabcdefgh-', env)).toBe(4); + }); +}); From bb389ca6e8f2fc395ae294ca178e00bf7b6e0364 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 22 Sep 2026 22:14:20 -0700 Subject: [PATCH 3/5] fix(sdk): hold incomplete named-credential tails across live-log polls followLog released any trailing partial line unless it matched an environment-secret opener. A poll that split inside a named credential header (authorization:, Bearer, x-callback-token:, x-nightcto-evidence-token:) or inside a JSON credential field emitted the prefix in clear, and the completing poll then bypassed whole-log redaction semantics. Add openCredentialStart() mirroring openSecretStart(): retain the tail whenever it opens one of those contexts without its terminator, and hold a bare trailing 'Bearer' since the scheme alone discloses nothing but its continuation does. Regression tests split each reviewer repro across polls and assert zero secret bytes before the completing poll. --- packages/sdk/src/cli/cloud-live.ts | 12 +++-- packages/sdk/src/redact.ts | 65 +++++++++++++++++++++++++++ packages/sdk/tests/cloud-live.test.ts | 43 ++++++++++++++++++ packages/sdk/tests/redact.test.ts | 53 +++++++++++++++++++++- 4 files changed, 168 insertions(+), 5 deletions(-) diff --git a/packages/sdk/src/cli/cloud-live.ts b/packages/sdk/src/cli/cloud-live.ts index d3ed51783..d6e9b4a08 100644 --- a/packages/sdk/src/cli/cloud-live.ts +++ b/packages/sdk/src/cli/cloud-live.ts @@ -27,7 +27,7 @@ import { type CloudRunDetail, type CloudRunLog, type CloudStep, } from '../cloud-read.js'; import { isCloudRunActive, type CloudRunState } from '../cloud-run-record.js'; -import { openSecretStart, redact } from '../redact.js'; +import { openCredentialStart, openSecretStart, redact } from '../redact.js'; import { thousands } from './cloud-format.js'; import { fail, isTransientRead, refusalFor, RUN_ID_REQUIRED, type CloudReadOptions } from './cloud-refusal.js'; import { renderCloudStatus, scrubRun, scrubSteps } from './cloud-status-view.js'; @@ -193,9 +193,13 @@ function lineBoundaryAtOrBefore(text: string, at: number): number { /** * Release every line the log has settled on, redacted as one block. * - * Two things hold a line back. `openSecretStart` marks where a secret value + * Three things hold a line back. `openSecretStart` marks where a secret value * has begun and not ended, so nothing at or past it can be shown until the - * next poll completes it. And a candidate block is released only when + * next poll completes it. `openCredentialStart` marks a trailing credential + * header — `authorization:`, `Bearer`, a callback-token name, or an + * unterminated credential JSON field — whose value may only arrive on a later + * poll; releasing the header first would leave the value without the context + * the redactor needs. And a candidate block is released only when * redacting it alone gives the same text as the front of the whole redacted * remainder: if the cut fell inside something the redactor would have caught — * the second line of a private key, a header whose value is on the next line — @@ -208,7 +212,7 @@ function releaseLines(state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): const raw = state.consumed; if (state.released >= raw.length) return; const remainder = redact(raw.slice(state.released), env); - let cut = lineBoundaryAtOrBefore(raw, openSecretStart(raw, env)); + let cut = lineBoundaryAtOrBefore(raw, Math.min(openSecretStart(raw, env), openCredentialStart(raw))); while (cut > state.released) { const block = redact(raw.slice(state.released, cut), env); if (remainder.startsWith(block)) { diff --git a/packages/sdk/src/redact.ts b/packages/sdk/src/redact.ts index 6c47252b4..3ee8b48b1 100644 --- a/packages/sdk/src/redact.ts +++ b/packages/sdk/src/redact.ts @@ -105,6 +105,71 @@ export function redactRelayError(message: string, env: NodeJS.ProcessEnv = proce return redact(message, env); } +/** + * Credential heads whose value may continue past a line boundary. + * + * Every NAMED_VALUE pattern puts `\s` between the name and the value, and + * `\s` matches newlines: `authorization:` alone at the end of a poll is a + * complete, unredacted line, while the opaque value that lands on the next + * poll arrives with no recognizable credential context and would print + * verbatim. A trailing header — the colon plus only whitespace, or for + * `authorization` one scheme word — is therefore an open context. A header + * already followed by a non-space word has its value on that line and is not + * open. `Bearer` is the one bare scheme word the patterns treat as a name. + */ +const OPEN_HEADER_PATTERNS: readonly RegExp[] = [ + /\bauthorization:(?:\s+\w+)?\s*$/i, + /\bx-callback-token:\s*$/i, + /\bx-nightcto-evidence-token:\s*$/i, + /\bBearer\s*$/, +]; + +/** A quoted JSON field name; the value side is inspected by the scanner. */ +const JSON_FIELD_HEAD = /"([A-Za-z0-9_.\-]{1,64})"/g; + +/** + * Where a named-credential context may still be half-arrived at the end of + * `text`. + * + * `openSecretStart` covers secret env *values*; this covers the credential + * *names* whose patterns can span a line break: a header released on one poll + * leaves its late-arriving value unrecognizable on the next, and a credential + * JSON field whose value string is not yet closed leaks its first fragment. + * Returns the earliest index at which such a context is open — the point past + * which nothing may be released until more of the stream arrives — or + * `text.length` when none is open. Holding back is line-granular upstream, so + * a false positive delays one line one poll rather than dropping it. + */ +export function openCredentialStart(text: string): number { + let hold = text.length; + for (const pattern of OPEN_HEADER_PATTERNS) { + const match = pattern.exec(text); + if (match && match.index < hold) hold = match.index; + } + JSON_FIELD_HEAD.lastIndex = 0; + let head: RegExpExecArray | null; + while ((head = JSON_FIELD_HEAD.exec(text)) !== null) { + if (!isCredentialName(head[1]!)) continue; + let i = head.index + head[0].length; + while (i < text.length && /\s/.test(text[i]!)) i++; + if (i >= text.length) { hold = Math.min(hold, head.index); continue; } + if (text[i] !== ':') continue; + i++; + while (i < text.length && /\s/.test(text[i]!)) i++; + if (i >= text.length) { hold = Math.min(hold, head.index); continue; } + if (text.startsWith('[redacted]', i) || text[i] !== '"') continue; + i++; + let closed = false; + while (i < text.length) { + if (text[i] === '\\') { i += 2; continue; } + if (text[i] === '"') { closed = true; break; } + i++; + } + if (!closed) hold = Math.min(hold, head.index); + } + return hold; +} + /** * Where a secret value may still be half-arrived at the end of `text`. * diff --git a/packages/sdk/tests/cloud-live.test.ts b/packages/sdk/tests/cloud-live.test.ts index b54daefc8..22c22fe8f 100644 --- a/packages/sdk/tests/cloud-live.test.ts +++ b/packages/sdk/tests/cloud-live.test.ts @@ -515,6 +515,49 @@ describe('flows logs --follow', () => { expect(rendered).toContain('[redacted:CI_TOKEN]'); }); + it.each(['x-callback-token:', 'Bearer', 'authorization:'])( + 'never lets a credential value arrive a poll after its %s header', + async (prefix) => { + stagedCloud([ + { run: RUNNING, log: logBody(`${prefix}\n`) }, + { run: COMPLETED, log: logBody(`${prefix}\nopaque-sensitive-value\n`, { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('opaque-sensitive-value'); + expect(rendered).toContain('[redacted]'); + }, + ); + + it('holds an authorization scheme word whose credential value is still coming', async () => { + stagedCloud([ + { run: RUNNING, log: logBody('authorization: Bearer\n') }, + { run: COMPLETED, log: logBody('authorization: Bearer\ntoken-material\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + const rendered = out.stdout.join('\n'); + // The scheme name stays visible — whole-log redaction keeps it too — but + // the value that completed it on the second poll must never print. + expect(rendered).toContain('Bearer'); + expect(rendered).not.toContain('token-material'); + expect(rendered).toContain('[redacted]'); + }); + + it('never lets a credential JSON field value arrive a poll after its name', async () => { + stagedCloud([ + { run: RUNNING, log: logBody('posting {"authToken": "opa') }, + { run: COMPLETED, log: logBody('posting {"authToken": "opaque-sensitive-value"}\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('opaque-sensitive-value'); + expect(rendered).not.toContain('"opa'); + expect(rendered).toContain('{"authToken": "[redacted]"}'); + }); + // A PEM in the environment is the shape a line-at-a-time redactor cannot // catch: no single line of it equals the value, so every line of key // material goes straight to stdout. The value is fake. diff --git a/packages/sdk/tests/redact.test.ts b/packages/sdk/tests/redact.test.ts index 8a732f7c8..f5417d0c2 100644 --- a/packages/sdk/tests/redact.test.ts +++ b/packages/sdk/tests/redact.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { openSecretStart, redact, redactRelayError } from '../src/redact.js'; +import { openCredentialStart, openSecretStart, redact, redactRelayError } from '../src/redact.js'; const ENV: NodeJS.ProcessEnv = { GITHUB_TOKEN: 'ghp_abcdefghijklmnopqrstuvwxyz0123', @@ -156,3 +156,54 @@ describe('openSecretStart', () => { expect(openSecretStart('tailabcdefgh-', env)).toBe(4); }); }); + +describe('openCredentialStart', () => { + it('answers the length when no credential context is open', () => { + expect(openCredentialStart('plain log line\n')).toBe(15); + expect(openCredentialStart('')).toBe(0); + expect(openCredentialStart('authorization: Bearer tok\n')).toBe(26); + expect(openCredentialStart('x-callback-token: tok\n')).toBe(22); + }); + + it.each([ + 'x-callback-token:', + 'x-callback-token:\n', + 'x-callback-token: \n', + 'x-nightcto-evidence-token:\n', + 'Bearer\n', + 'Bearer \n\n', + 'authorization:', + 'authorization:\n', + 'authorization: Bearer\n', + 'authorization:\nBearer\n', + 'authorization: Bearer \n', + ])('marks a trailing %j header as open at the name', (tail) => { + const text = `ok line\n${tail}`; + expect(openCredentialStart(text)).toBe(8); + }); + + it('does not hold a header whose value already landed on its line', () => { + // `authorization:Bearer` keeps `Bearer` held — a bare scheme word is an + // open context anywhere — which is conservative, not wrong: whole-log + // redaction would emit `authorization:[redacted]` either way. + expect(openCredentialStart('authorization:Bearer\n')).toBe(14); + expect(openCredentialStart('x-callback-token: opaque\n')).toBe(25); + expect(openCredentialStart('Bearer opaque\n')).toBe(14); + expect(openCredentialStart('authorization: Bearer opaque\nrest\n')).toBe(34); + }); + + it('marks an unterminated credential JSON field as open at the name', () => { + expect(openCredentialStart('posting {"authToken": "opa')).toBe(9); + expect(openCredentialStart('posting {"authToken":')).toBe(9); + expect(openCredentialStart('posting {"authToken"')).toBe(9); + expect(openCredentialStart('{"nested": {"password": "sec')).toBe(12); + }); + + it('does not hold closed, redacted, or non-credential JSON fields', () => { + expect(openCredentialStart('{"authToken": "opaque"}\n')).toBe(24); + expect(openCredentialStart('{"authToken": "[redacted]"}\n')).toBe(28); + expect(openCredentialStart('{"authToken": 12')).toBe(16); + expect(openCredentialStart('{"note": "unterminated')).toBe(22); + expect(openCredentialStart('{"monkey": "tail')).toBe(16); + }); +}); From ebd76b1c9a293a606d7cabdc87ad189094b2cc24 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 22 Sep 2026 23:14:52 -0700 Subject: [PATCH 4/5] fix(sdk): close two live-log follow holes from review - require a post-terminal log read before a done snapshot drains the follow, so a stale upload marker cannot truncate the run's last bytes - never end a released block on a bare credential header; its value may begin on the next line, which the block-prefix check cannot see --- packages/sdk/src/cli/cloud-live.ts | 36 ++++++++++++++++++++----- packages/sdk/src/redact.ts | 19 +++++++++++++ packages/sdk/tests/cloud-live.test.ts | 39 ++++++++++++++++++++++++++- 3 files changed, 87 insertions(+), 7 deletions(-) diff --git a/packages/sdk/src/cli/cloud-live.ts b/packages/sdk/src/cli/cloud-live.ts index d6e9b4a08..2f90fd6db 100644 --- a/packages/sdk/src/cli/cloud-live.ts +++ b/packages/sdk/src/cli/cloud-live.ts @@ -27,7 +27,7 @@ import { type CloudRunDetail, type CloudRunLog, type CloudStep, } from '../cloud-read.js'; import { isCloudRunActive, type CloudRunState } from '../cloud-run-record.js'; -import { openCredentialStart, openSecretStart, redact } from '../redact.js'; +import { endsWithOpenCredentialHeader, openCredentialStart, openSecretStart, redact } from '../redact.js'; import { thousands } from './cloud-format.js'; import { fail, isTransientRead, refusalFor, RUN_ID_REQUIRED, type CloudReadOptions } from './cloud-refusal.js'; import { renderCloudStatus, scrubRun, scrubSteps } from './cloud-status-view.js'; @@ -193,17 +193,20 @@ function lineBoundaryAtOrBefore(text: string, at: number): number { /** * Release every line the log has settled on, redacted as one block. * - * Three things hold a line back. `openSecretStart` marks where a secret value + * Four things hold a line back. `openSecretStart` marks where a secret value * has begun and not ended, so nothing at or past it can be shown until the * next poll completes it. `openCredentialStart` marks a trailing credential * header — `authorization:`, `Bearer`, a callback-token name, or an * unterminated credential JSON field — whose value may only arrive on a later * poll; releasing the header first would leave the value without the context - * the redactor needs. And a candidate block is released only when + * the redactor needs. A block is never ended on a bare credential header even + * when more text follows, because the redactor folds the header's newline + * into its own match — the block-prefix check cannot see that the header's + * value lives on the next line. And a candidate block is released only when * redacting it alone gives the same text as the front of the whole redacted * remainder: if the cut fell inside something the redactor would have caught — - * the second line of a private key, a header whose value is on the next line — - * the two disagree, and the cut moves back a line and is tried again. + * the second line of a private key — the two disagree, and the cut moves back + * a line and is tried again. * * A poll that can release nothing prints nothing; the bytes are not lost, they * are held until a later poll or the drain flush can redact them whole. @@ -214,6 +217,19 @@ function releaseLines(state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): const remainder = redact(raw.slice(state.released), env); let cut = lineBoundaryAtOrBefore(raw, Math.min(openSecretStart(raw, env), openCredentialStart(raw))); while (cut > state.released) { + if (endsWithOpenCredentialHeader(raw.slice(state.released, cut))) { + // A bare credential header as the released block's last line strands + // the value that starts on the next line — redact needs them together. + // Take the next line into the block when it is complete; when it is + // not, the header waits with it and the lines before it still release. + const next = raw.indexOf('\n', cut); + if (next === -1) { + cut = lineBoundaryAtOrBefore(raw, cut - 1); + continue; + } + cut = next + 1; + continue; + } const block = redact(raw.slice(state.released, cut), env); if (remainder.startsWith(block)) { writeLines(block, state, io); @@ -280,6 +296,13 @@ export async function runCloudLogsFollow( const state: LogFollowState = { consumed: '', released: 0, emitted: false }; let header = false; let failures = 0; + // A snapshot that satisfies every drain fact can still predate terminality: + // `done` can describe an upload finished while the run kept writing, and the + // log route can keep serving that snapshot past the terminal transition. + // The read that ends the follow has to postdate the terminal observation — + // which on the first sighting it cannot — so a terminal run always gets one + // more poll before its log counts as drained. + let terminalObserved = false; for (;;) { if (isAborted(signal)) return aborted('following', runId, args.json, io); let outcome: CloudRunState; @@ -312,7 +335,8 @@ export async function runCloudLogsFollow( } state.consumed = log.content; const terminal = !isCloudRunActive(outcome.status); - const drained = logDrained(log, terminal, log.content); + const drained = terminalObserved && logDrained(log, terminal, log.content); + terminalObserved ||= terminal; if (!args.json) { if (!header) { diff --git a/packages/sdk/src/redact.ts b/packages/sdk/src/redact.ts index 3ee8b48b1..7f5e2d923 100644 --- a/packages/sdk/src/redact.ts +++ b/packages/sdk/src/redact.ts @@ -170,6 +170,25 @@ export function openCredentialStart(text: string): number { return hold; } +/** + * Does `text` end on a bare credential header — a named credential line whose + * value, if it has one, begins on the NEXT line? + * + * The stream releaser needs this apart from `openCredentialStart`: that check + * sees only a header at the very end of the consumed text, but the danger is + * wider — a released block that *ends* on `x-callback-token:` prints the + * header now and its value, arriving on the next line in some later poll, + * without the context `redact` needs. `redact` folds the header's newline into + * its own named-pattern match, so the releaser's block-prefix check cannot + * detect the dependency on its own. Trailing whitespace-only lines after the + * header count as part of it — the value is still to come. + */ +export function endsWithOpenCredentialHeader(text: string): boolean { + const stripped = text.replace(/\s+$/, ''); + const lastLine = stripped.slice(stripped.lastIndexOf('\n') + 1); + return OPEN_HEADER_PATTERNS.some((pattern) => pattern.test(lastLine)); +} + /** * Where a secret value may still be half-arrived at the end of `text`. * diff --git a/packages/sdk/tests/cloud-live.test.ts b/packages/sdk/tests/cloud-live.test.ts index 22c22fe8f..c916f03db 100644 --- a/packages/sdk/tests/cloud-live.test.ts +++ b/packages/sdk/tests/cloud-live.test.ts @@ -411,9 +411,12 @@ describe('flows logs --follow', () => { `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, + // The first terminal sighting cannot prove the snapshot postdates the + // transition; the drained read is always one poll later. + `/api/v1/workflows/runs/${RUN}`, `/api/v1/workflows/runs/${RUN}/logs`, ]); // The runner log is followed whole; no offset is guessed at. - expect(server.requests.map((request) => request.query)).toEqual(['', '', '', '', '', '']); + expect(server.requests.map((request) => request.query)).toEqual(['', '', '', '', '', '', '', '']); }); it('preserves blank lines and repeats, and prints nothing for a poll that added nothing', async () => { @@ -478,6 +481,21 @@ describe('flows logs --follow', () => { expect(server.requests).toHaveLength(4); }); + it('waits for a post-terminal log snapshot before trusting a done upload', async () => { + // `done: true` on the active poll covered an upload, not the run's last + // bytes; the first terminal read still serves that stale snapshot, and + // draining on it would drop the tail published a poll later. + const server = stagedCloud([ + { run: RUNNING, log: logBody('a\n', { done: true }) }, + { run: COMPLETED, log: logBody('a\n', { done: true }) }, + { run: COMPLETED, log: logBody('a\nb\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + expect(out.stdout.slice(1, -1)).toEqual(['a', 'b']); + expect(server.requests).toHaveLength(6); + }); + it('keeps following a done log while the run is still going, and a live log after it ends', async () => { const server = stagedCloud([ // `done` with an active run: the log upload finished, the run did not. @@ -530,6 +548,25 @@ describe('flows logs --follow', () => { }, ); + it('never releases a credential header line before the line carrying its value completes', async () => { + // Poll two serves the value as an unterminated fragment: releasing the + // header line there leaves the completed value with no context on poll + // three, where it would print verbatim. The header must wait for the + // line that carries its value. + stagedCloud([ + { run: RUNNING, log: logBody('x-callback-token:\n') }, + { run: RUNNING, log: logBody('x-callback-token:\nopaque-sensitive-value') }, + { run: RUNNING, log: logBody('x-callback-token:\nopaque-sensitive-value\n') }, + { run: COMPLETED, log: logBody('x-callback-token:\nopaque-sensitive-value\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('opaque-sensitive-value'); + expect(rendered).toContain('x-callback-token:'); + expect(rendered).toContain('[redacted]'); + }); + it('holds an authorization scheme word whose credential value is still coming', async () => { stagedCloud([ { run: RUNNING, log: logBody('authorization: Bearer\n') }, From be836baf3d5b0eef14fd7728cafff28f037a4058 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 22 Sep 2026 23:47:35 -0700 Subject: [PATCH 5/5] fix(sdk): never grow the release cut past a held credential header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-on findings on the header-hold: the grow-then-recheck could bounce cut between a bare credential header and its next line forever (header + blank line or Bearer + unterminated tail), and the grown cut could pass openSecretStart's bound, releasing a header plus the first line of a still-arriving multiline env secret — leaving the rest of the value unredactable on the next poll. The fix removes growth entirely: a candidate block that ends on a bare credential header simply shrinks until the header is inside the held remainder; it releases together with its value once that line completes. Mutation-verified: buggy version hangs on the unterminated-tail test (exit 124) and prints beta/gamma verbatim on the multiline test. --- packages/sdk/src/cli/cloud-live.ts | 11 +++------ packages/sdk/tests/cloud-live.test.ts | 33 +++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/packages/sdk/src/cli/cloud-live.ts b/packages/sdk/src/cli/cloud-live.ts index 2f90fd6db..ee4252264 100644 --- a/packages/sdk/src/cli/cloud-live.ts +++ b/packages/sdk/src/cli/cloud-live.ts @@ -220,14 +220,9 @@ function releaseLines(state: LogFollowState, io: CliIo, env: NodeJS.ProcessEnv): if (endsWithOpenCredentialHeader(raw.slice(state.released, cut))) { // A bare credential header as the released block's last line strands // the value that starts on the next line — redact needs them together. - // Take the next line into the block when it is complete; when it is - // not, the header waits with it and the lines before it still release. - const next = raw.indexOf('\n', cut); - if (next === -1) { - cut = lineBoundaryAtOrBefore(raw, cut - 1); - continue; - } - cut = next + 1; + // Shrink until the header is inside the held remainder; it releases + // with its value line once that line is complete. + cut = lineBoundaryAtOrBefore(raw, cut - 1); continue; } const block = redact(raw.slice(state.released, cut), env); diff --git a/packages/sdk/tests/cloud-live.test.ts b/packages/sdk/tests/cloud-live.test.ts index c916f03db..cf1ce3ef7 100644 --- a/packages/sdk/tests/cloud-live.test.ts +++ b/packages/sdk/tests/cloud-live.test.ts @@ -567,6 +567,39 @@ describe('flows logs --follow', () => { expect(rendered).toContain('[redacted]'); }); + it('does not spin when a held credential header precedes an unterminated tail', async () => { + // `authorization:` + a bare `Bearer` line + a still-arriving tail made + // the release cut grow onto `Bearer\n`, see the header again, and shrink + // — forever, inside one poll. The follow must terminate and hold the + // incomplete context instead. + stagedCloud([ + { run: RUNNING, log: logBody('authorization:\nBearer\npartial-tail') }, + { run: COMPLETED, log: logBody('authorization:\nBearer\npartial-tail\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live())).toBe(0); + const rendered = out.stdout.join('\n'); + expect(rendered).not.toContain('partial-tail'); + expect(rendered).toContain('[redacted]'); + }); + + it('holds a credential header that precedes a still-arriving multiline env secret', async () => { + const env = { SERVICE_TOKEN: 'alpha\nbeta\ngamma' }; + stagedCloud([ + { run: RUNNING, log: logBody('x-callback-token:\nalpha\n') }, + { run: COMPLETED, log: logBody('x-callback-token:\nalpha\nbeta\ngamma\n', { done: true }) }, + ]); + const out = io(); + expect(await runCloudLogsFollow({ runId: RUN, step: undefined, json: false }, out.io, live({ env }))).toBe(0); + const rendered = out.stdout.join('\n'); + // Releasing the header with `alpha` leaves `beta\ngamma` without the + // whole value the redactor needs — every fragment must stay held. + expect(rendered).not.toContain('alpha'); + expect(rendered).not.toContain('beta'); + expect(rendered).not.toContain('gamma'); + expect(rendered).toContain('[redacted:SERVICE_TOKEN]'); + }); + it('holds an authorization scheme word whose credential value is still coming', async () => { stagedCloud([ { run: RUNNING, log: logBody('authorization: Bearer\n') },