feat: Step Function + hardning

2026-09-27 23:26:00 -04:00
parent ce8400757c
commit 0d613ae256
11 changed files with 672 additions and 1 deletions
@@ -3,7 +3,12 @@ import {
    CreateStateMachineCommand,
    DeleteStateMachineCommand,
    DescribeStateMachineCommand,
+    DescribeExecutionCommand,
+    GetExecutionHistoryCommand,
+    ListExecutionsCommand,
    ListStateMachinesCommand,
+    StartExecutionCommand,
+    StopExecutionCommand,
    type SFNClient,
 } from '@aws-sdk/client-sfn'
 import {AwsStepFunctionsAdapter} from './AwsStepFunctionsAdapter'
@@ -221,4 +226,39 @@ describe('AwsStepFunctionsAdapter', () => {

        expect((sent[0] as DeleteStateMachineCommand).input.stateMachineArn).toBe(ARN)
    })
+
+    test('uses AWS execution commands with input data and history pagination', async () => {
+        const executionArn = 'arn:aws:states:us-east-1:000000000000:execution:orders:run-1'
+        const {client, sent} = stubSfn((command) => {
+            if (command instanceof StartExecutionCommand) return {executionArn, startDate: new Date('2026-07-28T10:01:00.000Z')}
+            if (command instanceof ListExecutionsCommand) return {executions: [{executionArn, name: 'run-1', status: 'RUNNING', startDate: new Date('2026-07-28T10:01:00.000Z')}]}
+            if (command instanceof DescribeExecutionCommand) return {executionArn, name: 'run-1', status: 'FAILED', startDate: new Date('2026-07-28T10:01:00.000Z'), stateMachineArn: ARN, input: '{}', error: 'States.TaskFailed', cause: 'boom'}
+            if (command instanceof GetExecutionHistoryCommand) return {events: [{id: 1, type: 'ExecutionStarted', timestamp: new Date()}], nextToken: 'page-2'}
+            return {}
+        })
+        const adapter = new AwsStepFunctionsAdapter(client)
+
+        await adapter.startExecution(ARN, {input: '{"orderId":"1"}', name: 'run-1'})
+        await adapter.listExecutions(ARN)
+        const detail = await adapter.describeExecution(executionArn)
+        const history = await adapter.getExecutionHistory(executionArn, 'previous-page')
+        await adapter.stopExecution(executionArn)
+
+        expect((sent[0] as StartExecutionCommand).input).toMatchObject({stateMachineArn: ARN, input: '{"orderId":"1"}', name: 'run-1'})
+        expect((sent[1] as ListExecutionsCommand).input.stateMachineArn).toBe(ARN)
+        expect((sent[2] as DescribeExecutionCommand).input.executionArn).toBe(executionArn)
+        expect((sent[3] as GetExecutionHistoryCommand).input).toMatchObject({executionArn, includeExecutionData: true, maxResults: 1_000, nextToken: 'previous-page'})
+        expect((sent[4] as StopExecutionCommand).input.executionArn).toBe(executionArn)
+        expect(detail).toMatchObject({input: '{}', error: 'States.TaskFailed', cause: 'boom'})
+        expect(history.nextToken).toBe('page-2')
+    })
+
+    test('rejects invalid execution JSON before calling AWS', async () => {
+        const {client, sent} = runtimeStub()
+
+        await expect(new AwsStepFunctionsAdapter(client).startExecution(ARN, {input: 'not json'})).rejects.toThrow(
+            new ValidationError('input must be valid JSON'),
+        )
+        expect(sent).toHaveLength(0)
+    })
 })
@@ -2,12 +2,18 @@ import {
    CreateStateMachineCommand,
    DeleteStateMachineCommand,
    DescribeStateMachineCommand,
+    DescribeExecutionCommand,
+    GetExecutionHistoryCommand,
+    ListExecutionsCommand,
    ListStateMachinesCommand,
+    StartExecutionCommand,
+    StopExecutionCommand,
    type SFNClient,
    type StateMachineListItem,
 } from '@aws-sdk/client-sfn'
 import {RuntimeError, ValidationError} from '../cloud-spi/errors'
 import {STATE_MACHINE_NAME_PATTERN, awsWorkflowsSchema} from '../cloud-spi/workflowsSchema'
