From aa0a1fe104865258da1eddf030b10f92925ed744 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 18 Aug 2026 08:58:40 -0700 Subject: [PATCH 1/4] feat(cli): follow a run, wait for one, and tail the log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three commands the surface was missing, each polling or streaming something the generated command layer cannot express. `workflows run --follow` renders the SSE the execute route already emits, so a multi-minute agent run stops printing nothing until it ends. It rides on the generated `run` leaf rather than a sibling command — same operation, one different response encoding — and delegates to the handler it replaced, so every non-follow invocation still runs the generated path. Answer text, thinking and tool calls go to stderr; only the final envelope reaches stdout, so redirecting still yields the result. Reasoning and tool frames need the `X-Sim-Stream-Protocol` header, which is sent only when asked for, because negotiating also switches answer text to live chunks the server may retract. `workflows runs wait` closes the loop `--async` opens. Terminal is completed, failed or cancelled; `redacting` is not, since a run whose output is still being scrubbed is not yet a run you can read. A time pause keeps polling because the server resumes it, and a human pause stops with the resume command rather than burning the bound and calling it a timeout. Distinct exit codes keep cancelled and paused from reading as failure. The bound is `--wait-timeout` and not `--timeout`, because SIM_TIMEOUT_SECONDS already bounds one request and two knobs of the same name hide each other. `logs follow` tails runs as they arrive. Dedup keys on run id, not on the timestamp: a schedule fan-out starts many runs in the same millisecond, so a timestamp watermark either drops the siblings or reprints them. JSON output is one object per line, because a follow never closes an array, and the table header is printed once so columns stay aligned across polls. --- apps/docs/content/docs/en/cli/logs.mdx | 22 + apps/docs/content/docs/en/cli/reference.mdx | 56 ++ apps/docs/content/docs/en/cli/workflows.mdx | 30 + .../sim-cli/src/commands/protocol/index.ts | 13 +- .../src/commands/protocol/logs-follow.test.ts | 311 ++++++++++ .../src/commands/protocol/logs-follow.ts | 560 ++++++++++++++++++ .../protocol/workflow-run-follow.test.ts | 368 ++++++++++++ .../commands/protocol/workflow-run-follow.ts | 338 +++++++++++ .../protocol/workflow-run-wait.test.ts | 289 +++++++++ .../commands/protocol/workflow-run-wait.ts | 288 +++++++++ packages/sim-cli/src/contract/commands.ts | 8 +- 11 files changed, 2278 insertions(+), 5 deletions(-) create mode 100644 packages/sim-cli/src/commands/protocol/logs-follow.test.ts create mode 100644 packages/sim-cli/src/commands/protocol/logs-follow.ts create mode 100644 packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts create mode 100644 packages/sim-cli/src/commands/protocol/workflow-run-follow.ts create mode 100644 packages/sim-cli/src/commands/protocol/workflow-run-wait.test.ts create mode 100644 packages/sim-cli/src/commands/protocol/workflow-run-wait.ts diff --git a/apps/docs/content/docs/en/cli/logs.mdx b/apps/docs/content/docs/en/cli/logs.mdx index d2c3420c550..d6334d9db42 100644 --- a/apps/docs/content/docs/en/cli/logs.mdx +++ b/apps/docs/content/docs/en/cli/logs.mdx @@ -66,3 +66,25 @@ sim logs list [options] | `--folder ` | No | Folder path as shown in the app; the leading / is optional (space-separated, or @path / @- with one value per line). | + +## Watch runs as they arrive, printing each new run once + +```bash +sim logs follow [options] +``` + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--workflow ` | No | Only follow runs of this workflow (repeatable). Defaults to ``. | +| `--folder ` | No | Only follow runs of workflows in this folder (repeatable). Defaults to ``. | +| `--trigger ` | No | Only follow runs with this trigger type (repeatable). Defaults to ``. | +| `--level ` | No | Only follow runs at this severity. Accepted values: `info`, `error`. | +| `--details ` | No | Response detail level; full names each run’s workflow. Accepted values: `basic`, `full`. Defaults to `full`. | +| `-n, --lines ` | No | Recent runs to print before watching. Defaults to `10`. | +| `--interval ` | No | Seconds between polls. Defaults to `3`. | + + diff --git a/apps/docs/content/docs/en/cli/reference.mdx b/apps/docs/content/docs/en/cli/reference.mdx index 0edf783ac28..3208568d6cc 100644 --- a/apps/docs/content/docs/en/cli/reference.mdx +++ b/apps/docs/content/docs/en/cli/reference.mdx @@ -1538,6 +1538,30 @@ sim logs list [options] +### sim logs follow + +Watch runs as they arrive, printing each new run once + +```bash +sim logs follow [options] +``` + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--workflow ` | No | Only follow runs of this workflow (repeatable). Defaults to ``. | +| `--folder ` | No | Only follow runs of workflows in this folder (repeatable). Defaults to ``. | +| `--trigger ` | No | Only follow runs with this trigger type (repeatable). Defaults to ``. | +| `--level ` | No | Only follow runs at this severity. Accepted values: `info`, `error`. | +| `--details ` | No | Response detail level; full names each run’s workflow. Accepted values: `basic`, `full`. Defaults to `full`. | +| `-n, --lines ` | No | Recent runs to print before watching. Defaults to `10`. | +| `--interval ` | No | Seconds between polls. Defaults to `3`. | + + + ## sim mcp-servers Also spelled `sim mcp-server`. @@ -3190,6 +3214,35 @@ sim workflows runs resume [options] +### sim workflows runs wait + +Wait for a run to reach a terminal state, then show it + +```bash +sim workflows runs wait [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `runId` | Yes | Unique workflow run identifier. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--workflow ` | Yes | Workflow ID. | +| `--wait-timeout ` | No | Give up after this many seconds, or 0 to wait indefinitely (default: 3600). Bounds the whole wait; SIM_TIMEOUT_SECONDS bounds one request. | + + + ### sim workflows create Create Workflow @@ -3389,6 +3442,9 @@ sim workflows run [options] | `--include-file-base64` | No | Inline eligible output files as base64 content. Rejected when `async` is true. | | `--no-include-file-base64` | No | Send --include-file-base64 as false. | | `--base64-max-bytes ` | No | Maximum total bytes of file content to inline as base64. Rejected when `async` is true. | +| `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. | +| `--include-thinking` | No | Show model reasoning while following (requires --follow). | +| `--include-tool-calls` | No | Show tool calls while following (requires --follow). | diff --git a/apps/docs/content/docs/en/cli/workflows.mdx b/apps/docs/content/docs/en/cli/workflows.mdx index 19ffd718bde..9741d563b0e 100644 --- a/apps/docs/content/docs/en/cli/workflows.mdx +++ b/apps/docs/content/docs/en/cli/workflows.mdx @@ -117,6 +117,33 @@ Resume a paused run (output is included in JSON or YAML output) +## Wait for a run to reach a terminal state, then show it + +```bash +sim workflows runs wait [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `runId` | Yes | Unique workflow run identifier. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--workflow ` | Yes | Workflow ID. | +| `--wait-timeout ` | No | Give up after this many seconds, or 0 to wait indefinitely (default: 3600). Bounds the whole wait; SIM_TIMEOUT_SECONDS bounds one request. | + + + ## Create workflow ```bash @@ -300,6 +327,9 @@ sim workflows run [options] | `--include-file-base64` | No | Inline eligible output files as base64 content. Rejected when `async` is true. | | `--no-include-file-base64` | No | Send --include-file-base64 as false. | | `--base64-max-bytes ` | No | Maximum total bytes of file content to inline as base64. Rejected when `async` is true. | +| `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. | +| `--include-thinking` | No | Show model reasoning while following (requires --follow). | +| `--include-tool-calls` | No | Show tool calls while following (requires --follow). | diff --git a/packages/sim-cli/src/commands/protocol/index.ts b/packages/sim-cli/src/commands/protocol/index.ts index 8be3088630c..9af2f47c747 100644 --- a/packages/sim-cli/src/commands/protocol/index.ts +++ b/packages/sim-cli/src/commands/protocol/index.ts @@ -2,8 +2,11 @@ import { Command } from 'commander' import { attachFileGet } from './files-get' import { attachFileUpload } from './files-upload' import { attachKnowledgeDocumentUpload } from './knowledge-document-upload' +import { attachLogsFollow } from './logs-follow' import { attachResourceDirectoryCommands } from './resource-directory' import { attachTableImport } from './tables-import' +import { attachWorkflowRunFollow } from './workflow-run-follow' +import { attachWorkflowRunWait } from './workflow-run-wait' function group(program: Command, name: string): Command { const existing = program.commands.find((command) => command.name() === name) @@ -43,10 +46,18 @@ export function attachProtocolCommands(program: Command): void { createFolder: 'createTableFolder', }) - attachResourceDirectoryCommands(group(program, 'workflows'), { + const workflows = group(program, 'workflows') + attachResourceDirectoryCommands(workflows, { kind: 'workflow', resources: 'listWorkflows', folders: 'listWorkflowFolders', createFolder: 'createWorkflowFolder', }) + // Both augment commands the generated pass already built — `run` gains + // `--follow`, and `runs` gains `wait` — so they must attach after it, which is + // the order `buildProgram` calls them in. + attachWorkflowRunFollow(workflows) + attachWorkflowRunWait(group(workflows, 'runs')) + + attachLogsFollow(group(program, 'logs')) } diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts new file mode 100644 index 00000000000..05818a6b210 --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts @@ -0,0 +1,311 @@ +/** + * @vitest-environment node + */ +import { Command } from 'commander' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { ListLogsResponse } from '../../generated/v2-api' +import { SimApiError } from '../../http/client' +import { attachLogsFollow, type LogRow } from './logs-follow' + +const { mockRequest, mockSleep, profile } = vi.hoisted(() => ({ + mockRequest: vi.fn(), + mockSleep: vi.fn(() => Promise.resolve()), + profile: { output: 'json' as string }, +})) + +vi.mock('../../helpers', () => ({ sleep: mockSleep })) + +vi.mock('../../context', () => ({ + clientFrom: () => ({ + client: { request: mockRequest, requireWorkspace: () => 'ws_1' }, + profile: { + name: 'default', + endpoint: 'https://sim.example', + apiKey: 'k', + workspaceId: 'ws_1', + output: profile.output, + }, + }), +})) + +/** Stops a runaway follow before it can hang the suite. */ +const MAX_POLLS = 50 + +const originalStderrIsTTY = process.stderr.isTTY + +let stdout: string[] +let stderr: string[] + +function row(runId: string, startedAt: string): LogRow { + return { + runId, + workflowId: 'wf_1', + deploymentVersionId: null, + status: 'completed', + level: 'info', + trigger: 'api', + startedAt, + endedAt: startedAt, + totalDurationMs: 12, + cost: { total: 0.5 }, + files: null, + workflow: { id: 'wf_1', name: 'Nightly sync', description: null, deleted: false }, + } +} + +function page(rows: LogRow[], nextCursor: string | null = null): ListLogsResponse { + return { data: rows, nextCursor } +} + +/** + * Answers each poll from `responses`, then ends the follow the way a user does. + * + * Ctrl-C is the only clean exit a follow has, so the tests stop it the same way + * rather than by unwinding the loop with an error. + */ +function respondWith(responses: Array): void { + let polls = 0 + mockRequest.mockImplementation(async () => { + polls += 1 + if (polls > MAX_POLLS) throw new Error('follow did not stop') + const next = responses.shift() + if (next === undefined) { + process.emit('SIGINT') + return page([]) + } + if (next instanceof Error) throw next + return next + }) +} + +function follow(...argv: string[]): Promise { + const root = new Command('sim').exitOverride() + const logs = new Command('logs').exitOverride() + root.addCommand(logs) + attachLogsFollow(logs) + for (const command of logs.commands) command.exitOverride() + return root.parseAsync(['node', 'sim', 'logs', 'follow', ...argv]) +} + +/** The run ids printed to stdout, in the order they were printed. */ +function printedRunIds(): string[] { + return stdout.map((line) => JSON.parse(line).runId as string) +} + +beforeEach(() => { + stdout = [] + stderr = [] + profile.output = 'json' + mockRequest.mockReset() + mockSleep.mockClear() + vi.spyOn(console, 'log').mockImplementation((line: unknown) => { + stdout.push(String(line)) + }) + vi.spyOn(process.stderr, 'write').mockImplementation((chunk: unknown) => { + stderr.push(String(chunk)) + return true + }) +}) + +afterEach(() => { + vi.restoreAllMocks() + Object.defineProperty(process.stderr, 'isTTY', { + value: originalStderrIsTTY, + configurable: true, + }) +}) + +describe('sim logs follow', () => { + it('prints the backlog oldest first and never reprints it', async () => { + const rows = [ + row('run_3', '2026-08-17T10:00:03.000Z'), + row('run_2', '2026-08-17T10:00:02.000Z'), + row('run_1', '2026-08-17T10:00:01.000Z'), + ] + respondWith([page(rows), page(rows), page(rows)]) + + await follow('-n', '3') + + expect(printedRunIds()).toEqual(['run_1', 'run_2', 'run_3']) + }) + + it('prints only the runs that arrived since the last poll', async () => { + const first = row('run_1', '2026-08-17T10:00:01.000Z') + const second = row('run_2', '2026-08-17T10:00:02.000Z') + respondWith([page([first]), page([second, first])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_1', 'run_2']) + }) + + it('prints a sibling run that shares a start time with one already printed', async () => { + const sameInstant = '2026-08-17T10:00:01.000Z' + const first = row('run_1', sameInstant) + const sibling = row('run_2', sameInstant) + respondWith([page([first]), page([sibling, first])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_1', 'run_2']) + }) + + it('prints a burst of runs oldest first, as a terminal reads', async () => { + const first = row('run_1', '2026-08-17T10:00:01.000Z') + const second = row('run_2', '2026-08-17T10:00:02.000Z') + const third = row('run_3', '2026-08-17T10:00:03.000Z') + respondWith([page([first]), page([third, second, first])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_1', 'run_2', 'run_3']) + }) + + it('does not replay history a later page reaches back into', async () => { + const newest = row('run_2', '2026-08-17T10:00:02.000Z') + const older = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([page([newest]), page([newest, older])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_2']) + }) + + it('prints nothing from the backlog at -n 0 but still anchors to now', async () => { + const existing = row('run_1', '2026-08-17T10:00:01.000Z') + const arrived = row('run_2', '2026-08-17T10:00:02.000Z') + respondWith([page([existing]), page([arrived, existing])]) + + await follow('-n', '0') + + expect(printedRunIds()).toEqual(['run_2']) + }) + + it('retries a transient failure and keeps following', async () => { + const first = row('run_1', '2026-08-17T10:00:01.000Z') + const second = row('run_2', '2026-08-17T10:00:02.000Z') + respondWith([page([first]), new SimApiError('Service Unavailable', 503), page([second, first])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_1', 'run_2']) + }) + + it('stops immediately on an authentication failure', async () => { + const first = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([page([first]), new SimApiError('Unauthorized', 401), page([first])]) + + await expect(follow('-n', '1')).rejects.toThrow('Unauthorized') + expect(mockRequest).toHaveBeenCalledTimes(2) + }) + + it('stops immediately on a rejected filter', async () => { + respondWith([new SimApiError('triggers contains an empty entry', 400)]) + + await expect(follow('--trigger', '', '-n', '1')).rejects.toThrow('empty entry') + expect(mockRequest).toHaveBeenCalledTimes(1) + }) + + it('emits one JSON object per line rather than an array', async () => { + const rows = [ + row('run_2', '2026-08-17T10:00:02.000Z'), + row('run_1', '2026-08-17T10:00:01.000Z'), + ] + respondWith([page(rows)]) + + await follow('-n', '2') + + expect(stdout).toHaveLength(2) + for (const line of stdout) { + expect(line.startsWith('{')).toBe(true) + expect(line).not.toContain('\n') + expect(JSON.parse(line)).toMatchObject({ workflow: { name: 'Nightly sync' } }) + } + }) + + it('prints the table header once, not once per poll', async () => { + profile.output = 'table' + const first = row('run_1', '2026-08-17T10:00:01.000Z') + const second = row('run_2', '2026-08-17T10:00:02.000Z') + respondWith([page([first]), page([second, first])]) + + await follow('-n', '1') + + expect(stdout.filter((line) => line.includes('STARTED'))).toHaveLength(1) + expect(stdout).toHaveLength(3) + }) + + it('keeps rows on stdout and retry notices on stderr', async () => { + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const first = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([page([first]), new SimApiError('Too Many Requests', 429), page([first])]) + + await follow('-n', '1') + + expect(printedRunIds()).toEqual(['run_1']) + expect(stderr.join('')).toContain('retrying in') + expect(stdout.join('')).not.toContain('retrying in') + }) + + it('stays silent on stderr when it is not a terminal', async () => { + Object.defineProperty(process.stderr, 'isTTY', { value: false, configurable: true }) + const first = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([page([first]), new SimApiError('Too Many Requests', 429), page([first])]) + + await follow('-n', '1') + + expect(stderr).toEqual([]) + }) + + it('exits cleanly on Ctrl-C and leaves no signal listeners behind', async () => { + const before = process.listenerCount('SIGINT') + respondWith([page([row('run_1', '2026-08-17T10:00:01.000Z')])]) + + await expect(follow('-n', '1')).resolves.toBeDefined() + + expect(process.listenerCount('SIGINT')).toBe(before) + }) + + it('rejects an interval that would hammer the API', async () => { + respondWith([]) + + await expect(follow('--interval', '0.001')).rejects.toThrow('--interval must be at least') + expect(mockRequest).not.toHaveBeenCalled() + }) + + it('rejects a backlog count that is not a whole number of runs', async () => { + respondWith([]) + + await expect(follow('-n', '-1')).rejects.toThrow('--lines must be a non-negative integer') + expect(mockRequest).not.toHaveBeenCalled() + }) + + it('asks for the detail level that names each run’s workflow, and the newest page first', async () => { + respondWith([page([row('run_1', '2026-08-17T10:00:01.000Z')])]) + + await follow('-n', '1', '--workflow', 'wf_1', '--workflow', 'wf_2', '--folder', '/Q1 2026') + + expect(mockRequest).toHaveBeenCalledWith('/api/v2/logs', { + query: expect.objectContaining({ + workspaceId: 'ws_1', + details: 'full', + order: 'desc', + workflowIds: 'wf_1,wf_2', + folderPaths: '/Q1%202026', + limit: 1, + }), + }) + }) + + it('waits between polls instead of spinning', async () => { + const first = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([page([first]), page([first])]) + + await follow('-n', '1') + + expect(mockSleep).toHaveBeenCalled() + for (const [ms] of mockSleep.mock.calls as unknown as Array<[number]>) { + expect(ms).toBeGreaterThan(0) + } + }) +}) diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.ts b/packages/sim-cli/src/commands/protocol/logs-follow.ts new file mode 100644 index 00000000000..b1dcf8a541f --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/logs-follow.ts @@ -0,0 +1,560 @@ +import chalk from 'chalk' +import { type Command, Option } from 'commander' +import { dump } from 'js-yaml' +import type { OutputFormat } from '../../config/index' +import { clientFrom } from '../../context' +import { CLI_CONTRACT } from '../../contract/commands' +import type { ColumnSpec } from '../../contract/types' +import { type ListLogsResponse, V2_OPERATIONS } from '../../generated/v2-api' +import { sleep } from '../../helpers' +import { SimApiError, type SimClient } from '../../http/client' +import { + bool, + bytes, + type Column, + duration, + printList, + text, + timestamp, + visibleWidth, +} from '../../output/render' +import { encodeFolderPath } from '../../runtime/request' + +/** One run, as `GET /api/v2/logs` returns it. */ +export type LogRow = ListLogsResponse['data'][number] + +/** Recent runs printed before the follow starts watching, as `tail -f` does. */ +const DEFAULT_BACKLOG = 10 + +/** Seconds between polls when `--interval` is not given. */ +const DEFAULT_INTERVAL_SECONDS = 3 + +/** + * Shortest poll interval accepted. + * + * A follow is an unattended loop against a rate-limited API, so the floor is + * what stops a typo — `--interval 0.001` — from turning one terminal into a + * request flood the workspace is billed for. + */ +const MIN_INTERVAL_SECONDS = 0.1 + +/** Longest a backoff may stretch the poll interval while the API is unhappy. */ +const MAX_BACKOFF_MS = 30_000 + +/** Rows asked for per page once the follow is watching. */ +const POLL_PAGE_SIZE = 100 + +/** + * Pages one poll walks back before it stops catching up. + * + * A burst larger than this is not lost — the next poll resumes from the same + * high-water mark — but it bounds how long a single poll can hold the loop. + */ +const MAX_PAGES_PER_POLL = 10 + +/** + * Run ids the follow remembers. + * + * The set is what makes a run print exactly once, so it has to be bounded for a + * follow left running for days. Eviction also raises the floor, so an id that is + * forgotten cannot come back as a new row. + */ +const MAX_REMEMBERED_RUNS = 5000 + +/** Widest a table column renders, matching the `logs list` table. */ +const MAX_CELL_WIDTH = 60 + +/** How often an interruptible wait checks whether Ctrl-C has arrived. */ +const WAIT_SLICE_MS = 250 + +/** + * Client statuses worth retrying. + * + * Every other 4xx describes the request itself — a rejected filter, a key that + * cannot read this workspace — and retrying one every few seconds repeats the + * same rejection forever instead of showing it once and stopping. + */ +const RETRYABLE_CLIENT_STATUSES = new Set([408, 425, 429]) + +/** + * Erases from the cursor to the end of the line. + * + * Built from a char code so the source carries no raw ESC byte, which an editor + * or a patch tool can silently eat; `output/render` builds its patterns the same + * way for the same reason. + */ +const ERASE_LINE = `${String.fromCharCode(27)}[K` + +interface FollowOptions { + workflow?: string[] + folder?: string[] + trigger?: string[] + level?: string + details: string + lines: string + interval: string +} + +/** Collects a repeatable flag, as the generated `list` flags do. */ +function collect(value: string, previous: string[]): string[] { + return [...previous, value] +} + +function at(row: unknown, path: string): unknown { + return path + .split('.') + .reduce( + (value, key) => (value && typeof value === 'object' ? (value as never)[key] : undefined), + row + ) +} + +/** + * Renders one cell the way `logs list` renders it. + * + * `runtime/result` keeps its equivalent private and only exposes whole-list + * printing, which a follow cannot use: it prints a header once and then rows + * against locked widths, so it needs cells rather than a finished table. The + * column *definitions* still come from the same contract entry, so the two views + * cannot drift in what they show — only in how the rows are laid out. + */ +function renderCell(value: unknown, format: ColumnSpec['format']): string { + switch (format) { + case 'timestamp': + return timestamp(value as string | null) + case 'duration': + return duration(value as number | null) + case 'bytes': + return bytes(value as number | null) + case 'bool': + return bool(value as boolean | null) + case 'cost': + return typeof value === 'number' ? `$${value.toFixed(4)}` : text(null) + default: + return text(typeof value === 'object' && value !== null ? JSON.stringify(value) : value) + } +} + +const COLUMNS: Column[] = (CLI_CONTRACT.listLogs?.columns ?? []).map((spec) => ({ + header: spec.header, + value: (row) => renderCell(at(row, spec.path ?? spec.header), spec.format), +})) + +/** + * Flattens a finished cell onto one line. + * + * The shared `sanitize` cannot be used on a finished cell — it strips escape + * sequences, including the colour the CLI itself just added. The server-supplied + * text inside the cell was already sanitized by the formatters above. + */ +function oneLine(value: string): string { + return value.replace(/\s*[\r\n\t]+\s*/g, ' ') +} + +function pad(value: string, width: number): string { + return value + ' '.repeat(Math.max(0, width - visibleWidth(value))) +} + +/** + * Truncates a cell to its locked column width. + * + * Skipped unless the visible width equals the string length, which is only true + * of text carrying neither an escape sequence nor a wide character — slicing + * either would cut an escape in half or mis-measure the result. + */ +function clamp(value: string, width: number): string { + if (visibleWidth(value) <= width || visibleWidth(value) !== value.length) return value + return `${value.slice(0, Math.max(1, width - 1))}…` +} + +type RowWriter = (rows: LogRow[]) => void + +/** + * Prints the header once, then rows against widths locked by the first batch. + * + * A follow has no end, so the alternative — one finished table per poll — + * repeats the header every few seconds and, worse, recomputes the column widths + * per batch, so no two batches line up and the stream stops reading as one + * table. Locking the widths is what `kubectl get -w` does, for the same reason. + * The header prints on the first call even when that call carries no rows, so + * the columns are labelled from the start rather than from the first run. + */ +function createTableWriter(): RowWriter { + let widths: number[] | null = null + + return (rows) => { + const lines = rows.map((row) => COLUMNS.map((column) => oneLine(column.value(row)))) + + if (!widths) { + widths = COLUMNS.map((column, index) => + Math.min( + MAX_CELL_WIDTH, + Math.max(visibleWidth(column.header), ...lines.map((line) => visibleWidth(line[index]))) + ) + ) + const header = widths + console.log( + chalk.dim( + COLUMNS.map((column, index) => pad(column.header.toUpperCase(), header[index])) + .join(' ') + .trimEnd() + ) + ) + } + + const locked = widths + for (const line of lines) { + console.log( + line + .map((cell, index) => pad(clamp(cell, locked[index]), locked[index])) + .join(' ') + .trimEnd() + ) + } + } +} + +/** + * Picks the writer for the profile's output format. + * + * `json` emits one object per line rather than an array, because a follow never + * closes the bracket that would make an array valid — JSONL is the shape `jq` + * and `while read` can consume as it arrives. `yaml` becomes a `---` separated + * document stream for the same reason, and `text` reuses the shared + * tab-separated rendering, which already prints no header. + */ +function createWriter(format: OutputFormat): RowWriter { + if (format === 'json') { + return (rows) => { + for (const row of rows) console.log(JSON.stringify(row)) + } + } + if (format === 'yaml') { + return (rows) => { + for (const row of rows) { + console.log(`---\n${dump(row, { lineWidth: 0, noRefs: true }).trimEnd()}`) + } + } + } + if (format === 'text') { + return (rows) => { + if (rows.length > 0) printList('text', rows, COLUMNS) + } + } + return createTableWriter() +} + +export interface FollowStatus { + /** Replaces the status line, if there is a terminal to draw it on. */ + note: (message: string) => void + /** Erases the line, if anything was ever written to it. */ + clear: () => void +} + +/** + * Reports what the follow is doing, on stderr. + * + * The rule `pageProgress` follows: stdout is the stream of rows and gets piped, + * so anything that is not a row goes to stderr, and only to a terminal. + */ +export function followStatus(): FollowStatus { + let reported = false + return { + note: (message) => { + if (!process.stderr.isTTY) return + reported = true + process.stderr.write(`\r${chalk.dim(message)}${ERASE_LINE}`) + }, + clear: () => { + if (!reported) return + reported = false + process.stderr.write(`\r${ERASE_LINE}`) + }, + } +} + +interface Interruption { + interrupted: () => boolean + dispose: () => void +} + +/** + * Turns Ctrl-C into a loop exit rather than a process kill. + * + * A follow is meant to be ended by the user, so ending it is a success: the loop + * finishes the row it is on, drops its listeners and returns, which exits 0 with + * nothing half-written. Under the default signal disposition the process is torn + * down wherever it happens to be, which can be mid-row. + */ +function watchForInterrupt(): Interruption { + let stopped = false + const stop = () => { + stopped = true + } + process.on('SIGINT', stop) + process.on('SIGTERM', stop) + return { + interrupted: () => stopped, + dispose: () => { + process.off('SIGINT', stop) + process.off('SIGTERM', stop) + }, + } +} + +/** + * Sleeps in slices, so Ctrl-C is felt within a slice rather than within a poll. + * + * The shared `sleep` is not cancellable, and a `--interval 30` follow that only + * noticed the signal when its timer elapsed would look hung for half a minute + * after the user asked it to stop. + */ +async function waitFor(ms: number, interrupted: () => boolean): Promise { + let remaining = ms + while (remaining > 0 && !interrupted()) { + const step = Math.min(WAIT_SLICE_MS, remaining) + await sleep(step) + remaining -= step + } +} + +/** + * What the follow has already accounted for. + * + * Two facts, because neither is sufficient alone. `seen` is the authority on + * duplicates: `startedAt` has millisecond resolution and one schedule fan-out + * starts many runs inside the same millisecond, so a `startedAt > last` + * watermark silently drops every sibling but one while `>=` reprints them all — + * and a run persisted out of order would fall below the watermark and never + * print at all. `floor` is the authority on history: `seen` only holds runs + * observed since start, so without it any older run reachable on a later page + * would be printed as though it had just arrived. + */ +interface FollowState { + seen: Map + floor: string | null +} + +function isUnprinted(state: FollowState, row: LogRow): boolean { + if (state.seen.has(row.runId)) return false + return state.floor === null || row.startedAt >= state.floor +} + +/** + * Records rows as printed, forgetting the oldest once the cap is reached. + * + * Eviction raises the floor to the forgotten runs' own start times, so a run + * whose id is no longer remembered is excluded by the floor instead of + * reappearing as new. + */ +function remember(state: FollowState, rows: LogRow[]): void { + for (const row of rows) state.seen.set(row.runId, row.startedAt) + + let excess = state.seen.size - MAX_REMEMBERED_RUNS + if (excess <= 0) return + + for (const [runId, startedAt] of state.seen) { + if (excess <= 0) break + if (state.floor === null || startedAt > state.floor) state.floor = startedAt + state.seen.delete(runId) + excess -= 1 + } +} + +/** + * Walks the newest-first list back to the first row already accounted for, and + * returns everything above it, still newest first. + * + * Paging continues only while a whole page was new, which is the only shape that + * says there may be more above the previous poll's high-water mark. A page + * holding one known row proves the rest of the list is older still, so walking + * further would only re-read history the floor rejects anyway. + */ +async function collectUnprinted( + client: Pick, + path: string, + query: Record, + state: FollowState, + pageSize: number, + maxPages: number +): Promise { + const fresh: LogRow[] = [] + let cursor: string | null = null + + for (let page = 0; page < maxPages; page += 1) { + const response: ListLogsResponse = await client.request(path, { + query: { ...query, limit: pageSize, cursor }, + }) + const rows = response?.data ?? [] + const unprinted = rows.filter((row) => isUnprinted(state, row)) + fresh.push(...unprinted) + + cursor = response?.nextCursor ?? null + if (!cursor || rows.length === 0 || unprinted.length < rows.length) break + } + + return fresh +} + +/** + * Whether a failed poll is worth retrying. + * + * Status 0 is a transport failure — a dropped connection, a laptop that slept — + * which is exactly what a follow left running overnight has to survive. An + * expired key or a rejected filter is not: it fails identically on every poll, + * so it surfaces once and stops the command instead of scrolling forever. + */ +function isTransient(error: unknown): boolean { + if (!(error instanceof SimApiError)) return false + if (error.status === 0 || error.status >= 500) return true + return RETRYABLE_CLIENT_STATUSES.has(error.status) +} + +function nonNegativeInteger(raw: string, flag: string): number { + const value = Number(raw) + if (!Number.isSafeInteger(value) || value < 0) { + throw new SimApiError(`${flag} must be a non-negative integer`, 0) + } + return value +} + +function intervalMs(raw: string): number { + const seconds = Number(raw) + if (!Number.isFinite(seconds) || seconds < MIN_INTERVAL_SECONDS) { + throw new SimApiError(`--interval must be at least ${MIN_INTERVAL_SECONDS} seconds`, 0) + } + return Math.round(seconds * 1000) +} + +/** Seconds, to one decimal, for a delay reported to the reader. */ +function inSeconds(ms: number): number { + return Math.round(ms / 100) / 10 +} + +/** + * Adds `sim logs follow`. + * + * A sibling command rather than `logs list --follow`, because `logs list` is + * generated from the CLI contract, which describes the HTTP surface: `--follow` + * has no wire counterpart, and its paging is the inverse of the contract's — + * re-reading the newest page forever, rather than walking a cursor to the end + * once. The filters are the same ones `logs list` exposes, so the two commands + * take the same arguments and render the same columns. + */ +export function attachLogsFollow(logs: Command): void { + logs + .command('follow') + .description('Watch runs as they arrive, printing each new run once') + .option('--workflow ', 'Only follow runs of this workflow (repeatable)', collect, []) + .option( + '--folder ', + 'Only follow runs of workflows in this folder (repeatable)', + collect, + [] + ) + .option('--trigger ', 'Only follow runs with this trigger type (repeatable)', collect, []) + .addOption( + new Option('--level ', 'Only follow runs at this severity').choices([ + ...V2_OPERATIONS.listLogs.query.level.values, + ]) + ) + .addOption( + new Option('--details ', 'Response detail level; full names each run’s workflow') + .choices([...V2_OPERATIONS.listLogs.query.details.values]) + .default('full') + ) + .option('-n, --lines ', 'Recent runs to print before watching', String(DEFAULT_BACKLOG)) + .option('--interval ', 'Seconds between polls', String(DEFAULT_INTERVAL_SECONDS)) + .addHelpText( + 'after', + ` +Each run prints once, when it is first seen, so its status is the status it had +at that moment. With --output json every run is a JSON object on its own line +(JSONL) rather than a member of an array, because a follow never ends and so can +never close one; --output yaml emits a --- separated document stream. Progress +and retries go to stderr, leaving stdout a clean stream of rows. Ctrl-C stops the +follow. + +Examples: + $ sim logs follow --level error + $ sim logs follow --workflow wf_123 -n 0 + $ sim --output json logs follow | jq -r '.runId' +` + ) + .action(async (options: FollowOptions, command: Command) => { + const lines = nonNegativeInteger(options.lines, '--lines') + const delay = intervalMs(options.interval) + + const { client, profile } = clientFrom(command) + const path = V2_OPERATIONS.listLogs.path + const query = { + workspaceId: client.requireWorkspace(), + workflowIds: options.workflow?.length ? options.workflow.join(',') : undefined, + // Encoded exactly as the contract-driven `--folder` encodes it, so a + // path that works on `logs list` works here; encoding first is also what + // keeps the comma-joined form unambiguous. + folderPaths: options.folder?.length + ? options.folder.map(encodeFolderPath).join(',') + : undefined, + triggers: options.trigger?.length ? options.trigger.join(',') : undefined, + level: options.level, + details: options.details, + order: 'desc', + } + + const write = createWriter(profile.output) + const status = followStatus() + const interrupt = watchForInterrupt() + const state: FollowState = { seen: new Map(), floor: null } + + try { + // The backlog page doubles as the seed: every run on it is recorded and + // its oldest start time becomes the floor, so even `-n 0` anchors the + // follow to now instead of replaying the workspace's whole history. + const backlog = await collectUnprinted(client, path, query, state, Math.max(lines, 1), 1) + remember(state, backlog) + state.floor = backlog.at(-1)?.startedAt ?? null + write(lines > 0 ? backlog.slice(0, lines).reverse() : []) + + let failures = 0 + while (!interrupt.interrupted()) { + await waitFor( + failures === 0 ? delay : Math.min(delay * 2 ** failures, MAX_BACKOFF_MS), + interrupt.interrupted + ) + if (interrupt.interrupted()) break + + let fresh: LogRow[] + try { + fresh = await collectUnprinted( + client, + path, + query, + state, + POLL_PAGE_SIZE, + MAX_PAGES_PER_POLL + ) + } catch (error) { + if (!isTransient(error)) throw error + failures += 1 + const next = Math.min(delay * 2 ** failures, MAX_BACKOFF_MS) + status.note( + `poll failed (${(error as SimApiError).message}); retrying in ${inSeconds(next)}s…` + ) + continue + } + + failures = 0 + if (fresh.length === 0) continue + status.clear() + remember(state, fresh) + // Reversed because the API answers newest-first while a terminal reads + // downwards: the newest run has to end up on the last line. + write(fresh.reverse()) + } + } finally { + status.clear() + interrupt.dispose() + } + }) +} diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts new file mode 100644 index 00000000000..7ada9f8e46c --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts @@ -0,0 +1,368 @@ +/** + * @vitest-environment node + */ +import { Command } from 'commander' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { sleep } from '../../helpers' +import { SimApiError } from '../../http/client' +import { buildGeneratedCommands } from '../../runtime/build' +import { attachWorkflowRunFollow, renderRunStream } from './workflow-run-follow' + +const { output, request, requestRaw } = vi.hoisted(() => ({ + output: { format: 'json' }, + request: vi.fn(), + requestRaw: vi.fn(), +})) + +vi.mock('../../context', () => ({ + clientFrom: () => ({ + client: { request, requestRaw, requireWorkspace: () => 'ws_local' }, + profile: { + workspaceId: 'ws_local', + output: output.format, + name: 'default', + apiKey: 'k', + endpoint: 'https://sim.example', + }, + }), +})) + +/** Collects commentary the way `process.stderr` would receive it. */ +function writer() { + const written: string[] = [] + return { + write: (text: string) => { + written.push(text) + return true + }, + get text() { + return written.join('') + }, + } +} + +function bodyOf(chunks: string[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk)) + controller.close() + }, + }) +} + +function sse(...frames: unknown[]): ReadableStream { + return bodyOf(frames.map((frame) => `data: ${JSON.stringify(frame)}\n\n`)) +} + +function streamResponse(body: ReadableStream): Response { + return { + body, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + } as unknown as Response +} + +function options(overrides: Partial[1]> = {}) { + return { includeThinking: false, includeToolCalls: false, stderr: writer(), ...overrides } +} + +beforeEach(() => { + output.format = 'json' + request.mockReset() + requestRaw.mockReset() +}) + +afterEach(() => { + vi.restoreAllMocks() + vi.unstubAllGlobals() + vi.unstubAllEnvs() +}) + +describe('renderRunStream', () => { + it('returns the final envelope and writes answer text to the commentary sink', async () => { + const stderr = writer() + const final = await renderRunStream( + sse( + { blockId: 'agent-1', chunk: 'Hello' }, + { blockId: 'agent-1', chunk: ' world' }, + { event: 'final', data: { success: true, output: { answer: 'Hello world' } } }, + '[DONE]' + ), + options({ stderr }) + ) + + expect(final).toEqual({ success: true, output: { answer: 'Hello world' } }) + expect(stderr.text).toContain('Hello world') + }) + + it('reassembles a frame split across chunk boundaries', async () => { + const final = await renderRunStream( + bodyOf(['data: {"event":"fin', 'al","data":{"success":true}}\n\n', 'data: "[DONE]"\n\n']), + options() + ) + + expect(final).toEqual({ success: true }) + }) + + it('renders answer text while the run is still open, not only once it ends', async () => { + const held: { source?: ReadableStreamDefaultController } = {} + const body = new ReadableStream({ + start(source) { + held.source = source + }, + }) + const source = held.source + if (!source) throw new Error('stream controller was not captured') + + const encode = (text: string) => new TextEncoder().encode(text) + const stderr = writer() + const rendered = renderRunStream(body, options({ stderr })) + + // Deliberately without the trailing blank line: a reader keyed on the + // `\n\n` event separator would hold this frame until the next one arrived. + source.enqueue(encode('data: {"blockId":"agent-1","chunk":"live"}\n')) + await sleep(0) + expect(stderr.text).toContain('live') + + source.enqueue(encode('\ndata: {"event":"final","data":{"success":true}}\n\n')) + source.close() + await expect(rendered).resolves.toEqual({ success: true }) + }) + + it('hides thinking and tool frames unless they were asked for', async () => { + const stderr = writer() + await renderRunStream( + sse( + { event: 'thinking', blockId: 'agent-1', data: 'weighing options' }, + { event: 'tool', blockId: 'agent-1', phase: 'start', id: 't1', name: 'http_request' }, + { event: 'final', data: { success: true } } + ), + options({ stderr }) + ) + + expect(stderr.text).not.toContain('weighing options') + expect(stderr.text).not.toContain('http_request') + }) + + it('renders thinking and tool frames when they were asked for', async () => { + const stderr = writer() + await renderRunStream( + sse( + { event: 'thinking', blockId: 'agent-1', data: 'weighing options' }, + { event: 'tool', blockId: 'agent-1', phase: 'start', id: 't1', name: 'http_request' }, + { + event: 'tool', + blockId: 'agent-1', + phase: 'end', + id: 't1', + name: 'http_request', + status: 'error', + }, + { event: 'final', data: { success: true } } + ), + options({ stderr, includeThinking: true, includeToolCalls: true }) + ) + + expect(stderr.text).toContain('weighing options') + expect(stderr.text).toContain('http_request') + expect(stderr.text).toContain('(error)') + }) + + it('reports a retraction so streamed text is never silently wrong', async () => { + const stderr = writer() + await renderRunStream( + sse( + { blockId: 'agent-1', chunk: 'draft answer' }, + { event: 'chunk_reset', blockId: 'agent-1' }, + { event: 'final', data: { success: true } } + ), + options({ stderr }) + ) + + expect(stderr.text).toContain('retracted') + }) + + it('keeps reading after a non-terminal stream_error and warns about it', async () => { + const stderr = writer() + const final = await renderRunStream( + sse( + { event: 'stream_error', blockId: 'agent-1', error: 'partial read' }, + { event: 'final', data: { success: true, output: {} } } + ), + options({ stderr }) + ) + + expect(stderr.text).toContain('warning: partial read') + expect(final).toEqual({ success: true, output: {} }) + }) + + it('fails with the server message on a terminal error frame', async () => { + await expect( + renderRunStream(sse({ event: 'error', error: 'Agent block timed out' }), options()) + ).rejects.toThrow(/Agent block timed out/) + }) + + it('fails rather than reporting an empty success when the stream is truncated', async () => { + await expect( + renderRunStream(sse({ blockId: 'agent-1', chunk: 'partial' }), options()) + ).rejects.toThrow(/ended before the workflow reported a result/) + }) + + it('stops at the terminal sentinel and ignores anything after it', async () => { + const final = await renderRunStream( + sse({ event: 'final', data: { success: true, output: { a: 1 } } }, '[DONE]', { + event: 'final', + data: { success: false }, + }), + options() + ) + + expect(final).toEqual({ success: true, output: { a: 1 } }) + }) + + it('skips a frame shape it does not understand instead of failing the run', async () => { + const final = await renderRunStream( + bodyOf([ + 'data: not json at all\n\n', + 'data: {"event":"invented_later","blockId":"b"}\n\n', + 'data: {"event":"final","data":{"success":true}}\n\n', + ]), + options() + ) + + expect(final).toEqual({ success: true }) + }) + + it('strips terminal control sequences out of server-supplied answer text', async () => { + const stderr = writer() + await renderRunStream( + sse( + { blockId: 'agent-1', chunk: '\u001b[2Joops' }, + { event: 'final', data: { success: true } } + ), + options({ stderr }) + ) + + expect(stderr.text).not.toContain('\u001b[2J') + expect(stderr.text).toContain('oops') + }) +}) + +function program(): Command { + const root = new Command() + root.exitOverride() + for (const command of buildGeneratedCommands()) root.addCommand(command) + + const workflows = root.commands.find((command) => command.name() === 'workflows') + if (!workflows) throw new Error('workflows group missing') + attachWorkflowRunFollow(workflows) + return root +} + +async function run(...argv: string[]): Promise { + await program().parseAsync(['node', 'sim', 'workflows', 'run', ...argv]) +} + +describe('sim workflows run --follow', () => { + it('refuses --follow together with --async', async () => { + await expect(run('wf_1', '--follow', '--async')).rejects.toThrow(/pass one, not both/) + expect(requestRaw).not.toHaveBeenCalled() + }) + + it('refuses stream-only flags without --follow', async () => { + await expect(run('wf_1', '--include-thinking')).rejects.toThrow(/add --follow/) + expect(request).not.toHaveBeenCalled() + }) + + it('leaves the generated non-streaming path untouched', async () => { + request.mockResolvedValue({ data: { success: true, output: {} } }) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await run('wf_1', '--input', '{"topic":"otters"}') + + expect(requestRaw).not.toHaveBeenCalled() + expect(request).toHaveBeenCalledTimes(1) + expect(request.mock.calls[0][1].body).toEqual({ input: { topic: 'otters' } }) + }) + + it('asks for a stream and does not negotiate agent events by default', async () => { + requestRaw.mockResolvedValue(streamResponse(sse({ event: 'final', data: { success: true } }))) + vi.spyOn(console, 'log').mockImplementation(() => {}) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + + await run('wf_1', '--follow', '--input', '{"topic":"otters"}') + + const [path, init] = requestRaw.mock.calls[0] + expect(path).toBe('/api/v2/workflows/wf_1/execute') + expect(init.body).toEqual({ input: { topic: 'otters' }, stream: true }) + expect(init.headers.accept).toBe('text/event-stream') + expect(init.headers['x-sim-stream-protocol']).toBeUndefined() + }) + + it('negotiates the agent-event protocol when event frames are requested', async () => { + requestRaw.mockResolvedValue(streamResponse(sse({ event: 'final', data: { success: true } }))) + vi.spyOn(console, 'log').mockImplementation(() => {}) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + + await run('wf_1', '--follow', '--include-thinking', '--include-tool-calls') + + const init = requestRaw.mock.calls[0][1] + expect(init.body).toMatchObject({ stream: true, includeThinking: true, includeToolCalls: true }) + expect(init.headers['x-sim-stream-protocol']).toBe('agent-events-v1') + }) + + it('prints the final envelope on stdout and the chatter on stderr', async () => { + requestRaw.mockResolvedValue( + streamResponse( + sse( + { blockId: 'agent-1', chunk: 'thinking out loud' }, + { event: 'final', data: { success: true, output: { answer: 42 } } }, + '[DONE]' + ) + ) + ) + const stdout = vi.spyOn(console, 'log').mockImplementation(() => {}) + const stderr = vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + + await run('wf_1', '--follow') + + const printed = stdout.mock.calls.map((call) => String(call[0])).join('\n') + expect(JSON.parse(printed)).toEqual({ success: true, output: { answer: 42 } }) + expect(printed).not.toContain('thinking out loud') + expect(stderr.mock.calls.map((call) => String(call[0])).join('')).toContain('thinking out loud') + }) + + it('still prints the envelope, then fails, when the run itself failed', async () => { + requestRaw.mockResolvedValue( + streamResponse( + sse({ event: 'final', data: { success: false, error: 'Block agent_1 failed', output: {} } }) + ) + ) + const stdout = vi.spyOn(console, 'log').mockImplementation(() => {}) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + + await expect(run('wf_1', '--follow')).rejects.toThrow(/Block agent_1 failed/) + expect(stdout).toHaveBeenCalled() + }) + + it('explains a deployment that answers JSON instead of an event stream', async () => { + const cancel = vi.fn().mockResolvedValue(undefined) + requestRaw.mockResolvedValue({ + body: { cancel }, + status: 200, + headers: new Headers({ 'content-type': 'application/json' }), + } as unknown as Response) + + await expect(run('wf_1', '--follow')).rejects.toThrow(/instead of an event stream/) + expect(cancel).toHaveBeenCalled() + }) + + it('reports a mid-stream failure as an explainable error, not a crash', async () => { + requestRaw.mockResolvedValue( + streamResponse(sse({ event: 'error', error: 'Execution cancelled' }, '[DONE]')) + ) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + + await expect(run('wf_1', '--follow')).rejects.toBeInstanceOf(SimApiError) + }) +}) diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts new file mode 100644 index 00000000000..811f5adbfce --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts @@ -0,0 +1,338 @@ +import chalk from 'chalk' +import type { Command } from 'commander' +import { clientFrom } from '../../context' +import { CLI_CONTRACT } from '../../contract/commands' +import { V2_OPERATIONS } from '../../generated/v2-api' +import { SimApiError } from '../../http/client' +import { safeOneLine, sanitize } from '../../output/render' +import { executeOperation } from '../../runtime/execute' +import { buildRequest } from '../../runtime/request' +import { renderResult } from '../../runtime/result' +import type { OperationSpec } from '../../runtime/types' + +/** + * Declares that this client understands agent-event framing. + * + * The server refuses `includeThinking` / `includeToolCalls` outright without it + * rather than silently downgrading, so the header is not optional decoration — + * it is the difference between the flags working and a 400. It is sent *only* + * when one of those flags is set, because negotiating also switches answer text + * to live streaming, which the server may later retract with `chunk_reset`. + * A plain `--follow` therefore stays on settled final-turn text, and nothing + * printed to the terminal can ever turn out to have been withdrawn. + */ +const AGENT_STREAM_PROTOCOL_HEADER = 'x-sim-stream-protocol' +const AGENT_STREAM_PROTOCOL_V1 = 'agent-events-v1' + +/** Terminal marker. Sent JSON-encoded, so the raw payload carries its quotes. */ +const DONE_SENTINEL = '[DONE]' + +/** The sink live commentary is written to; `process.stderr` satisfies it. */ +export interface CommentaryWriter { + write(text: string): unknown +} + +export interface FollowOptions { + includeThinking: boolean + includeToolCalls: boolean + stderr: CommentaryWriter +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function stringField(frame: Record, key: string): string | null { + const value = frame[key] + return typeof value === 'string' ? value : null +} + +/** + * Yields the payload of every `data:` line in an SSE body. + * + * Split on `\n` rather than on the `\n\n` event separator: a chunk boundary can + * fall between the two newlines, and a parser keyed on the pair would hold the + * completed event until the next one arrived — turning a live token stream into + * one that always lags a frame behind. + */ +async function* sseData(body: ReadableStream): AsyncGenerator { + const reader = body.getReader() + const decoder = new TextDecoder() + let buffer = '' + + try { + while (true) { + const { done, value } = await reader.read() + buffer += done ? decoder.decode() : decoder.decode(value, { stream: true }) + + const lines = buffer.split('\n') + buffer = done ? '' : (lines.pop() ?? '') + + for (const rawLine of lines) { + const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine + if (!line.startsWith('data:')) continue + const payload = line.slice(5).startsWith(' ') ? line.slice(6) : line.slice(5) + if (payload.length > 0) yield payload + } + + if (done) return + } + } finally { + reader.releaseLock() + } +} + +/** + * Writes run commentary while keeping raw answer text and line-oriented notices + * from colliding. + * + * Answer and thinking text arrive as deltas with no framing of their own, so a + * tool notice emitted mid-token would be appended to whatever half-sentence was + * already on the line. The cursor position is tracked instead of guessed. + */ +class Commentary { + private atLineStart = true + + constructor(private readonly sink: CommentaryWriter) {} + + inline(text: string): void { + if (text.length === 0) return + this.sink.write(text) + this.atLineStart = text.endsWith('\n') + } + + line(text: string): void { + this.sink.write(`${this.atLineStart ? '' : '\n'}${text}\n`) + this.atLineStart = true + } + + /** Closes a half-written delta line without inventing a blank one. */ + endLine(): void { + if (this.atLineStart) return + this.sink.write('\n') + this.atLineStart = true + } +} + +function toolNotice(frame: Record): string { + const name = safeOneLine(stringField(frame, 'name') ?? 'tool') + if (frame.phase === 'start') return chalk.dim(`→ ${name}`) + + const status = stringField(frame, 'status') + if (status && status !== 'success') return chalk.yellow(`✗ ${name} (${safeOneLine(status)})`) + return chalk.dim(`✓ ${name}`) +} + +/** + * Renders one execute stream, returning the `final` envelope. + * + * Everything rendered here is commentary and goes to `stderr`: the payload a + * script captures is the final envelope, which the caller prints to stdout in + * the profile's output format. Streaming the answer text to stdout as well + * would put the same content in the redirect twice, once unparseable. + * + * Throws on a terminal `error` frame and on a stream that stops before either + * terminal frame arrives — a truncated stream is a failed run, and reporting it + * as an empty success is the one outcome a caller cannot detect afterwards. + */ +export async function renderRunStream( + body: ReadableStream, + options: FollowOptions +): Promise> { + const commentary = new Commentary(options.stderr) + let final: Record | null = null + + for await (const payload of sseData(body)) { + let frame: unknown + try { + frame = JSON.parse(payload) + } catch { + // A `data:` line the CLI cannot parse is a protocol the CLI does not + // speak yet, not a failed run. Skipping keeps a server that adds a frame + // shape from breaking every older client that only wants the outcome. + continue + } + + if (frame === DONE_SENTINEL) break + if (!isRecord(frame)) continue + + if (frame.event === undefined && typeof frame.chunk === 'string') { + commentary.inline(sanitize(frame.chunk)) + continue + } + + switch (frame.event) { + case 'chunk_reset': + commentary.line(chalk.dim('… retracted; that turn resolved to tool calls')) + break + case 'thinking': + if (options.includeThinking && typeof frame.data === 'string') { + commentary.inline(chalk.dim(sanitize(frame.data))) + } + break + case 'tool': + if (options.includeToolCalls) commentary.line(toolNotice(frame)) + break + case 'stream_error': + commentary.line( + chalk.yellow( + `warning: ${safeOneLine(stringField(frame, 'error') ?? 'stream read failed')}` + ) + ) + break + case 'error': + commentary.endLine() + throw new SimApiError( + safeOneLine(stringField(frame, 'error') ?? 'The workflow run failed.'), + 0 + ) + case 'final': + if (isRecord(frame.data)) final = frame.data + break + default: + break + } + } + + commentary.endLine() + if (!final) { + throw new SimApiError( + 'The run stream ended before the workflow reported a result. The run may still be in progress — check: sim workflows runs list', + 0 + ) + } + return final +} + +/** + * Sends the run and renders it. Kept apart from the Commander wiring so the + * whole protocol can be exercised without parsing argv. + */ +async function followRun(workflowId: string, command: Command): Promise { + const flags = command.optsWithGlobals() as Record + + if (flags.async === true) { + throw new SimApiError( + '--follow streams a run as it happens and --async returns before it starts; pass one, not both', + 0 + ) + } + + const includeThinking = flags.includeThinking === true + const includeToolCalls = flags.includeToolCalls === true + const negotiates = includeThinking || includeToolCalls + + const { client, profile } = clientFrom(command) + const operation = V2_OPERATIONS.executeWorkflow as OperationSpec + const request = buildRequest('executeWorkflow', [workflowId], flags, profile.workspaceId) + + const response = await client.requestRaw(request.path, { + method: 'POST', + query: request.query, + body: { + ...(request.body ?? {}), + stream: true, + ...(includeThinking ? { includeThinking: true } : {}), + ...(includeToolCalls ? { includeToolCalls: true } : {}), + }, + headers: { + accept: 'text/event-stream', + ...(negotiates ? { [AGENT_STREAM_PROTOCOL_HEADER]: AGENT_STREAM_PROTOCOL_V1 } : {}), + }, + }) + + const contentType = response.headers.get('content-type') ?? '' + if (!contentType.toLowerCase().includes('text/event-stream')) { + await response.body?.cancel() + throw new SimApiError( + `${operation.path} answered ${contentType || 'an unknown content type'} instead of an event stream. This deployment may predate streaming runs — re-run without --follow.`, + response.status + ) + } + if (!response.body) { + throw new SimApiError('The run stream had no body.', response.status) + } + + const final = await renderRunStream(response.body, { + includeThinking, + includeToolCalls, + stderr: process.stderr, + }) + + renderResult('executeWorkflow', profile.output, final, CLI_CONTRACT.executeWorkflow ?? {}) + + // Printed first, then failed: the envelope carries the block outputs that + // explain *why* the run failed, and exiting before writing it would leave a + // piped consumer with an exit code and nothing to read. + if (final.success === false) { + throw new SimApiError( + safeOneLine(typeof final.error === 'string' ? final.error : 'The workflow run failed.'), + 0 + ) + } +} + +/** + * The handler Commander invokes for the generated `run` leaf, given the handler + * it is replacing. + */ +function followOrDelegate(previous: ((args: unknown[]) => unknown) | null) { + return async (workflowId: string, _options: unknown, command: Command): Promise => { + const flags = command.optsWithGlobals() as Record + + if (flags.follow !== true) { + if (flags.includeThinking === true || flags.includeToolCalls === true) { + throw new SimApiError( + '--include-thinking and --include-tool-calls describe a stream; add --follow', + 0 + ) + } + // Whatever was installed before wins, so a second augmentation of the + // same leaf composes with this one instead of replacing it. + if (previous) { + await previous(command.processedArgs) + return + } + await executeOperation( + 'executeWorkflow', + CLI_CONTRACT.executeWorkflow ?? {}, + V2_OPERATIONS.executeWorkflow as OperationSpec, + [workflowId, command.opts(), command] + ) + return + } + + await followRun(workflowId, command) + } +} + +/** + * Teaches the generated `workflows run` leaf to stream. + * + * `--follow` rides on `run` rather than standing up a sibling command because + * it is the same operation with a different response encoding: the input, + * output-selection, and `--async` flags all still apply, and a second command + * would have to restate every one of them and then drift. + * + * Commander offers no way to read the action it already holds, so the existing + * handler is captured and delegated to — every non-`--follow` invocation still + * runs the generated path byte for byte. + */ +export function attachWorkflowRunFollow(workflows: Command): void { + const run = workflows.commands.find((command) => command.name() === 'run') + if (!run) { + throw new Error('workflows run must be registered before --follow can be attached to it') + } + + const held = (run as Command & { _actionHandler?: unknown })._actionHandler + const previous = typeof held === 'function' ? (held as (args: unknown[]) => unknown) : null + + run + .option( + '--follow', + 'Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns' + ) + .option('--include-thinking', 'Show model reasoning while following (requires --follow)') + .option('--include-tool-calls', 'Show tool calls while following (requires --follow)') + .action(followOrDelegate(previous)) +} diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-wait.test.ts b/packages/sim-cli/src/commands/protocol/workflow-run-wait.test.ts new file mode 100644 index 00000000000..6174458ab29 --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/workflow-run-wait.test.ts @@ -0,0 +1,289 @@ +/** + * @vitest-environment node + */ +import { Command } from 'commander' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { attachWorkflowRunWait } from './workflow-run-wait' + +const { mockRequest, sleeps, clock, output } = vi.hoisted(() => ({ + mockRequest: vi.fn(), + sleeps: [] as number[], + clock: { now: 0 }, + output: { format: 'text' }, +})) + +vi.mock('../../context', () => ({ + clientFrom: () => ({ + client: { request: mockRequest, requireWorkspace: () => 'ws_local' }, + profile: { + workspaceId: 'ws_local', + output: output.format, + name: 'default', + apiKey: 'k', + endpoint: 'https://sim.example', + }, + }), +})) + +/** + * Waiting is simulated rather than timed: the mock advances the same clock the + * command reads its deadline from, so a `--wait-timeout 3600` test costs + * nothing and the poll schedule is exactly what the assertions say it is. + */ +vi.mock('../../helpers', () => ({ + sleep: (ms: number) => { + sleeps.push(ms) + clock.now += ms + return Promise.resolve() + }, +})) + +let logged: string[] +let errored: string[] +let progress: string[] + +const stderr = process.stderr as unknown as { isTTY: boolean } +const realIsTTY = stderr.isTTY + +beforeEach(() => { + vi.restoreAllMocks() + mockRequest.mockReset() + sleeps.length = 0 + clock.now = 1_000_000 + output.format = 'text' + logged = [] + errored = [] + progress = [] + process.exitCode = undefined + stderr.isTTY = false + + vi.spyOn(Date, 'now').mockImplementation(() => clock.now) + vi.spyOn(console, 'log').mockImplementation((line: string) => { + logged.push(line) + }) + vi.spyOn(console, 'error').mockImplementation((line: string) => { + errored.push(line) + }) + vi.spyOn(process.stderr, 'write').mockImplementation((chunk: string | Uint8Array) => { + progress.push(String(chunk)) + return true + }) +}) + +afterEach(() => { + vi.restoreAllMocks() + stderr.isTTY = realIsTTY + process.exitCode = undefined +}) + +interface RunPayload { + status: string + paused?: { + contextId?: string | null + pauseKind?: string | null + resumeAt?: string | null + } | null +} + +function run(payload: RunPayload) { + return { + data: { + runId: 'run_1', + workflowId: 'wf_1', + startedAt: '2026-08-17T00:00:00.000Z', + endedAt: null, + durationMs: null, + cost: null, + error: payload.status === 'failed' ? { message: 'Block agent_1 exploded' } : null, + paused: null, + ...payload, + }, + } +} + +/** Answers each poll in order, repeating the last payload once the script runs out. */ +function respondWith(...payloads: RunPayload[]) { + let index = 0 + mockRequest.mockImplementation(() => { + const payload = payloads[Math.min(index, payloads.length - 1)] + index += 1 + return Promise.resolve(run(payload)) + }) +} + +async function wait(argv: string[] = []): Promise { + const root = new Command('sim').exitOverride() + const workflows = new Command('workflows').exitOverride() + const runs = new Command('runs').exitOverride() + workflows.addCommand(runs) + root.addCommand(workflows) + attachWorkflowRunWait(runs) + runs.commands.forEach((command) => command.exitOverride()) + + await root.parseAsync([ + 'node', + 'sim', + 'workflows', + 'runs', + 'wait', + 'run_1', + '--workflow', + 'wf_1', + ...argv, + ]) +} + +describe('workflows runs wait', () => { + it('polls until the run reaches a terminal state', async () => { + respondWith({ status: 'queued' }, { status: 'running' }, { status: 'completed' }) + + await wait() + + expect(mockRequest).toHaveBeenCalledTimes(3) + expect(mockRequest).toHaveBeenCalledWith('/api/v2/workflows/wf_1/runs/run_1', { method: 'GET' }) + expect(process.exitCode).toBe(0) + expect(logged.join('\n')).toContain('completed') + }) + + it('keeps waiting through the states the server leaves on its own', async () => { + respondWith({ status: 'pending' }, { status: 'redacting' }, { status: 'completed' }) + + await wait() + + expect(mockRequest).toHaveBeenCalledTimes(3) + expect(process.exitCode).toBe(0) + }) + + it('exits non-zero when the run failed', async () => { + respondWith({ status: 'running' }, { status: 'failed' }) + + await wait() + + expect(process.exitCode).toBe(1) + expect(errored.join('\n')).toContain('Run run_1 failed') + expect(logged.join('\n')).toContain('Block agent_1 exploded') + }) + + it('distinguishes a cancelled run from a failed one', async () => { + respondWith({ status: 'cancelled' }) + + await wait() + + expect(process.exitCode).toBe(2) + expect(errored.join('\n')).toContain('cancelled') + }) + + it('keeps polling a run paused until a time it will resume itself', async () => { + respondWith( + { status: 'paused', paused: { pauseKind: 'time', resumeAt: '2026-08-17T00:01:00.000Z' } }, + { status: 'completed' } + ) + + await wait() + + expect(mockRequest).toHaveBeenCalledTimes(2) + expect(process.exitCode).toBe(0) + }) + + it('stops on a pause that is waiting for a human, and says how to resume it', async () => { + respondWith({ status: 'paused', paused: { pauseKind: 'human', contextId: 'ctx_9' } }) + + await wait() + + expect(mockRequest).toHaveBeenCalledTimes(1) + expect(process.exitCode).toBe(3) + expect(errored.join('\n')).toContain( + 'sim workflows runs resume run_1 --workflow wf_1 --context ctx_9' + ) + }) + + it('stops on a pause whose kind the server did not name', async () => { + respondWith({ status: 'paused', paused: { pauseKind: null } }) + + await wait() + + expect(process.exitCode).toBe(3) + }) + + it('spaces the polls out, backing off to a ceiling', async () => { + respondWith( + { status: 'running' }, + { status: 'running' }, + { status: 'running' }, + { status: 'running' }, + { status: 'running' }, + { status: 'completed' } + ) + + await wait() + + expect(sleeps).toEqual([2000, 4000, 8000, 15000, 15000]) + }) + + it('gives up when the wait bound elapses, without overshooting it', async () => { + respondWith({ status: 'running' }) + + await wait(['--wait-timeout', '3']) + + expect(sleeps).toEqual([2000, 1000]) + expect(process.exitCode).toBe(4) + expect(errored.join('\n')).toContain('Timed out after 3s') + expect(errored.join('\n')).toContain('status: running') + expect(logged.join('\n')).toContain('running') + }) + + it('waits indefinitely when the bound is zero', async () => { + respondWith( + { status: 'running' }, + { status: 'running' }, + { status: 'running' }, + { status: 'completed' } + ) + + await wait(['--wait-timeout', '0']) + + expect(process.exitCode).toBe(0) + expect(mockRequest).toHaveBeenCalledTimes(4) + }) + + it('rejects a wait bound that is not a number of seconds', async () => { + respondWith({ status: 'completed' }) + + await expect(wait(['--wait-timeout', 'soon'])).rejects.toThrow(/--wait-timeout/) + await expect(wait(['--wait-timeout', '-5'])).rejects.toThrow(/non-negative/) + expect(mockRequest).not.toHaveBeenCalled() + }) + + it('requires the workflow the run belongs to', async () => { + const root = new Command('sim').exitOverride() + const runs = new Command('runs').exitOverride() + root.addCommand(runs) + attachWorkflowRunWait(runs) + runs.commands.forEach((command) => command.exitOverride()) + + await expect(root.parseAsync(['node', 'sim', 'runs', 'wait', 'run_1'])).rejects.toThrow( + /--workflow/ + ) + }) + + it('keeps progress on stderr and the result on stdout', async () => { + stderr.isTTY = true + output.format = 'json' + respondWith({ status: 'running' }, { status: 'completed' }) + + await wait() + + expect(progress.join('')).toContain('running') + expect(logged).toHaveLength(1) + expect(JSON.parse(logged[0])).toMatchObject({ runId: 'run_1', status: 'completed' }) + expect(logged.join('')).not.toContain('waiting') + }) + + it('writes no progress when stderr is not a terminal', async () => { + respondWith({ status: 'running' }, { status: 'completed' }) + + await wait() + + expect(progress).toEqual([]) + }) +}) diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts b/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts new file mode 100644 index 00000000000..13751c61418 --- /dev/null +++ b/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts @@ -0,0 +1,288 @@ +import chalk from 'chalk' +import { type Command, Option } from 'commander' +import { clientFrom } from '../../context' +import { CLI_CONTRACT } from '../../contract/commands' +import type { CommandSpec } from '../../contract/types' +import { V2_OPERATIONS } from '../../generated/v2-api' +import { sleep } from '../../helpers' +import { resolvePath, SimApiError } from '../../http/client' +import { renderResult } from '../../runtime/result' + +/** + * Statuses a run never leaves. + * + * Taken from the reported enum of `GET /api/v2/workflows/[id]/runs/[runId]`, + * which is `PERSISTED_WORKFLOW_EXECUTION_STATUSES` plus `queued`. The four + * absent from this set are all states the server moves out of on its own: + * `queued` and `pending` precede execution, `running` is it, and `redacting` is + * the window where a finished run's output is scrubbed — stopping there would + * report a run as done while its record is still being rewritten. + */ +const TERMINAL_STATUSES = new Set(['completed', 'failed', 'cancelled']) + +/** + * How a wait ended, and the exit status that says so. + * + * Exit codes are the whole point of this command: a CI step runs it precisely so + * that a failed run fails the build. They follow `whoami`'s split — 1 is the + * CLI's blanket "explained failure" and here means the run itself failed, while + * everything above it is a distinct outcome a script has to branch on. + * Cancelling is somebody's decision, a pause is a prompt to act, and giving up + * on the clock says nothing about the run at all; collapsing any of them into 1 + * would make all three read as "the workflow broke". + */ +const WAIT_EXIT_CODES = { + completed: 0, + failed: 1, + cancelled: 2, + paused: 3, + timeout: 4, +} as const + +type WaitOutcome = keyof typeof WAIT_EXIT_CODES + +/** + * How long to wait between polls, in milliseconds. + * + * A run that has already finished answers on the first request, so the first + * delay is only ever paid by a run that is genuinely still going. The interval + * then backs off to a ceiling: an hour-long run should not cost a request every + * two seconds, and nobody watching a `wait` in CI can tell a two-second refresh + * from a fifteen-second one. + * + * Deliberately without jitter, unlike the retry pacing elsewhere in the + * monorepo. Jitter exists to break up a herd of clients hammering the same + * failed dependency in lockstep; one terminal polling one run is not that, and a + * predictable schedule is one fewer thing to explain when someone counts the + * requests. + */ +const FIRST_POLL_DELAY_MS = 2_000 +const MAX_POLL_DELAY_MS = 15_000 +const POLL_BACKOFF_FACTOR = 2 + +/** + * How long to wait in total before giving up, in seconds. + * + * Matches the ceiling a paid plan puts on a single run, so the default never + * abandons a run the server would still have finished. `0` waits indefinitely, + * the same escape hatch `SIM_TIMEOUT_SECONDS` offers. + */ +const DEFAULT_WAIT_TIMEOUT_SECONDS = 3600 + +/** + * Spelled `--wait-timeout` rather than `--timeout`, because `SIM_TIMEOUT_SECONDS` + * already bounds a single HTTP request while this bounds the whole wait across + * many of them. Two knobs both called "timeout" would be read as one, and the + * failure that follows is silent — raising the wrong one changes nothing. + */ +const WAIT_TIMEOUT_FLAG = '--wait-timeout ' + +interface RunSnapshot { + status: string + /** `time` resumes itself; `human` and null wait for a person. */ + pauseKind: string | null + resumeAt: string | null + contextId: string | null +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function optionalString(value: unknown): string | null { + return typeof value === 'string' && value !== '' ? value : null +} + +/** + * Reads the fields the loop branches on out of the run response. + * + * v2 answers `{ data: … }`, but the payload is unwrapped tolerantly so a bare + * record — what a self-hosted deployment behind an unwrapping proxy hands back — + * still polls, rather than refusing to find a status that is right there. + */ +function readRun(raw: unknown): RunSnapshot { + const run = isRecord(raw) && isRecord(raw.data) ? raw.data : raw + if (!isRecord(run) || typeof run.status !== 'string') { + throw new SimApiError('Run status response carried no status.', 0) + } + const paused = isRecord(run.paused) ? run.paused : null + return { + status: run.status, + pauseKind: paused ? optionalString(paused.pauseKind) : null, + resumeAt: paused ? optionalString(paused.resumeAt) : null, + contextId: paused ? optionalString(paused.contextId) : null, + } +} + +/** + * The outcome a snapshot settles the wait as, or null to keep polling. + * + * A `paused` run is the judgement call here. A pause waiting on time resumes + * itself, so the run is still making progress and the loop keeps going. A pause + * waiting on a human never resolves without one — polling it to the deadline + * would burn the entire budget and then report a timeout, describing a run doing + * exactly what it was asked to as a run that failed to answer. So a human pause + * ends the wait and names the command that resumes it. An unspecified + * `pauseKind` is treated the same way: the response makes no promise that it + * will move on by itself, and a wait that might never end is worse than one that + * stops and explains itself. + */ +function classify(snapshot: RunSnapshot): WaitOutcome | null { + if (snapshot.status === 'paused') return snapshot.pauseKind === 'time' ? null : 'paused' + if (!TERMINAL_STATUSES.has(snapshot.status)) return null + return snapshot.status === 'completed' + ? 'completed' + : snapshot.status === 'cancelled' + ? 'cancelled' + : 'failed' +} + +interface WaitProgress { + advance: (status: string, elapsedMs: number) => void + finish: () => void +} + +/** + * Reports what the run is doing while the wait continues. + * + * stderr and TTY-only, the rule `pageProgress` already follows: stdout carries + * the finished run so `sim workflows runs wait … > result.json` stays a usable + * file, and a redirected stderr should not collect a log of half-erased status + * lines. + */ +function waitProgress(): WaitProgress { + let reported = false + return { + advance: (status, elapsedMs) => { + if (!process.stderr.isTTY) return + reported = true + process.stderr.write( + `\r${chalk.dim(`${status} — waiting ${Math.round(elapsedMs / 1000)}s…`)}\u001b[K` + ) + }, + // Idempotent: the loop clears the line before printing a result, and the + // `finally` clears it again for the path that threw. Erasing twice would + // write a stray control sequence onto output already on the stream. + finish: () => { + if (!reported) return + reported = false + process.stderr.write('\r\u001b[K') + }, + } +} + +function parseWaitTimeout(raw: string): number { + const seconds = Number(raw) + if (!Number.isFinite(seconds) || seconds < 0) { + throw new SimApiError( + `Invalid ${WAIT_TIMEOUT_FLAG} "${raw}". Use a non-negative number of seconds, or 0 to wait indefinitely.`, + 0 + ) + } + return seconds +} + +/** The one line explaining an outcome the caller is about to see as a non-zero exit. */ +function explain( + outcome: WaitOutcome, + runId: string, + workflowId: string, + snapshot: RunSnapshot +): string | null { + if (outcome === 'completed') return null + if (outcome === 'failed') return `Run ${runId} failed.` + if (outcome === 'cancelled') return `Run ${runId} was cancelled.` + const context = snapshot.contextId ? ` --context ${snapshot.contextId}` : '' + return `Run ${runId} is paused waiting for input. Resume it: sim workflows runs resume ${runId} --workflow ${workflowId}${context}` +} + +/** + * The fields `workflows runs get` renders, so a waited run and a polled one read + * identically. Read from the contract rather than restated here: a field added + * to the sibling should reach both without anyone remembering this file exists. + */ +function runSpec(): CommandSpec { + return CLI_CONTRACT.getWorkflowRun ?? {} +} + +/** Adds `workflows runs wait` — poll one run until it stops moving. */ +export function attachWorkflowRunWait(runs: Command): void { + runs + .command('wait') + .argument('', V2_OPERATIONS.getWorkflowRun.pathParamDocs?.runId) + .description('Wait for a run to reach a terminal state, then show it') + .addOption( + new Option('--workflow ', 'Workflow ID (required)').makeOptionMandatory() + ) + .addOption( + new Option( + WAIT_TIMEOUT_FLAG, + `Give up after this many seconds, or 0 to wait indefinitely (default: ${DEFAULT_WAIT_TIMEOUT_SECONDS}). Bounds the whole wait; SIM_TIMEOUT_SECONDS bounds one request` + ) + ) + .action( + async ( + runId: string, + options: { workflow: string; waitTimeout?: string }, + command: Command + ) => { + const timeoutSeconds = + options.waitTimeout === undefined + ? DEFAULT_WAIT_TIMEOUT_SECONDS + : parseWaitTimeout(options.waitTimeout) + + const { client, profile } = clientFrom(command) + const operation = V2_OPERATIONS.getWorkflowRun + const path = resolvePath(operation.path, { id: options.workflow, runId }) + + const startedAt = Date.now() + const deadline = + timeoutSeconds === 0 ? Number.POSITIVE_INFINITY : startedAt + timeoutSeconds * 1000 + const progress = waitProgress() + let delayMs = FIRST_POLL_DELAY_MS + + // `finally`, because a request that throws part-way through would + // otherwise leave `running — waiting 12s…` sitting on the line the error + // is then written onto. + try { + while (true) { + const raw = await client.request(path, { method: operation.method }) + const snapshot = readRun(raw) + const outcome = classify(snapshot) + + if (outcome) { + progress.finish() + renderResult('getWorkflowRun', profile.output, raw, runSpec()) + const message = explain(outcome, runId, options.workflow, snapshot) + if (message) console.error(chalk.red(message)) + process.exitCode = WAIT_EXIT_CODES[outcome] + return + } + + const remainingMs = deadline - Date.now() + if (remainingMs <= 0) { + progress.finish() + renderResult('getWorkflowRun', profile.output, raw, runSpec()) + console.error( + chalk.red( + `Timed out after ${timeoutSeconds}s waiting for run ${runId} (status: ${snapshot.status}${ + snapshot.resumeAt ? `, resuming at ${snapshot.resumeAt}` : '' + }). Raise ${WAIT_TIMEOUT_FLAG}, or set it to 0 to wait indefinitely.` + ) + ) + process.exitCode = WAIT_EXIT_CODES.timeout + return + } + + progress.advance(snapshot.status, Date.now() - startedAt) + // Clamped to the time left so the last sleep of a bounded wait ends + // at the deadline instead of overshooting it by a whole interval. + await sleep(Math.min(delayMs, remainingMs)) + delayMs = Math.min(delayMs * POLL_BACKOFF_FACTOR, MAX_POLL_DELAY_MS) + } + } finally { + progress.finish() + } + } + ) +} diff --git a/packages/sim-cli/src/contract/commands.ts b/packages/sim-cli/src/contract/commands.ts index f276a443305..c90fd06067e 100644 --- a/packages/sim-cli/src/contract/commands.ts +++ b/packages/sim-cli/src/contract/commands.ts @@ -814,10 +814,10 @@ export const CLI_CONTRACT: CliContract = { describe: 'Return blockName.field values (e.g. agent_1.content); missing fields are omitted', }, - // SSE, not JSON — the generic client cannot consume it. A `sim workflows - // run --follow` that renders the stream is a separate, hand-written - // command; advertising a flag that breaks the response is worse than - // not offering it yet. + // SSE, not JSON — the generic client cannot consume it, so the response + // encoding is chosen by `--follow`, which `workflow-run-follow.ts` adds to + // this same leaf and renders by hand. These stay omitted because sending + // them down the generated path would still break it. stream: { omit: true }, includeThinking: { omit: true }, includeToolCalls: { omit: true }, From 218eca119feba9da2fa75e250f8ba62fe2ce2be0 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 18 Aug 2026 11:15:16 -0700 Subject: [PATCH 2/4] fix(cli): disclose a truncated burst, and clear a stale retry notice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings in `logs follow`, both verified against the code first. The page budget bounds one poll so an enormous burst cannot stall the follow, but on reaching it the live cursor was discarded: the remainder is older than everything collected and the next poll restarts at the newest page, so those runs were never printed and nothing said so. The budget stays — draining without one trades a bounded poll for unbounded buffering in a process meant to run for hours — but hitting it now warns on stderr, naming the count and pointing at `sim logs list`. That notice is written even off a terminal, because a piped log is where an unexplained hole is hardest to spot. The retry notice was cleared after the empty-rows check, so a poll that recovered but found nothing left "retrying in Ns…" on screen while the follow was already healthy. Clearing now happens as soon as a poll succeeds. The second test needed two failures to be worth anything: the teardown clears the line either way, so what separates fixed from broken is whether a bare erase lands before the second notice or only at the end. The first version passed against the bug. --- .../src/commands/protocol/logs-follow.test.ts | 45 ++++++++++++ .../src/commands/protocol/logs-follow.ts | 73 +++++++++++++++---- 2 files changed, 103 insertions(+), 15 deletions(-) diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts index 05818a6b210..18619d94052 100644 --- a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts +++ b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts @@ -247,6 +247,51 @@ describe('sim logs follow', () => { expect(stdout.join('')).not.toContain('retrying in') }) + it('says so when a burst is larger than one poll may read', async () => { + // The page budget bounds one poll so an enormous burst cannot stall the + // follow, but the remainder is older than everything collected and the next + // poll restarts at the newest page — so those runs are never coming, and a + // hole the reader cannot see is worse than a slow poll. + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const seed = row('seed', '2026-08-17T10:00:00.000Z') + const budgeted = Array.from({ length: 10 }, (_, index) => + page([row(`burst_${index}`, `2026-08-17T10:01:0${index}.000Z`)], `cursor_${index}`) + ) + respondWith([page([seed]), ...budgeted]) + + await follow('-n', '1') + + expect(stderr.join('')).toContain('older ones were skipped') + expect(stdout.join('')).not.toContain('older ones were skipped') + }) + + it('clears a retry notice on the first healthy poll, even an empty one', async () => { + // The clear used to sit after the empty check, so a poll that recovered but + // found nothing left "retrying in Ns…" up while the follow was already + // healthy. Asserted between two failures because the teardown clears the + // line either way: what separates the two is whether a bare erase lands + // BEFORE the second notice, or only at the end. + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const first = row('run_1', '2026-08-17T10:00:01.000Z') + respondWith([ + page([first]), + new SimApiError('Service Unavailable', 503), + page([first]), + new SimApiError('Service Unavailable', 503), + page([first]), + ]) + + await follow('-n', '1') + + const bareErase = stderr.indexOf(`\r${String.fromCharCode(27)}[K`) + const notices = stderr + .map((line, index) => (line.includes('retrying in') ? index : -1)) + .filter((index) => index >= 0) + expect(notices).toHaveLength(2) + expect(bareErase).toBeGreaterThan(notices[0]) + expect(bareErase).toBeLessThan(notices[1]) + }) + it('stays silent on stderr when it is not a terminal', async () => { Object.defineProperty(process.stderr, 'isTTY', { value: false, configurable: true }) const first = row('run_1', '2026-08-17T10:00:01.000Z') diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.ts b/packages/sim-cli/src/commands/protocol/logs-follow.ts index b1dcf8a541f..7944271abf9 100644 --- a/packages/sim-cli/src/commands/protocol/logs-follow.ts +++ b/packages/sim-cli/src/commands/protocol/logs-follow.ts @@ -247,6 +247,15 @@ function createWriter(format: OutputFormat): RowWriter { export interface FollowStatus { /** Replaces the status line, if there is a terminal to draw it on. */ note: (message: string) => void + /** + * Reports something the reader has to know, on its own line. + * + * Unlike {@link note} this is not progress and is never erased: it records + * that rows are missing, which stays true after the follow moves on. It is + * written even when stderr is not a terminal, because a piped log is exactly + * where an unexplained hole is hardest to spot. + */ + warn: (message: string) => void /** Erases the line, if anything was ever written to it. */ clear: () => void } @@ -265,6 +274,13 @@ export function followStatus(): FollowStatus { reported = true process.stderr.write(`\r${chalk.dim(message)}${ERASE_LINE}`) }, + warn: (message) => { + if (reported) { + reported = false + process.stderr.write(`\r${ERASE_LINE}`) + } + process.stderr.write(`warning: ${message}\n`) + }, clear: () => { if (!reported) return reported = false @@ -370,6 +386,19 @@ function remember(state: FollowState, rows: LogRow[]): void { * holding one known row proves the rest of the list is older still, so walking * further would only re-read history the floor rejects anyway. */ +/** + * One poll's worth of new rows, and whether the page budget cut it short. + * + * A follow reads a bounded number of pages per poll so one enormous burst + * cannot stall it indefinitely or buffer without limit. Reaching that bound + * means older runs from the same burst will never be printed, which is worth + * saying out loud rather than leaving the reader to notice a hole later. + */ +interface PollBatch { + rows: LogRow[] + truncated: boolean +} + async function collectUnprinted( client: Pick, path: string, @@ -377,23 +406,28 @@ async function collectUnprinted( state: FollowState, pageSize: number, maxPages: number -): Promise { - const fresh: LogRow[] = [] +): Promise { + const rows: LogRow[] = [] let cursor: string | null = null + let truncated = false for (let page = 0; page < maxPages; page += 1) { const response: ListLogsResponse = await client.request(path, { query: { ...query, limit: pageSize, cursor }, }) - const rows = response?.data ?? [] - const unprinted = rows.filter((row) => isUnprinted(state, row)) - fresh.push(...unprinted) + const page_rows = response?.data ?? [] + const unprinted = page_rows.filter((row) => isUnprinted(state, row)) + rows.push(...unprinted) cursor = response?.nextCursor ?? null - if (!cursor || rows.length === 0 || unprinted.length < rows.length) break + if (!cursor || page_rows.length === 0 || unprinted.length < page_rows.length) break + // Every page so far was new and another is waiting, so the burst is larger + // than one poll may read. The remainder is older than everything collected + // here and the next poll restarts at the newest page, so it is not coming. + if (page === maxPages - 1) truncated = true } - return fresh + return { rows, truncated } } /** @@ -511,10 +545,10 @@ Examples: // The backlog page doubles as the seed: every run on it is recorded and // its oldest start time becomes the floor, so even `-n 0` anchors the // follow to now instead of replaying the workspace's whole history. - const backlog = await collectUnprinted(client, path, query, state, Math.max(lines, 1), 1) - remember(state, backlog) - state.floor = backlog.at(-1)?.startedAt ?? null - write(lines > 0 ? backlog.slice(0, lines).reverse() : []) + const seed = await collectUnprinted(client, path, query, state, Math.max(lines, 1), 1) + remember(state, seed.rows) + state.floor = seed.rows.at(-1)?.startedAt ?? null + write(lines > 0 ? seed.rows.slice(0, lines).reverse() : []) let failures = 0 while (!interrupt.interrupted()) { @@ -524,7 +558,7 @@ Examples: ) if (interrupt.interrupted()) break - let fresh: LogRow[] + let fresh: PollBatch try { fresh = await collectUnprinted( client, @@ -544,13 +578,22 @@ Examples: continue } + // Cleared on success rather than after the empty check: a poll that + // recovers but finds nothing still ends the retry, and leaving the + // notice up until rows happen to arrive reports a healthy follow as + // still failing. failures = 0 - if (fresh.length === 0) continue status.clear() - remember(state, fresh) + if (fresh.truncated) { + status.warn( + `more than ${MAX_PAGES_PER_POLL * POLL_PAGE_SIZE} runs arrived at once; older ones were skipped — see sim logs list` + ) + } + if (fresh.rows.length === 0) continue + remember(state, fresh.rows) // Reversed because the API answers newest-first while a terminal reads // downwards: the newest run has to end up on the last line. - write(fresh.reverse()) + write(fresh.rows.reverse()) } } finally { status.clear() From f677acb17b84e83e38d55dd1eddddfe19c86cb7f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 18 Aug 2026 11:23:12 -0700 Subject: [PATCH 3/4] test(cli): pin that a mixed page is the watermark, not a truncation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A page holding a run already printed proves the follow caught up, so the truncation warning must not fire there — that is how every healthy poll terminates, and warning would report a hole on the ordinary path. The straggler sharing that page is still collected, because the filter takes every unprinted row on it rather than only those above the known one. --- .../src/commands/protocol/logs-follow.test.ts | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts index 18619d94052..06e65bc48b2 100644 --- a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts +++ b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts @@ -265,6 +265,26 @@ describe('sim logs follow', () => { expect(stdout.join('')).not.toContain('older ones were skipped') }) + it('does not warn when the last budgeted page proves the follow caught up', async () => { + // A page holding a run already printed is the watermark: everything below it + // is older, so stopping there is the correct terminus, not a truncation. + // Warning here would report a hole on the ordinary steady-state path, and + // the run sharing that page is still collected because the filter takes + // every unprinted row on it, not only those above the known one. + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const seen = row('seed', '2026-08-17T10:00:00.000Z') + const budgeted = Array.from({ length: 9 }, (_, index) => + page([row(`burst_${index}`, `2026-08-17T10:01:0${index}.000Z`)], `cursor_${index}`) + ) + const mixed = page([row('straggler', '2026-08-17T10:00:30.000Z'), seen], null) + respondWith([page([seen]), ...budgeted, mixed]) + + await follow('-n', '1') + + expect(stderr.join('')).not.toContain('older ones were skipped') + expect(printedRunIds()).toContain('straggler') + }) + it('clears a retry notice on the first healthy poll, even an empty one', async () => { // The clear used to sit after the empty check, so a poll that recovered but // found nothing left "retrying in Ns…" up while the follow was already From 0d60b708a3d79119b2947ef246c6879e0a2e6844 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 18 Aug 2026 11:32:22 -0700 Subject: [PATCH 4/4] fix(cli): say when the requested backlog was larger than a page holds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The logs API clamps `limit` into 1–1000 rather than rejecting it, so `logs follow -n 5000` came back with 1000 rows, anchored the floor to that partial page, and said nothing. The seed already knew — it computes whether a live cursor remained — but the caller discarded the answer. Guarded on both halves. Fewer rows than asked for is only a shortfall when more were waiting: a workspace holding ten runs answers `-n 50` with ten and nothing is missing, so warning on the row count alone would fire on every small workspace. The cursor is what separates the two. --- .../src/commands/protocol/logs-follow.test.ts | 27 +++++++++++++++++++ .../src/commands/protocol/logs-follow.ts | 9 +++++++ 2 files changed, 36 insertions(+) diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts index 06e65bc48b2..0281da56c49 100644 --- a/packages/sim-cli/src/commands/protocol/logs-follow.test.ts +++ b/packages/sim-cli/src/commands/protocol/logs-follow.test.ts @@ -265,6 +265,33 @@ describe('sim logs follow', () => { expect(stdout.join('')).not.toContain('older ones were skipped') }) + it('says so when the requested backlog is larger than a page holds', async () => { + // The API clamps limit into 1–1000 instead of rejecting, so -n above that + // comes back short and the follow anchors its floor to the partial page. + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const rows = Array.from({ length: 3 }, (_, index) => + row(`run_${index}`, `2026-08-17T10:00:0${index}.000Z`) + ) + respondWith([page(rows, 'more'), page(rows, 'more')]) + + await follow('-n', '50') + + expect(stderr.join('')).toContain('asked for 50 earlier runs') + expect(stdout.join('')).not.toContain('asked for 50') + }) + + it('stays quiet when the workspace simply holds fewer runs than asked for', async () => { + // Short because there is no more to give is not a shortfall, and warning + // there would fire on every small workspace. + Object.defineProperty(process.stderr, 'isTTY', { value: true, configurable: true }) + const rows = [row('run_0', '2026-08-17T10:00:00.000Z')] + respondWith([page(rows, null), page(rows, null)]) + + await follow('-n', '50') + + expect(stderr.join('')).not.toContain('asked for 50') + }) + it('does not warn when the last budgeted page proves the follow caught up', async () => { // A page holding a run already printed is the watermark: everything below it // is older, so stopping there is the correct terminus, not a truncation. diff --git a/packages/sim-cli/src/commands/protocol/logs-follow.ts b/packages/sim-cli/src/commands/protocol/logs-follow.ts index 7944271abf9..5ab2d498696 100644 --- a/packages/sim-cli/src/commands/protocol/logs-follow.ts +++ b/packages/sim-cli/src/commands/protocol/logs-follow.ts @@ -548,6 +548,15 @@ Examples: const seed = await collectUnprinted(client, path, query, state, Math.max(lines, 1), 1) remember(state, seed.rows) state.floor = seed.rows.at(-1)?.startedAt ?? null + // The API clamps `limit` into 1–1000 rather than rejecting it, so a + // larger `-n` comes back short with no indication. Fewer rows than asked + // for is only a shortfall when more were waiting: a workspace holding + // ten runs answers `-n 50` with ten and nothing is missing. + if (seed.truncated && seed.rows.length < lines) { + status.warn( + `asked for ${lines} earlier runs but a page holds ${seed.rows.length}; following from there — see sim logs list for more` + ) + } write(lines > 0 ? seed.rows.slice(0, lines).reverse() : []) let failures = 0