Lambda Destinations: Route Failed Agent Tool Invokes Without Silent Drops

// ❌ Fire-and-forget tool — failures evaporate
import { LambdaClient, InvokeCommand } from "@aws-sdk/client-lambda";

const lambda = new LambdaClient({});

export async function runSandboxTool(input: {
 sessionId: string;
 patch: string;
}) {
 await lambda.send(
 new InvokeCommand({
 FunctionName: process.env.SANDBOX_FN!,
 InvocationType: "Event",
 Payload: Buffer.from(JSON.stringify(input)),
 })
 );
 // Model hears "started" — if sandbox crashes, nobody knows
 return { status: "started" };
}

// ✅ CDK — destinations for agent worker
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as destinations from "aws-cdk-lib/aws-lambda-destinations";
import * as sqs from "aws-cdk-lib/aws-sqs";
import * as events from "aws-cdk-lib/aws-events";
import * as cdk from "aws-cdk-lib";

const failureQ = new sqs.Queue(this, "AgentToolFailureQ", {
 retentionPeriod: cdk.Duration.days(14),
});
const bus = new events.EventBus(this, "AgentBus");

const sandbox = new lambda.Function(this, "SandboxWorker", {
 // runtime, handler, ...
});

sandbox.configureAsyncInvoke({
 maxEventAge: cdk.Duration.hours(2),
 retryAttempts: 1, // ✅ agents: low retries + destination > thrash
 onFailure: new destinations.SqsDestination(failureQ),
 onSuccess: new destinations.EventBridgeDestination(bus),
});

// ✅ Failure consumer — notify session + park for replay
import type { SQSHandler } from "aws-lambda";
import {
 ApiGatewayManagementApiClient,
 PostToConnectionCommand,
} from "@aws-sdk/client-apigatewaymanagementapi";

const apigw = new ApiGatewayManagementApiClient({
 endpoint: process.env.WS_CALLBACK_URL!,
});

export const onToolFailure: SQSHandler = async (event) => {
 for (const rec of event.Records) {
 const dest = JSON.parse(rec.body);
 // Lambda destination envelope
 const original = dest.requestPayload ?? dest;
 const sessionId = original.sessionId as string | undefined;
 const connectionId = original.connectionId as string | undefined;
 const err =
 dest.responsePayload?.errorMessage ??
 dest.responseContext?.statusCode ??
 "tool_failed";

 if (connectionId) {
 try {
 await apigw.send(
 new PostToConnectionCommand({
 ConnectionId: connectionId,
 Data: Buffer.from(
 JSON.stringify({
 type: "tool_failed",
 sessionId,
 error: String(err),
 })
 ),
 })
 );
 } catch {
 // stale connection — session table TTL will clean up
 }
 }
 // persist for operator replay / idempotent redrive
 }
};

// ✅ Per-invoke DestinationConfig when the default is not enough
await lambda.send(
 new InvokeCommand({
 FunctionName: process.env.SANDBOX_FN!,
 InvocationType: "Event",
 Payload: Buffer.from(JSON.stringify(input)),
 // SDK v3: DestinationConfig supported on Invoke for async
 DestinationConfig: {
 OnFailure: { Destination: process.env.FAILURE_QUEUE_ARN! },
 OnSuccess: { Destination: process.env.SUCCESS_TOPIC_ARN! },
 },
 })
);