+import type {StartWorkflowExecutionInput, WorkflowExecutionDetail, WorkflowExecutionHistoryPage, WorkflowExecutionSummary} from '../cloud-spi/workflowExecutions'
 import type {
    CloudResource,
    CloudServiceAdapter,
@@ -104,6 +110,65 @@ export class AwsStepFunctionsAdapter implements CloudServiceAdapter {
    async delete(id: string): Promise<void> {
        await this.sfn.send(new DeleteStateMachineCommand({stateMachineArn: id}))
    }
+
+    async startExecution(stateMachineArn: string, input: StartWorkflowExecutionInput): Promise<{executionArn: string; startDate: Date}> {
+        const result = await this.sfn.send(new StartExecutionCommand({
+            stateMachineArn,
+            input: validExecutionInput(input.input),
+            ...(input.name ? {name: input.name} : {}),
+        }))
+        if (!result.executionArn || !result.startDate) throw new RuntimeError('Step Functions did not return an execution ARN and start date')
+        return {executionArn: result.executionArn, startDate: result.startDate}
+    }
+
+    async listExecutions(stateMachineArn: string): Promise<WorkflowExecutionSummary[]> {
+        const executions: WorkflowExecutionSummary[] = []
+        let nextToken: string | undefined
+        do {
+            const result = await this.sfn.send(new ListExecutionsCommand({stateMachineArn, ...(nextToken ? {nextToken} : {})}))
+            for (const execution of result.executions ?? []) {
+                if (execution.executionArn && execution.name && execution.status && execution.startDate) {
+                    executions.push({
+                        executionArn: execution.executionArn,
+                        name: execution.name,
+                        status: execution.status,
+                        startDate: execution.startDate,
+                        ...(execution.stopDate ? {stopDate: execution.stopDate} : {}),
+                    })
+                }
+            }
+            nextToken = result.nextToken
+        } while (nextToken)
+        return executions
+    }
+
+    async describeExecution(executionArn: string): Promise<WorkflowExecutionDetail> {
+        const result = await this.sfn.send(new DescribeExecutionCommand({executionArn}))
+        if (!result.executionArn || !result.name || !result.status || !result.startDate || !result.stateMachineArn) {
+            throw new RuntimeError('Step Functions returned an incomplete execution description')
+        }
+        return {
+            executionArn: result.executionArn,
+            name: result.name,
+            status: result.status,
+            startDate: result.startDate,
+            stateMachineArn: result.stateMachineArn,
+            ...(result.stopDate ? {stopDate: result.stopDate} : {}),
+            ...(result.input !== undefined ? {input: result.input} : {}),
+            ...(result.output !== undefined ? {output: result.output} : {}),
+            ...(result.error !== undefined ? {error: result.error} : {}),
+            ...(result.cause !== undefined ? {cause: result.cause} : {}),
+        }
+    }
+
+    async getExecutionHistory(executionArn: string, nextToken?: string): Promise<WorkflowExecutionHistoryPage> {
+        const result = await this.sfn.send(new GetExecutionHistoryCommand({executionArn, includeExecutionData: true, maxResults: 1_000, ...(nextToken ? {nextToken} : {})}))
+        return {events: (result.events ?? []).map((event) => event as unknown as Record<string, unknown>), ...(result.nextToken ? {nextToken: result.nextToken} : {})}
+    }
+
+    async stopExecution(executionArn: string): Promise<void> {
+        await this.sfn.send(new StopExecutionCommand({executionArn}))
+    }
 }

 function toResource(machine: StateMachineListItem | DescribedMachine): CloudResource {
@@ -172,6 +237,16 @@ function requiredDefinition(value: unknown): string {
    return raw
 }

+function validExecutionInput(value: string): string {
+    const input = requiredString(value, 'input')
+    try {
+        JSON.parse(input)
+    } catch {
+        throw new ValidationError('input must be valid JSON')
+    }
+    return input
+}
+
 function isMissing(error: unknown): boolean {
    if (typeof error !== 'object' || error === null) return false
    return (error as {name?: string}).name === 'StateMachineDoesNotExist'
@@ -0,0 +1,25 @@
+export interface StartWorkflowExecutionInput {
+    input: string
+    name?: string
+}
+
+export interface WorkflowExecutionSummary {
+    executionArn: string
+    name: string
+    status: string
+    startDate: Date
+    stopDate?: Date
+}
+
+export interface WorkflowExecutionDetail extends WorkflowExecutionSummary {
+    stateMachineArn: string
+    input?: string
+    output?: string
+    error?: string
+    cause?: string
+}
+
+export interface WorkflowExecutionHistoryPage {
+    events: Array<Record<string, unknown>>
+    nextToken?: string
+}
@@ -729,6 +729,44 @@ export function createCloudRoutes(injectedService?: CloudProxyService) {
        })
    })

