Bedrock Agents + EventBridge: Async Tools Without Blocking the Session

// ❌ Blocking tool — looks simple, dies in prod
export async function runCdkSynth(_args: { stack: string }) {
  // 60–180s later...
  return { ok: true, logs: "..." };
}

// tools/start-cdk-synth.ts
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";
import { DynamoDBClient, PutItemCommand, GetItemCommand } from "@aws-sdk/client-dynamodb";
import { randomUUID } from "node:crypto";
import { z } from "zod";

const eb = new EventBridgeClient({});
const ddb = new DynamoDBClient({});
const TABLE = process.env.JOBS_TABLE!;
const BUS = process.env.APP_BUS!;

export const StartSynthArgs = z.object({
  schemaVersion: z.literal(1),
  idempotencyKey: z.string().min(8),
  stackName: z.string().min(1).max(128),
  ref: z.string().min(1), // git sha or branch
});

export async function startCdkSynth(raw: unknown) {
  const args = StartSynthArgs.parse(raw);
  const existing = await ddb.send(
    new GetItemCommand({
      TableName: TABLE,
      Key: { pk: { S: `IDEM#${args.idempotencyKey}` } },
    }),
  );
  if (existing.Item?.jobId?.S) {
    // ✅ Same idempotencyKey returns same jobId
    return {
      ok: true,
      jobId: existing.Item.jobId.S,
      status: existing.Item.status?.S ?? "UNKNOWN",
      resumed: true,
    };
  }

  const jobId = randomUUID();
  await ddb.send(
    new PutItemCommand({
      TableName: TABLE,
      Item: {
        pk: { S: `JOB#${jobId}` },
        gsi1pk: { S: `IDEM#${args.idempotencyKey}` },
        status: { S: "QUEUED" },
        stackName: { S: args.stackName },
        ref: { S: args.ref },
        createdAt: { S: new Date().toISOString() },
      },
      ConditionExpression: "attribute_not_exists(pk)",
    }),
  );
  // secondary idempotency pointer
  await ddb.send(
    new PutItemCommand({
      TableName: TABLE,
      Item: {
        pk: { S: `IDEM#${args.idempotencyKey}` },
        jobId: { S: jobId },
        status: { S: "QUEUED" },
      },
      ConditionExpression: "attribute_not_exists(pk)",
    }),
  );

  await eb.send(
    new PutEventsCommand({
      Entries: [
        {
          EventBusName: BUS,
          Source: "com.cheatcoders.agents",
          DetailType: "CdkSynthRequested",
          Detail: JSON.stringify({
            jobId,
            stackName: args.stackName,
            ref: args.ref,
            schemaVersion: 1,
          }),
        },
      ],
    }),
  );

  return {
    ok: true,
    jobId,
    status: "QUEUED",
    // Tell the model how to continue — do not invent busy-waits of 100 tool calls
    nextTool: "get_cdk_synth_status",
    pollAfterSec: 15,
  };
}

// tools/get-cdk-synth-status.ts
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
import { z } from "zod";

const ddb = new DynamoDBClient({});
const TABLE = process.env.JOBS_TABLE!;

export const StatusArgs = z.object({
  schemaVersion: z.literal(1),
  jobId: z.string().uuid(),
});

export async function getCdkSynthStatus(raw: unknown) {
  const { jobId } = StatusArgs.parse(raw);
  const out = await ddb.send(
    new GetItemCommand({
      TableName: TABLE,
      Key: { pk: { S: `JOB#${jobId}` } },
    }),
  );
  if (!out.Item) {
    return {
      ok: false,
      error: { code: "NOT_FOUND", retryable: false, message: "unknown jobId" },
    };
  }

  const status = out.Item.status.S!;
  return {
    ok: true,
    jobId,
    status,
    // ✅ Terminal payloads only when done — keep interim small
    artifactUri: out.Item.artifactUri?.S,
    errorMessage: out.Item.errorMessage?.S,
    resumeHint:
      status === "QUEUED" || status === "RUNNING"
        ? { action: "poll", pollAfterSec: 15 }
        : status === "SUCCEEDED"
          ? { action: "continue", summaryTool: "summarize_synth_artifact" }
          : { action: "replan" },
  };
}

export type JobRecord = {
  jobId: string;
  principalId: string;
  sessionId: string;
  status: "QUEUED" | "RUNNING" | "SUCCEEDED" | "FAILED";
  createdAt: string;
};