test(aws-lambda): compare every CDK Lambda payload against the SAM template (#5282)

* test(aws-lambda): compare every CDK Lambda payload against the SAM template

#5259 shipped because CDK drops null-valued payload fields when it renders the
state machine, so the construct's source and its deployed definition disagreed
with nothing to notice. This test reads all six Lambda task payloads from the
synthesized definition and requires each to equal the SAM template's, reading
`States.StringToJson('null')` as the null it sends. A field dropped or renamed
on either side now fails CI, not just AssembleV2's audio.

Refs #5259

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* test(aws-lambda): discover Lambda tasks for the payload parity check

The parity test took its tasks from helpers that name six states, so a seventh
Lambda task on either side would be skipped and the check would stay green.
It now walks both definitions, Map and Parallel processors included, collects
every lambda:invoke task by name, requires the two name sets to match, then
compares each payload.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* test(aws-lambda): key discovered Lambda tasks by their nested state path

Keying tasks by state name let two branches that each hold a RenderChunk
collapse into one entry, so the other's payload went unchecked. Tasks are now
keyed by path (a Map processor as `Map/`, a Parallel branch as `Parallel/<i>/`),
and a regression test keeps two same-named branch tasks apart.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts

@@ -264,6 +264,38 @@ describe("HyperframesRenderStack — snapshot", () => {
     expect(v2).not.toContain("PlanS3Uri");
   });
 
+  it("sends the same payload from every Lambda task as the SAM template", () => {
+    // CDK drops null-valued payload fields when it renders the definition (#5259), so compare
+    // what each deployed task sends rather than the construct source.
+    const cdkTasks = lambdaTasks(SYNTHED.definition.States);
+    const samTasks = lambdaTasks(readSamDefinition().States);
+    expect(Object.keys(cdkTasks).sort()).toEqual(Object.keys(samTasks).sort());
+    expect(Object.keys(cdkTasks).length).toBeGreaterThanOrEqual(6);
+    for (const [taskName, cdkTask] of Object.entries(cdkTasks)) {
+      expect({ taskName, payload: sentPayload(cdkTask) }).toEqual({
+        taskName,
+        payload: sentPayload(samTasks[taskName]),
+      });
+    }
+  });
+
+  it("keeps same-named Lambda tasks in different branches apart", () => {
+    const task = (format: string) => ({
+      Type: "Task",
+      Resource: "arn:aws:states:::lambda:invoke",
+      Parameters: { Payload: { Action: "renderChunk", Format: format } },
+    });
+    const branch = (format: string) => ({
+      StartAt: "RenderChunk",
+      States: { RenderChunk: task(format) },
+    });
+    const tasks = lambdaTasks({
+      Fan: { Type: "Parallel", Branches: [branch("mp4"), branch("webm")] },
+    });
+    expect(Object.keys(tasks).sort()).toEqual(["Fan/0/RenderChunk", "Fan/1/RenderChunk"]);
+    expect(sentPayload(tasks["Fan/1/RenderChunk"]).Format).toBe("webm");
+  });
+
   it("materializes null audio in the synthesized CDK v2 assembly payload", () => {
     const state = requireRecord(SYNTHED.definition.States.AssembleV2, "AssembleV2 state");
     const parameters = requireRecordProperty(state, "Parameters", "AssembleV2 parameters");
@@ -318,6 +350,59 @@ function requireRecordProperty(
   return requireRecord(record[property], label);
 }
 
+const isLambdaTask = (state: Record<string, unknown>): boolean =>
+  state.Type === "Task" && String(state.Resource).endsWith(":states:::lambda:invoke");
+
+/** The state maps nested in a Map processor or Parallel branches, keyed by their path. */
+function nestedStates(name: string, state: Record<string, unknown>): [string, unknown][] {
+  const processor = isRecord(state.Iterator) ? state.Iterator : state.ItemProcessor;
+  const branches: unknown[] = Array.isArray(state.Branches) ? state.Branches : [];
+  const nested: [string, unknown][] = [
+    [`${name}/`, isRecord(processor) && processor.States],
+    ...branches.map((branch, i): [string, unknown] => [
+      `${name}/${i}/`,
+      isRecord(branch) && branch.States,
+    ]),
+  ];
+  return nested.filter(([, states]) => isRecord(states));
+}
+
+/** Every Lambda-invoking task by its full state path, so same-named tasks in two branches stay apart. */
+function lambdaTasks(states: Record<string, unknown>, path = ""): Record<string, unknown> {
+  const records = Object.entries(states).filter(
+    (entry): entry is [string, Record<string, unknown>] => isRecord(entry[1]),
+  );
+  return Object.assign(
+    Object.fromEntries(
+      records
+        .filter(([, state]) => isLambdaTask(state))
+        .map(([name, state]) => [path + name, state]),
+    ),
+    ...records.flatMap(([name, state]) =>
+      nestedStates(path + name, state).map(([nestedPath, nested]) =>
+        lambdaTasks(nested as Record<string, unknown>, nestedPath),
+      ),
+    ),
+  );
+}
+
+/** A task's payload as Step Functions sends it: a `States.StringToJson('null')` field is a null. */
+function sentPayload(state: unknown): Record<string, unknown> {
+  const parameters = requireRecordProperty(
+    requireRecord(state, "task state"),
+    "Parameters",
+    "task parameters",
+  );
+  const payload = requireRecordProperty(parameters, "Payload", "task payload");
+  return Object.fromEntries(
+    Object.entries(payload).map(([key, value]) =>
+      key.endsWith(".$") && value === "States.StringToJson('null')"
+        ? [key.slice(0, -2), null]
+        : [key, value],
+    ),
+  );
+}
+
 function getV2TaskStates(definition: {
   States: Record<string, unknown>;
 }): Record<"PlanV2" | "RenderChunkV2" | "AssembleV2", unknown> {