+    app.post('/:cloud/services/workflows/resources/:id/executions', async (c) => {
+        if (c.req.param('cloud') !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
+        return withRuntime(c, async () => {
+            const body = await jsonBody<{input?: unknown; name?: unknown}>(c)
+            if (typeof body.input !== 'string') throw new ValidationError('input must be a JSON string')
+            try {
+                JSON.parse(body.input)
+            } catch {
+                throw new ValidationError('input must be valid JSON')
+            }
+            if (body.name !== undefined && (typeof body.name !== 'string' || !body.name.trim())) throw new ValidationError('name must be a non-empty string')
+            return c.json(await svc(c).startWorkflowExecution('aws', c.req.param('id'), {input: body.input, ...(typeof body.name === 'string' ? {name: body.name} : {})}), 201)
+        })
+    })
+
+    app.get('/:cloud/services/workflows/resources/:id/executions', async (c) => {
+        if (c.req.param('cloud') !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
+        return withRuntime(c, async () => c.json(await svc(c).listWorkflowExecutions('aws', c.req.param('id'))))
+    })
+
+    app.get('/:cloud/services/workflows/resources/:id/executions/:executionArn', async (c) => {
+        if (c.req.param('cloud') !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
+        return withRuntime(c, async () => c.json(await svc(c).describeWorkflowExecution('aws', c.req.param('executionArn'))))
+    })
+
+    app.get('/:cloud/services/workflows/resources/:id/executions/:executionArn/history', async (c) => {
+        if (c.req.param('cloud') !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
+        return withRuntime(c, async () => c.json(await svc(c).getWorkflowExecutionHistory('aws', c.req.param('executionArn'), c.req.query('nextToken'))))
+    })
+
+    app.post('/:cloud/services/workflows/resources/:id/executions/:executionArn/stop', async (c) => {
+        if (c.req.param('cloud') !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
+        return withRuntime(c, async () => {
+            await svc(c).stopWorkflowExecution('aws', c.req.param('executionArn'))
+            return c.json({ok: true})
+        })
+    })
+
    app.get('/:cloud/services/serverless/resources/:id/sqs-event-source-mappings', async (c) => {
        const cloud = c.req.param('cloud') as CloudProvider
        if (cloud !== 'aws') return c.json({error: 'Unknown cloud or service'}, 404)
@@ -0,0 +1,59 @@
+import {describe, expect, test} from 'bun:test'
+import {createCloudRoutes} from './clouds'
+import type {CloudProxyService} from '../service/CloudProxyService'
+
+const machineArn = 'arn:aws:states:us-east-1:000000000000:stateMachine:orders'
+const executionArn = 'arn:aws:states:us-east-1:000000000000:execution:orders:run-1'
+
+describe('workflow execution routes', () => {
+    test('routes execution debugging through the AWS workflow contract', async () => {
+        const calls: Array<[string, ...unknown[]]> = []
+        const service = {
+            async startWorkflowExecution(...args: unknown[]) {
+                calls.push(['start', ...args])
+                return {executionArn, startDate: new Date('2026-07-28T10:01:00.000Z')}
+            },
+            async listWorkflowExecutions(...args: unknown[]) {
+                calls.push(['list', ...args])
+                return [{executionArn, name: 'run-1', status: 'RUNNING', startDate: new Date('2026-07-28T10:01:00.000Z')}]
+            },
+            async describeWorkflowExecution(...args: unknown[]) {
+                calls.push(['describe', ...args])
+                return {executionArn, name: 'run-1', status: 'RUNNING', startDate: new Date(), stateMachineArn: machineArn, input: '{}'}
+            },
+            async getWorkflowExecutionHistory(...args: unknown[]) {
+                calls.push(['history', ...args])
+                return {events: [{id: 1, type: 'ExecutionStarted'}], nextToken: 'next'}
+            },
+            async stopWorkflowExecution(...args: unknown[]) {
+                calls.push(['stop', ...args])
+            },
+        } as unknown as CloudProxyService
+        const app = createCloudRoutes(service)
+        const resource = encodeURIComponent(machineArn)
+        const execution = encodeURIComponent(executionArn)
+
+        const started = await app.request(`/aws/services/workflows/resources/${resource}/executions`, {method: 'POST', body: JSON.stringify({input: '{}', name: 'run-1'}), headers: {'content-type': 'application/json'}})
+        const listed = await app.request(`/aws/services/workflows/resources/${resource}/executions`)
+        const described = await app.request(`/aws/services/workflows/resources/${resource}/executions/${execution}`)
+        const history = await app.request(`/aws/services/workflows/resources/${resource}/executions/${execution}/history?nextToken=previous`)
+        const stopped = await app.request(`/aws/services/workflows/resources/${resource}/executions/${execution}/stop`, {method: 'POST'})
+
+        expect([started.status, listed.status, described.status, history.status, stopped.status]).toEqual([201, 200, 200, 200, 200])
+        expect(calls).toEqual([
+            ['start', 'aws', machineArn, {input: '{}', name: 'run-1'}],
+            ['list', 'aws', machineArn],
+            ['describe', 'aws', executionArn],
+            ['history', 'aws', executionArn, 'previous'],
+            ['stop', 'aws', executionArn],
+        ])
+    })
+
+    test('rejects invalid execution input before calling the adapter', async () => {
+        const app = createCloudRoutes({} as CloudProxyService)
+        const response = await app.request(`/aws/services/workflows/resources/${encodeURIComponent(machineArn)}/executions`, {method: 'POST', body: JSON.stringify({input: 'not json'}), headers: {'content-type': 'application/json'}})
+
+        expect(response.status).toBe(400)
+        await expect(response.json()).resolves.toMatchObject({error: 'input must be valid JSON'})
+    })
+})
@@ -75,6 +75,7 @@ import type {
 } from '../cloud-spi/childCollections'
 import {NotSupportedError} from '../cloud-spi/errors'
 import {CloudAdapterRegistry} from '../registry/CloudAdapterRegistry'
+import type {StartWorkflowExecutionInput, WorkflowExecutionDetail, WorkflowExecutionHistoryPage, WorkflowExecutionSummary} from '../cloud-spi/workflowExecutions'
 import {SERVICE_CATALOG_ENTRIES, displayNameFor, routeFor} from '../cloud-spi/serviceCatalog'
 import {toHttpError} from '../cloud-spi/errors'
 import {mapAwsSdkError} from '../adapter-aws/awsErrors'
@@ -761,6 +762,34 @@ export class CloudProxyService {
        return this.appConfigAdapter(cloud).getAppConfigDeployment!(applicationId, environmentId, deploymentNumber)
    }

+    async startWorkflowExecution(cloud: CloudProvider, stateMachineArn: string, input: StartWorkflowExecutionInput): Promise<{executionArn: string; startDate: Date}> {
+        return this.workflowAdapter(cloud).startExecution(stateMachineArn, input)
+    }
+
+    async listWorkflowExecutions(cloud: CloudProvider, stateMachineArn: string): Promise<WorkflowExecutionSummary[]> {
+        return this.workflowAdapter(cloud).listExecutions(stateMachineArn)
+    }
+
+    async describeWorkflowExecution(cloud: CloudProvider, executionArn: string): Promise<WorkflowExecutionDetail> {
+        return this.workflowAdapter(cloud).describeExecution(executionArn)
+    }
+
+    async getWorkflowExecutionHistory(cloud: CloudProvider, executionArn: string, nextToken?: string): Promise<WorkflowExecutionHistoryPage> {
+        return this.workflowAdapter(cloud).getExecutionHistory(executionArn, nextToken)
+    }
+
+    async stopWorkflowExecution(cloud: CloudProvider, executionArn: string): Promise<void> {
+        await this.workflowAdapter(cloud).stopExecution(executionArn)
+    }
+
+    private workflowAdapter(cloud: CloudProvider) {
+        const adapter = this.requireAdapter(cloud, 'workflows')
+        if (!('startExecution' in adapter) || !('listExecutions' in adapter) || !('describeExecution' in adapter) || !('getExecutionHistory' in adapter) || !('stopExecution' in adapter)) {
+            throw new NotSupportedError(`Workflow executions are not supported for ${cloud}`)
+        }
+        return adapter as import('../adapter-aws/AwsStepFunctionsAdapter').AwsStepFunctionsAdapter
+    }
+
    private requireAdapter(cloud: CloudProvider, service: CloudServiceType) {
        const adapter = this.registry.get(cloud, service)
        if (!adapter) throw new NotSupportedError(`No adapter registered for ${cloud}/${service}`)