mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
feat(cli): follow a run, wait for one, and tail the log (#6813)
* feat(cli): follow a run, wait for one, and tail the log 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. * fix(cli): disclose a truncated burst, and clear a stale retry notice 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. * test(cli): pin that a mixed page is the watermark, not a truncation 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. * fix(cli): say when the requested backlog was larger than a page holds 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.
This commit is contained in:
Waleed
@@ -66,3 +66,25 @@ sim logs list [options]
|
||||
| `--folder <value...>` | No | Folder path as shown in the app; the leading / is optional (space-separated, or @path / @- with one value per line). |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
## Watch runs as they arrive, printing each new run once
|
||||
|
||||
```bash
|
||||
sim logs follow [options]
|
||||
```
|
||||
|
||||
**Options**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Option | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `--workflow <id>` | No | Only follow runs of this workflow (repeatable). Defaults to ``. |
|
||||
| `--folder <path>` | No | Only follow runs of workflows in this folder (repeatable). Defaults to ``. |
|
||||
| `--trigger <type>` | No | Only follow runs with this trigger type (repeatable). Defaults to ``. |
|
||||
| `--level <level>` | No | Only follow runs at this severity. Accepted values: `info`, `error`. |
|
||||
| `--details <level>` | No | Response detail level; full names each run’s workflow. Accepted values: `basic`, `full`. Defaults to `full`. |
|
||||
| `-n, --lines <count>` | No | Recent runs to print before watching. Defaults to `10`. |
|
||||
| `--interval <seconds>` | No | Seconds between polls. Defaults to `3`. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
@@ -1538,6 +1538,30 @@ sim logs list [options]
|
||||
|
||||
</CommandTable>
|
||||
|
||||
### sim logs follow
|
||||
|
||||
Watch runs as they arrive, printing each new run once
|
||||
|
||||
```bash
|
||||
sim logs follow [options]
|
||||
```
|
||||
|
||||
**Options**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Option | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `--workflow <id>` | No | Only follow runs of this workflow (repeatable). Defaults to ``. |
|
||||
| `--folder <path>` | No | Only follow runs of workflows in this folder (repeatable). Defaults to ``. |
|
||||
| `--trigger <type>` | No | Only follow runs with this trigger type (repeatable). Defaults to ``. |
|
||||
| `--level <level>` | No | Only follow runs at this severity. Accepted values: `info`, `error`. |
|
||||
| `--details <level>` | No | Response detail level; full names each run’s workflow. Accepted values: `basic`, `full`. Defaults to `full`. |
|
||||
| `-n, --lines <count>` | No | Recent runs to print before watching. Defaults to `10`. |
|
||||
| `--interval <seconds>` | No | Seconds between polls. Defaults to `3`. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
## sim mcp-servers
|
||||
|
||||
Also spelled `sim mcp-server`.
|
||||
@@ -3190,6 +3214,35 @@ sim workflows runs resume <runId> [options]
|
||||
|
||||
</CommandTable>
|
||||
|
||||
### sim workflows runs wait
|
||||
|
||||
Wait for a run to reach a terminal state, then show it
|
||||
|
||||
```bash
|
||||
sim workflows runs wait <runId> [options]
|
||||
```
|
||||
|
||||
**Arguments**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Argument | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `runId` | Yes | Unique workflow run identifier. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
**Options**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Option | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `--workflow <workflowId>` | Yes | Workflow ID. |
|
||||
| `--wait-timeout <seconds>` | No | Give up after this many seconds, or 0 to wait indefinitely (default: 3600). Bounds the whole wait; SIM_TIMEOUT_SECONDS bounds one request. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
### sim workflows create
|
||||
|
||||
Create Workflow
|
||||
@@ -3389,6 +3442,9 @@ sim workflows run <id> [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 <value>` | 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). |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
|
||||
@@ -117,6 +117,33 @@ Resume a paused run (output is included in JSON or YAML output)
|
||||
|
||||
</CommandTable>
|
||||
|
||||
## Wait for a run to reach a terminal state, then show it
|
||||
|
||||
```bash
|
||||
sim workflows runs wait <runId> [options]
|
||||
```
|
||||
|
||||
**Arguments**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Argument | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `runId` | Yes | Unique workflow run identifier. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
**Options**
|
||||
|
||||
<CommandTable>
|
||||
|
||||
| Option | Required | Description |
|
||||
| --- | --- | --- |
|
||||
| `--workflow <workflowId>` | Yes | Workflow ID. |
|
||||
| `--wait-timeout <seconds>` | No | Give up after this many seconds, or 0 to wait indefinitely (default: 3600). Bounds the whole wait; SIM_TIMEOUT_SECONDS bounds one request. |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
## Create workflow
|
||||
|
||||
```bash
|
||||
@@ -300,6 +327,9 @@ sim workflows run <id> [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 <value>` | 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). |
|
||||
|
||||
</CommandTable>
|
||||
|
||||
|
||||
@@ -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'))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,403 @@
|
||||
/**
|
||||
* @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<ListLogsResponse | Error>): 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<unknown> {
|
||||
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('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('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.
|
||||
// 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
|
||||
// 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')
|
||||
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)
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,612 @@
|
||||
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<unknown>(
|
||||
(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<LogRow>[] = (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
|
||||
/**
|
||||
* 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
|
||||
}
|
||||
|
||||
/**
|
||||
* 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}`)
|
||||
},
|
||||
warn: (message) => {
|
||||
if (reported) {
|
||||
reported = false
|
||||
process.stderr.write(`\r${ERASE_LINE}`)
|
||||
}
|
||||
process.stderr.write(`warning: ${message}\n`)
|
||||
},
|
||||
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<void> {
|
||||
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<string, string>
|
||||
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.
|
||||
*/
|
||||
/**
|
||||
* 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<SimClient, 'request'>,
|
||||
path: string,
|
||||
query: Record<string, string | number | undefined>,
|
||||
state: FollowState,
|
||||
pageSize: number,
|
||||
maxPages: number
|
||||
): Promise<PollBatch> {
|
||||
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<ListLogsResponse>(path, {
|
||||
query: { ...query, limit: pageSize, cursor },
|
||||
})
|
||||
const page_rows = response?.data ?? []
|
||||
const unprinted = page_rows.filter((row) => isUnprinted(state, row))
|
||||
rows.push(...unprinted)
|
||||
|
||||
cursor = response?.nextCursor ?? null
|
||||
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 { rows, truncated }
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 <id>', 'Only follow runs of this workflow (repeatable)', collect, [])
|
||||
.option(
|
||||
'--folder <path>',
|
||||
'Only follow runs of workflows in this folder (repeatable)',
|
||||
collect,
|
||||
[]
|
||||
)
|
||||
.option('--trigger <type>', 'Only follow runs with this trigger type (repeatable)', collect, [])
|
||||
.addOption(
|
||||
new Option('--level <level>', 'Only follow runs at this severity').choices([
|
||||
...V2_OPERATIONS.listLogs.query.level.values,
|
||||
])
|
||||
)
|
||||
.addOption(
|
||||
new Option('--details <level>', 'Response detail level; full names each run’s workflow')
|
||||
.choices([...V2_OPERATIONS.listLogs.query.details.values])
|
||||
.default('full')
|
||||
)
|
||||
.option('-n, --lines <count>', 'Recent runs to print before watching', String(DEFAULT_BACKLOG))
|
||||
.option('--interval <seconds>', '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 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
|
||||
while (!interrupt.interrupted()) {
|
||||
await waitFor(
|
||||
failures === 0 ? delay : Math.min(delay * 2 ** failures, MAX_BACKOFF_MS),
|
||||
interrupt.interrupted
|
||||
)
|
||||
if (interrupt.interrupted()) break
|
||||
|
||||
let fresh: PollBatch
|
||||
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
|
||||
}
|
||||
|
||||
// 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
|
||||
status.clear()
|
||||
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.rows.reverse())
|
||||
}
|
||||
} finally {
|
||||
status.clear()
|
||||
interrupt.dispose()
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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<Uint8Array> {
|
||||
return new ReadableStream({
|
||||
start(controller) {
|
||||
for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk))
|
||||
controller.close()
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
function sse(...frames: unknown[]): ReadableStream<Uint8Array> {
|
||||
return bodyOf(frames.map((frame) => `data: ${JSON.stringify(frame)}\n\n`))
|
||||
}
|
||||
|
||||
function streamResponse(body: ReadableStream<Uint8Array>): Response {
|
||||
return {
|
||||
body,
|
||||
status: 200,
|
||||
headers: new Headers({ 'content-type': 'text/event-stream' }),
|
||||
} as unknown as Response
|
||||
}
|
||||
|
||||
function options(overrides: Partial<Parameters<typeof renderRunStream>[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<Uint8Array> } = {}
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
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<void> {
|
||||
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)
|
||||
})
|
||||
})
|
||||
@@ -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<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
|
||||
function stringField(frame: Record<string, unknown>, 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<Uint8Array>): AsyncGenerator<string> {
|
||||
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, unknown>): 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<Uint8Array>,
|
||||
options: FollowOptions
|
||||
): Promise<Record<string, unknown>> {
|
||||
const commentary = new Commentary(options.stderr)
|
||||
let final: Record<string, unknown> | 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<void> {
|
||||
const flags = command.optsWithGlobals() as Record<string, unknown>
|
||||
|
||||
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<void> => {
|
||||
const flags = command.optsWithGlobals() as Record<string, unknown>
|
||||
|
||||
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))
|
||||
}
|
||||
@@ -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<void> {
|
||||
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([])
|
||||
})
|
||||
})
|
||||
@@ -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 <seconds>'
|
||||
|
||||
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<string, unknown> {
|
||||
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('<runId>', V2_OPERATIONS.getWorkflowRun.pathParamDocs?.runId)
|
||||
.description('Wait for a run to reach a terminal state, then show it')
|
||||
.addOption(
|
||||
new Option('--workflow <workflowId>', '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<unknown>(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()
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
@@ -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 },
|
||||
|
||||
Reference in New Issue
Block a user