diff --git a/docs/CLOUD.md b/docs/CLOUD.md index 1fc6bc83..548e1710 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,91 @@ 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. + +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. 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 +323,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 8b98dad6..3f23962f 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 0bf94474..be064990 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 202a69a4..61e2351e 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 00000000..2adae2f4 --- /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 00000000..d3ed5178 --- /dev/null +++ b/packages/sdk/src/cli/cloud-live.ts @@ -0,0 +1,343 @@ +// `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 { 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'; +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 where the rest starts. + * + * 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; + /** How much of `consumed` has been printed; the rest is still redactable. */ + released: number; + emitted: boolean; +} + +/** + * 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. 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 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? + * + * 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: '', released: 0, 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); + } + 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; + } + releaseLines(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.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: ` + + `${'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 5e946208..2776434d 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 00000000..8e06169d --- /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 00000000..e0014441 --- /dev/null +++ b/packages/sdk/src/cli/cloud-status-view.ts @@ -0,0 +1,213 @@ +// 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); + // 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)}`); + 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 5567b3de..4458e616 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 4a21ab20..fa8f6220 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 00000000..bc919625 --- /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 fb92081e..ef4c2930 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/src/redact.ts b/packages/sdk/src/redact.ts index 27c48f80..6c47252b 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 new file mode 100644 index 00000000..b54daefc --- /dev/null +++ b/packages/sdk/tests/cloud-live.test.ts @@ -0,0 +1,802 @@ +// `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('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, + 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]'); + }); + + // 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') }, + { 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 54877253..76cf2df3 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/redact.test.ts b/packages/sdk/tests/redact.test.ts index e3154fb4..8a732f7c 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); + }); +}); diff --git a/packages/sdk/tests/relay-cli-surface.test.ts b/packages/sdk/tests/relay-cli-surface.test.ts index ef8b02b6..c626f35c 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' }, {