Modify Cdk Workflows

Modify CDK Step Functions Workflows

Workflow Architecture

cdk/lib/constructs/
├── review-processor.ts         # Review workflow (3-step)
├── checklist-processor.ts      # Checklist workflow (5-step)
├── agent.ts                    # AgentCore infrastructure
└── lambda/invoke-agent/        # Agent invocation Lambda

ReviewProcessor Flow

1. Prepare Review - Fetch checklist items 2. Process All Items (Map State) - Parallel: pre-process -> AgentCore -> post-process 3. Finalize Review - Aggregate results

Config: maxConcurrency default 1, timeout 2 hours

ChecklistProcessor Flow

1. Process Document - File format detection, page extraction 2. Process All Pages Inline (Map State) - Parallel page processing 3. Aggregate Results - Combine page results 4. Store to Database - Persist findings 5. Detect Ambiguity - Identify ambiguities

Config: inlineMapConcurrency default 1, timeout 24 hours, thresholds: 40 pages (medium), 100 pages (large)

Common Modifications

1. Adding/Removing Workflow Steps

// Create task
const newTask = new tasks.LambdaInvoke(this, "NewTaskId", {
  lambdaFunction: processorLambda,
  payload: sfn.TaskInput.fromObject({
    action: "newAction",
    dataParam: sfn.JsonPath.stringAt("$.previous.result"),
  }),
  resultPath: "$.newResult",
  resultSelector: { "Payload.$": "$.Payload" },
});

// Add error handling
newTask.addCatch(handleErrorTask, {
  errors: ["States.ALL"],
  resultPath: "$.error",
});

// Chain into workflow
const definition = prepareTask.next(newTask).next(processTask).next(finalizeTask);

Look for definitionBody: sfn.DefinitionBody.fromChainable() in processor files.

2. Modifying Map State Concurrency

const processItemsMap = new sfn.Map(this, "ProcessAllItems", {
  maxConcurrency: maxConcurrency,
  itemsPath: sfn.JsonPath.stringAt("$.prepareResult.Payload.checkItems"),
  resultPath: "$.processedItems",
});

Set via CDK parameters: cdk deploy -c rapid.reviewMapConcurrency=5

Trade-offs: Higher = faster but more cost/throttling. Lower = slower but predictable.

3. Adding Retry Logic

task.addRetry({
  errors: ["RetryException", "ThrottlingException", "ServiceQuotaExceededException"],
  interval: cdk.Duration.seconds(2),
  maxAttempts: 5,
  backoffRate: 2,
});

4. Adding Parameters

1. Define in parameter-schema.ts: reviewMapConcurrency: z.number().int().min(1).optional() 2. Pass to construct in rapid-stack.ts 3. Use in construct constructor

5. Modifying Timeouts

Task: timeout: cdk.Duration.minutes(15) State machine: timeout: cdk.Duration.hours(2)

For JsonPath patterns and state machine creation templates, see references/CDK-PATTERNS.md.

Quick Reference

Verification

cd cdk && npx cdk synth

After verification, deploy with /deploy-cdk-stack.

Success Criteria

• cdk synth completes without errors
• No circular dependencies
• All tasks have proper error handling
• Concurrency and timeout settings are appropriate

CDK Step Functions Patterns

Reference for common CDK Step Functions patterns used in RAPID workflows.

Data Flow with JsonPath

// From execution input
sfn.JsonPath.stringAt("$$.Execution.Input.userId")

// From Map state item
sfn.JsonPath.stringAt("$$.Map.Item.Value.fieldName")

// From previous task result
sfn.JsonPath.stringAt("$.previousTask.Payload.field")

// Entire execution context
sfn.JsonPath.entirePayload

Item Selector for Map State:

itemSelector: {
  "reviewJobId.$": "$.reviewJobId",
  "checkId.$": "$$.Map.Item.Value.checkId",
  "itemData.$": "$$.Map.Item.Value",
}

Result Selector:

resultSelector: {
  "Payload.$": "$.Payload",
  "StatusCode.$": "$.StatusCode",
}

State Machine Creation Pattern

// Create IAM role
const stateMachineRole = new iam.Role(this, "Role", {
  assumedBy: new iam.ServicePrincipal("states.amazonaws.com"),
});

// Grant permissions
stateMachineRole.addToPolicy(
  new iam.PolicyStatement({
    actions: ["bedrock:InvokeModel"],
    resources: ["*"],
  })
);

// Create log group
const logGroup = new logs.LogGroup(this, "LogGroup", {
  retention: logs.RetentionDays.ONE_WEEK,
});

// Define workflow
const definition = taskA.next(taskB).next(taskC);

// Create state machine
this.stateMachine = new sfn.StateMachine(this, "WorkflowName", {
  definitionBody: sfn.DefinitionBody.fromChainable(definition),
  role: stateMachineRole,
  timeout: cdk.Duration.hours(2),
  tracingEnabled: true,
  logs: {
    destination: logGroup,
    level: sfn.LogLevel.ALL,
    includeExecutionData: true,
  },
});