chore: add the upstream helpers the composite alarm change builds on

Composite alarms from upstream (#6408) use SmithyValidationErrors for AWS's
validation messages, and their tests use the CfnStackWaits, PartitionCleanup
and ThreadStates test helpers and ReplacementCleanup.owedPhysicalIds. These
are copied from upstream as they were at that commit; none brings in other
behaviour.

compatibility-tests/sdk-test-java/src/test/java/com/floci/test/CloudWatchMetricMathAlarmTest.java

@@ -0,0 +1,117 @@
+package com.floci.test;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.cloudwatch.CloudWatchClient;
+import software.amazon.awssdk.services.cloudwatch.model.ComparisonOperator;
+import software.amazon.awssdk.services.cloudwatch.model.Dimension;
+import software.amazon.awssdk.services.cloudwatch.model.Metric;
+import software.amazon.awssdk.services.cloudwatch.model.MetricAlarm;
+import software.amazon.awssdk.services.cloudwatch.model.MetricDataQuery;
+import software.amazon.awssdk.services.cloudwatch.model.MetricStat;
+import software.amazon.awssdk.services.cloudwatch.model.StandardUnit;
+import software.amazon.awssdk.services.cloudwatch.model.Statistic;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@DisplayName("CloudWatch - Metric-math alarm definitions")
+class CloudWatchMetricMathAlarmTest {
+
+    private CloudWatchClient cloudWatch;
+    private String alarmName;
+
+    @BeforeEach
+    void setup() {
+        cloudWatch = TestFixtures.cloudWatchClient();
+        alarmName = TestFixtures.uniqueName("sdk-metric-math-alarm");
+    }
+
+    @AfterEach
+    void cleanup() {
+        try {
+            cloudWatch.deleteAlarms(request -> request.alarmNames(alarmName));
+        } finally {
+            cloudWatch.close();
+        }
+    }
+
+    @Test
+    void metricsRoundTripWithoutInventingSingleMetricFields() {
+        List<MetricDataQuery> metrics = metrics();
+        putMathAlarm(metrics);
+        assertMathAlarm(metrics);
+    }
+
+    @Test
+    void updatesReplaceQueriesAndSwitchBackToASingleMetric() {
+        putSingleMetricAlarm(300);
+        List<MetricDataQuery> original = metrics();
+        putMathAlarm(original);
+        assertMathAlarm(original);
+
+        List<MetricDataQuery> replacement = List.of(
+                original.get(0).toBuilder().expression("errors / requests").label("Ratio").build(),
+                original.get(1), original.get(2));
+        putMathAlarm(replacement);
+        assertMathAlarm(replacement);
+
+        putSingleMetricAlarm(120);
+        MetricAlarm alarm = describe();
+        assertThat(alarm.hasMetrics()).isFalse();
+        assertThat(alarm.period()).isEqualTo(120);
+        assertThat(alarm.metricName()).isEqualTo("5xx");
+        assertThat(alarm.namespace()).isEqualTo("AWS/ApiGateway");
+    }
+
+    private void putMathAlarm(List<MetricDataQuery> metrics) {
+        cloudWatch.putMetricAlarm(request -> request.alarmName(alarmName)
+                .comparisonOperator(ComparisonOperator.GREATER_THAN_THRESHOLD)
+                .evaluationPeriods(2).threshold(5.0).metrics(metrics));
+    }
+
+    private void putSingleMetricAlarm(int period) {
+        cloudWatch.putMetricAlarm(request -> request.alarmName(alarmName)
+                .comparisonOperator(ComparisonOperator.GREATER_THAN_THRESHOLD)
+                .evaluationPeriods(2).threshold(5.0).namespace("AWS/ApiGateway")
+                .metricName("5xx").statistic(Statistic.SUM).period(period));
+    }
+
+    private MetricAlarm describe() {
+        List<MetricAlarm> alarms = cloudWatch.describeAlarms(request -> request.alarmNames(alarmName)).metricAlarms();
+        assertThat(alarms).hasSize(1);
+        return alarms.get(0);
+    }
+
+    private void assertMathAlarm(List<MetricDataQuery> metrics) {
+        MetricAlarm alarm = describe();
+        assertThat(alarm.metrics()).containsExactlyElementsOf(metrics);
+        assertThat(alarm.period()).isNull();
+        assertThat(alarm.metricName()).isNull();
+        assertThat(alarm.namespace()).isNull();
+        assertThat(alarm.statistic()).isNull();
+        assertThat(alarm.extendedStatistic()).isNull();
+        assertThat(alarm.unit()).isNull();
+        assertThat(alarm.hasDimensions()).isFalse();
+        assertThat(alarm.evaluationPeriods()).isEqualTo(2);
+        assertThat(alarm.threshold()).isEqualTo(5.0);
+    }
+
+    private static List<MetricDataQuery> metrics() {
+        return List.of(
+                MetricDataQuery.builder().id("rate").expression("100 * errors / requests")
+                        .returnData(true).period(300).label("Error <rate> & requests").build(),
+                MetricDataQuery.builder().id("errors").returnData(false).accountId("123456789012")
+                        .metricStat(MetricStat.builder().period(300).stat("Sum").unit(StandardUnit.COUNT)
+                                .metric(Metric.builder().namespace("AWS/ApiGateway").metricName("5xx")
+                                        .dimensions(Dimension.builder().name("ApiName").value("api<&>").build())
+                                        .build()).build()).build(),
+                MetricDataQuery.builder().id("requests").returnData(false)
+                        .metricStat(MetricStat.builder().period(300).stat("Sum")
+                                .metric(Metric.builder().namespace("AWS/ApiGateway").metricName("Count").build())
+                                .build()).build());
+    }
+}

compatibility-tests/sdk-test-java/src/test/java/com/floci/test/CloudWatchTest.java

@@ -6,6 +6,7 @@ import software.amazon.awssdk.services.cloudwatch.model.*;
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.*;
 
@@ -185,4 +186,67 @@ class CloudWatchTest {
         );
         assertThat(noMatch.datapoints()).isEmpty();
     }
+
+    @Test
+    @DisplayName("PutMetricData stores datums nested under EntityMetricData")
+    void testPutMetricDataWithEntityMetricData() {
+        String namespace = "JavaTestEntityMetricData";
+        Instant now = Instant.now();
+
+        cloudWatchClient.putMetricData(request -> request
+                .namespace(namespace)
+                .metricData(datum -> datum
+                        .metricName("TopLevel")
+                        .value(1.0)
+                        .timestamp(now))
+                .entityMetricData(entityMetricData -> entityMetricData
+                        .entity(entity -> entity
+                                .keyAttributes(Map.of(
+                                        "Type", "AWS::Resource",
+                                        "ResourceType", "AWS::EC2::Instance",
+                                        "Identifier", "i-0123456789abcdef0")))
+                        .metricData(MetricDatum.builder()
+                                .metricName("mem_used_percent")
+                                .value(6.5)
+                                .unit(StandardUnit.PERCENT)
+                                .storageResolution(60)
+                                .timestamp(now)
+                                .dimensions(Dimension.builder().name("host").value("h1").build())
+                                .build()))
+                .strictEntityValidation(false)
+        );
+
+        ListMetricsResponse listed = cloudWatchClient.listMetrics(request -> request.namespace(namespace));
+        assertThat(listed.metrics())
+                .extracting(Metric::metricName)
+                .containsExactlyInAnyOrder("TopLevel", "mem_used_percent");
+
+        GetMetricStatisticsResponse stats = cloudWatchClient.getMetricStatistics(request -> request
+                .namespace(namespace)
+                .metricName("mem_used_percent")
+                .dimensions(Dimension.builder().name("host").value("h1").build())
+                .startTime(now.minus(1, ChronoUnit.HOURS))
+                .endTime(now.plus(1, ChronoUnit.MINUTES))
+                .period(3600)
+                .statistics(Statistic.SUM)
+        );
+        assertThat(stats.datapoints()).hasSize(1);
+        assertThat(stats.datapoints().get(0).sum()).isEqualTo(6.5);
+    }
+
+    @Test
+    @DisplayName("PutMetricData without MetricName throws MissingRequiredParameterException")
+    void testPutMetricDataWithoutMetricNameIsRejected() {
+        String namespace = "JavaTestMissingMetricName";
+
+        assertThatThrownBy(() -> cloudWatchClient.putMetricData(request -> request
+                .namespace(namespace)
+                .entityMetricData(entityMetricData -> entityMetricData
+                        .metricData(MetricDatum.builder().value(1.0).build()))))
+                .isInstanceOf(MissingRequiredParameterException.class)
+                .satisfies(e -> assertThat(((MissingRequiredParameterException) e).awsErrorDetails().errorCode())
+                        .isEqualTo("MissingParameter"));
+
+        assertThat(cloudWatchClient.listMetrics(request -> request.namespace(namespace)).metrics()).isEmpty();
+    }
 }

compatibility-tests/sdk-test-python/tests/test_cloudwatch.py

@@ -203,6 +203,8 @@ class TestCloudWatchAlarms:
 
             alarm = next(a for a in alarms if a["AlarmName"] == alarm_name)
             assert alarm["StateValue"] == "INSUFFICIENT_DATA"
+            assert response["CompositeAlarms"] == []
+            assert response["LogAlarms"] == []
         finally:
             cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])
 

docs/services/cloudformation.md

@@ -160,7 +160,7 @@ cross-resource references.
 | Kinesis Data Firehose | `DeliveryStream` |
 | IoT Core | `DomainConfiguration` (`ServerCertificates` resolves to a JSON string), `Policy` (deleted after detaching it from its principals; on AWS the delete fails with `DeleteConflictException` while the policy is attached), `Thing`, `TopicRule` |
 | CloudFront | `CachePolicy`, `Distribution`, `OriginAccessControl`, `OriginRequestPolicy`, `ResponseHeadersPolicy` |
-| CloudWatch | `Alarm`, `Dashboard` |
+| CloudWatch | `Alarm`, `CompositeAlarm`, `Dashboard`. Composite alarms have backing definitions with name Ref and Arn GetAtt, rollback, and replacement cleanup. Composite actions are stored only; suppression properties are rejected. |
 | CloudWatch Logs | `LogGroup`, `LogStream`, `MetricFilter` |
 | WAFv2 | `WebACL` |
 | Config | `ConfigRule` |

docs/services/cloudwatch.md

@@ -311,6 +311,7 @@ aws logs put-retention-policy \
 | `GetMetricStatistics` | Get metric statistics (Average, Sum, etc.) |
 | `GetMetricData` | Query metrics with math expressions |
 | `PutMetricAlarm` | Create a metric alarm |
+| `PutCompositeAlarm` | Create or update a composite alarm rule |
 | `DescribeAlarms` | List alarms |
 | `DeleteAlarms` | Delete alarms |
 | `SetAlarmState` | Manually set alarm state |
@@ -331,6 +332,31 @@ aws logs put-retention-policy \
 Metric streams are stored as definitions with their `running` or `stopped` state. Floci never
 delivers metrics to the Firehose delivery stream a metric stream names.
 
+Composite alarms support boolean rules over alarm states. Request `AlarmTypes` containing
+`CompositeAlarm` when describing them; the default describes metric alarms. Composite
+action lists are stored but not executed. Action suppression is not implemented and is
+rejected with `ValidationError`.
+
+### SNS alarm actions
+
+State transitions publish to same-account standard SNS topics in `AlarmActions`, `OKActions`, or
+`InsufficientDataActions`, including transitions from `SetAlarmState`. Unchanged states and
+`ActionsEnabled=false` suppress notifications. Scaling action retries remain independent.
+Failures are logged per target without blocking other targets. Notifications include a
+single-metric `Trigger`; durable retries, FIFO topics, and metric-math payloads are not covered.
+
+### Metric-math alarm definitions
+
+`PutMetricAlarm` stores the `Metrics` query definitions and `DescribeAlarms` returns them over
+both JSON and Query protocols. Expressions, labels, `ReturnData`, account IDs, query periods,
+and nested metric statistics (including dimensions and units) survive storage reloads.
+Metric-math alarm responses omit the single-metric fields, including top-level `Period`.
+Updating an alarm replaces its metric definition, so changing between `Metrics` and a single
+metric does not retain the previous definition.
+
+This is management-plane support: the alarm evaluator currently evaluates only single-metric
+alarms, not metric-math expressions. `SetAlarmState` remains available for manual transitions.
+
 ### Examples
 
 ```bash

src/main/java/io/github/hectorvent/floci/core/common/SmithyValidationErrors.java

@@ -0,0 +1,97 @@
+package io.github.hectorvent.floci.core.common;
+
+import java.util.List;
+
+/**
+ * The message AWS's Smithy validation layer answers when a request breaks its model's constraints:
+ * "1 validation error detected: a" or "N validation errors detected: a; b", each violation reading
+ * "Value at 'path' failed to satisfy constraint: ...". The wording is Smithy's, shared by every
+ * service built on it; only the error code differs by protocol.
+ */
+public final class SmithyValidationErrors {
+
+    private SmithyValidationErrors() {}
+
+    /**
+     * The message for one or more violations. An empty list is the caller's bug, so it fails rather
+     * than answering "0 validation errors"; check {@code isEmpty()} before throwing.
+     */
+    public static String message(List<String> violations) {
+        if (violations.isEmpty()) {
+            throw new IllegalArgumentException("no violations to report");
+        }
+        String count = violations.size() == 1
+                ? "1 validation error detected: "
+                : violations.size() + " validation errors detected: ";
+        return count + String.join("; ", violations);
+    }
+
+    /** The violation for a required member the request leaves out. */
+    public static String notNull(String path) {
+        return "Value null at '" + path + "' failed to satisfy constraint: Member must not be null";
+    }
+
+    /** The 400 error carrying {@link #message} under the protocol's validation error code. */
+    public static AwsException exception(String errorCode, List<String> violations) {
+        return new AwsException(errorCode, message(violations), 400);
+    }
+
+    public static AwsException exception(String errorCode, String violation) {
+        return exception(errorCode, List.of(violation));
+    }
+
+    /** The code most JSON services use, on both the awsJson and the restJson protocols. */
+    public static AwsException validationException(List<String> violations) {
+        return exception("ValidationException", violations);
+    }
+
+    public static AwsException validationException(String violation) {
+        return validationException(List.of(violation));
+    }
+
+    /** The awsQuery protocol's code. */
+    public static AwsException validationError(List<String> violations) {
+        return exception("ValidationError", violations);
+    }
+
+    public static AwsException validationError(String violation) {
+        return validationError(List.of(violation));
+    }
+
+    /** The code of REST JSON services that model their validation failure as BadRequestException. */
+    public static AwsException badRequest(List<String> violations) {
+        return exception("BadRequestException", violations);
+    }
+
+    public static AwsException badRequest(String violation) {
+        return badRequest(List.of(violation));
+    }
+
+    /** The violation for a number outside {@code min} to {@code max}, or {@code null} for one inside. */
+    public static String outOfRange(String path, long value, long min, long max) {
+        if (value < min) {
+            return "Value '" + value + "' at '" + path + "' failed to satisfy constraint: "
+                    + "Member must have value greater than or equal to " + min;
+        }
+        if (value > max) {
+            return "Value '" + value + "' at '" + path + "' failed to satisfy constraint: "
+                    + "Member must have value less than or equal to " + max;
+        }
+        return null;
+    }
+
+    /** The violation for a value outside its enum, listing the allowed values in the model's order. */
+    public static String notInEnum(String value, String path, List<String> allowed) {
+        return "Value '" + value + "' at '" + path + "' failed to satisfy constraint: "
+                + "Member must satisfy enum value set: " + allowed;
+    }
+
+    /**
+     * The violation for a list holding a value outside its members' enum. The list is checked as a
+     * whole, so the whole list is the offending value and the enum constraint is nested.
+     */
+    public static String notAllInEnum(List<String> values, String path, List<String> allowed) {
+        return "Value '" + values + "' at '" + path + "' failed to satisfy constraint: "
+                + "Member must satisfy constraint: [Member must satisfy enum value set: " + allowed + "]";
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudformation/provisioners/CloudWatchCompositeAlarmCfnProvisioner.java

@@ -0,0 +1,183 @@
+package io.github.hectorvent.floci.services.cloudformation.provisioners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.hectorvent.floci.core.common.AwsException;
+import io.github.hectorvent.floci.services.cloudformation.model.StackResource;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.CloudWatchMetricsService;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
+import jakarta.enterprise.context.ApplicationScoped;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+@ApplicationScoped
+public class CloudWatchCompositeAlarmCfnProvisioner implements CfnResourceProvisioner {
+    private static final String TYPE = "AWS::CloudWatch::CompositeAlarm";
+    private static final String SNAPSHOT = "FlociCompositeAlarmSnapshot";
+    private static final String TAG_KEYS = "FlociCompositeAlarmTagKeys";
+    private static final String EXPLICIT_NAME = "FlociCompositeAlarmExplicitName";
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private final CloudWatchMetricsService service;
+
+    public CloudWatchCompositeAlarmCfnProvisioner(CloudWatchMetricsService service) {
+        this.service = service;
+    }
+
+    @Override
+    public Set<String> resourceTypes() {
+        return Set.of(TYPE);
+    }
+
+    @Override
+    public void provision(StackResource resource, JsonNode props, ProvisionContext ctx) {
+        for (String unsupported : List.of("ActionsSuppressor", "ActionsSuppressorWaitPeriod",
+                "ActionsSuppressorExtensionPeriod")) {
+            if (props.has(unsupported)) {
+                throw new AwsException("ValidationError", "Composite action suppression is not supported", 400);
+            }
+        }
+        if (resource.getAttributes().containsKey(SNAPSHOT)) {
+            throw new AwsException("ResourceConflictException", "Alarm update cleanup is still pending", 400);
+        }
+        String explicit = ctx.resolveOptional(props, "AlarmName");
+        boolean removedName = explicit == null && "true".equals(resource.getAttributes().get(EXPLICIT_NAME));
+        String name = removedName ? ctx.generatePhysicalName(resource.getLogicalId(), 255, false)
+                : ctx.stablePhysicalName(explicit, resource.getLogicalId(), 255, false);
+        CompositeAlarm alarm = new CompositeAlarm();
+        alarm.setAlarmName(name);
+        alarm.setAlarmDescription(ctx.resolveOptional(props, "AlarmDescription"));
+        alarm.setAlarmRule(ctx.resolveOptional(props, "AlarmRule"));
+        alarm.setActionsEnabled(!"false".equals(ctx.resolveOptional(props, "ActionsEnabled")));
+        alarm.setTags(ctx.resolveTags(props, "Tags"));
+        actions(props, "AlarmActions", ctx, alarm.getAlarmActions());
+        actions(props, "OKActions", ctx, alarm.getOkActions());
+        actions(props, "InsufficientDataActions", ctx, alarm.getInsufficientDataActions());
+        Map<String, String> desiredTags = Map.copyOf(alarm.getTags());
+        Map<String, String> priorAttributes = Map.copyOf(resource.getAttributes());
+        boolean inPlace = ctx.reusesPriorEntity(name);
+        String updateSnapshot = null;
+        if (inPlace) {
+            ObjectNode snapshot = MAPPER.createObjectNode().put("region", ctx.region()).put("name", name);
+            List<CompositeAlarm> previous = service.describeCompositeAlarms(List.of(name), null, ctx.region());
+            if (!previous.isEmpty()) {
+                snapshot.set("alarm", MAPPER.valueToTree(previous.getFirst()));
+            }
+            snapshot.set("attributes", MAPPER.valueToTree(priorAttributes));
+            Set<String> changedKeys = new TreeSet<>(desiredTags.keySet());
+            changedKeys.addAll(tagKeys(resource.getAttributes().get(TAG_KEYS)));
+            snapshot.set("changedTagKeys", MAPPER.valueToTree(changedKeys));
+            updateSnapshot = snapshot.toString();
+        }
+        service.putCompositeAlarm(alarm, ctx.region(), inPlace);
+        if (inPlace) {
+            resource.getAttributes().put(SNAPSHOT, updateSnapshot);
+            reconcileTags(alarm.getAlarmArn(), tagKeys(priorAttributes.get(TAG_KEYS)), desiredTags, ctx.region());
+        }
+        resource.getAttributes().put(TAG_KEYS, MAPPER.valueToTree(desiredTags.keySet()).toString());
+        resource.setPhysicalId(name);
+        resource.getAttributes().put("Arn", alarm.getAlarmArn());
+        resource.getAttributes().put(EXPLICIT_NAME, Boolean.toString(explicit != null));
+        ReplacementCleanup.record(resource, ctx, priorAttributes);
+    }
+
+    private static void actions(JsonNode props, String field, ProvisionContext ctx, List<String> target) {
+        JsonNode values = ctx.engine().resolveNode(props.path(field));
+        if (values != null && values.isArray()) {
+            values.forEach(value -> target.add(value.asText()));
+        }
+    }
+
+    private static Set<String> tagKeys(String json) {
+        Set<String> keys = new TreeSet<>();
+        if (json != null) {
+            try {
+                MAPPER.readTree(json).forEach(value -> keys.add(value.asText()));
+            } catch (JsonProcessingException error) {
+                throw new IllegalStateException("Cannot read composite alarm tag ownership", error);
+            }
+        }
+        return keys;
+    }
+
+    private void reconcileTags(String arn, Set<String> owned, Map<String, String> desired, String region) {
+        List<String> removed = owned.stream().filter(key -> !desired.containsKey(key)).toList();
+        if (!removed.isEmpty()) {
+            service.untagResource(arn, removed, region);
+        }
+        if (!desired.isEmpty()) {
+            service.tagResource(arn, desired, region);
+        }
+    }
+
+    @Override
+    public void delete(String type, String physicalId, String region) {
+        service.deleteAlarms(List.of(physicalId), region);
+    }
+
+    @Override
+    public boolean retainsFailedUpdateState(StackResource resource) {
+        return resource.getAttributes().containsKey(SNAPSHOT);
+    }
+
+    @Override
+    public boolean rollbackUpdate(StackResource resource) {
+        if (ReplacementCleanup.rollback(resource, this::delete)) {
+            return true;
+        }
+        String raw = resource.getAttributes().get(SNAPSHOT);
+        if (raw == null) {
+            return false;
+        }
+        try {
+            JsonNode snapshot = MAPPER.readTree(raw);
+            String region = snapshot.path("region").asText();
+            if (snapshot.has("alarm")) {
+                CompositeAlarm original = MAPPER.treeToValue(snapshot.get("alarm"), CompositeAlarm.class);
+                Map<String, String> originalTags = Map.copyOf(original.getTags());
+                service.putCompositeAlarm(original, region);
+                Set<String> changed = tagKeys(snapshot.path("changedTagKeys").toString());
+                reconcileTags(original.getAlarmArn(), changed, originalTags.entrySet().stream()
+                        .filter(entry -> changed.contains(entry.getKey()))
+                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), region);
+            } else {
+                delete(TYPE, snapshot.path("name").asText(), region);
+            }
+            resource.getAttributes().clear();
+            snapshot.path("attributes").fields().forEachRemaining(entry ->
+                    resource.getAttributes().put(entry.getKey(), entry.getValue().asText()));
+        } catch (JsonProcessingException error) {
+            throw new IllegalStateException("Cannot read composite alarm rollback state", error);
+        }
+        return true;
+    }
+
+    @Override
+    public boolean hasReplacementUpdate(StackResource resource) {
+        return ReplacementCleanup.hasReplacement(resource);
+    }
+
+    @Override
+    public String updateCleanupPhysicalId(StackResource resource) {
+        return ReplacementCleanup.cleanupPhysicalId(resource);
+    }
+
+    @Override
+    public UpdateCleanupResult completeUpdate(StackResource resource) {
+        if (resource.getAttributes().containsKey(SNAPSHOT) && !ReplacementCleanup.hasReplacement(resource)) {
+            return new UpdateCleanupResult(true, true, null, 0, null);
+        }
+        return ReplacementCleanup.complete(resource, this::delete);
+    }
+
+    @Override
+    public void clearUpdate(StackResource resource) {
+        resource.getAttributes().remove(SNAPSHOT);
+        ReplacementCleanup.clear(resource);
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudformation/provisioners/ReplacementCleanup.java

@@ -9,9 +9,11 @@ import io.github.hectorvent.floci.services.cloudformation.model.StackResource;
 import org.jboss.logging.Logger;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * The replacement lifecycle for provisioners whose replacing update simply creates the new entity
@@ -121,6 +123,22 @@ public final class ReplacementCleanup {
         return cleanup != null && cleanup.path("displaced").size() > 0;
     }
 
+    /** The physical IDs of replaced resources this resource still has to delete. */
+    static Set<String> owedPhysicalIds(StackResource r) {
+        Set<String> owed = new HashSet<>();
+        ObjectNode cleanup = read(r);
+        if (cleanup == null) {
+            return owed;
+        }
+        for (JsonNode entry : cleanup.path("displaced")) {
+            String physicalId = entry.path("physicalId").asText(null);
+            if (physicalId != null) {
+                owed.add(physicalId);
+            }
+        }
+        return owed;
+    }
+
     /**
      * The entity the cleanup is working on, or null when none is owed or {@code UpdateReplacePolicy:
      * Retain} keeps it. The engine announces one physical id per resource, so a resource owing more

src/main/java/io/github/hectorvent/floci/services/cloudwatch/dashboards/CloudWatchDashboardsService.java

@@ -179,29 +179,82 @@ public class CloudWatchDashboardsService {
     }
 
     public Map<String, String> listTagsForResource(String resourceArn, String region) {
-        return findByArn(resourceArn, region).map(Dashboard::getTags).orElse(Map.of());
+        return requireDashboard(resourceArn, region).getTags();
     }
 
     public void tagResource(String resourceArn, Map<String, String> tags, String region) {
-        findByArn(resourceArn, region).ifPresent(dashboard -> {
+        Dashboard dashboard = requireDashboard(resourceArn, region);
         dashboard.getTags().putAll(tags);
         dashboardStore.put(key(region, dashboard.getDashboardName()), dashboard);
-        });
     }
 
     public void untagResource(String resourceArn, List<String> tagKeys, String region) {
-        findByArn(resourceArn, region).ifPresent(dashboard -> {
+        Dashboard dashboard = requireDashboard(resourceArn, region);
         tagKeys.forEach(dashboard.getTags()::remove);
         dashboardStore.put(key(region, dashboard.getDashboardName()), dashboard);
-        });
+    }
+
+    /**
+     * Resolves the dashboard an ARN names, or reports that nothing does. The tag operations
+     * declare {@code ResourceNotFoundException}, which is a different shape from the
+     * {@code ResourceNotFound} that {@code GetDashboard} declares,
+     * so this is not the {@link #notFound(String)} used by the rest of this service.
+     */
+    private Dashboard requireDashboard(String resourceArn, String region) {
+        return findByArn(resourceArn, region).orElseThrow(() ->
+                new AwsException("ResourceNotFoundException",
+                        "Dashboard does not exist: " + resourceArn, 404));
     }
 
     private java.util.Optional<Dashboard> findByArn(String resourceArn, String region) {
         return dashboardStore.scan(k -> k.startsWith(region + "::")).stream()
-                .filter(d -> d.getDashboardArn() != null && d.getDashboardArn().equals(resourceArn))
+                .filter(d -> namesDashboard(resourceArn, d.getDashboardArn()))
                 .findFirst();
     }
 
+    /**
+     * Whether {@code candidate} is an ARN for the dashboard whose own ARN is {@code stored}.
+     *
+     * <p>Both the region-ful and the regionless form resolve, deliberately, because the two
+     * disagree here and a caller may hold either. Dashboards are a global resource and AWS
+     * documents their ARN without a region on both the TagResource and the ListTagsForResource
+     * pages, {@code arn:aws:cloudwatch::<account-id>:dashboard/<dashboard-name>}, while Floci
+     * mints a region-ful one. That divergence used to be invisible on this path: an ARN
+     * matching nothing was a silent success answering an empty tag map. Now that it is an
+     * error, a caller sending the documented ARN for a dashboard that exists would be told the
+     * dashboard does not, so the lookup accepts both shapes.
+     *
+     * <p>What Floci mints is left alone here on purpose. Clients may already hold the ARNs it
+     * has handed out, so bringing the minted shape into line with AWS is a separate change
+     * from making the lookup tolerant, and only the second one is needed to keep this path
+     * correct.
+     *
+     * <p>Only an absent region is accepted in place of the dashboard's own, so an ARN naming
+     * some other region is still a miss rather than resolving to a same-named dashboard
+     * elsewhere.
+     */
+    private static boolean namesDashboard(String candidate, String stored) {
+        if (candidate == null || stored == null) {
+            return false;
+        }
+        return candidate.equals(stored) || candidate.equals(withoutRegion(stored));
+    }
+
+    /**
+     * Blanks the region field of an ARN, turning
+     * {@code arn:aws:cloudwatch:us-east-1:123:dashboard/x} into
+     * {@code arn:aws:cloudwatch::123:dashboard/x}. An ARN with fewer than six fields is
+     * returned unchanged, since there is no region field to blank.
+     */
+    private static String withoutRegion(String arn) {
+        String[] fields = arn.split(":", 6);
+        if (fields.length < 6) {
+            return arn;
+        }
+        fields[3] = "";
+        return String.join(":", fields);
+    }
+
     private static String key(String region, String dashboardName) {
         return region + "::" + dashboardName;
     }

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmEvaluator.java

@@ -275,9 +275,10 @@ public class AlarmEvaluator {
 
     private static boolean isEvaluable(MetricAlarm alarm) {
         return alarm.getRegion() != null
+                && alarm.getMetrics().isEmpty()
                 && alarm.getNamespace() != null && !alarm.getNamespace().isBlank()
                 && alarm.getMetricName() != null && !alarm.getMetricName().isBlank()
-                && alarm.getPeriod() > 0
+                && alarm.getPeriod() != null && alarm.getPeriod() > 0
                 && alarm.getEvaluationPeriods() > 0
                 && alarm.getComparisonOperator() != null;
     }

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifier.java

@@ -0,0 +1,155 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.AwsArnUtils;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.sns.SnsService;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.jboss.logging.Logger;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/** Publishes SNS actions once per state transition, independently of scaling action retries. */
+@ApplicationScoped
+public class AlarmSnsNotifier {
+
+    /** The names CloudWatch writes in the alarm message's Region field; other regions use their ID. */
+    private static final Map<String, String> REGION_NAMES = Map.ofEntries(
+            Map.entry("us-east-1", "US East (N. Virginia)"),
+            Map.entry("us-east-2", "US East (Ohio)"),
+            Map.entry("us-west-1", "US West (N. California)"),
+            Map.entry("us-west-2", "US West (Oregon)"),
+            Map.entry("ca-central-1", "Canada (Central)"),
+            Map.entry("ca-west-1", "Canada West (Calgary)"),
+            Map.entry("sa-east-1", "South America (Sao Paulo)"),
+            Map.entry("eu-west-1", "EU (Ireland)"),
+            Map.entry("eu-west-2", "EU (London)"),
+            Map.entry("eu-west-3", "EU (Paris)"),
+            Map.entry("eu-central-1", "EU (Frankfurt)"),
+            Map.entry("eu-central-2", "EU (Zurich)"),
+            Map.entry("eu-north-1", "EU (Stockholm)"),
+            Map.entry("eu-south-1", "EU (Milan)"),
+            Map.entry("eu-south-2", "EU (Spain)"),
+            Map.entry("me-south-1", "Middle East (Bahrain)"),
+            Map.entry("me-central-1", "Middle East (UAE)"),
+            Map.entry("il-central-1", "Israel (Tel Aviv)"),
+            Map.entry("af-south-1", "Africa (Cape Town)"),
+            Map.entry("ap-east-1", "Asia Pacific (Hong Kong)"),
+            Map.entry("ap-south-1", "Asia Pacific (Mumbai)"),
+            Map.entry("ap-south-2", "Asia Pacific (Hyderabad)"),
+            Map.entry("ap-northeast-1", "Asia Pacific (Tokyo)"),
+            Map.entry("ap-northeast-2", "Asia Pacific (Seoul)"),
+            Map.entry("ap-northeast-3", "Asia Pacific (Osaka)"),
+            Map.entry("ap-southeast-1", "Asia Pacific (Singapore)"),
+            Map.entry("ap-southeast-2", "Asia Pacific (Sydney)"),
+            Map.entry("ap-southeast-3", "Asia Pacific (Jakarta)"),
+            Map.entry("ap-southeast-4", "Asia Pacific (Melbourne)"));
+
+    private static final Logger LOG = Logger.getLogger(AlarmSnsNotifier.class);
+    private static final DateTimeFormatter TIMESTAMP_FORMAT = DateTimeFormatter
+            .ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ", Locale.ROOT).withZone(ZoneOffset.UTC);
+    private final SnsService snsService;
+    private final ObjectMapper objectMapper;
+
+    public AlarmSnsNotifier(SnsService snsService, ObjectMapper objectMapper) {
+        this.snsService = snsService;
+        this.objectMapper = objectMapper;
+    }
+
+    record Notification(String topicArn, String message, String subject, String region) {}
+
+    List<Notification> prepareTransition(MetricAlarm alarm, String oldState) {
+        if (!alarm.isActionsEnabled() || alarm.getStateValue().equals(oldState)) {
+            return List.of();
+        }
+        List<String> actions = switch (alarm.getStateValue()) {
+            case "ALARM" -> alarm.getAlarmActions();
+            case "OK" -> alarm.getOkActions();
+            case "INSUFFICIENT_DATA" -> alarm.getInsufficientDataActions();
+            default -> List.of();
+        };
+        if (actions == null) {
+            return List.of();
+        }
+        List<Notification> notifications = new ArrayList<>();
+        for (String action : actions) {
+            if (!AwsArnUtils.isArnFor(action, "sns")) {
+                continue;
+            }
+            try {
+                AwsArnUtils.Arn topic = AwsArnUtils.parse(action);
+                String subject = alarm.getStateValue() + ": \"" + alarm.getAlarmName() + "\"";
+                if (subject.length() > 100) {
+                    subject = subject.substring(0, 100);
+                }
+                notifications.add(new Notification(action, message(alarm, oldState), subject, topic.region()));
+            } catch (Exception e) {
+                LOG.warnv(e, "Unable to publish alarm {0} state {1} to SNS action {2}",
+                        alarm.getAlarmName(), alarm.getStateValue(), action);
+            }
+        }
+        return notifications;
+    }
+
+    void publish(List<Notification> notifications) {
+        for (Notification notification : notifications) {
+            try {
+                snsService.publish(notification.topicArn(), null, notification.message(),
+                        notification.subject(), notification.region());
+            } catch (Exception error) {
+                LOG.warnv(error, "Unable to publish alarm transition {0} to SNS action {1}",
+                        notification.subject(), notification.topicArn());
+            }
+        }
+    }
+
+    private String message(MetricAlarm alarm, String oldState) throws JsonProcessingException {
+        AwsArnUtils.Arn arn = AwsArnUtils.parse(alarm.getAlarmArn());
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("AlarmName", alarm.getAlarmName());
+        payload.put("AlarmDescription", alarm.getAlarmDescription());
+        payload.put("AWSAccountId", arn.accountId());
+        payload.put("AlarmConfigurationUpdatedTimestamp", timestamp(alarm.getAlarmConfigurationUpdatedTimestamp()));
+        payload.put("NewStateValue", alarm.getStateValue());
+        payload.put("NewStateReason", alarm.getStateReason());
+        payload.put("StateChangeTime", timestamp(alarm.getStateUpdatedTimestamp()));
+        payload.put("Region", REGION_NAMES.getOrDefault(arn.region(), arn.region()));
+        payload.put("AlarmArn", alarm.getAlarmArn());
+        payload.put("OldStateValue", oldState);
+        payload.put("OKActions", alarm.getOkActions());
+        payload.put("AlarmActions", alarm.getAlarmActions());
+        payload.put("InsufficientDataActions", alarm.getInsufficientDataActions());
+        Map<String, Object> trigger = new LinkedHashMap<>();
+        trigger.put("MetricName", alarm.getMetricName());
+        trigger.put("Namespace", alarm.getNamespace());
+        trigger.put("StatisticType", "Statistic");
+        trigger.put("Statistic", alarm.getStatistic() != null
+                ? alarm.getStatistic().toUpperCase(Locale.ROOT) : null);
+        trigger.put("Unit", alarm.getUnit());
+        trigger.put("Dimensions", alarm.getDimensions());
+        trigger.put("Period", alarm.getPeriod());
+        trigger.put("EvaluationPeriods", alarm.getEvaluationPeriods());
+        if (alarm.getDatapointsToAlarm() != null) {
+            trigger.put("DatapointsToAlarm", alarm.getDatapointsToAlarm());
+        }
+        trigger.put("ComparisonOperator", alarm.getComparisonOperator());
+        trigger.put("Threshold", alarm.getThreshold());
+        trigger.put("TreatMissingData", alarm.getTreatMissingData());
+        trigger.put("EvaluateLowSampleCountPercentile", alarm.getEvaluateLowSampleCountPercentile() != null
+                ? alarm.getEvaluateLowSampleCountPercentile() : "");
+        payload.put("Trigger", trigger);
+        return objectMapper.writeValueAsString(payload);
+    }
+
+    private String timestamp(long epochSeconds) {
+        return TIMESTAMP_FORMAT.format(Instant.ofEpochSecond(epochSeconds));
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsException.java

@@ -0,0 +1,34 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import io.github.hectorvent.floci.core.common.AwsException;
+
+/**
+ * A CloudWatch error whose Query code and JSON shape name differ.
+ *
+ * <p>CloudWatch is awsQueryCompatible: the Query {@code <Code>} and the
+ * {@code x-amzn-query-error} header carry the {@code awsQueryError} code
+ * ({@code MissingParameter}), while a JSON 1.0 or CBOR body's {@code __type} names the
+ * model shape ({@code MissingRequiredParameterException}). The AWS SDK v2 picks the typed
+ * exception from {@code __type}, so the code alone yields a generic {@code CloudWatchException}.
+ * Keeping {@link #getErrorCode()} as the Query code and overriding {@link #jsonType()}
+ * reproduces both surfaces.
+ */
+class CloudWatchMetricsException extends AwsException {
+
+    private final String jsonType;
+
+    private CloudWatchMetricsException(String errorCode, String jsonType, String message, int httpStatus) {
+        super(errorCode, message, httpStatus);
+        this.jsonType = jsonType;
+    }
+
+    static CloudWatchMetricsException missingParameter(String parameter) {
+        return new CloudWatchMetricsException("MissingParameter", "MissingRequiredParameterException",
+                "The parameter " + parameter + " is required.", 400);
+    }
+
+    @Override
+    public String jsonType() {
+        return jsonType;
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsJsonHandler.java

@@ -5,13 +5,21 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.github.hectorvent.floci.core.common.AwsErrorResponse;
+import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.PaginatedResult;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.model.Dashboard;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.AlarmMetricDataQuery;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Dimension;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.CloudWatchMetricStreamsService;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStream;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStreamFilter;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStreamStatisticsConfiguration;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.AlarmMetricDataQuery;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Dimension;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
@@ -56,6 +64,7 @@ public class CloudWatchMetricsJsonHandler {
             case "ListMetrics" -> handleListMetrics(request, region);
             case "GetMetricStatistics" -> handleGetMetricStatistics(request, region);
             case "PutMetricAlarm" -> handlePutMetricAlarm(request, region);
+            case "PutCompositeAlarm" -> handlePutCompositeAlarm(request, region);
             case "DescribeAlarms" -> handleDescribeAlarms(request, region);
             case "DeleteAlarms" -> handleDeleteAlarms(request, region);
             case "SetAlarmState" -> handleSetAlarmState(request, region);
@@ -81,10 +90,14 @@ public class CloudWatchMetricsJsonHandler {
 
     private Response handlePutMetricData(JsonNode request, String region) {
         String namespace = request.path("Namespace").asText();
-        LOG.infov("JSON PutMetricData raw: {0}", request);
+        List<MetricDatum> datums = parseMetricDataJson(request.path("MetricData"), "MetricData");
-        List<MetricDatum> datums = parseMetricDataJson(request.path("MetricData"));
+        int entityIndex = 0;
-        LOG.infov("JSON PutMetricData parsed {0} datums, sums={1}", datums.size(),
+        for (JsonNode entityMetricData : request.path("EntityMetricData")) {
-                datums.stream().map(d -> d.getMetricName() + "=" + d.getSum()).toList());
+            entityIndex++;
+            datums.addAll(parseMetricDataJson(entityMetricData.path("MetricData"),
+                    "EntityMetricData.member." + entityIndex + ".MetricData"));
+        }
+        LOG.debugv("JSON PutMetricData parsed {0} datums for namespace {1}", datums.size(), namespace);
         metricsService.putMetricData(namespace, datums, region);
         return Response.ok(objectMapper.createObjectNode()).build();
     }
@@ -151,12 +164,17 @@ public class CloudWatchMetricsJsonHandler {
 
     private Response handlePutMetricAlarm(JsonNode request, String region) {
         MetricAlarm alarm = new MetricAlarm();
-        alarm.setAlarmName(request.path("AlarmName").asText());
+        parseAlarmFields(request, alarm);
-        alarm.setAlarmDescription(request.path("AlarmDescription").asText(null));
         alarm.setMetricName(request.path("MetricName").asText(null));
         alarm.setNamespace(request.path("Namespace").asText(null));
         alarm.setStatistic(request.path("Statistic").asText(null));
-        alarm.setPeriod(request.path("Period").asInt(60));
+        JsonNode metrics = request.path("Metrics");
+        if (metrics.isArray()) {
+            for (JsonNode query : metrics) {
+                alarm.getMetrics().add(objectMapper.convertValue(query, AlarmMetricDataQuery.class));
+            }
+        }
+        alarm.setPeriod(alarm.getMetrics().isEmpty() ? request.path("Period").asInt(60) : null);
         alarm.setUnit(request.path("Unit").asText(null));
         alarm.setEvaluationPeriods(request.path("EvaluationPeriods").asInt(1));
         alarm.setDatapointsToAlarm(request.hasNonNull("DatapointsToAlarm")
@@ -164,9 +182,33 @@ public class CloudWatchMetricsJsonHandler {
         alarm.setThreshold(request.path("Threshold").asDouble(0));
         alarm.setComparisonOperator(request.path("ComparisonOperator").asText(null));
         alarm.setTreatMissingData(request.path("TreatMissingData").asText(null));
-        alarm.setActionsEnabled(request.path("ActionsEnabled").asBoolean(true));
         alarm.setDimensions(parseDimensionsJson(request.path("Dimensions")));
 
+        metricsService.putMetricAlarm(alarm, region);
+        return Response.ok(objectMapper.createObjectNode()).build();
+    }
+
+    private Response handlePutCompositeAlarm(JsonNode request, String region) {
+        rejectSuppressor(request);
+        CompositeAlarm alarm = new CompositeAlarm();
+        parseAlarmFields(request, alarm);
+        alarm.setAlarmRule(request.path("AlarmRule").asText(null));
+        metricsService.putCompositeAlarm(alarm, region);
+        return Response.ok(objectMapper.createObjectNode()).build();
+    }
+
+    private static void rejectSuppressor(JsonNode request) {
+        if (request.has("ActionsSuppressor") || request.has("ActionsSuppressorWaitPeriod")
+                || request.has("ActionsSuppressorExtensionPeriod")) {
+            throw new AwsException("ValidationError", "Composite action suppression is not supported", 400);
+        }
+    }
+
+    private void parseAlarmFields(JsonNode request, Alarm alarm) {
+        alarm.setAlarmName(request.path("AlarmName").asText());
+        alarm.setAlarmDescription(request.path("AlarmDescription").asText(null));
+        alarm.setActionsEnabled(request.path("ActionsEnabled").asBoolean(true));
+
         JsonNode alarmActions = request.path("AlarmActions");
         if (alarmActions.isArray()) {
             alarmActions.forEach(a -> alarm.getAlarmActions().add(a.asText()));
@@ -187,8 +229,6 @@ public class CloudWatchMetricsJsonHandler {
             alarm.setTags(tags);
         }
 
-        metricsService.putMetricAlarm(alarm, region);
-        return Response.ok(objectMapper.createObjectNode()).build();
     }
 
     private Response handleDescribeAlarms(JsonNode request, String region) {
@@ -199,32 +239,42 @@ public class CloudWatchMetricsJsonHandler {
         }
         String prefix = request.has("AlarmNamePrefix") ? request.path("AlarmNamePrefix").asText() : null;
 
-        List<MetricAlarm> alarms = metricsService.describeAlarms(alarmNames, prefix, region);
+        List<String> types = new ArrayList<>();
+        request.path("AlarmTypes").forEach(value -> types.add(value.asText()));
+        metricsService.validateAlarmTypes(types);
+        List<MetricAlarm> alarms = types.isEmpty() || types.contains("MetricAlarm")
+                ? metricsService.describeAlarms(alarmNames, prefix, region) : List.of();
 
         ObjectNode response = objectMapper.createObjectNode();
         ArrayNode arr = response.putArray("MetricAlarms");
         for (MetricAlarm a : alarms) {
             ObjectNode node = arr.addObject();
-            node.put("AlarmName", a.getAlarmName());
+            describeAlarmFields(node, a);
-            if (a.getAlarmArn() != null) node.put("AlarmArn", a.getAlarmArn());
+            if (!a.getMetrics().isEmpty()) {
-            if (a.getAlarmDescription() != null) node.put("AlarmDescription", a.getAlarmDescription());
+                node.set("Metrics", objectMapper.valueToTree(a.getMetrics()));
-            ArrayNode alarmActions = node.putArray("AlarmActions");
+            } else {
-            a.getAlarmActions().forEach(alarmActions::add);
+                if (a.getMetricName() != null) {
-            ArrayNode okActions = node.putArray("OKActions");
+                    node.put("MetricName", a.getMetricName());
-            a.getOkActions().forEach(okActions::add);
+                }
-            ArrayNode insufficientDataActions = node.putArray("InsufficientDataActions");
+                if (a.getNamespace() != null) {
-            a.getInsufficientDataActions().forEach(insufficientDataActions::add);
+                    node.put("Namespace", a.getNamespace());
-            if (a.getMetricName() != null) node.put("MetricName", a.getMetricName());
+                }
-            if (a.getNamespace() != null) node.put("Namespace", a.getNamespace());
+                if (a.getStatistic() != null) {
-            if (a.getStatistic() != null) node.put("Statistic", a.getStatistic());
+                    node.put("Statistic", a.getStatistic());
+                }
                 ArrayNode dimensions = node.putArray("Dimensions");
                 a.getDimensions().forEach(d -> {
                     ObjectNode dimNode = dimensions.addObject();
                     dimNode.put("Name", d.name());
                     dimNode.put("Value", d.value());
                 });
+                if (a.getPeriod() != null) {
                     node.put("Period", a.getPeriod());
-            if (a.getUnit() != null) node.put("Unit", a.getUnit());
+                }
+                if (a.getUnit() != null) {
+                    node.put("Unit", a.getUnit());
+                }
+            }
             node.put("EvaluationPeriods", a.getEvaluationPeriods());
             if (a.getDatapointsToAlarm() != null) {
                 node.put("DatapointsToAlarm", a.getDatapointsToAlarm());
@@ -232,15 +282,48 @@ public class CloudWatchMetricsJsonHandler {
             node.put("Threshold", a.getThreshold());
             if (a.getComparisonOperator() != null) node.put("ComparisonOperator", a.getComparisonOperator());
             if (a.getTreatMissingData() != null) node.put("TreatMissingData", a.getTreatMissingData());
-            node.put("ActionsEnabled", a.isActionsEnabled());
+
-            if (a.getStateValue() != null) node.put("StateValue", a.getStateValue());
-            if (a.getStateReason() != null) node.put("StateReason", a.getStateReason());
-            if (a.getStateReasonData() != null) node.put("StateReasonData", a.getStateReasonData());
-            node.put("StateUpdatedTimestamp", a.getStateUpdatedTimestamp());
         }
+        ArrayNode composites = response.putArray("CompositeAlarms");
+        if (types.contains("CompositeAlarm")) {
+            for (CompositeAlarm alarm : metricsService.describeCompositeAlarms(alarmNames, prefix, region)) {
+                ObjectNode node = composites.addObject();
+                describeAlarmFields(node, alarm);
+                node.put("AlarmRule", alarm.getAlarmRule());
+            }
+        }
+        response.putArray("LogAlarms");
         return Response.ok(response).build();
     }
 
+    private void describeAlarmFields(ObjectNode node, Alarm a) {
+        node.put("AlarmName", a.getAlarmName());
+        if (a.getAlarmArn() != null) {
+            node.put("AlarmArn", a.getAlarmArn());
+        }
+        if (a.getAlarmDescription() != null) {
+            node.put("AlarmDescription", a.getAlarmDescription());
+        }
+        ArrayNode alarmActions = node.putArray("AlarmActions");
+        a.getAlarmActions().forEach(alarmActions::add);
+        ArrayNode okActions = node.putArray("OKActions");
+        a.getOkActions().forEach(okActions::add);
+        ArrayNode insufficientDataActions = node.putArray("InsufficientDataActions");
+        a.getInsufficientDataActions().forEach(insufficientDataActions::add);
+        node.put("ActionsEnabled", a.isActionsEnabled());
+        if (a.getStateValue() != null) {
+            node.put("StateValue", a.getStateValue());
+        }
+        if (a.getStateReason() != null) {
+            node.put("StateReason", a.getStateReason());
+        }
+        if (a.getStateReasonData() != null) {
+            node.put("StateReasonData", a.getStateReasonData());
+        }
+        node.put("StateUpdatedTimestamp", a.getStateUpdatedTimestamp());
+        node.put("AlarmConfigurationUpdatedTimestamp", a.getAlarmConfigurationUpdatedTimestamp());
+    }
+
     private Response handleDeleteAlarms(JsonNode request, String region) {
         List<String> alarmNames = new ArrayList<>();
         JsonNode namesNode = request.path("AlarmNames");
@@ -586,12 +669,18 @@ public class CloudWatchMetricsJsonHandler {
         return configurations;
     }
 
-    private List<MetricDatum> parseMetricDataJson(JsonNode node) {
+    private List<MetricDatum> parseMetricDataJson(JsonNode node, String listPath) {
         List<MetricDatum> datums = new ArrayList<>();
         if (!node.isArray()) return datums;
+        int index = 0;
         for (JsonNode item : node) {
+            index++;
+            JsonNode metricName = item.path("MetricName");
+            if (!metricName.isTextual() || metricName.asText().isEmpty()) {
+                throw CloudWatchMetricsException.missingParameter(listPath + ".member." + index + ".MetricName");
+            }
             MetricDatum datum = new MetricDatum();
-            datum.setMetricName(item.path("MetricName").asText());
+            datum.setMetricName(metricName.asText());
             datum.setValue(item.path("Value").asDouble(0));
             datum.setUnit(item.path("Unit").asText(null));
             JsonNode ts = item.path("Timestamp");

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsQueryHandler.java

@@ -1,5 +1,6 @@
 package io.github.hectorvent.floci.services.cloudwatch.metrics;
 
+import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.AwsNamespaces;
 import io.github.hectorvent.floci.core.common.AwsQueryResponse;
 import io.github.hectorvent.floci.core.common.PaginatedResult;
@@ -7,10 +8,17 @@ import io.github.hectorvent.floci.core.common.Pagination;
 import io.github.hectorvent.floci.core.common.XmlBuilder;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.model.Dashboard;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.AlarmMetricDataQuery;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Dimension;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.CloudWatchMetricStreamsService;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStream;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStreamFilter;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.model.MetricStreamStatisticsConfiguration;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.AlarmMetricDataQuery;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Dimension;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
@@ -25,11 +33,16 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import org.jboss.logging.Logger;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 @ApplicationScoped
 public class CloudWatchMetricsQueryHandler {
 
     private static final Logger LOG = Logger.getLogger(CloudWatchMetricsQueryHandler.class);
+    private static final Pattern ENTITY_METRIC_DATA_MEMBER = Pattern.compile("EntityMetricData\\.member\\.(\\d{1,9})\\.");
 
     private CloudWatchMetricsService metricsService;
     private final CloudWatchDashboardsService dashboardsService;
@@ -52,6 +65,7 @@ public class CloudWatchMetricsQueryHandler {
             case "GetMetricStatistics" -> handleGetMetricStatistics(params, region);
             case "GetMetricData" -> handleGetMetricData(params, region);
             case "PutMetricAlarm" -> handlePutMetricAlarm(params, region);
+            case "PutCompositeAlarm" -> handlePutCompositeAlarm(params, region);
             case "DescribeAlarms" -> handleDescribeAlarms(params, region);
             case "DeleteAlarms" -> handleDeleteAlarms(params, region);
             case "SetAlarmState" -> handleSetAlarmState(params, region);
@@ -153,7 +167,8 @@ public class CloudWatchMetricsQueryHandler {
             xml.elem("Unit", dp.unit()).end("member");
         }
         xml.end("Datapoints");
-        return Response.ok(AwsQueryResponse.envelope("GetMetricStatistics", AwsNamespaces.CW, xml.build())).build();
+        return Response.ok(AwsQueryResponse.envelope("GetMetricStatistics",
+                AwsNamespaces.CW, xml.build())).build();
     }
 
     private Response handleGetMetricData(MultivaluedMap<String, String> params, String region) {
@@ -237,6 +252,26 @@ public class CloudWatchMetricsQueryHandler {
         return Response.ok(AwsQueryResponse.envelopeNoResult("PutMetricAlarm", null)).build();
     }
 
+    private Response handlePutCompositeAlarm(MultivaluedMap<String, String> params, String region) {
+        if (params.containsKey("ActionsSuppressor") || params.containsKey("ActionsSuppressorWaitPeriod")
+                || params.containsKey("ActionsSuppressorExtensionPeriod")) {
+            throw new AwsException("ValidationError", "Composite action suppression is not supported", 400);
+        }
+        CompositeAlarm alarm = new CompositeAlarm();
+        parseAlarmFields(params, alarm);
+        alarm.setAlarmRule(params.getFirst("AlarmRule"));
+        metricsService.putCompositeAlarm(alarm, region);
+        return Response.ok(AwsQueryResponse.envelopeNoResult("PutCompositeAlarm", null)).build();
+    }
+
+    private static List<String> alarmMembers(MultivaluedMap<String, String> params, String key) {
+        List<String> values = new ArrayList<>();
+        for (int i = 1; params.getFirst(key + ".member." + i) != null; i++) {
+            values.add(params.getFirst(key + ".member." + i));
+        }
+        return values;
+    }
+
     private Response handleDescribeAlarms(MultivaluedMap<String, String> params, String region) {
         List<String> alarmNames = new ArrayList<>();
         for (int i = 1; ; i++) {
@@ -246,13 +281,24 @@ public class CloudWatchMetricsQueryHandler {
         }
         String prefix = params.getFirst("AlarmNamePrefix");
 
-        List<MetricAlarm> alarms = metricsService.describeAlarms(alarmNames, prefix, region);
+        List<String> types = alarmMembers(params, "AlarmTypes");
+        metricsService.validateAlarmTypes(types);
+        List<MetricAlarm> alarms = types.isEmpty() || types.contains("MetricAlarm")
+                ? metricsService.describeAlarms(alarmNames, prefix, region) : List.of();
 
         var xml = new XmlBuilder().start("MetricAlarms");
         for (MetricAlarm a : alarms) {
             toAlarmXml(xml, a);
         }
-        xml.end("MetricAlarms");
+        xml.end("MetricAlarms").start("CompositeAlarms");
+        if (types.contains("CompositeAlarm")) {
+            for (CompositeAlarm alarm : metricsService.describeCompositeAlarms(alarmNames, prefix, region)) {
+                xml.start("member");
+                describeAlarmFields(xml, alarm);
+                xml.elem("AlarmRule", alarm.getAlarmRule()).end("member");
+            }
+        }
+        xml.end("CompositeAlarms").start("LogAlarms").end("LogAlarms");
         return Response.ok(AwsQueryResponse.envelope("DescribeAlarms", null, xml.build())).build();
     }
 
@@ -590,42 +636,63 @@ public class CloudWatchMetricsQueryHandler {
     // ──────────────────────────── Parsing Helpers ────────────────────────────
 
     private List<MetricDatum> parseMetricData(MultivaluedMap<String, String> params) {
-        LOG.infov("Query PutMetricData params: {0}", params);
+        List<MetricDatum> datums = parseMetricData(params, "MetricData");
+        // An empty EntityMetricData entry serializes to no keys, so its index is absent
+        // while later entries keep theirs: iterate the indexes present, not 1..first gap.
+        SortedSet<Integer> entityIndexes = new TreeSet<>();
+        for (String key : params.keySet()) {
+            Matcher matcher = ENTITY_METRIC_DATA_MEMBER.matcher(key);
+            if (matcher.lookingAt()) {
+                entityIndexes.add(Integer.parseInt(matcher.group(1)));
+            }
+        }
+        for (int index : entityIndexes) {
+            datums.addAll(parseMetricData(params, "EntityMetricData.member." + index + ".MetricData"));
+        }
+        LOG.debugv("Query PutMetricData parsed {0} datums", datums.size());
+        return datums;
+    }
+
+    private List<MetricDatum> parseMetricData(MultivaluedMap<String, String> params, String listPrefix) {
         List<MetricDatum> datums = new ArrayList<>();
         for (int i = 1; ; i++) {
-            String metricName = params.getFirst("MetricData.member." + i + ".MetricName");
+            String memberPrefix = listPrefix + ".member." + i + ".";
-            if (metricName == null) {
+            String metricName = params.getFirst(memberPrefix + "MetricName");
+            if (metricName == null || metricName.isEmpty()) {
+                if (metricName == null && params.keySet().stream().noneMatch(key -> key.startsWith(memberPrefix))) {
                     break;
                 }
+                throw CloudWatchMetricsException.missingParameter(memberPrefix + "MetricName");
+            }
             MetricDatum datum = new MetricDatum();
             datum.setMetricName(metricName);
-            datum.setUnit(params.getFirst("MetricData.member." + i + ".Unit"));
+            datum.setUnit(params.getFirst(listPrefix + ".member." + i + ".Unit"));
 
-            String valueStr = params.getFirst("MetricData.member." + i + ".Value");
+            String valueStr = params.getFirst(listPrefix + ".member." + i + ".Value");
             if (valueStr != null) {
                 datum.setValue(parseDouble(valueStr, 0));
             }
 
-            String tsStr = params.getFirst("MetricData.member." + i + ".Timestamp");
+            String tsStr = params.getFirst(listPrefix + ".member." + i + ".Timestamp");
             if (tsStr != null) {
                 Instant ts = parseInstant(tsStr);
                 datum.setTimestamp(ts != null ? ts.getEpochSecond() : 0);
             }
 
             // StatisticValues
-            String sc = params.getFirst("MetricData.member." + i + ".StatisticValues.SampleCount");
+            String sc = params.getFirst(listPrefix + ".member." + i + ".StatisticValues.SampleCount");
             if (sc != null) {
                 datum.setSampleCount(parseDouble(sc, 0));
-                datum.setSum(parseDouble(params.getFirst("MetricData.member." + i + ".StatisticValues.Sum"), 0));
+                datum.setSum(parseDouble(params.getFirst(listPrefix + ".member." + i + ".StatisticValues.Sum"), 0));
-                datum.setMinimum(parseDouble(params.getFirst("MetricData.member." + i + ".StatisticValues.Minimum"), 0));
+                datum.setMinimum(parseDouble(params.getFirst(listPrefix + ".member." + i + ".StatisticValues.Minimum"), 0));
-                datum.setMaximum(parseDouble(params.getFirst("MetricData.member." + i + ".StatisticValues.Maximum"), 0));
+                datum.setMaximum(parseDouble(params.getFirst(listPrefix + ".member." + i + ".StatisticValues.Maximum"), 0));
             }
 
             // Dimensions
             List<Dimension> dims = new ArrayList<>();
             for (int j = 1; ; j++) {
-                String dimName = params.getFirst("MetricData.member." + i + ".Dimensions.member." + j + ".Name");
+                String dimName = params.getFirst(listPrefix + ".member." + i + ".Dimensions.member." + j + ".Name");
-                String dimValue = params.getFirst("MetricData.member." + i + ".Dimensions.member." + j + ".Value");
+                String dimValue = params.getFirst(listPrefix + ".member." + i + ".Dimensions.member." + j + ".Value");
                 if (dimName == null) {
                     break;
                 }
@@ -653,14 +720,12 @@ public class CloudWatchMetricsQueryHandler {
 
     private MetricAlarm parseAlarm(MultivaluedMap<String, String> params) {
         MetricAlarm a = new MetricAlarm();
-        a.setAlarmName(params.getFirst("AlarmName"));
+        parseAlarmFields(params, a);
-        a.setAlarmDescription(params.getFirst("AlarmDescription"));
-        String actionsEnabled = params.getFirst("ActionsEnabled");
-        a.setActionsEnabled(actionsEnabled == null || Boolean.parseBoolean(actionsEnabled));
         a.setMetricName(params.getFirst("MetricName"));
         a.setNamespace(params.getFirst("Namespace"));
         a.setStatistic(params.getFirst("Statistic"));
-        a.setPeriod(parseIntParam(params, "Period", 60));
+        a.setMetrics(parseAlarmMetricQueries(params));
+        a.setPeriod(a.getMetrics().isEmpty() ? parseIntParam(params, "Period", 60) : null);
         a.setUnit(params.getFirst("Unit"));
         a.setEvaluationPeriods(parseIntParam(params, "EvaluationPeriods", 1));
         String datapointsToAlarm = params.getFirst("DatapointsToAlarm");
@@ -680,38 +745,100 @@ public class CloudWatchMetricsQueryHandler {
         }
         a.setDimensions(dims);
 
-        // Actions
-        for (int i = 1; ; i++) {
-            String act = params.getFirst("OKActions.member." + i);
-            if (act == null) break;
-            a.getOkActions().add(act);
-        }
-        for (int i = 1; ; i++) {
-            String act = params.getFirst("AlarmActions.member." + i);
-            if (act == null) break;
-            a.getAlarmActions().add(act);
-        }
-        for (int i = 1; ; i++) {
-            String act = params.getFirst("InsufficientDataActions.member." + i);
-            if (act == null) break;
-            a.getInsufficientDataActions().add(act);
-        }
-
-        // Tags
-        Map<String, String> tags = new LinkedHashMap<>();
-        for (int i = 1; ; i++) {
-            String key = params.getFirst("Tags.member." + i + ".Key");
-            if (key == null) break;
-            tags.put(key, params.getFirst("Tags.member." + i + ".Value"));
-        }
-        a.setTags(tags);
-
         return a;
     }
 
+    private void parseAlarmFields(MultivaluedMap<String, String> params, Alarm alarm) {
+        alarm.setAlarmName(params.getFirst("AlarmName"));
+        alarm.setAlarmDescription(params.getFirst("AlarmDescription"));
+        String actionsEnabled = params.getFirst("ActionsEnabled");
+        alarm.setActionsEnabled(actionsEnabled == null || Boolean.parseBoolean(actionsEnabled));
+        alarm.setAlarmActions(alarmMembers(params, "AlarmActions"));
+        alarm.setOkActions(alarmMembers(params, "OKActions"));
+        alarm.setInsufficientDataActions(alarmMembers(params, "InsufficientDataActions"));
+        Map<String, String> tags = new LinkedHashMap<>();
+        for (int i = 1; params.getFirst("Tags.member." + i + ".Key") != null; i++) {
+            tags.put(params.getFirst("Tags.member." + i + ".Key"), params.getFirst("Tags.member." + i + ".Value"));
+        }
+        alarm.setTags(tags);
+    }
+
+    private List<AlarmMetricDataQuery> parseAlarmMetricQueries(MultivaluedMap<String, String> params) {
+        List<AlarmMetricDataQuery> queries = new ArrayList<>();
+        for (int index = 1; ; index++) {
+            String prefix = "Metrics.member." + index;
+            String id = params.getFirst(prefix + ".Id");
+            if (id == null) {
+                break;
+            }
+            AlarmMetricDataQuery.MetricStat metricStat = null;
+            String metricPrefix = prefix + ".MetricStat.Metric";
+            if (params.getFirst(metricPrefix + ".Namespace") != null) {
+                List<AlarmMetricDataQuery.MetricDimension> dimensions = null;
+                for (int dimensionIndex = 1; ; dimensionIndex++) {
+                    String dimensionPrefix = metricPrefix + ".Dimensions.member." + dimensionIndex;
+                    String name = params.getFirst(dimensionPrefix + ".Name");
+                    if (name == null) {
+                        break;
+                    }
+                    if (dimensions == null) {
+                        dimensions = new ArrayList<>();
+                    }
+                    dimensions.add(new AlarmMetricDataQuery.MetricDimension(
+                            name, params.getFirst(dimensionPrefix + ".Value")));
+                }
+                AlarmMetricDataQuery.Metric metric = new AlarmMetricDataQuery.Metric(
+                        params.getFirst(metricPrefix + ".Namespace"),
+                        params.getFirst(metricPrefix + ".MetricName"), dimensions);
+                metricStat = new AlarmMetricDataQuery.MetricStat(metric,
+                        optionalInteger(params.getFirst(prefix + ".MetricStat.Period")),
+                        params.getFirst(prefix + ".MetricStat.Stat"), params.getFirst(prefix + ".MetricStat.Unit"));
+            }
+            String returnData = params.getFirst(prefix + ".ReturnData");
+            queries.add(new AlarmMetricDataQuery(id, params.getFirst(prefix + ".Expression"),
+                    params.getFirst(prefix + ".Label"), returnData == null ? null : Boolean.valueOf(returnData),
+                    optionalInteger(params.getFirst(prefix + ".Period")),
+                    params.getFirst(prefix + ".AccountId"), metricStat));
+        }
+        return queries;
+    }
+
+    private static Integer optionalInteger(String value) {
+        return value == null ? null : Integer.valueOf(value);
+    }
+
     private void toAlarmXml(XmlBuilder xml, MetricAlarm a) {
-        xml.start("member")
+        xml.start("member");
-                .elem("AlarmName", a.getAlarmName())
+        describeAlarmFields(xml, a);
+        if (!a.getMetrics().isEmpty()) {
+            toAlarmMetricsXml(xml, a.getMetrics());
+        } else {
+            xml.start("Dimensions");
+            for (Dimension d : a.getDimensions()) {
+                xml.start("member").elem("Name", d.name()).elem("Value", d.value()).end("member");
+            }
+            xml.end("Dimensions")
+                    .elem("MetricName", a.getMetricName())
+                    .elem("Namespace", a.getNamespace())
+                    .elem("Statistic", a.getStatistic())
+                    .elem("Period", a.getPeriod() == null ? null : a.getPeriod().toString())
+                    .elem("Unit", a.getUnit());
+        }
+
+        xml.elem("EvaluationPeriods", String.valueOf(a.getEvaluationPeriods()))
+                .elem("Threshold", String.valueOf(a.getThreshold()))
+                .elem("ComparisonOperator", a.getComparisonOperator())
+                .elem("TreatMissingData", a.getTreatMissingData());
+        // Only an alarm the caller gave an M for carries the member, matching what AWS returns.
+        if (a.getDatapointsToAlarm() != null) {
+            xml.elem("DatapointsToAlarm", String.valueOf(a.getDatapointsToAlarm()));
+        }
+
+        xml.end("member");
+    }
+
+    private void describeAlarmFields(XmlBuilder xml, Alarm a) {
+        xml.elem("AlarmName", a.getAlarmName())
                 .elem("AlarmArn", a.getAlarmArn())
                 .elem("AlarmDescription", a.getAlarmDescription());
         xml.start("AlarmActions");
@@ -726,32 +853,42 @@ public class CloudWatchMetricsQueryHandler {
         xml.start("InsufficientDataActions");
         a.getInsufficientDataActions().forEach(act -> xml.elem("member", act));
         xml.end("InsufficientDataActions");
-        xml.start("Dimensions");
-        for (Dimension d : a.getDimensions()) {
-            xml.start("member").elem("Name", d.name()).elem("Value", d.value()).end("member");
-        }
-        xml.end("Dimensions");
-
         xml.elem("StateValue", a.getStateValue())
                 .elem("StateReason", a.getStateReason())
                 .elem("StateReasonData", a.getStateReasonData())
-                .elem("StateUpdatedTimestamp", Instant.ofEpochSecond(a.getStateUpdatedTimestamp()).toString())
+                .elem("StateUpdatedTimestamp", Instant.ofEpochSecond(a.getStateUpdatedTimestamp()).toString());
-                .elem("MetricName", a.getMetricName())
-                .elem("Namespace", a.getNamespace())
-                .elem("Statistic", a.getStatistic())
-                .elem("Period", String.valueOf(a.getPeriod()))
-                .elem("Unit", a.getUnit())
-                .elem("EvaluationPeriods", String.valueOf(a.getEvaluationPeriods()))
-                .elem("Threshold", String.valueOf(a.getThreshold()))
-                .elem("ComparisonOperator", a.getComparisonOperator())
-                .elem("TreatMissingData", a.getTreatMissingData());
-        // Only an alarm the caller gave an M for carries the member, matching what AWS returns.
-        if (a.getDatapointsToAlarm() != null) {
-            xml.elem("DatapointsToAlarm", String.valueOf(a.getDatapointsToAlarm()));
     }
 
+    private void toAlarmMetricsXml(XmlBuilder xml, List<AlarmMetricDataQuery> metrics) {
+        xml.start("Metrics");
+        for (AlarmMetricDataQuery query : metrics) {
+            xml.start("member").elem("Id", query.id()).elem("Expression", query.expression())
+                    .elem("Label", query.label()).elem("AccountId", query.accountId())
+                    .elem("Period", query.period() == null ? null : query.period().toString())
+                    .elem("ReturnData", query.returnData() == null ? null : query.returnData().toString());
+            AlarmMetricDataQuery.MetricStat stat = query.metricStat();
+            if (stat != null) {
+                xml.start("MetricStat");
+                AlarmMetricDataQuery.Metric metric = stat.metric();
+                if (metric != null) {
+                    xml.start("Metric").elem("Namespace", metric.namespace()).elem("MetricName", metric.metricName());
+                    if (metric.dimensions() != null) {
+                        xml.start("Dimensions");
+                        for (AlarmMetricDataQuery.MetricDimension dimension : metric.dimensions()) {
+                            xml.start("member").elem("Name", dimension.name()).elem("Value", dimension.value())
+                                    .end("member");
+                        }
+                        xml.end("Dimensions");
+                    }
+                    xml.end("Metric");
+                }
+                xml.elem("Period", stat.period() == null ? null : stat.period().toString())
+                        .elem("Stat", stat.stat()).elem("Unit", stat.unit()).end("MetricStat");
+            }
             xml.end("member");
         }
+        xml.end("Metrics");
+    }
 
     private Instant parseInstant(String value) {
         if (value == null || value.isBlank()) {

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java

@@ -1,10 +1,15 @@
 package io.github.hectorvent.floci.services.cloudwatch.metrics;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import io.github.hectorvent.floci.core.common.AwsArnUtils;
 import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.common.SmithyValidationErrors;
 import io.github.hectorvent.floci.core.storage.AccountAwareStorageBackend;
 import io.github.hectorvent.floci.core.storage.StorageBackend;
 import io.github.hectorvent.floci.core.storage.StorageFactory;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Dimension;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
@@ -14,11 +19,16 @@ import jakarta.inject.Inject;
 import org.jboss.logging.Logger;
 
 import java.time.Instant;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
@@ -27,25 +37,41 @@ public class CloudWatchMetricsService {
 
     private static final Logger LOG = Logger.getLogger(CloudWatchMetricsService.class);
 
+    private static final List<String> ALARM_TYPES = List.of("CompositeAlarm", "MetricAlarm", "LogAlarm");
+
     private final StorageBackend<String, MetricDatum> metricStore;
-    private final StorageBackend<String, MetricAlarm> alarmStore;
+    private final StorageBackend<String, Alarm> alarmStore;
     private final RegionResolver regionResolver;
+    private final AlarmSnsNotifier alarmSnsNotifier;
+
+    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver) {
+        this(storageFactory, regionResolver, null);
+    }
 
     @Inject
-    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver) {
+    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver,
+                                    AlarmSnsNotifier alarmSnsNotifier) {
         this.metricStore = storageFactory.create("cloudwatchmetrics", "cwmetrics.json",
                 new TypeReference<Map<String, MetricDatum>>() {});
         this.alarmStore = storageFactory.create("cloudwatchmetrics", "cwalarms.json",
-                new TypeReference<Map<String, MetricAlarm>>() {});
+                new TypeReference<Map<String, Alarm>>() {});
         this.regionResolver = regionResolver;
+        this.alarmSnsNotifier = alarmSnsNotifier;
     }
 
     CloudWatchMetricsService(StorageBackend<String, MetricDatum> metricStore,
-                             StorageBackend<String, MetricAlarm> alarmStore,
+                             StorageBackend<String, Alarm> alarmStore,
                              RegionResolver regionResolver) {
+        this(metricStore, alarmStore, regionResolver, null);
+    }
+
+    CloudWatchMetricsService(StorageBackend<String, MetricDatum> metricStore,
+                             StorageBackend<String, Alarm> alarmStore,
+                             RegionResolver regionResolver, AlarmSnsNotifier alarmSnsNotifier) {
         this.metricStore = metricStore;
         this.alarmStore = alarmStore;
         this.regionResolver = regionResolver;
+        this.alarmSnsNotifier = alarmSnsNotifier;
     }
 
     public void putMetricData(String namespace, List<MetricDatum> datums, String region) {
@@ -282,85 +308,293 @@ public class CloudWatchMetricsService {
         };
     }
 
-    public void putMetricAlarm(MetricAlarm alarm, String region) {
+    public synchronized void putMetricAlarm(MetricAlarm alarm, String region) {
+        putAlarm(alarm, region);
+        evaluateComposites(region, Set.of(alarm.getAlarmName()), null);
+    }
+
+    public void putCompositeAlarm(CompositeAlarm alarm, String region) {
+        putCompositeAlarm(alarm, region, true);
+    }
+
+    public synchronized void putCompositeAlarm(CompositeAlarm alarm, String region, boolean replaceExisting) {
+        if (!replaceExisting && alarmStore.get(region + "::" + alarm.getAlarmName()).isPresent()) {
+            throw new AwsException("AlreadyExists", "Alarm already exists: " + alarm.getAlarmName(), 400);
+        }
+        Set<String> references = alarmReferences(alarm, region);
+        if (references.size() > 100) {
+            throw new AwsException("LimitExceeded", "AlarmRule references more than 100 child alarms", 400);
+        }
+        List<CompositeAlarm> existing = describeCompositeAlarms(null, null, region);
+        for (String reference : references) {
+            long parents = existing.stream().filter(parent -> !parent.getAlarmName().equals(alarm.getAlarmName()))
+                    .filter(parent -> alarmReferences(parent, region).contains(reference))
+                    .count();
+            if (parents >= 150) {
+                throw new AwsException("LimitExceeded", "An alarm can be referenced by at most 150 composite alarms", 400);
+            }
+        }
+        boolean created = alarmStore.get(region + "::" + alarm.getAlarmName()).isEmpty();
+        putAlarm(alarm, region);
+        if (created) {
+            evaluateComposites(region, Set.of(), alarm.getAlarmName());
+        }
+    }
+
+    private Set<String> alarmReferences(CompositeAlarm alarm, String region) {
+        return CompositeAlarmRule.parse(alarm.getAlarmRule()).references().stream()
+                .map(reference -> AwsArnUtils.isArn(reference) ? reference
+                        : regionResolver.buildArn("cloudwatch", region, "alarm:" + reference))
+                .collect(Collectors.toSet());
+    }
+
+    private void putAlarm(Alarm alarm, String region) {
+        String name = alarm.getAlarmName();
+        if (name == null || name.isBlank() || name.length() > 255) {
+            throw SmithyValidationErrors.validationError(name == null ? SmithyValidationErrors.notNull("alarmName")
+                    : "Value '" + name + "' at 'alarmName' failed to satisfy constraint: Member must have length "
+                            + (name.length() > 255 ? "less than or equal to 255" : "greater than or equal to 1"));
+        }
+        Alarm previous = alarmStore.get(region + "::" + name).orElse(null);
+        if (previous != null) {
+            if (!previous.getClass().equals(alarm.getClass())) {
+                throw new AwsException("ValidationError", "An alarm of another type already has this name", 400);
+            }
+            if (alarm instanceof CompositeAlarm) {
+                alarm.setStateValue(previous.getStateValue());
+                alarm.setStateReason(previous.getStateReason());
+                alarm.setStateReasonData(previous.getStateReasonData());
+                alarm.setStateUpdatedTimestamp(previous.getStateUpdatedTimestamp());
+                alarm.setTags(new HashMap<>(previous.getTags()));
+            }
+        }
         if (alarm.getAlarmArn() == null) {
-            alarm.setAlarmArn(regionResolver.buildArn("cloudwatch", region, "alarm:" + alarm.getAlarmName()));
+            alarm.setAlarmArn(regionResolver.buildArn("cloudwatch", region, "alarm:" + name));
         }
         alarm.setRegion(region);
         alarm.setAlarmConfigurationUpdatedTimestamp(Instant.now().getEpochSecond());
-        alarmStore.put(region + "::" + alarm.getAlarmName(), alarm);
+        alarmStore.put(region + "::" + name, alarm);
-        LOG.infov("PutMetricAlarm: {0} in {1}", alarm.getAlarmName(), region);
+        LOG.infov("Put {0}: {1} in {2}", alarm.getClass().getSimpleName(), name, region);
     }
 
-    /** Every stored alarm, across all regions. Used by the background {@link AlarmEvaluator}
+    /** Metric alarms evaluated by the existing metric evaluator. */
-     * tick, which has no per-request region to scope a lookup to. */
     public List<MetricAlarm> allAlarms() {
-        return alarmStore.scan(k -> true);
+        return alarmStore.scan(k -> true).stream().filter(MetricAlarm.class::isInstance)
+                .map(MetricAlarm.class::cast).toList();
     }
 
-    public List<MetricAlarm> describeAlarms(List<String> alarmNames, String alarmNamePrefix, String region) {
+    public List<MetricAlarm> describeAlarms(List<String> names, String prefix, String region) {
-        String prefix = region + "::";
+        return matchingAlarms(names, prefix, region).stream().filter(MetricAlarm.class::isInstance)
-        List<MetricAlarm> all = alarmStore.scan(k -> k.startsWith(prefix));
+                .map(MetricAlarm.class::cast).toList();
-
-        if (alarmNames != null && !alarmNames.isEmpty()) {
-            return all.stream().filter(a -> alarmNames.contains(a.getAlarmName())).toList();
-        }
-        if (alarmNamePrefix != null && !alarmNamePrefix.isBlank()) {
-            return all.stream().filter(a -> a.getAlarmName().startsWith(alarmNamePrefix)).toList();
-        }
-        return all;
     }
 
-    public void deleteAlarms(List<String> alarmNames, String region) {
+    public void validateAlarmTypes(List<String> types) {
+        if (!ALARM_TYPES.containsAll(types)) {
+            throw SmithyValidationErrors.validationError(
+                    SmithyValidationErrors.notAllInEnum(types, "alarmTypes", ALARM_TYPES));
+        }
+    }
+
+    public List<CompositeAlarm> describeCompositeAlarms(List<String> names, String prefix, String region) {
+        return matchingAlarms(names, prefix, region).stream().filter(CompositeAlarm.class::isInstance)
+                .map(CompositeAlarm.class::cast).toList();
+    }
+
+    private List<Alarm> matchingAlarms(List<String> names, String prefix, String region) {
+        return alarmStore.scan(k -> k.startsWith(region + "::")).stream()
+                .filter(a -> names == null || names.isEmpty() || names.contains(a.getAlarmName()))
+                .filter(a -> prefix == null || a.getAlarmName().startsWith(prefix)).toList();
+    }
+
+    private Map<String, Alarm> alarmIndex(String region) {
+        Map<String, Alarm> alarms = new HashMap<>();
+        for (Alarm alarm : alarmStore.scan(k -> k.startsWith(region + "::"))) {
+            alarms.put(alarm.getAlarmName(), alarm);
+            alarms.put(alarm.getAlarmArn(), alarm);
+        }
+        return alarms;
+    }
+
+    private void evaluateComposites(String region, Set<String> changedAlarmNames, String newAlarm) {
+        if (changedAlarmNames.isEmpty() && newAlarm == null) {
+            return;
+        }
+        Map<String, Alarm> alarms = alarmIndex(region);
+        Set<String> changed = new HashSet<>(changedAlarmNames);
+        for (String name : changedAlarmNames) {
+            changed.add(regionResolver.buildArn("cloudwatch", region, "alarm:" + name));
+        }
+        Map<String, CompositeAlarmRule.Rule> candidates = new HashMap<>();
+        for (Alarm alarm : new HashSet<>(alarms.values())) {
+            if (alarm instanceof CompositeAlarm composite && !changedAlarmNames.contains(alarm.getAlarmName())) {
+                candidates.put(alarm.getAlarmName(), CompositeAlarmRule.parse(composite.getAlarmRule()));
+            }
+        }
+        Set<String> affected = new HashSet<>(changed);
+        Map<String, CompositeAlarmRule.Rule> pending = new HashMap<>();
+        boolean discovered;
+        do {
+            discovered = false;
+            Iterator<Map.Entry<String, CompositeAlarmRule.Rule>> iterator = candidates.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Map.Entry<String, CompositeAlarmRule.Rule> next = iterator.next();
+                if (next.getKey().equals(newAlarm) || next.getValue().references().stream().anyMatch(affected::contains)) {
+                    pending.put(next.getKey(), next.getValue());
+                    affected.add(next.getKey());
+                    affected.add(alarms.get(next.getKey()).getAlarmArn());
+                    iterator.remove();
+                    discovered = true;
+                }
+            }
+        } while (discovered);
+        Set<String> cycles = pending.keySet().stream()
+                .filter(name -> isCyclic(name, alarms)).collect(Collectors.toSet());
+        boolean progress;
+        do {
+            progress = false;
+            Iterator<Map.Entry<String, CompositeAlarmRule.Rule>> iterator = pending.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Map.Entry<String, CompositeAlarmRule.Rule> next = iterator.next();
+                boolean blocked = next.getValue().references().stream().map(alarms::get)
+                        .anyMatch(child -> child != null && pending.containsKey(child.getAlarmName()));
+                if (blocked || cycles.contains(next.getKey())) {
+                    continue;
+                }
+                if (!next.getKey().equals(newAlarm)
+                        && next.getValue().references().stream().noneMatch(changed::contains)) {
+                    iterator.remove();
+                    progress = true;
+                    continue;
+                }
+                Alarm alarm = alarms.get(next.getKey());
+                boolean triggered = next.getValue().evaluate(reference -> {
+                    Alarm child = alarms.get(reference);
+                    return child == null ? null : child.getStateValue();
+                });
+                String state = triggered ? "ALARM" : "OK";
+                if (!state.equals(alarm.getStateValue())) {
+                    alarm.setStateValue(state);
+                    alarm.setStateReason("AlarmRule evaluated to " + state);
+                    alarm.setStateReasonData(null);
+                    alarm.setStateUpdatedTimestamp(Instant.now().getEpochSecond());
+                    alarmStore.put(region + "::" + alarm.getAlarmName(), alarm);
+                    changed.add(alarm.getAlarmName());
+                    changed.add(alarm.getAlarmArn());
+                }
+                iterator.remove();
+                progress = true;
+            }
+        } while (progress && !pending.isEmpty());
+        if (!pending.isEmpty()) {
+            LOG.debugv("Composite alarm evaluation stopped at cyclic dependencies: {0}", pending.keySet());
+        }
+    }
+
+    public synchronized void deleteAlarms(List<String> alarmNames, String region) {
+        Map<String, Alarm> alarms = alarmIndex(region);
+        long composites = alarmNames.stream().map(alarms::get).filter(CompositeAlarm.class::isInstance).count();
+        if (composites > 1) {
+            throw new AwsException("ValidationError", "DeleteAlarms accepts at most one composite alarm", 400);
+        }
         for (String name : alarmNames) {
+            if (isCyclic(name, alarms)) {
+                throw new AwsException("ValidationError", "Break the composite alarm cycle before deleting " + name, 400);
+            }
+        }
+        Set<String> deletedNames = new HashSet<>();
+        for (String name : alarmNames) {
+            if (alarmStore.get(region + "::" + name).isPresent()) {
+                deletedNames.add(name);
+            }
             alarmStore.delete(region + "::" + name);
         }
+        evaluateComposites(region, deletedNames, null);
         LOG.infov("Deleted alarms: {0} in {1}", alarmNames, region);
     }
 
+    private boolean isCyclic(String name, Map<String, Alarm> alarms) {
+        Set<String> visited = new HashSet<>();
+        ArrayDeque<String> pending = new ArrayDeque<>();
+        pending.add(name);
+        while (!pending.isEmpty()) {
+            Alarm next = alarms.get(pending.remove());
+            if (next instanceof CompositeAlarm composite && visited.add(next.getAlarmName())) {
+                for (String reference : CompositeAlarmRule.parse(composite.getAlarmRule()).references()) {
+                    Alarm child = alarms.get(reference);
+                    if (child != null && child.getAlarmName().equals(name)) {
+                        return true;
+                    }
+                    if (child != null) {
+                        pending.add(child.getAlarmName());
+                    }
+                }
+            }
+        }
+        return false;
+    }
+
     public void setAlarmState(String alarmName, String stateValue, String stateReason, String stateReasonData, String region) {
+        List<AlarmSnsNotifier.Notification> notifications = List.of();
+        synchronized (this) {
             String key = region + "::" + alarmName;
-        MetricAlarm alarm = alarmStore.get(key)
+            Alarm alarm = alarmStore.get(key)
                     .orElseThrow(() -> new AwsException("ResourceNotFound", "Alarm not found: " + alarmName, 404));
 
+            String oldState = alarm.getStateValue();
             alarm.setStateValue(stateValue);
             alarm.setStateReason(stateReason);
             alarm.setStateReasonData(stateReasonData);
             alarm.setStateUpdatedTimestamp(Instant.now().getEpochSecond());
 
             alarmStore.put(key, alarm);
+            evaluateComposites(region, Set.of(alarmName), null);
             LOG.infov("SetAlarmState: {0} -> {1}", alarmName, stateValue);
+            if (alarmSnsNotifier != null && alarm instanceof MetricAlarm metricAlarm) {
+                notifications = alarmSnsNotifier.prepareTransition(metricAlarm, oldState);
+            }
+        }
+        if (alarmSnsNotifier != null) {
+            alarmSnsNotifier.publish(notifications);
+        }
     }
 
-    public Map<String, String> listTagsForResource(String resourceArn, String region) {
+    /**
+     * Resolves the alarm an ARN names, or reports that nothing does.
+     *
+     * <p>CloudWatch's three tag operations each declare {@code ResourceNotFoundException},
+     * which is a different shape from the {@code ResourceNotFound} that {@code SetAlarmState}
+     * and {@code GetDashboard} declare; the SDK maps the two codes to two exception classes,
+     * so the tag path uses the longer one rather than the code used elsewhere in this service.
+     *
+     * <p>The message names the ARN rather than asserting it was an alarm, because both tag
+     * handlers route every ARN that is not a dashboard and not a metric stream here. That
+     * includes kinds AWS considers taggable and Floci does not serve, a Contributor Insights
+     * {@code insight-rule/} ARN being the one AWS documents; reporting that no resource
+     * matches the ARN is true of those, where "alarm not found" would not be.
+     */
+    private Alarm requireAlarm(String resourceArn, String region) {
         return alarmStore.scan(k -> k.startsWith(region + "::"))
                 .stream()
-                .filter(a -> resourceArn.equals(a.getAlarmArn()))
+                .filter(a -> a.getAlarmArn() != null && a.getAlarmArn().equals(resourceArn))
                 .findFirst()
-                .map(MetricAlarm::getTags)
+                .orElseThrow(() -> new AwsException("ResourceNotFoundException",
-                .orElse(Map.of());
+                        "No CloudWatch resource matches the ARN " + resourceArn + ".", 404));
     }
 
-    public void tagResource(String resourceArn, Map<String, String> tags, String region) {
+    public synchronized Map<String, String> listTagsForResource(String resourceArn, String region) {
-        alarmStore.scan(k -> k.startsWith(region + "::"))
+        return new HashMap<>(requireAlarm(resourceArn, region).getTags());
-                .stream()
+    }
-                .filter(a -> resourceArn.equals(a.getAlarmArn()))
+
-                .findFirst()
+    public synchronized void tagResource(String resourceArn, Map<String, String> tags, String region) {
-                .ifPresent(alarm -> {
+        Alarm alarm = requireAlarm(resourceArn, region);
         alarm.getTags().putAll(tags);
         alarmStore.put(region + "::" + alarm.getAlarmName(), alarm);
-                });
     }
 
-    public void untagResource(String resourceArn, List<String> tagKeys, String region) {
+    public synchronized void untagResource(String resourceArn, List<String> tagKeys, String region) {
-        alarmStore.scan(k -> k.startsWith(region + "::"))
+        Alarm alarm = requireAlarm(resourceArn, region);
-                .stream()
-                .filter(a -> resourceArn.equals(a.getAlarmArn()))
-                .findFirst()
-                .ifPresent(alarm -> {
         tagKeys.forEach(alarm.getTags()::remove);
         alarmStore.put(region + "::" + alarm.getAlarmName(), alarm);
-                });
     }
 
     // ──────────────────────────── Helpers ────────────────────────────

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CompositeAlarmRule.java

@@ -0,0 +1,184 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import io.github.hectorvent.floci.core.common.AwsException;
+import io.github.hectorvent.floci.core.common.SmithyValidationErrors;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Function;
+
+/** CloudWatch's boolean alarm language, parsed completely before any alarm is stored. */
+final class CompositeAlarmRule {
+    private final String input;
+    private final Set<String> references = new HashSet<>();
+    private int offset;
+    private int elements;
+
+    private CompositeAlarmRule(String input) {
+        this.input = input;
+    }
+
+    static Rule parse(String input) {
+        if (input == null || input.isBlank() || input.length() > 10240) {
+            throw SmithyValidationErrors.validationError(input == null ? SmithyValidationErrors.notNull("alarmRule")
+                    : "Value '" + input + "' at 'alarmRule' failed to satisfy constraint: Member must have length "
+                            + (input.length() > 10240 ? "less than or equal to 10240" : "greater than or equal to 1"));
+        }
+        CompositeAlarmRule parser = new CompositeAlarmRule(input);
+        Expression expression = parser.or();
+        parser.whitespace();
+        if (parser.offset != input.length()) {
+            throw invalid("Invalid AlarmRule");
+        }
+        return new Rule(expression, Set.copyOf(parser.references));
+    }
+
+    private Expression or() {
+        Expression left = and();
+        while (word("OR")) {
+            Expression prior = left;
+            Expression right = and();
+            left = states -> prior.evaluate(states) || right.evaluate(states);
+        }
+        return left;
+    }
+
+    private Expression and() {
+        Expression left = atom();
+        while (word("AND")) {
+            Expression prior = left;
+            Expression right = atom();
+            left = states -> prior.evaluate(states) && right.evaluate(states);
+        }
+        return left;
+    }
+
+    private Expression atom() {
+        boolean negate = false;
+        while (word("NOT")) {
+            negate = !negate;
+        }
+        Expression operand = primary();
+        return negate ? states -> !operand.evaluate(states) : operand;
+    }
+
+    private Expression primary() {
+        element();
+        if (character('(')) {
+            Expression nested = or();
+            require(')');
+            element();
+            return nested;
+        }
+        if (word("TRUE")) {
+            return states -> true;
+        }
+        if (word("FALSE")) {
+            return states -> false;
+        }
+        String state;
+        if (word("ALARM")) {
+            state = "ALARM";
+        } else if (word("OK")) {
+            state = "OK";
+        } else if (word("INSUFFICIENT_DATA")) {
+            state = "INSUFFICIENT_DATA";
+        } else {
+            throw invalid("Expected an alarm state, boolean, NOT, or parenthesis");
+        }
+        require('(');
+        whitespace();
+        String name;
+        if (character('"')) {
+            StringBuilder quoted = new StringBuilder();
+            boolean closed = false;
+            while (offset < input.length()) {
+                char next = input.charAt(offset++);
+                if (next == '"') {
+                    closed = true;
+                    break;
+                }
+                if (next == '\\') {
+                    if (offset == input.length()) {
+                        throw invalid("Unterminated quoted alarm name");
+                    }
+                    next = input.charAt(offset++);
+                    if (next != '\\' && next != '"') {
+                        throw invalid("Invalid quoted alarm name escape");
+                    }
+                }
+                quoted.append(next);
+            }
+            if (!closed) {
+                throw invalid("Unterminated quoted alarm name");
+            }
+            name = quoted.toString();
+        } else {
+            int start = offset;
+            while (offset < input.length() && input.charAt(offset) != ')') {
+                offset++;
+            }
+            name = input.substring(start, offset).trim();
+        }
+        if (name.isEmpty()) {
+            throw invalid("An alarm name or ARN is required");
+        }
+        require(')');
+        references.add(name);
+        return states -> state.equals(states.apply(name));
+    }
+
+    private void element() {
+        if (++elements > 500) {
+            throw invalid("AlarmRule exceeds 500 child alarms, booleans, and parentheses");
+        }
+    }
+
+    private boolean word(String word) {
+        whitespace();
+        if (!input.startsWith(word, offset)) {
+            return false;
+        }
+        int end = offset + word.length();
+        if (end < input.length() && (Character.isLetterOrDigit(input.charAt(end)) || input.charAt(end) == '_')) {
+            return false;
+        }
+        offset = end;
+        return true;
+    }
+
+    private boolean character(char expected) {
+        whitespace();
+        if (offset < input.length() && input.charAt(offset) == expected) {
+            offset++;
+            return true;
+        }
+        return false;
+    }
+
+    private void require(char expected) {
+        if (!character(expected)) {
+            throw invalid("Expected '" + expected + "' in AlarmRule");
+        }
+    }
+
+    private void whitespace() {
+        while (offset < input.length() && Character.isWhitespace(input.charAt(offset))) {
+            offset++;
+        }
+    }
+
+    private static AwsException invalid(String message) {
+        return new AwsException("ValidationError", message, 400);
+    }
+
+    record Rule(Expression expression, Set<String> references) {
+        boolean evaluate(Function<String, String> states) {
+            return expression.evaluate(states);
+        }
+    }
+
+    interface Expression {
+        boolean evaluate(Function<String, String> states);
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/model/Alarm.java

@@ -0,0 +1,87 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@RegisterForReflection
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXISTING_PROPERTY,
+        property = "alarmType", defaultImpl = MetricAlarm.class)
+@JsonSubTypes({@JsonSubTypes.Type(value = MetricAlarm.class, name = "MetricAlarm"),
+        @JsonSubTypes.Type(value = CompositeAlarm.class, name = "CompositeAlarm")})
+public abstract class Alarm {
+    private String alarmName;
+    private String alarmArn;
+    private String alarmDescription;
+    private long alarmConfigurationUpdatedTimestamp;
+    private boolean actionsEnabled;
+    private List<String> okActions = new ArrayList<>();
+    private List<String> alarmActions = new ArrayList<>();
+    private List<String> insufficientDataActions = new ArrayList<>();
+    private String stateValue = "INSUFFICIENT_DATA";
+    private String stateReason = "Unchecked";
+    private String stateReasonData;
+    private long stateUpdatedTimestamp;
+    private Map<String, String> tags = new HashMap<>();
+    private String region;
+
+
+    protected Alarm() {
+        long now = Instant.now().getEpochSecond();
+        this.alarmConfigurationUpdatedTimestamp = now;
+        this.stateUpdatedTimestamp = now;
+    }
+
+    public abstract String getAlarmType();
+
+    public String getAlarmName() { return alarmName; }
+    public void setAlarmName(String alarmName) { this.alarmName = alarmName; }
+
+    public String getAlarmArn() { return alarmArn; }
+    public void setAlarmArn(String alarmArn) { this.alarmArn = alarmArn; }
+
+    public String getAlarmDescription() { return alarmDescription; }
+    public void setAlarmDescription(String alarmDescription) { this.alarmDescription = alarmDescription; }
+
+    public long getAlarmConfigurationUpdatedTimestamp() { return alarmConfigurationUpdatedTimestamp; }
+    public void setAlarmConfigurationUpdatedTimestamp(long timestamp) { this.alarmConfigurationUpdatedTimestamp = timestamp; }
+
+    public boolean isActionsEnabled() { return actionsEnabled; }
+    public void setActionsEnabled(boolean actionsEnabled) { this.actionsEnabled = actionsEnabled; }
+
+    public List<String> getOkActions() { return okActions; }
+    public void setOkActions(List<String> okActions) { this.okActions = okActions; }
+
+    public List<String> getAlarmActions() { return alarmActions; }
+    public void setAlarmActions(List<String> alarmActions) { this.alarmActions = alarmActions; }
+
+    public List<String> getInsufficientDataActions() { return insufficientDataActions; }
+    public void setInsufficientDataActions(List<String> insufficientDataActions) { this.insufficientDataActions = insufficientDataActions; }
+
+    public String getStateValue() { return stateValue; }
+    public void setStateValue(String stateValue) { this.stateValue = stateValue; }
+
+    public String getStateReason() { return stateReason; }
+    public void setStateReason(String stateReason) { this.stateReason = stateReason; }
+
+    public String getStateReasonData() { return stateReasonData; }
+    public void setStateReasonData(String stateReasonData) { this.stateReasonData = stateReasonData; }
+
+    public long getStateUpdatedTimestamp() { return stateUpdatedTimestamp; }
+    public void setStateUpdatedTimestamp(long stateUpdatedTimestamp) { this.stateUpdatedTimestamp = stateUpdatedTimestamp; }
+
+    public Map<String, String> getTags() { return tags; }
+    public void setTags(Map<String, String> tags) { this.tags = tags; }
+
+
+    public String getRegion() { return region; }
+    public void setRegion(String region) { this.region = region; }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/model/AlarmMetricDataQuery.java

@@ -0,0 +1,46 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+import java.util.List;
+
+/** The persisted AWS wire shape of a metric-math alarm query, including optional members. */
+@RegisterForReflection
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record AlarmMetricDataQuery(
+        @JsonProperty("Id") String id,
+        @JsonProperty("Expression") String expression,
+        @JsonProperty("Label") String label,
+        @JsonProperty("ReturnData") Boolean returnData,
+        @JsonProperty("Period") Integer period,
+        @JsonProperty("AccountId") String accountId,
+        @JsonProperty("MetricStat") MetricStat metricStat) {
+
+    @RegisterForReflection
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public record MetricStat(
+            @JsonProperty("Metric") Metric metric,
+            @JsonProperty("Period") Integer period,
+            @JsonProperty("Stat") String stat,
+            @JsonProperty("Unit") String unit) { }
+
+    @RegisterForReflection
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public record Metric(
+            @JsonProperty("Namespace") String namespace,
+            @JsonProperty("MetricName") String metricName,
+            @JsonProperty("Dimensions") List<MetricDimension> dimensions) { }
+
+    @RegisterForReflection
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public record MetricDimension(
+            @JsonProperty("Name") String name,
+            @JsonProperty("Value") String value) { }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/model/CompositeAlarm.java

@@ -0,0 +1,14 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics.model;
+
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+@RegisterForReflection
+public class CompositeAlarm extends Alarm {
+    private String alarmRule;
+
+    @Override
+    public String getAlarmType() { return "CompositeAlarm"; }
+
+    public String getAlarmRule() { return alarmRule; }
+    public void setAlarmRule(String alarmRule) { this.alarmRule = alarmRule; }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/model/MetricAlarm.java

@@ -3,32 +3,22 @@ package io.github.hectorvent.floci.services.cloudwatch.metrics.model;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import io.quarkus.runtime.annotations.RegisterForReflection;
 
-import java.time.Instant;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 @RegisterForReflection
 @JsonIgnoreProperties(ignoreUnknown = true)
-public class MetricAlarm {
+public class MetricAlarm extends Alarm {
-    private String alarmName;
+    @Override
-    private String alarmArn;
+    public String getAlarmType() { return "MetricAlarm"; }
-    private String alarmDescription;
+
-    private long alarmConfigurationUpdatedTimestamp;
-    private boolean actionsEnabled;
-    private List<String> okActions = new ArrayList<>();
-    private List<String> alarmActions = new ArrayList<>();
-    private List<String> insufficientDataActions = new ArrayList<>();
-    private String stateValue = "INSUFFICIENT_DATA";
-    private String stateReason = "Unchecked";
-    private String stateReasonData;
-    private long stateUpdatedTimestamp;
     private String metricName;
     private String namespace;
     private String statistic;
     private List<Dimension> dimensions = new ArrayList<>();
-    private int period;
+    /** Single-metric period; absent for alarms whose periods are defined in Metrics. */
+    private Integer period;
+    private List<AlarmMetricDataQuery> metrics = new ArrayList<>();
     private String unit;
     private int evaluationPeriods;
     /**
@@ -43,50 +33,6 @@ public class MetricAlarm {
     private String comparisonOperator;
     private String treatMissingData = "missing";
     private String evaluateLowSampleCountPercentile;
-    private Map<String, String> tags = new HashMap<>();
-    private String region;
-
-    public MetricAlarm() {
-        long now = Instant.now().getEpochSecond();
-        this.alarmConfigurationUpdatedTimestamp = now;
-        this.stateUpdatedTimestamp = now;
-    }
-
-    public String getAlarmName() { return alarmName; }
-    public void setAlarmName(String alarmName) { this.alarmName = alarmName; }
-
-    public String getAlarmArn() { return alarmArn; }
-    public void setAlarmArn(String alarmArn) { this.alarmArn = alarmArn; }
-
-    public String getAlarmDescription() { return alarmDescription; }
-    public void setAlarmDescription(String alarmDescription) { this.alarmDescription = alarmDescription; }
-
-    public long getAlarmConfigurationUpdatedTimestamp() { return alarmConfigurationUpdatedTimestamp; }
-    public void setAlarmConfigurationUpdatedTimestamp(long timestamp) { this.alarmConfigurationUpdatedTimestamp = timestamp; }
-
-    public boolean isActionsEnabled() { return actionsEnabled; }
-    public void setActionsEnabled(boolean actionsEnabled) { this.actionsEnabled = actionsEnabled; }
-
-    public List<String> getOkActions() { return okActions; }
-    public void setOkActions(List<String> okActions) { this.okActions = okActions; }
-
-    public List<String> getAlarmActions() { return alarmActions; }
-    public void setAlarmActions(List<String> alarmActions) { this.alarmActions = alarmActions; }
-
-    public List<String> getInsufficientDataActions() { return insufficientDataActions; }
-    public void setInsufficientDataActions(List<String> insufficientDataActions) { this.insufficientDataActions = insufficientDataActions; }
-
-    public String getStateValue() { return stateValue; }
-    public void setStateValue(String stateValue) { this.stateValue = stateValue; }
-
-    public String getStateReason() { return stateReason; }
-    public void setStateReason(String stateReason) { this.stateReason = stateReason; }
-
-    public String getStateReasonData() { return stateReasonData; }
-    public void setStateReasonData(String stateReasonData) { this.stateReasonData = stateReasonData; }
-
-    public long getStateUpdatedTimestamp() { return stateUpdatedTimestamp; }
-    public void setStateUpdatedTimestamp(long stateUpdatedTimestamp) { this.stateUpdatedTimestamp = stateUpdatedTimestamp; }
 
     public String getMetricName() { return metricName; }
     public void setMetricName(String metricName) { this.metricName = metricName; }
@@ -100,8 +46,13 @@ public class MetricAlarm {
     public List<Dimension> getDimensions() { return dimensions; }
     public void setDimensions(List<Dimension> dimensions) { this.dimensions = dimensions; }
 
-    public int getPeriod() { return period; }
+    public Integer getPeriod() { return period; }
-    public void setPeriod(int period) { this.period = period; }
+    public void setPeriod(Integer period) { this.period = period; }
+
+    public List<AlarmMetricDataQuery> getMetrics() { return metrics; }
+    public void setMetrics(List<AlarmMetricDataQuery> metrics) {
+        this.metrics = metrics == null ? new ArrayList<>() : new ArrayList<>(metrics);
+    }
 
     public String getUnit() { return unit; }
     public void setUnit(String unit) { this.unit = unit; }
@@ -124,11 +75,4 @@ public class MetricAlarm {
     public String getEvaluateLowSampleCountPercentile() { return evaluateLowSampleCountPercentile; }
     public void setEvaluateLowSampleCountPercentile(String percentile) { this.evaluateLowSampleCountPercentile = percentile; }
 
-    public Map<String, String> getTags() { return tags; }
-    public void setTags(Map<String, String> tags) { this.tags = tags; }
-
-    /** Not part of the AWS shape; set by {@code putMetricAlarm} so a background evaluator
-     * that only holds the alarm object still knows which region's metric data to query. */
-    public String getRegion() { return region; }
-    public void setRegion(String region) { this.region = region; }
 }

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metricstreams/CloudWatchMetricStreamsService.java

@@ -196,21 +196,30 @@ public class CloudWatchMetricStreamsService {
     }
 
     public Map<String, String> listTagsForResource(String resourceArn, String region) {
-        return findByArn(resourceArn, region).map(MetricStream::getTags).orElse(Map.of());
+        return requireStream(resourceArn, region).getTags();
     }
 
     public void tagResource(String resourceArn, Map<String, String> tags, String region) {
-        findByArn(resourceArn, region).ifPresent(stream -> {
+        MetricStream stream = requireStream(resourceArn, region);
         stream.getTags().putAll(tags);
         streamStore.put(key(region, stream.getName()), stream);
-        });
     }
 
     public void untagResource(String resourceArn, List<String> tagKeys, String region) {
-        findByArn(resourceArn, region).ifPresent(stream -> {
+        MetricStream stream = requireStream(resourceArn, region);
         tagKeys.forEach(stream.getTags()::remove);
         streamStore.put(key(region, stream.getName()), stream);
-        });
+    }
+
+    /**
+     * Resolves the stream an ARN names, or reports that nothing does. The tag operations declare
+     * {@code ResourceNotFoundException}, the same code {@code GetMetricStream} already answers
+     * with, so an unknown ARN here reads the same way an unknown name does.
+     */
+    private MetricStream requireStream(String resourceArn, String region) {
+        return findByArn(resourceArn, region).orElseThrow(() ->
+                new AwsException("ResourceNotFoundException",
+                        "Metric stream " + resourceArn + " does not exist.", 404));
     }
 
     private Optional<MetricStream> findByArn(String resourceArn, String region) {

src/test/java/io/github/hectorvent/floci/core/common/AwsJsonCborCloudWatchTimestampIntegrationTest.java

@@ -6,11 +6,13 @@ import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
 import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.response.Response;
 import org.junit.jupiter.api.Test;
 
 import java.time.Instant;
 
 import static io.restassured.RestAssured.given;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -82,6 +84,29 @@ class AwsJsonCborCloudWatchTimestampIntegrationTest {
                         + driftSeconds + "s): " + datapoint);
     }
 
+    // CloudWatch is awsQueryCompatible: the header keeps the Query code while the body names
+    // the model shape, which is what the AWS SDK v2 uses to pick the typed exception.
+    @Test
+    void putMetricDataWithoutMetricNameReturnsTheModelledExceptionShape() throws Exception {
+        ObjectNode putRequest = JSON.createObjectNode();
+        putRequest.put("Namespace", "CborRpcV2MissingName-" + System.nanoTime());
+        putRequest.putArray("MetricData").addObject().put("Value", 1.0);
+
+        Response response = given()
+                .contentType("application/cbor")
+                .accept("application/cbor")
+                .header("smithy-protocol", "rpc-v2-cbor")
+                .body(AwsJsonCborController.nodeToSmithyCbor(putRequest))
+                .when().post("/service/GraniteServiceVersion20100801/operation/PutMetricData");
+
+        assertEquals(400, response.statusCode());
+        assertEquals("MissingParameter;Sender", response.header("x-amzn-query-error"));
+        JsonNode body = CBOR.readTree(response.asByteArray());
+        assertEquals("MissingRequiredParameterException", body.path("__type").asText(), body.toString());
+        assertEquals("The parameter MetricData.member.1.MetricName is required.",
+                body.path("message").asText(), body.toString());
+    }
+
     private static JsonNode rpcV2Call(String action, ObjectNode body) throws Exception {
         byte[] responseCbor = given()
                 .contentType("application/cbor")

src/test/java/io/github/hectorvent/floci/services/cloudformation/CfnProvisionerFixture.java

@@ -45,6 +45,7 @@ import io.github.hectorvent.floci.services.cloudformation.provisioners.AppSyncCf
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CdkMetadataCfnProvisioner;
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CloudTrailCfnProvisioner;
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CloudWatchCfnProvisioner;
+import io.github.hectorvent.floci.services.cloudformation.provisioners.CloudWatchCompositeAlarmCfnProvisioner;
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CloudWatchDashboardCfnProvisioner;
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CognitoCfnProvisioner;
 import io.github.hectorvent.floci.services.cloudformation.provisioners.CognitoUserPoolGroupCfnProvisioner;
@@ -300,6 +301,7 @@ final class CfnProvisionerFixture {
             }
             if (cloudWatchMetricsService != null) {
                 discovered.add(new CloudWatchCfnProvisioner(cloudWatchMetricsService));
+                discovered.add(new CloudWatchCompositeAlarmCfnProvisioner(cloudWatchMetricsService));
             }
             if (cloudWatchDashboardsService != null) {
                 discovered.add(new CloudWatchDashboardCfnProvisioner(cloudWatchDashboardsService));

src/test/java/io/github/hectorvent/floci/services/cloudformation/CloudFormationCompositeAlarmIntegrationTest.java

@@ -0,0 +1,190 @@
+package io.github.hectorvent.floci.services.cloudformation;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.testing.CfnStackWaits;
+import io.github.hectorvent.floci.testing.PartitionCleanup;
+import io.github.hectorvent.floci.testing.RestAssuredJsonUtils;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.response.Response;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.net.URI;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static io.restassured.RestAssured.given;
+import static org.junit.jupiter.api.Assertions.*;
+
+@QuarkusTest
+class CloudFormationCompositeAlarmIntegrationTest {
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    @TestHTTPResource
+    URI baseUri;
+
+    @RegisterExtension
+    final PartitionCleanup cleanup = new PartitionCleanup();
+
+    @BeforeAll
+    static void configureRestAssured() {
+        RestAssuredJsonUtils.configureAwsContentTypes();
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void failedUpdateRestoresRuleTagsAndIdentity(boolean rename) throws Exception {
+        String stack = unique("composite-rollback");
+        String name = stack + "-alarm";
+        String next = rename ? name + "-new" : name;
+        deploy("CreateStack", stack, template(name, "TRUE", Map.of("team", "old", "drop", "restore"), false, false));
+        assertEquals("CREATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        String arn = alarm(name).get("AlarmArn").toString();
+        Response described = given().contentType("application/x-www-form-urlencoded")
+                .formParam("Action", "DescribeStacks").formParam("StackName", stack).post("/");
+        assertEquals(name, described.xmlPath().getString(
+                "DescribeStacksResponse.DescribeStacksResult.Stacks.member.Outputs.member.find { it.OutputKey == 'Name' }.OutputValue"));
+        assertEquals(arn, described.xmlPath().getString(
+                "DescribeStacksResponse.DescribeStacksResult.Stacks.member.Outputs.member.find { it.OutputKey == 'Arn' }.OutputValue"));
+        cw("TagResource", Map.of("ResourceARN", arn, "Tags", List.of(Map.of("Key", "external", "Value", "keep"))))
+                .then().statusCode(200);
+        deploy("UpdateStack", stack, template(next, "FALSE", Map.of("team", "new", "added", "remove"), true, false));
+        assertEquals("UPDATE_ROLLBACK_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        assertEquals("TRUE", alarm(name).get("AlarmRule"));
+        List<Map<String, String>> tags = cw("ListTagsForResource", Map.of("ResourceARN", arn))
+                .then().statusCode(200).extract().jsonPath().getList("Tags");
+        assertEquals(Map.of("team", "old", "drop", "restore", "external", "keep"),
+                tags.stream().collect(Collectors.toMap(t -> t.get("Key"), t -> t.get("Value"))));
+        if (rename) {
+            assertMissing(next);
+        }
+        deleteStack(stack);
+        assertMissing(name);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void replacementHonorsRetentionAndDeletesItsCurrentBackingAlarm(boolean retain) throws Exception {
+        String stack = unique("composite-replace");
+        String original = stack + "-old";
+        String next = stack + "-new";
+        deploy("CreateStack", stack, template(original, "TRUE", Map.of(), false, retain));
+        assertEquals("CREATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        deploy("UpdateStack", stack, template(next, "FALSE", Map.of(), false, retain));
+        assertEquals("UPDATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        assertEquals("FALSE", alarm(next).get("AlarmRule"));
+        if (retain) {
+            assertEquals("TRUE", alarm(original).get("AlarmRule"));
+        } else {
+            assertMissing(original);
+        }
+        deleteStack(stack);
+        assertMissing(next);
+        if (retain) {
+            assertEquals("TRUE", alarm(original).get("AlarmRule"));
+            cw("DeleteAlarms", Map.of("AlarmNames", List.of(original))).then().statusCode(200);
+        }
+    }
+
+    @Test
+    void committedInPlaceUpdatesAllowAnotherUpdateAndReplacement() throws Exception {
+        String stack = unique("composite-sequence");
+        String name = stack + "-alarm";
+        String replacement = name + "-new";
+        deploy("CreateStack", stack, template(name, "TRUE", Map.of("team", "initial"), false, false));
+        assertEquals("CREATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        for (String rule : List.of("FALSE", "TRUE")) {
+            deploy("UpdateStack", stack, template(name, rule, Map.of("team", rule), false, false));
+            assertEquals("UPDATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+            assertEquals(rule, alarm(name).get("AlarmRule"));
+        }
+        deploy("UpdateStack", stack, template(replacement, "FALSE", Map.of(), false, false));
+        assertEquals("UPDATE_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        assertMissing(name);
+        assertEquals("FALSE", alarm(replacement).get("AlarmRule"));
+        deleteStack(stack);
+        assertMissing(replacement);
+    }
+
+    @Test
+    void createCollisionCannotOverwriteOrDeleteAnUnmanagedAlarm() throws Exception {
+        String stack = unique("composite-foreign");
+        String name = stack + "-alarm";
+        cw("PutCompositeAlarm", Map.of("AlarmName", name, "AlarmRule", "TRUE")).then().statusCode(200);
+        deploy("CreateStack", stack, template(name, "FALSE", Map.of(), false, false));
+        assertEquals("ROLLBACK_COMPLETE", CfnStackWaits.awaitTerminal(stack).status());
+        assertEquals("TRUE", alarm(name).get("AlarmRule"));
+        deleteStack(stack);
+        assertEquals("TRUE", alarm(name).get("AlarmRule"));
+        cw("DeleteAlarms", Map.of("AlarmNames", List.of(name))).then().statusCode(200);
+    }
+
+    private String template(String name, String rule, Map<String, String> tags, boolean fail, boolean retain)
+            throws JsonProcessingException {
+        cleanup.register(() -> cw("DeleteAlarms", Map.of("AlarmNames", List.of(name))).then().statusCode(200));
+        Map<String, Object> resources = new LinkedHashMap<>();
+        resources.put("Alarm", Map.of("Type", "AWS::CloudWatch::CompositeAlarm",
+                "UpdateReplacePolicy", retain ? "Retain" : "Delete",
+                "Properties", Map.of("AlarmName", name, "AlarmRule", rule,
+                        "Tags", tags.entrySet().stream().map(e -> Map.of("Key", e.getKey(), "Value", e.getValue())).toList())));
+        if (fail) {
+            resources.put("Failure", Map.of("Type", "AWS::SecretsManager::Secret", "DependsOn", "Alarm",
+                    "Properties", Map.of("SecretString", "explicit", "GenerateSecretString", Map.of("PasswordLength", 32))));
+        }
+        return MAPPER.writeValueAsString(Map.of("Resources", resources, "Outputs", Map.of(
+                "Name", Map.of("Value", Map.of("Ref", "Alarm")),
+                "Arn", Map.of("Value", Map.of("Fn::GetAtt", List.of("Alarm", "Arn"))))));
+    }
+
+    private void deploy(String action, String stack, String template) {
+        if ("CreateStack".equals(action)) {
+            cleanup.register(() -> given().port(baseUri.getPort())
+                    .contentType("application/x-www-form-urlencoded")
+                    .formParam("Action", "DeleteStack").formParam("StackName", stack)
+                    .post("/").then().statusCode(200));
+        }
+        given().contentType("application/x-www-form-urlencoded").formParam("Action", action)
+                .formParam("StackName", stack).formParam("TemplateBody", template).post("/").then().statusCode(200);
+    }
+
+    private Response cw(String operation, Map<String, ?> body) {
+        if ("PutCompositeAlarm".equals(operation)) {
+            String name = body.get("AlarmName").toString();
+            cleanup.register(() -> cw("DeleteAlarms", Map.of("AlarmNames", List.of(name))).then().statusCode(200));
+        }
+        return given().port(baseUri.getPort()).contentType("application/x-amz-json-1.0")
+                .header("X-Amz-Target", "GraniteServiceVersion20100801." + operation)
+                .body(MAPPER.valueToTree(body).toString()).post("/");
+    }
+
+    private Map<String, Object> alarm(String name) throws JsonProcessingException {
+        List<Map<String, Object>> alarms = cw("DescribeAlarms", Map.of("AlarmNames", List.of(name),
+                "AlarmTypes", List.of("CompositeAlarm"))).then().statusCode(200)
+                .extract().jsonPath().getList("CompositeAlarms");
+        assertEquals(1, alarms.size(), name);
+        return alarms.getFirst();
+    }
+
+    private void assertMissing(String name) throws JsonProcessingException {
+        assertEquals(List.of(), cw("DescribeAlarms", Map.of("AlarmNames", List.of(name),
+                "AlarmTypes", List.of("CompositeAlarm"))).then().statusCode(200)
+                .extract().jsonPath().getList("CompositeAlarms"));
+    }
+
+    private void deleteStack(String stack) {
+        given().contentType("application/x-www-form-urlencoded").formParam("Action", "DeleteStack")
+                .formParam("StackName", stack).post("/").then().statusCode(200);
+        CfnStackWaits.awaitStackDeleted(stack);
+    }
+
+    private static String unique(String prefix) {
+        return prefix + "-" + Long.toString(System.nanoTime(), 36);
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudformation/provisioners/CloudWatchCompositeAlarmCfnProvisionerTest.java

@@ -0,0 +1,89 @@
+package io.github.hectorvent.floci.services.cloudformation.provisioners;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.AwsException;
+import io.github.hectorvent.floci.services.cloudformation.CloudFormationTemplateEngine;
+import io.github.hectorvent.floci.services.cloudformation.model.StackResource;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.CloudWatchMetricsService;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class CloudWatchCompositeAlarmCfnProvisionerTest {
+    @Test
+    void rejectedBackingUpdateDoesNotArmRollbackToDeleteAnUnmanagedReplacement() throws Exception {
+        CloudWatchMetricsService service = mock(CloudWatchMetricsService.class);
+        CloudWatchCompositeAlarmCfnProvisioner provisioner = new CloudWatchCompositeAlarmCfnProvisioner(service);
+        CloudFormationTemplateEngine engine = mock(CloudFormationTemplateEngine.class);
+        when(engine.resolve(any())).thenAnswer(call -> {
+            JsonNode node = call.getArgument(0);
+            return node == null || node.isMissingNode() ? null : node.asText();
+        });
+        when(engine.resolveNode(any())).thenAnswer(call -> call.getArgument(0));
+        StackResource resource = new StackResource();
+        resource.setLogicalId("Alarm");
+        resource.setResourceType("AWS::CloudWatch::CompositeAlarm");
+        resource.setPhysicalId("kept");
+        resource.getAttributes().put("Arn", "prior-arn");
+        AwsException collision = new AwsException("ValidationError", "A metric alarm now has this name", 400);
+        doThrow(collision).when(service).putCompositeAlarm(any(), eq("us-east-1"), eq(true));
+        ProvisionContext context = new ProvisionContext(engine, "us-east-1", "000000000000", "stack", "kept");
+        JsonNode props = new ObjectMapper().readTree("{\"AlarmName\":\"kept\",\"AlarmRule\":\"TRUE\"}");
+        assertSame(collision, assertThrows(AwsException.class, () -> provisioner.provision(resource, props, context)));
+        assertFalse(provisioner.retainsFailedUpdateState(resource));
+        assertFalse(provisioner.rollbackUpdate(resource));
+        assertEquals("kept", resource.getPhysicalId());
+        assertEquals(Map.of("Arn", "prior-arn"), resource.getAttributes());
+        verify(service, never()).deleteAlarms(any(), anyString());
+    }
+
+    @Test
+    void inPlaceUpdateRetriesPreviouslyDisplacedAlarmCleanup() throws Exception {
+        CloudWatchMetricsService service = mock(CloudWatchMetricsService.class);
+        CloudWatchCompositeAlarmCfnProvisioner provisioner = new CloudWatchCompositeAlarmCfnProvisioner(service);
+        CloudFormationTemplateEngine engine = mock(CloudFormationTemplateEngine.class);
+        when(engine.resolve(any())).thenAnswer(call -> {
+            JsonNode node = call.getArgument(0);
+            return node == null || node.isMissingNode() ? null : node.asText();
+        });
+        when(engine.resolveNode(any())).thenAnswer(call -> call.getArgument(0));
+        StackResource resource = new StackResource();
+        resource.setLogicalId("Alarm");
+        resource.setResourceType("AWS::CloudWatch::CompositeAlarm");
+        resource.setPhysicalId("current");
+        ReplacementCleanup.record(resource,
+                new ProvisionContext(engine, "us-east-1", "000000000000", "stack", "displaced"), Map.of());
+        doThrow(new IllegalStateException("delete unavailable"))
+                .doThrow(new IllegalStateException("delete still unavailable")).doNothing()
+                .when(service).deleteAlarms(List.of("displaced"), "us-east-1");
+        assertFalse(provisioner.completeUpdate(resource).complete());
+        provisioner.clearUpdate(resource);
+        assertEquals(Set.of("displaced"), ReplacementCleanup.owedPhysicalIds(resource));
+
+        ProvisionContext context = new ProvisionContext(engine, "us-east-1", "000000000000", "stack", "current");
+        JsonNode props = new ObjectMapper().readTree("{\"AlarmName\":\"current\",\"AlarmRule\":\"TRUE\"}");
+        provisioner.provision(resource, props, context);
+        assertTrue(provisioner.retainsFailedUpdateState(resource));
+        UpdateCleanupResult retry = provisioner.completeUpdate(resource);
+        assertFalse(retry.complete());
+        assertEquals(2, retry.attempts());
+        assertEquals("displaced", retry.previousPhysicalId());
+        assertEquals("delete still unavailable", retry.failureReason());
+        assertEquals(Set.of("displaced"), ReplacementCleanup.owedPhysicalIds(resource));
+        assertTrue(provisioner.completeUpdate(resource).complete());
+        provisioner.clearUpdate(resource);
+        assertFalse(provisioner.retainsFailedUpdateState(resource));
+        assertTrue(ReplacementCleanup.owedPhysicalIds(resource).isEmpty());
+        assertEquals("current", resource.getPhysicalId());
+        verify(service, times(3)).deleteAlarms(List.of("displaced"), "us-east-1");
+        verify(service, never()).deleteAlarms(List.of("current"), "us-east-1");
+    }
+
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/CloudWatchTagUnknownResourceIntegrationTest.java

@@ -0,0 +1,349 @@
+package io.github.hectorvent.floci.services.cloudwatch;
+
+import io.github.hectorvent.floci.testing.RestAssuredJsonUtils;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.specification.RequestSpecification;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+
+/**
+ * Tagging a CloudWatch resource whose ARN names nothing has to fail, on every one of the three
+ * tag operations and for every taggable kind CloudWatch owns.
+ *
+ * <p>Each kind is checked twice: once on an ARN that resolves, so the operation is known to work
+ * at all, and once on the same ARN with the resource name swapped for one that was never created.
+ * The two ARNs are byte-identical in shape, so "the ARN was malformed" cannot explain the second
+ * answer. Without that pairing a test asserting only the failure would also pass against a build
+ * whose tag operations were broken outright.
+ *
+ * <p>The failure to look for is {@code ResourceNotFoundException}. CloudWatch's model declares
+ * that code, not the shorter {@code ResourceNotFound} that {@code GetDashboard} and
+ * {@code SetAlarmState} answer with, for {@code TagResource}, {@code UntagResource} and
+ * {@code ListTagsForResource}; the SDK maps the two to different exception classes, so a caller
+ * catching the documented one would not see the other.
+ */
+@QuarkusTest
+class CloudWatchTagUnknownResourceIntegrationTest {
+
+    private static final String CW_SCOPE =
+            "AWS4-HMAC-SHA256 Credential=test/20260205/us-east-1/monitoring/aws4_request";
+    private static final String LOGS_SCOPE =
+            "AWS4-HMAC-SHA256 Credential=test/20260205/us-east-1/logs/aws4_request";
+
+    private static final String JSON_1_0 = "application/x-amz-json-1.0";
+    private static final String JSON_1_1 = "application/x-amz-json-1.1";
+
+    private static final String PREFIX = "arn:aws:cloudwatch:us-east-1:000000000000:";
+    private static final String FIREHOSE_ARN =
+            "arn:aws:firehose:us-east-1:000000000000:deliverystream/metrics";
+    private static final String ROLE_ARN = "arn:aws:iam::000000000000:role/metric-stream-role";
+
+    @BeforeAll
+    static void configureRestAssured() {
+        RestAssuredJsonUtils.configureAwsContentTypes();
+    }
+
+    private static RequestSpecification query(String action) {
+        return given()
+                .contentType("application/x-www-form-urlencoded")
+                .header("Authorization", CW_SCOPE)
+                .formParam("Action", action);
+    }
+
+    private static RequestSpecification json(String contentType, String scope, String target, String body) {
+        return given()
+                .contentType(contentType)
+                .header("Authorization", scope)
+                .header("X-Amz-Target", target)
+                .body(body);
+    }
+
+    private static RequestSpecification metricsJson(String target, String body) {
+        return json(JSON_1_0, CW_SCOPE, "GraniteServiceVersion20100801." + target, body);
+    }
+
+    private static RequestSpecification logsJson(String target, String body) {
+        return json(JSON_1_1, LOGS_SCOPE, "Logs_20140328." + target, body);
+    }
+
+    /** The three Query tag operations against one ARN, asserting the same outcome for each. */
+    private static void assertQueryTagOpsNotFound(String arn) {
+        query("TagResource")
+                .formParam("ResourceARN", arn)
+                .formParam("Tags.member.1.Key", "env")
+                .formParam("Tags.member.1.Value", "prod")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("ErrorResponse.Error.Code", equalTo("ResourceNotFoundException"));
+
+        query("UntagResource")
+                .formParam("ResourceARN", arn)
+                .formParam("TagKeys.member.1", "env")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("ErrorResponse.Error.Code", equalTo("ResourceNotFoundException"));
+
+        query("ListTagsForResource")
+                .formParam("ResourceARN", arn)
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("ErrorResponse.Error.Code", equalTo("ResourceNotFoundException"));
+    }
+
+    /** The same three operations against an ARN that does resolve, read back through a list. */
+    private static void assertQueryTagOpsSucceed(String arn) {
+        query("TagResource")
+                .formParam("ResourceARN", arn)
+                .formParam("Tags.member.1.Key", "env")
+                .formParam("Tags.member.1.Value", "prod")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        query("ListTagsForResource")
+                .formParam("ResourceARN", arn)
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200)
+                .body("ListTagsForResourceResponse.ListTagsForResourceResult.Tags.member.Key",
+                        equalTo("env"))
+                .body("ListTagsForResourceResponse.ListTagsForResourceResult.Tags.member.Value",
+                        equalTo("prod"));
+
+        query("UntagResource")
+                .formParam("ResourceARN", arn)
+                .formParam("TagKeys.member.1", "env")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        query("ListTagsForResource")
+                .formParam("ResourceARN", arn)
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200)
+                .body(not(containsString("<Key>env</Key>")));
+    }
+
+    @Test
+    void alarmTagOperationsRejectAnArnThatNamesNoAlarm() {
+        query("PutMetricAlarm")
+                .formParam("AlarmName", "tag-unknown-alarm")
+                .formParam("MetricName", "CPUUtilization")
+                .formParam("Namespace", "AWS/EC2")
+                .formParam("ComparisonOperator", "GreaterThanThreshold")
+                .formParam("EvaluationPeriods", "1")
+                .formParam("Period", "60")
+                .formParam("Statistic", "Average")
+                .formParam("Threshold", "80")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        assertQueryTagOpsSucceed(PREFIX + "alarm:tag-unknown-alarm");
+        assertQueryTagOpsNotFound(PREFIX + "alarm:tag-unknown-alarm-nosuch");
+    }
+
+    @Test
+    void dashboardTagOperationsRejectAnArnThatNamesNoDashboard() {
+        query("PutDashboard")
+                .formParam("DashboardName", "tag-unknown-dashboard")
+                .formParam("DashboardBody", "{\"widgets\":[]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        assertQueryTagOpsSucceed(PREFIX + "dashboard/tag-unknown-dashboard");
+        assertQueryTagOpsNotFound(PREFIX + "dashboard/tag-unknown-dashboard-nosuch");
+    }
+
+    @Test
+    void metricStreamTagOperationsRejectAnArnThatNamesNoStream() {
+        query("PutMetricStream")
+                .formParam("Name", "tag-unknown-stream")
+                .formParam("FirehoseArn", FIREHOSE_ARN)
+                .formParam("RoleArn", ROLE_ARN)
+                .formParam("OutputFormat", "json")
+                .formParam("IncludeFilters.member.1.Namespace", "AWS/EC2")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        assertQueryTagOpsSucceed(PREFIX + "metric-stream/tag-unknown-stream");
+        assertQueryTagOpsNotFound(PREFIX + "metric-stream/tag-unknown-stream-nosuch");
+    }
+
+    /**
+     * The dashboard ARN AWS documents carries no region, and the one Floci mints does. A
+     * caller sending the documented form for a dashboard that exists must not be told the
+     * dashboard does not exist, which is what raising on a miss would otherwise have made of
+     * a shape that used to fall through to an empty tag map.
+     */
+    @Test
+    void theRegionlessDashboardArnAwsDocumentsAlsoResolves() {
+        query("PutDashboard")
+                .formParam("DashboardName", "tag-regionless-dashboard")
+                .formParam("DashboardBody", "{\"widgets\":[]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        assertQueryTagOpsSucceed("arn:aws:cloudwatch::000000000000:dashboard/tag-regionless-dashboard");
+
+        // Still only an absent region: another region's ARN remains a miss.
+        assertQueryTagOpsNotFound(
+                "arn:aws:cloudwatch:eu-west-1:000000000000:dashboard/tag-regionless-dashboard");
+    }
+
+    /**
+     * Both handlers route every ARN that is not a dashboard and not a metric stream to the
+     * alarm path, so a Contributor Insights ARN, which AWS documents as taggable and Floci
+     * does not serve, lands there. Reporting it as not found is right; calling it an alarm
+     * that was not found is not, so the message names the ARN instead.
+     */
+    @Test
+    void anArnOfAKindFlociDoesNotServeIsReportedAgainstTheArn() {
+        query("TagResource")
+                .formParam("ResourceARN", PREFIX + "insight-rule/no-such-rule")
+                .formParam("Tags.member.1.Key", "env")
+                .formParam("Tags.member.1.Value", "prod")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("ErrorResponse.Error.Code", equalTo("ResourceNotFoundException"))
+                .body("ErrorResponse.Error.Message",
+                        containsString("insight-rule/no-such-rule"))
+                .body("ErrorResponse.Error.Message",
+                        not(containsString("Alarm")));
+    }
+
+    /**
+     * CloudWatch Metrics serves the same three operations over JSON as well as Query, and
+     * AGENTS.md requires the two handlers not to drift. A fix applied to only one of them
+     * would leave the CLI and SDK v3 route still answering 200.
+     */
+    @Test
+    void theJsonProtocolRejectsTheSameArns() {
+        metricsJson("PutMetricAlarm", """
+                {"AlarmName": "tag-unknown-json-alarm", "MetricName": "CPUUtilization",
+                 "Namespace": "AWS/EC2", "ComparisonOperator": "GreaterThanThreshold",
+                 "EvaluationPeriods": 1, "Period": 60, "Statistic": "Average", "Threshold": 80}""")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        String real = PREFIX + "alarm:tag-unknown-json-alarm";
+        String ghost = PREFIX + "alarm:tag-unknown-json-alarm-nosuch";
+
+        metricsJson("TagResource",
+                "{\"ResourceARN\": \"" + real + "\", \"Tags\": [{\"Key\": \"env\", \"Value\": \"prod\"}]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        metricsJson("ListTagsForResource", "{\"ResourceARN\": \"" + real + "\"}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200)
+                .body("Tags[0].Key", equalTo("env"))
+                .body("Tags[0].Value", equalTo("prod"));
+
+        metricsJson("TagResource",
+                "{\"ResourceARN\": \"" + ghost + "\", \"Tags\": [{\"Key\": \"env\", \"Value\": \"prod\"}]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("__type", containsString("ResourceNotFoundException"));
+
+        metricsJson("UntagResource", "{\"ResourceARN\": \"" + ghost + "\", \"TagKeys\": [\"env\"]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("__type", containsString("ResourceNotFoundException"));
+
+        metricsJson("ListTagsForResource", "{\"ResourceARN\": \"" + ghost + "\"}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(404)
+                .body("__type", containsString("ResourceNotFoundException"));
+    }
+
+    /**
+     * CloudWatch Logs serves its own TagResource, UntagResource and ListTagsForResource over
+     * a separate handler, and already rejects an ARN that names no log group. Pinning it keeps
+     * the two halves of CloudWatch's tagging from diverging again in the other direction.
+     */
+    @Test
+    void logGroupTagOperationsRejectAnArnThatNamesNoGroup() {
+        String real = "arn:aws:logs:us-east-1:000000000000:log-group:tag-unknown-group:*";
+        String ghost = "arn:aws:logs:us-east-1:000000000000:log-group:tag-unknown-group-nosuch:*";
+
+        logsJson("CreateLogGroup", "{\"logGroupName\": \"tag-unknown-group\"}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        logsJson("TagResource",
+                "{\"resourceArn\": \"" + real + "\", \"tags\": {\"env\": \"prod\"}}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200);
+
+        logsJson("ListTagsForResource", "{\"resourceArn\": \"" + real + "\"}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(200)
+                .body("tags.env", equalTo("prod"));
+
+        logsJson("TagResource",
+                "{\"resourceArn\": \"" + ghost + "\", \"tags\": {\"env\": \"prod\"}}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(400)
+                .body("__type", containsString("ResourceNotFoundException"));
+
+        logsJson("UntagResource", "{\"resourceArn\": \"" + ghost + "\", \"tagKeys\": [\"env\"]}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(400)
+                .body("__type", containsString("ResourceNotFoundException"));
+
+        logsJson("ListTagsForResource", "{\"resourceArn\": \"" + ghost + "\"}")
+            .when()
+                .post("/")
+            .then()
+                .statusCode(400)
+                .body("__type", containsString("ResourceNotFoundException"));
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/dashboards/CloudWatchDashboardsServiceTest.java

@@ -8,6 +8,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -116,10 +117,10 @@ class CloudWatchDashboardsServiceTest {
 
     @Test
     void tagsGivenOnCreateAreReadableByArn() {
-        service.putDashboard("ops", BODY, java.util.Map.of("team", "platform"), REGION);
+        service.putDashboard("ops", BODY, Map.of("team", "platform"), REGION);
         String arn = service.getDashboard("ops", REGION).getDashboardArn();
 
-        assertEquals(java.util.Map.of("team", "platform"), service.listTagsForResource(arn, REGION));
+        assertEquals(Map.of("team", "platform"), service.listTagsForResource(arn, REGION));
         assertTrue(CloudWatchDashboardsService.isDashboardArn(arn));
     }
 
@@ -129,12 +130,12 @@ class CloudWatchDashboardsServiceTest {
      */
     @Test
     void tagsOnAReplacingPutDoNotOverwriteTheExistingOnes() {
-        service.putDashboard("ops", BODY, java.util.Map.of("team", "platform"), REGION);
+        service.putDashboard("ops", BODY, Map.of("team", "platform"), REGION);
         String arn = service.getDashboard("ops", REGION).getDashboardArn();
 
-        service.putDashboard("ops", BODY, java.util.Map.of("team", "someone-else"), REGION);
+        service.putDashboard("ops", BODY, Map.of("team", "someone-else"), REGION);
 
-        assertEquals(java.util.Map.of("team", "platform"), service.listTagsForResource(arn, REGION));
+        assertEquals(Map.of("team", "platform"), service.listTagsForResource(arn, REGION));
     }
 
     @Test
@@ -142,11 +143,75 @@ class CloudWatchDashboardsServiceTest {
         service.putDashboard("ops", BODY, REGION);
         String arn = service.getDashboard("ops", REGION).getDashboardArn();
 
-        service.tagResource(arn, java.util.Map.of("env", "dev"), REGION);
+        service.tagResource(arn, Map.of("env", "dev"), REGION);
-        assertEquals(java.util.Map.of("env", "dev"), service.listTagsForResource(arn, REGION));
+        assertEquals(Map.of("env", "dev"), service.listTagsForResource(arn, REGION));
 
         service.untagResource(arn, List.of("env"), REGION);
-        assertEquals(java.util.Map.of(), service.listTagsForResource(arn, REGION));
+        assertEquals(Map.of(), service.listTagsForResource(arn, REGION));
+    }
+
+    /**
+     * AWS documents the dashboard ARN without a region and Floci mints one with a region, so
+     * both forms have to resolve. Before the tag operations raised on a miss this divergence
+     * was invisible here: the documented ARN matched nothing and got an empty tag map back.
+     */
+    @Test
+    void bothTheRegionlessAndTheRegionfulDashboardArnResolve() {
+        service.putDashboard("ops", BODY, REGION);
+        String regionful = service.getDashboard("ops", REGION).getDashboardArn();
+        String regionless = "arn:aws:cloudwatch::000000000000:dashboard/ops";
+
+        assertEquals("arn:aws:cloudwatch:us-east-1:000000000000:dashboard/ops", regionful);
+        assertTrue(CloudWatchDashboardsService.isDashboardArn(regionless));
+
+        service.tagResource(regionless, Map.of("env", "dev"), REGION);
+        // Read back through the other form: one dashboard, reachable either way.
+        assertEquals(Map.of("env", "dev"), service.listTagsForResource(regionful, REGION));
+        assertEquals(Map.of("env", "dev"), service.listTagsForResource(regionless, REGION));
+
+        service.untagResource(regionless, List.of("env"), REGION);
+        assertEquals(Map.of(), service.listTagsForResource(regionful, REGION));
+    }
+
+    /**
+     * Only an absent region stands in for the dashboard's own. An ARN naming a different
+     * region must not resolve to the same-named dashboard in this one.
+     */
+    @Test
+    void anArnNamingAnotherRegionDoesNotResolve() {
+        service.putDashboard("ops", BODY, REGION);
+        String elsewhere = "arn:aws:cloudwatch:eu-west-1:000000000000:dashboard/ops";
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> service.listTagsForResource(elsewhere, REGION));
+        assertEquals("ResourceNotFoundException", e.getErrorCode());
+    }
+
+    /**
+     * An ARN naming no dashboard is an error on all three operations, not a silent no-op. The
+     * ARN used here differs from a working one only in the dashboard name, so the outcome
+     * cannot be explained by the ARN failing to parse as a dashboard ARN.
+     */
+    @Test
+    void tagOperationsRejectAnArnThatNamesNoDashboard() {
+        service.putDashboard("ops", BODY, REGION);
+        String arn = service.getDashboard("ops", REGION).getDashboardArn();
+        String ghost = arn.replace("dashboard/ops", "dashboard/nosuch");
+        assertTrue(CloudWatchDashboardsService.isDashboardArn(ghost));
+
+        for (Runnable call : List.<Runnable>of(
+                () -> service.tagResource(ghost, Map.of("env", "prod"), REGION),
+                () -> service.untagResource(ghost, List.of("env"), REGION),
+                () -> service.listTagsForResource(ghost, REGION))) {
+            AwsException e = assertThrows(AwsException.class, call::run);
+            assertEquals("ResourceNotFoundException", e.getErrorCode());
+            assertEquals(404, e.getHttpStatus());
+        }
+
+        // The failures wrote nothing: the real dashboard still carries only what it was given.
+        assertEquals(Map.of(), service.listTagsForResource(arn, REGION));
+        service.tagResource(arn, Map.of("env", "dev"), REGION);
+        assertEquals(Map.of("env", "dev"), service.listTagsForResource(arn, REGION));
     }
 
     @Test

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifierTest.java

@@ -0,0 +1,170 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.sns.SnsService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class AlarmSnsNotifierTest {
+    private static final String REGION = "us-east-1";
+    private static final String TOPIC = "arn:aws:sns:us-east-1:000000000000:alarms";
+    private final ObjectMapper mapper = new ObjectMapper();
+    private SnsService sns;
+    private CloudWatchMetricsService service;
+    private MetricAlarm alarm;
+
+    @BeforeEach
+    void setUp() {
+        sns = mock(SnsService.class);
+        service = new CloudWatchMetricsService(new InMemoryStorage<>(), new InMemoryStorage<>(),
+                new RegionResolver(REGION, "000000000000"), new AlarmSnsNotifier(sns, mapper));
+        alarm = new MetricAlarm();
+        alarm.setAlarmName("test-alarm");
+        alarm.setActionsEnabled(true);
+        alarm.setAlarmActions(List.of(TOPIC));
+        alarm.setOkActions(List.of(TOPIC));
+        alarm.setInsufficientDataActions(List.of(TOPIC));
+        alarm.setEvaluationPeriods(3);
+        alarm.setStatistic("Average");
+        alarm.setMetricName("Errors");
+        alarm.setNamespace("Local/PoC");
+        service.putMetricAlarm(alarm, REGION);
+    }
+
+    @Test
+    void publishesAwsPayloadForEachTransitionButNotRepeatedState() throws Exception {
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        service.setAlarmState("test-alarm", "ALARM", "still breached", null, REGION);
+        service.setAlarmState("test-alarm", "OK", "recovered", null, REGION);
+        service.setAlarmState("test-alarm", "INSUFFICIENT_DATA", "missing", null, REGION);
+        ArgumentCaptor<String> messages = ArgumentCaptor.forClass(String.class);
+        verify(sns, times(3)).publish(eq(TOPIC), isNull(), messages.capture(), anyString(), eq(REGION));
+        JsonNode first = mapper.readTree(messages.getAllValues().getFirst());
+        assertEquals("test-alarm", first.path("AlarmName").asText());
+        assertEquals("INSUFFICIENT_DATA", first.path("OldStateValue").asText());
+        assertEquals("ALARM", first.path("NewStateValue").asText());
+        assertEquals("000000000000", first.path("AWSAccountId").asText());
+        assertEquals(alarm.getAlarmArn(), first.path("AlarmArn").asText());
+        assertEquals("Errors", first.path("Trigger").path("MetricName").asText());
+        assertFalse(first.path("Trigger").has("DatapointsToAlarm"));
+        assertEquals("AVERAGE", first.path("Trigger").path("Statistic").asText());
+        assertEquals("", first.path("Trigger").path("EvaluateLowSampleCountPercentile").asText());
+        assertTrue(first.path("Trigger").path("EvaluateLowSampleCountPercentile").isTextual());
+        assertNull(alarm.getDatapointsToAlarm());
+        assertEquals("OK", mapper.readTree(messages.getAllValues().get(2)).path("OldStateValue").asText());
+    }
+
+    @Test
+    void formatsBothTimestampsWithMillisAndUtcOffset() throws Exception {
+        alarm.setAlarmConfigurationUpdatedTimestamp(1716280510L);
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        ArgumentCaptor<String> message = ArgumentCaptor.forClass(String.class);
+        verify(sns).publish(eq(TOPIC), isNull(), message.capture(), anyString(), eq(REGION));
+        JsonNode payload = mapper.readTree(message.getValue());
+        assertEquals("2024-05-21T08:35:10.000+0000",
+                payload.path("AlarmConfigurationUpdatedTimestamp").asText());
+        String stateChangeTime = payload.path("StateChangeTime").asText();
+        assertTrue(stateChangeTime
+                .matches("[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\\.[0-9]{3}\\+0000"));
+        assertEquals(alarm.getStateUpdatedTimestamp(),
+                Instant.parse(stateChangeTime.replace("+0000", "Z")).getEpochSecond());
+    }
+
+    @Test
+    void preservesExplicitDatapointsToAlarmInPayload() throws Exception {
+        alarm.setDatapointsToAlarm(2);
+        alarm.setStatistic("Maximum");
+        alarm.setEvaluateLowSampleCountPercentile("ignore");
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        ArgumentCaptor<String> message = ArgumentCaptor.forClass(String.class);
+        verify(sns).publish(eq(TOPIC), isNull(), message.capture(), anyString(), eq(REGION));
+        assertEquals(2, mapper.readTree(message.getValue()).path("Trigger").path("DatapointsToAlarm").intValue());
+        JsonNode trigger = mapper.readTree(message.getValue()).path("Trigger");
+        assertEquals("MAXIMUM", trigger.path("Statistic").asText());
+        assertEquals("ignore", trigger.path("EvaluateLowSampleCountPercentile").asText());
+        assertEquals(2, alarm.getDatapointsToAlarm());
+    }
+
+    @Test
+    void disabledActionsDoNotPublish() {
+        alarm.setActionsEnabled(false);
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        verifyNoInteractions(sns);
+        assertEquals("ALARM", alarm.getStateValue());
+    }
+
+    @Test
+    void failedTargetDoesNotPreventOtherTargetsOrStateChange() {
+        String missing = "arn:aws:sns:us-east-1:000000000000:missing";
+        alarm.setAlarmActions(List.of(missing, "arn:aws:autoscaling:us-east-1:000000000000:scalingPolicy:x", TOPIC));
+        doThrow(new IllegalStateException("missing topic"))
+                .when(sns).publish(eq(missing), isNull(), anyString(), anyString(), eq(REGION));
+        assertDoesNotThrow(() -> service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION));
+        verify(sns).publish(eq(TOPIC), isNull(), anyString(), anyString(), eq(REGION));
+        assertEquals("ALARM", alarm.getStateValue());
+    }
+
+    @Test
+    void usesTopicRegionAndPreservesChinaPartition() {
+        String topic = "arn:aws-cn:sns:cn-north-1:000000000000:alarms";
+        alarm.setAlarmArn("arn:aws-cn:cloudwatch:cn-north-1:000000000000:alarm:test-alarm");
+        alarm.setAlarmActions(List.of(topic));
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        verify(sns).publish(eq(topic), isNull(), contains("arn:aws-cn:cloudwatch:"), anyString(), eq("cn-north-1"));
+    }
+
+    @Test
+    void blockedNotificationDoesNotHoldTheAlarmMutationLock() throws Exception {
+        CountDownLatch publishing = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        doAnswer(call -> {
+            if (call.<String>getArgument(3).startsWith("ALARM:")) {
+                publishing.countDown();
+                assertTrue(release.await(5, TimeUnit.SECONDS));
+            }
+            return "message-id";
+        }).when(sns).publish(eq(TOPIC), isNull(), anyString(), anyString(), eq(REGION));
+
+        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
+            Future<?> transition = executor.submit(() ->
+                    service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION));
+            try {
+                assertTrue(publishing.await(2, TimeUnit.SECONDS));
+                Future<?> unrelated = executor.submit(() -> {
+                    CompositeAlarm composite = new CompositeAlarm();
+                    composite.setAlarmName("unrelated");
+                    composite.setAlarmRule("FALSE");
+                    service.putCompositeAlarm(composite, REGION);
+                    service.setAlarmState("test-alarm", "OK", "recovered", null, REGION);
+                });
+                unrelated.get(2, TimeUnit.SECONDS);
+                assertEquals("OK", alarm.getStateValue());
+            } finally {
+                release.countDown();
+            }
+            transition.get(2, TimeUnit.SECONDS);
+        }
+        ArgumentCaptor<String> messages = ArgumentCaptor.forClass(String.class);
+        verify(sns, times(2)).publish(eq(TOPIC), isNull(), messages.capture(), anyString(), eq(REGION));
+        assertEquals("ALARM", mapper.readTree(messages.getAllValues().getFirst()).path("NewStateValue").asText());
+        assertEquals("OK", mapper.readTree(messages.getAllValues().getLast()).path("NewStateValue").asText());
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchAlarmQueryHandlerTest.java

@@ -0,0 +1,45 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import jakarta.ws.rs.core.MultivaluedHashMap;
+import jakarta.ws.rs.core.MultivaluedMap;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class CloudWatchAlarmQueryHandlerTest {
+    @ParameterizedTest
+    @ValueSource(strings = {"PutMetricAlarm", "PutCompositeAlarm"})
+    void sharedFieldsReachBothAlarmTypes(String operation) {
+        String region = "us-east-1";
+        CloudWatchMetricsService service = new CloudWatchMetricsService(new InMemoryStorage<>(),
+                new InMemoryStorage<>(), new RegionResolver(region, "123456789012"));
+        CloudWatchMetricsQueryHandler handler = new CloudWatchMetricsQueryHandler(service, null, null);
+        MultivaluedMap<String, String> request = new MultivaluedHashMap<>();
+        request.putSingle("AlarmName", "shared");
+        request.putSingle("AlarmRule", "FALSE");
+        request.putSingle("AlarmDescription", "shared fields");
+        request.putSingle("ActionsEnabled", "false");
+        request.putSingle("AlarmActions.member.1", "alarm-action");
+        request.putSingle("OKActions.member.1", "ok-action");
+        request.putSingle("InsufficientDataActions.member.1", "missing-action");
+        request.putSingle("Tags.member.1.Key", "team");
+        request.putSingle("Tags.member.1.Value", "metrics");
+        assertEquals(200, handler.handle(operation, request, region).getStatus());
+        Alarm alarm = "PutMetricAlarm".equals(operation)
+                ? service.describeAlarms(List.of("shared"), null, region).getFirst()
+                : service.describeCompositeAlarms(List.of("shared"), null, region).getFirst();
+        assertEquals("shared fields", alarm.getAlarmDescription());
+        assertFalse(alarm.isActionsEnabled());
+        assertEquals(List.of("alarm-action"), alarm.getAlarmActions());
+        assertEquals(List.of("ok-action"), alarm.getOkActions());
+        assertEquals(List.of("missing-action"), alarm.getInsufficientDataActions());
+        assertEquals(Map.of("team", "metrics"), alarm.getTags());
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricMathAlarmTest.java

@@ -0,0 +1,218 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.common.XmlParser;
+import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.core.storage.PersistentStorage;
+import io.github.hectorvent.floci.core.storage.StorageBackend;
+import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import jakarta.ws.rs.core.MultivaluedHashMap;
+import jakarta.ws.rs.core.MultivaluedMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.w3c.dom.Document;
+
+import javax.xml.xpath.XPath;
+import javax.xml.xpath.XPathFactory;
+import java.nio.file.Path;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+class CloudWatchMetricMathAlarmTest {
+
+    private static final String REGION = "us-east-1";
+    private static final String ALARM_NAME = "metric-math-round-trip";
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final String XML_ALARM = "/DescribeAlarmsResponse/DescribeAlarmsResult/MetricAlarms/member";
+
+    private CloudWatchMetricsJsonHandler jsonHandler;
+    private CloudWatchMetricsQueryHandler queryHandler;
+
+    @BeforeEach
+    void setup() {
+        initialize(new InMemoryStorage<>());
+    }
+
+    private void initialize(StorageBackend<String, Alarm> alarmStore) {
+        RegionResolver resolver = new RegionResolver(REGION, "000000000000");
+        CloudWatchMetricsService service = new CloudWatchMetricsService(new InMemoryStorage<>(), alarmStore, resolver);
+        CloudWatchDashboardsService dashboards = new CloudWatchDashboardsService(new InMemoryStorage<>(), resolver);
+        jsonHandler = new CloudWatchMetricsJsonHandler(service, dashboards, null, MAPPER);
+        queryHandler = new CloudWatchMetricsQueryHandler(service, dashboards, null);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void metricMathQueriesRoundTripThroughBothProtocols(boolean queryProtocol) throws Exception {
+        ObjectNode request = metricMathRequest();
+        put(request, queryProtocol);
+        assertMetricMathAlarm(request);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void metricMathQueriesSurvivePersistentStorageReload(boolean queryProtocol, @TempDir Path directory)
+            throws Exception {
+        Path storageFile = directory.resolve("alarms.json");
+        TypeReference<Map<String, Alarm>> type = new TypeReference<>() { };
+        initialize(new PersistentStorage<>(storageFile, type));
+        ObjectNode request = metricMathRequest();
+        put(request, queryProtocol);
+
+        PersistentStorage<String, Alarm> reloaded = new PersistentStorage<>(storageFile, type);
+        reloaded.load();
+        assertEquals(1, reloaded.keys().size(), "the stored alarm must deserialize after a restart");
+        initialize(reloaded);
+        assertMetricMathAlarm(request);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void switchingBetweenSingleMetricAndMetricMathReplacesThePreviousDefinition(boolean queryProtocol)
+            throws Exception {
+        ObjectNode singleMetric = singleMetricRequest();
+        put(singleMetric, queryProtocol);
+        assertSingleMetricAlarm(300);
+
+        ObjectNode math = metricMathRequest();
+        put(math, !queryProtocol);
+        assertMetricMathAlarm(math);
+
+        singleMetric.put("Period", 120);
+        put(singleMetric, queryProtocol);
+        assertSingleMetricAlarm(120);
+    }
+
+    @Test
+    void existingSingleMetricDefaultPeriodIsPreserved() throws Exception {
+        ObjectNode singleMetric = singleMetricRequest();
+        singleMetric.remove("Period");
+        put(singleMetric, false);
+        assertSingleMetricAlarm(60);
+    }
+
+    private void put(ObjectNode request, boolean queryProtocol) {
+        if (queryProtocol) {
+            assertEquals(200, queryHandler.handle("PutMetricAlarm", toQuery(request), REGION).getStatus());
+        } else {
+            assertEquals(200, jsonHandler.handle("PutMetricAlarm", request, REGION).getStatus());
+        }
+    }
+
+    private JsonNode describeJson() {
+        ObjectNode request = MAPPER.createObjectNode();
+        request.putArray("AlarmNames").add(ALARM_NAME);
+        JsonNode response = (JsonNode) jsonHandler.handle("DescribeAlarms", request, REGION).getEntity();
+        assertEquals(1, response.path("MetricAlarms").size());
+        return response.path("MetricAlarms").get(0);
+    }
+
+    private Document describeXml() throws Exception {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        params.putSingle("AlarmNames.member.1", ALARM_NAME);
+        String xml = (String) queryHandler.handle("DescribeAlarms", params, REGION).getEntity();
+        return XmlParser.parseDocument(xml);
+    }
+
+    private void assertMetricMathAlarm(ObjectNode request) throws Exception {
+        JsonNode alarm = describeJson();
+        assertEquals(request.path("Metrics"), alarm.path("Metrics"));
+        Document document = describeXml();
+        XPath xpath = XPathFactory.newInstance().newXPath();
+        for (String field : new String[]{"Period", "MetricName", "Namespace", "Statistic", "ExtendedStatistic",
+                "Unit", "Dimensions"}) {
+            assertFalse(alarm.has(field), "metric-math JSON must not contain " + field);
+            assertEquals("0", xpath.evaluate("count(" + XML_ALARM + "/" + field + ")", document),
+                    "metric-math XML must not contain " + field);
+        }
+        assertEquals("3", xpath.evaluate("count(" + XML_ALARM + "/Metrics/member)", document));
+        assertEquals("5", xpath.evaluate(XML_ALARM + "/Threshold", document).replace(".0", ""));
+        assertEquals("2", xpath.evaluate(XML_ALARM + "/EvaluationPeriods", document));
+        assertEquals(5.0, alarm.path("Threshold").asDouble());
+        assertEquals(2, alarm.path("EvaluationPeriods").asInt());
+        assertXmlMatchesJson(request.path("Metrics"), XML_ALARM + "/Metrics", document, xpath);
+    }
+
+    private void assertSingleMetricAlarm(int period) throws Exception {
+        JsonNode alarm = describeJson();
+        assertEquals(period, alarm.path("Period").asInt());
+        assertEquals("AWS/ApiGateway", alarm.path("Namespace").asText());
+        assertEquals("5xx", alarm.path("MetricName").asText());
+        assertEquals("Sum", alarm.path("Statistic").asText());
+        assertFalse(alarm.has("Metrics"));
+        Document document = describeXml();
+        XPath xpath = XPathFactory.newInstance().newXPath();
+        assertEquals(Integer.toString(period), xpath.evaluate(XML_ALARM + "/Period", document));
+        assertEquals("0", xpath.evaluate("count(" + XML_ALARM + "/Metrics)", document));
+    }
+
+    private static void assertXmlMatchesJson(JsonNode value, String path, Document document, XPath xpath)
+            throws Exception {
+        if (value.isArray()) {
+            assertEquals(Integer.toString(value.size()), xpath.evaluate("count(" + path + "/member)", document));
+            for (int index = 0; index < value.size(); index++) {
+                assertXmlMatchesJson(value.get(index), path + "/member[" + (index + 1) + "]", document, xpath);
+            }
+        } else if (value.isObject()) {
+            for (Map.Entry<String, JsonNode> field : value.properties()) {
+                assertXmlMatchesJson(field.getValue(), path + "/" + field.getKey(), document, xpath);
+            }
+        } else {
+            assertEquals(value.asText(), xpath.evaluate(path, document), path);
+        }
+    }
+
+    private static MultivaluedMap<String, String> toQuery(JsonNode value) {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        flatten(value, "", params);
+        return params;
+    }
+
+    private static void flatten(JsonNode value, String prefix, MultivaluedMap<String, String> params) {
+        if (value.isArray()) {
+            for (int index = 0; index < value.size(); index++) {
+                flatten(value.get(index), prefix + ".member." + (index + 1), params);
+            }
+        } else if (value.isObject()) {
+            for (Map.Entry<String, JsonNode> field : value.properties()) {
+                flatten(field.getValue(), prefix.isEmpty() ? field.getKey() : prefix + "." + field.getKey(), params);
+            }
+        } else {
+            params.putSingle(prefix, value.asText());
+        }
+    }
+
+    private static ObjectNode metricMathRequest() throws Exception {
+        return (ObjectNode) MAPPER.readTree("""
+                {"AlarmName":"metric-math-round-trip","EvaluationPeriods":2,"Threshold":5,
+                 "ComparisonOperator":"GreaterThanThreshold","Metrics":[
+                  {"Id":"rate","Expression":"100 * errors / requests","ReturnData":true,
+                   "Period":300,"Label":"Error <rate> & requests"},
+                  {"Id":"errors","ReturnData":false,"AccountId":"123456789012","Label":"Errors",
+                   "MetricStat":{"Metric":{"Namespace":"AWS/ApiGateway","MetricName":"5xx",
+                    "Dimensions":[{"Name":"ApiName","Value":"api<&>"},{"Name":"Stage","Value":"test"}]},
+                    "Period":300,"Stat":"Sum","Unit":"Count"}},
+                  {"Id":"requests","ReturnData":false,"MetricStat":{
+                   "Metric":{"Namespace":"AWS/ApiGateway","MetricName":"Count"},"Period":300,"Stat":"Sum"}}
+                 ]}
+                """);
+    }
+
+    private static ObjectNode singleMetricRequest() throws Exception {
+        return (ObjectNode) MAPPER.readTree("""
+                {"AlarmName":"metric-math-round-trip","Namespace":"AWS/ApiGateway","MetricName":"5xx",
+                 "Statistic":"Sum","Period":300,"EvaluationPeriods":2,"Threshold":5,
+                 "ComparisonOperator":"GreaterThanThreshold"}
+                """);
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsEntityMetricDataIntegrationTest.java

@@ -0,0 +1,203 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.config.EncoderConfig;
+import io.restassured.http.ContentType;
+import io.restassured.parsing.Parser;
+import io.restassured.specification.RequestSpecification;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+
+import static io.restassured.RestAssured.config;
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+
+/**
+ * PutMetricData datums nested under EntityMetricData, as the CloudWatch Agent sends them,
+ * must be stored under the request namespace on every protocol path. floci-io/floci#6066.
+ */
+@QuarkusTest
+class CloudWatchMetricsEntityMetricDataIntegrationTest {
+
+    private static final String CW_AUTH =
+            "AWS4-HMAC-SHA256 Credential=test/20261020/us-east-1/monitoring/aws4_request";
+    private static final String JSON_1_0 = "application/x-amz-json-1.0";
+
+    @BeforeAll
+    static void registerJsonParser() {
+        RestAssured.registerParser(JSON_1_0, Parser.JSON);
+    }
+
+    private static RequestSpecification json(String operation) {
+        return given()
+                .header("Authorization", CW_AUTH)
+                .contentType(JSON_1_0)
+                .config(config().encoderConfig(EncoderConfig.encoderConfig()
+                        .encodeContentTypeAs(JSON_1_0, ContentType.TEXT)))
+                .header("X-Amz-Target", "GraniteServiceVersion20100801." + operation);
+    }
+
+    private static RequestSpecification query(String action) {
+        return given()
+                .header("Authorization", CW_AUTH)
+                .contentType("application/x-www-form-urlencoded")
+                .formParam("Action", action)
+                .formParam("Version", "2010-08-01");
+    }
+
+    @Test
+    void jsonEntityMetricDataOnlyIsListedAndAggregated() {
+        long now = Instant.now().getEpochSecond();
+        json("PutMetricData")
+                .body("""
+                        {"Namespace":"CWAgent/Json","MetricData":[],
+                         "EntityMetricData":[{
+                           "Entity":{"KeyAttributes":{"Type":"AWS::Resource",
+                             "ResourceType":"AWS::EC2::Instance","Identifier":"i-0123456789abcdef0"},
+                             "Attributes":{}},
+                           "MetricData":[
+                             {"MetricName":"mem_used_percent","Value":6.5,"Unit":"Percent",
+                              "StorageResolution":60,"Timestamp":%d,
+                              "Dimensions":[{"Name":"host","Value":"h1"}]},
+                             {"MetricName":"disk_used_percent","Value":0.75,"Unit":"Percent",
+                              "StorageResolution":60,"Timestamp":%d,
+                              "Dimensions":[{"Name":"host","Value":"h1"}]}]}],
+                         "StrictEntityValidation":false}
+                        """.formatted(now, now))
+        .when().post("/")
+        .then().statusCode(200);
+
+        json("ListMetrics")
+                .body("{\"Namespace\":\"CWAgent/Json\"}")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("Metrics.MetricName", containsInAnyOrder("mem_used_percent", "disk_used_percent"));
+
+        json("GetMetricStatistics")
+                .body("""
+                        {"Namespace":"CWAgent/Json","MetricName":"mem_used_percent",
+                         "Dimensions":[{"Name":"host","Value":"h1"}],
+                         "StartTime":%d,"EndTime":%d,"Period":60,"Statistics":["Sum"]}
+                        """.formatted(now - 120, now + 120))
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("Datapoints.Sum", hasItem(6.5f));
+    }
+
+    @Test
+    void jsonTopLevelAndEntityMetricDataAreBothStored() {
+        json("PutMetricData")
+                .body("""
+                        {"Namespace":"CWAgent/JsonMixed",
+                         "MetricData":[{"MetricName":"TopLevel","Value":1}],
+                         "EntityMetricData":[{"Entity":{"KeyAttributes":{"Type":"Service",
+                           "Name":"checkout","Environment":"prod"}},
+                           "MetricData":[{"MetricName":"FromEntity","Value":2}]}]}
+                        """)
+        .when().post("/")
+        .then().statusCode(200);
+
+        json("ListMetrics")
+                .body("{\"Namespace\":\"CWAgent/JsonMixed\"}")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("Metrics.MetricName", containsInAnyOrder("TopLevel", "FromEntity"));
+    }
+
+    @Test
+    void queryTopLevelAndEntityMetricDataAreListedAndAggregated() {
+        Instant now = Instant.now();
+        query("PutMetricData")
+                .formParam("Namespace", "CWAgent/Query")
+                .formParam("MetricData.member.1.MetricName", "TopLevel")
+                .formParam("MetricData.member.1.Value", "1")
+                .formParam("EntityMetricData.member.1.Entity.KeyAttributes.entry.1.key", "Type")
+                .formParam("EntityMetricData.member.1.Entity.KeyAttributes.entry.1.value", "AWS::Resource")
+                .formParam("EntityMetricData.member.1.MetricData.member.1.MetricName", "mem_used_percent")
+                .formParam("EntityMetricData.member.1.MetricData.member.1.Value", "6.5")
+                .formParam("EntityMetricData.member.1.MetricData.member.1.Timestamp", now.toString())
+                .formParam("EntityMetricData.member.1.MetricData.member.1.Dimensions.member.1.Name", "host")
+                .formParam("EntityMetricData.member.1.MetricData.member.1.Dimensions.member.1.Value", "h1")
+                .formParam("StrictEntityValidation", "false")
+        .when().post("/")
+        .then().statusCode(200);
+
+        query("ListMetrics")
+                .formParam("Namespace", "CWAgent/Query")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("ListMetricsResponse.ListMetricsResult.Metrics.member.MetricName",
+                        containsInAnyOrder("TopLevel", "mem_used_percent"));
+
+        query("GetMetricStatistics")
+                .formParam("Namespace", "CWAgent/Query")
+                .formParam("MetricName", "mem_used_percent")
+                .formParam("Dimensions.member.1.Name", "host")
+                .formParam("Dimensions.member.1.Value", "h1")
+                .formParam("StartTime", now.minusSeconds(120).toString())
+                .formParam("EndTime", now.plusSeconds(120).toString())
+                .formParam("Period", "60")
+                .formParam("Statistics.member.1", "Sum")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("GetMetricStatisticsResponse.GetMetricStatisticsResult.Datapoints.member.Sum",
+                        equalTo("6.5"));
+    }
+
+    @Test
+    void jsonNestedDatumWithoutMetricNameIsRejectedAndNothingIsStored() {
+        json("PutMetricData")
+                .body("""
+                        {"Namespace":"CWAgent/JsonInvalid",
+                         "MetricData":[{"MetricName":"TopLevel","Value":1}],
+                         "EntityMetricData":[{"MetricData":[{"Value":2}]}]}
+                        """)
+        .when().post("/")
+        .then()
+                .statusCode(400)
+                .header("x-amzn-query-error", equalTo("MissingParameter;Sender"))
+                .body("__type", equalTo("MissingRequiredParameterException"))
+                .body("message", equalTo(
+                        "The parameter EntityMetricData.member.1.MetricData.member.1.MetricName is required."));
+
+        json("ListMetrics")
+                .body("{\"Namespace\":\"CWAgent/JsonInvalid\"}")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("Metrics", empty());
+    }
+
+    @Test
+    void queryMemberWithoutMetricNameIsRejectedAndNothingIsStored() {
+        query("PutMetricData")
+                .formParam("Namespace", "CWAgent/QueryInvalid")
+                .formParam("MetricData.member.1.Value", "1")
+                .formParam("MetricData.member.2.MetricName", "Valid")
+                .formParam("MetricData.member.2.Value", "2")
+        .when().post("/")
+        .then()
+                .statusCode(400)
+                .body("ErrorResponse.Error.Code", equalTo("MissingParameter"))
+                .body("ErrorResponse.Error.Message",
+                        equalTo("The parameter MetricData.member.1.MetricName is required."));
+
+        query("ListMetrics")
+                .formParam("Namespace", "CWAgent/QueryInvalid")
+        .when().post("/")
+        .then()
+                .statusCode(200)
+                .body("ListMetricsResponse.ListMetricsResult.Metrics.member.size()", equalTo(0));
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsJsonHandlerTest.java

@@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.InMemoryStorage;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
@@ -17,6 +18,13 @@ import java.math.BigDecimal;
 import java.time.Instant;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Handler-level tests for CloudWatchMetricsJsonHandler.
@@ -146,6 +154,114 @@ class CloudWatchMetricsJsonHandlerTest {
                 "metric stored with 24h-ago timestamp must not appear in a 20-second window around now");
     }
 
+    // The CloudWatch Agent sends an empty top-level MetricData and nests every datum under
+    // EntityMetricData, floci-io/floci#6066.
+    @Test
+    void putMetricData_entityMetricDataOnly_storesNestedDatumsUnderTheRequestNamespace() {
+        ObjectNode req = MAPPER.createObjectNode();
+        req.put("Namespace", "CWAgent");
+        req.putArray("MetricData");
+        ObjectNode entityMetricData = req.putArray("EntityMetricData").addObject();
+        ObjectNode entity = entityMetricData.putObject("Entity");
+        entity.putObject("KeyAttributes")
+                .put("Type", "AWS::Resource")
+                .put("ResourceType", "AWS::EC2::Instance")
+                .put("Identifier", "i-0123456789abcdef0");
+        entity.putObject("Attributes");
+        ObjectNode datum = entityMetricData.putArray("MetricData").addObject();
+        datum.put("MetricName", "mem_used_percent");
+        datum.put("Value", 6.43);
+        datum.put("Unit", "Percent");
+        datum.put("StorageResolution", 60);
+        datum.put("Timestamp", sdkTimestamp(EPOCH_NOW));
+        datum.putArray("Dimensions").addObject().put("Name", "host").put("Value", "h1");
+        req.put("StrictEntityValidation", false);
+
+        assertEquals(200, handler.handle("PutMetricData", req, REGION).getStatus());
+
+        ObjectNode stats = getStats("CWAgent", "mem_used_percent", "host", "h1",
+                EPOCH_NOW.minusSeconds(60), EPOCH_NOW.plusSeconds(60), 60);
+        assertEquals(1, stats.get("Datapoints").size());
+        assertEquals(6.43, stats.get("Datapoints").get(0).get("Sum").asDouble());
+    }
+
+    @Test
+    void putMetricData_topLevelAndEntityMetricData_storesBoth() {
+        ObjectNode req = MAPPER.createObjectNode();
+        req.put("Namespace", "Mixed");
+        req.putArray("MetricData").addObject()
+                .put("MetricName", "TopLevel").put("Value", 1.0);
+        ArrayNode entityMetricData = req.putArray("EntityMetricData");
+        entityMetricData.addObject().putArray("MetricData").addObject()
+                .put("MetricName", "FirstEntity").put("Value", 2.0);
+        entityMetricData.addObject().putArray("MetricData").addObject()
+                .put("MetricName", "SecondEntity").put("Value", 3.0);
+
+        assertEquals(200, handler.handle("PutMetricData", req, REGION).getStatus());
+
+        ObjectNode listReq = MAPPER.createObjectNode();
+        listReq.put("Namespace", "Mixed");
+        ObjectNode listed = (ObjectNode) handler.handle("ListMetrics", listReq, REGION).getEntity();
+        List<String> names = new ArrayList<>();
+        listed.get("Metrics").forEach(m -> names.add(m.get("MetricName").asText()));
+        assertEquals(List.of("FirstEntity", "SecondEntity", "TopLevel"), names.stream().sorted().toList());
+    }
+
+    @Test
+    void putMetricData_entityDatumWithoutMetricName_rejectsTheWholeRequest() {
+        ObjectNode req = MAPPER.createObjectNode();
+        req.put("Namespace", "Invalid");
+        req.putArray("MetricData").addObject()
+                .put("MetricName", "TopLevel").put("Value", 1.0);
+        ArrayNode nested = req.putArray("EntityMetricData").addObject().putArray("MetricData");
+        nested.addObject().put("MetricName", "Valid").put("Value", 2.0);
+        nested.addObject().put("Value", 3.0);
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> handler.handle("PutMetricData", req, REGION));
+        assertEquals("MissingParameter", e.getErrorCode());
+        assertEquals("MissingRequiredParameterException", e.jsonType());
+        assertEquals(400, e.getHttpStatus());
+        assertEquals("The parameter EntityMetricData.member.1.MetricData.member.2.MetricName is required.",
+                e.getMessage());
+        assertEquals(0, listMetricNames("Invalid").size());
+    }
+
+    @Test
+    void putMetricData_topLevelDatumWithEmptyMetricName_rejectsTheWholeRequest() {
+        ObjectNode req = MAPPER.createObjectNode();
+        req.put("Namespace", "InvalidTop");
+        ArrayNode data = req.putArray("MetricData");
+        data.addObject().put("MetricName", "Valid").put("Value", 1.0);
+        data.addObject().put("MetricName", "").put("Value", 2.0);
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> handler.handle("PutMetricData", req, REGION));
+        assertEquals("MissingParameter", e.getErrorCode());
+        assertEquals("The parameter MetricData.member.2.MetricName is required.", e.getMessage());
+        assertEquals(0, listMetricNames("InvalidTop").size());
+    }
+
+    // AWS constrains MetricName to 1 to 255 characters with no pattern, so whitespace is a name.
+    @Test
+    void putMetricData_whitespaceMetricName_isStored() {
+        ObjectNode req = MAPPER.createObjectNode();
+        req.put("Namespace", "WhitespaceName");
+        req.putArray("MetricData").addObject().put("MetricName", " ").put("Value", 1.0);
+
+        assertEquals(200, handler.handle("PutMetricData", req, REGION).getStatus());
+        assertEquals(List.of(" "), listMetricNames("WhitespaceName"));
+    }
+
+    private List<String> listMetricNames(String namespace) {
+        ObjectNode listReq = MAPPER.createObjectNode();
+        listReq.put("Namespace", namespace);
+        ObjectNode listed = (ObjectNode) handler.handle("ListMetrics", listReq, REGION).getEntity();
+        List<String> names = new ArrayList<>();
+        listed.get("Metrics").forEach(m -> names.add(m.get("MetricName").asText()));
+        return names;
+    }
+
     @Test
     void setAlarmState_updatesFieldsCorrectly() {
         putMetric("NS", "M", "type", "current", 100.0, EPOCH_NOW);
@@ -234,6 +350,22 @@ class CloudWatchMetricsJsonHandlerTest {
         assertEquals("Count", alarm.get("Unit").asText());
     }
 
+    // AWS sends every alarm collection even when it is empty. The CLI's text output writes a
+    // None line for each paginator result key missing from the page (floci-io/floci#6362).
+    @Test
+    void describeAlarms_returnsEmptyCompositeAndLogAlarmCollections() {
+        handler.handle("PutMetricAlarm", alarmRequest("MetricOnly"), REGION);
+
+        ObjectNode response = (ObjectNode) handler.handle("DescribeAlarms", MAPPER.createObjectNode(), REGION)
+                .getEntity();
+
+        assertEquals(1, response.get("MetricAlarms").size());
+        assertTrue(response.get("CompositeAlarms").isArray());
+        assertTrue(response.get("CompositeAlarms").isEmpty());
+        assertTrue(response.get("LogAlarms").isArray());
+        assertTrue(response.get("LogAlarms").isEmpty());
+    }
+
     // An alarm the caller never gave an M for reports no DatapointsToAlarm at all. AWS omits the
     // member in that case, and echoing EvaluationPeriods instead reads as a change to any client
     // that re-plans against its own configuration, which is what floci-io/floci#3660 hit.

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsQueryHandlerTest.java

@@ -1,6 +1,9 @@
 package io.github.hectorvent.floci.services.cloudwatch.metrics;
 
+import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricDatum;
 import io.github.hectorvent.floci.services.cloudwatch.metricstreams.CloudWatchMetricStreamsService;
 import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
 import jakarta.ws.rs.core.MultivaluedHashMap;
@@ -12,12 +15,15 @@ import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -123,4 +129,128 @@ class CloudWatchMetricsQueryHandlerTest {
                 anyInt(), any(), unitCaptor.capture(), any());
         assertEquals("Count", unitCaptor.getValue());
     }
+
+    // The Query serialization of EntityMetricData nests each datum two lists deep;
+    // floci-io/floci#6066 is the JSON-side report of the same dropped datums.
+    @Test
+    void putMetricDataStoresTopLevelAndEntityMetricDataUnderTheRequestNamespace() {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        params.putSingle("Namespace", "CWAgent");
+        params.putSingle("MetricData.member.1.MetricName", "TopLevel");
+        params.putSingle("MetricData.member.1.Value", "1");
+        params.putSingle("EntityMetricData.member.1.Entity.KeyAttributes.entry.1.key", "Type");
+        params.putSingle("EntityMetricData.member.1.Entity.KeyAttributes.entry.1.value", "AWS::Resource");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.MetricName", "mem_used_percent");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.Value", "6.5");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.Unit", "Percent");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.Dimensions.member.1.Name", "host");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.Dimensions.member.1.Value", "h1");
+        params.putSingle("EntityMetricData.member.2.MetricData.member.1.MetricName", "disk_used_percent");
+        params.putSingle("EntityMetricData.member.2.MetricData.member.1.Value", "0.75");
+        params.putSingle("StrictEntityValidation", "false");
+
+        handler.handle("PutMetricData", params, REGION);
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<MetricDatum>> captor = ArgumentCaptor.forClass(List.class);
+        verify(metricsService).putMetricData(eq("CWAgent"), captor.capture(), eq(REGION));
+        List<MetricDatum> datums = captor.getValue();
+        assertEquals(List.of("TopLevel", "mem_used_percent", "disk_used_percent"),
+                datums.stream().map(MetricDatum::getMetricName).toList());
+        MetricDatum entityDatum = datums.get(1);
+        assertEquals(6.5, entityDatum.getValue());
+        assertEquals("Percent", entityDatum.getUnit());
+        assertEquals("host", entityDatum.getDimensions().get(0).name());
+        assertEquals("h1", entityDatum.getDimensions().get(0).value());
+    }
+
+    @Test
+    void putMetricDataStoresEntityMetricDataAfterAnEmptyEntityEntry() {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        params.putSingle("Namespace", "CWAgent");
+        params.putSingle("EntityMetricData.member.2.MetricData.member.1.MetricName", "M");
+        params.putSingle("EntityMetricData.member.2.MetricData.member.1.Value", "1");
+        params.putSingle("EntityMetricData.member.10.MetricData.member.1.MetricName", "N");
+        params.putSingle("EntityMetricData.member.10.MetricData.member.1.Value", "2");
+        params.putSingle("StrictEntityValidation", "false");
+
+        handler.handle("PutMetricData", params, REGION);
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<MetricDatum>> captor = ArgumentCaptor.forClass(List.class);
+        verify(metricsService).putMetricData(eq("CWAgent"), captor.capture(), eq(REGION));
+        List<MetricDatum> datums = captor.getValue();
+        assertEquals(List.of("M", "N"), datums.stream().map(MetricDatum::getMetricName).toList());
+        assertEquals(1.0, datums.get(0).getValue());
+        assertEquals(2.0, datums.get(1).getValue());
+    }
+
+    @Test
+    void putMetricDataRejectsAMemberWithoutMetricNameBeforeStoringAnything() {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        params.putSingle("Namespace", "NS");
+        params.putSingle("MetricData.member.1.Value", "1");
+        params.putSingle("MetricData.member.2.MetricName", "Valid");
+        params.putSingle("MetricData.member.2.Value", "2");
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> handler.handle("PutMetricData", params, REGION));
+        assertEquals("MissingParameter", e.getErrorCode());
+        assertEquals("MissingRequiredParameterException", e.jsonType());
+        assertEquals(400, e.getHttpStatus());
+        assertEquals("The parameter MetricData.member.1.MetricName is required.", e.getMessage());
+        verify(metricsService, never()).putMetricData(any(), any(), any());
+    }
+
+    @Test
+    void putMetricDataRejectsANestedMemberWithoutMetricNameBeforeStoringAnything() {
+        MultivaluedMap<String, String> params = new MultivaluedHashMap<>();
+        params.putSingle("Namespace", "NS");
+        params.putSingle("MetricData.member.1.MetricName", "TopLevel");
+        params.putSingle("MetricData.member.1.Value", "1");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.1.Unit", "Percent");
+        params.putSingle("EntityMetricData.member.1.MetricData.member.2.MetricName", "Valid");
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> handler.handle("PutMetricData", params, REGION));
+        assertEquals("The parameter EntityMetricData.member.1.MetricData.member.1.MetricName is required.",
+                e.getMessage());
+        verify(metricsService, never()).putMetricData(any(), any(), any());
+    }
+
+    @Test
+    void putMetricDataRejectsAnEmptyMetricNameButAcceptsWhitespace() {
+        MultivaluedMap<String, String> empty = new MultivaluedHashMap<>();
+        empty.putSingle("Namespace", "NS");
+        empty.putSingle("MetricData.member.1.MetricName", "");
+        empty.putSingle("MetricData.member.1.Value", "1");
+
+        AwsException e = assertThrows(AwsException.class,
+                () -> handler.handle("PutMetricData", empty, REGION));
+        assertEquals("The parameter MetricData.member.1.MetricName is required.", e.getMessage());
+        verify(metricsService, never()).putMetricData(any(), any(), any());
+
+        MultivaluedMap<String, String> whitespace = new MultivaluedHashMap<>();
+        whitespace.putSingle("Namespace", "NS");
+        whitespace.putSingle("MetricData.member.1.MetricName", " ");
+        whitespace.putSingle("MetricData.member.1.Value", "1");
+
+        handler.handle("PutMetricData", whitespace, REGION);
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<MetricDatum>> captor = ArgumentCaptor.forClass(List.class);
+        verify(metricsService).putMetricData(eq("NS"), captor.capture(), eq(REGION));
+        assertEquals(List.of(" "), captor.getValue().stream().map(MetricDatum::getMetricName).toList());
+    }
+
+    @Test
+    void describeAlarmsReturnsEmptyCompositeAndLogAlarmCollections() {
+        when(metricsService.describeAlarms(any(), any(), eq(REGION))).thenReturn(List.of());
+
+        String xml = (String) handler.handle("DescribeAlarms", new MultivaluedHashMap<>(), REGION).getEntity();
+
+        assertTrue(xml.contains("<MetricAlarms></MetricAlarms>"), xml);
+        assertTrue(xml.contains("<CompositeAlarms></CompositeAlarms>"), xml);
+        assertTrue(xml.contains("<LogAlarms></LogAlarms>"), xml);
+    }
 }

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsTagsTest.java

@@ -2,6 +2,7 @@ package io.github.hectorvent.floci.services.cloudwatch.metrics;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.InMemoryStorage;
 import io.github.hectorvent.floci.services.cloudwatch.dashboards.CloudWatchDashboardsService;
@@ -83,6 +84,34 @@ class CloudWatchMetricsTagsTest {
         assertNull(tags.get("team"));
     }
 
+    /**
+     * An ARN naming no alarm is an error on all three operations, not a silent no-op. The ARN
+     * used here differs from a working one only in the alarm name, so the outcome cannot be
+     * explained by the ARN failing to parse.
+     */
+    @Test
+    void tagOperationsRejectAnArnThatNamesNoAlarm() {
+        MetricAlarm alarm = new MetricAlarm();
+        alarm.setAlarmName("test-alarm");
+        alarm.setAlarmArn("arn:aws:cloudwatch:us-east-1:000000000000:alarm:test-alarm");
+        service.putMetricAlarm(alarm, REGION);
+        String ghost = "arn:aws:cloudwatch:us-east-1:000000000000:alarm:nosuch";
+
+        for (Runnable call : List.<Runnable>of(
+                () -> service.tagResource(ghost, Map.of("env", "prod"), REGION),
+                () -> service.untagResource(ghost, List.of("env"), REGION),
+                () -> service.listTagsForResource(ghost, REGION))) {
+            AwsException e = assertThrows(AwsException.class, call::run);
+            assertEquals("ResourceNotFoundException", e.getErrorCode());
+            assertEquals(404, e.getHttpStatus());
+        }
+
+        // The failures wrote nothing: the real alarm still carries only what it was given.
+        assertEquals(Map.of(), service.listTagsForResource(alarm.getAlarmArn(), REGION));
+        service.tagResource(alarm.getAlarmArn(), Map.of("env", "dev"), REGION);
+        assertEquals(Map.of("env", "dev"), service.listTagsForResource(alarm.getAlarmArn(), REGION));
+    }
+
     @Test
     void tagsViaPutMetricAlarmQuery() {
         MultivaluedMap<String, String> params = new MultivaluedHashMap<>();

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CompositeAlarmIntegrationTest.java

@@ -0,0 +1,160 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.testing.PartitionCleanup;
+import io.github.hectorvent.floci.testing.RestAssuredJsonUtils;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.response.Response;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.net.URI;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static io.restassured.RestAssured.given;
+import static org.junit.jupiter.api.Assertions.*;
+
+@QuarkusTest
+class CompositeAlarmIntegrationTest {
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final String SCOPE =
+            "AWS4-HMAC-SHA256 Credential=test/20261010/cn-north-1/monitoring/aws4_request";
+
+    @TestHTTPResource
+    URI baseUri;
+
+    @RegisterExtension
+    final PartitionCleanup cleanup = new PartitionCleanup();
+
+    @BeforeAll
+    static void configureRestAssured() {
+        RestAssuredJsonUtils.configureAwsContentTypes();
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void bothProtocolsCreateDescribeEvaluateAndDeleteRealAlarms(boolean json) throws Exception {
+        String prefix = "wire-composite-" + Long.toString(System.nanoTime(), 36);
+        String source = prefix + "-source";
+        String aggregate = prefix + "-aggregate";
+        call(json, "PutCompositeAlarm", Map.of("AlarmName", source, "AlarmRule", "TRUE")).then().statusCode(200);
+        call(json, "PutCompositeAlarm", Map.of("AlarmName", aggregate, "AlarmRule", "ALARM(\"" + source + "\")",
+                "ActionsEnabled", false, "AlarmDescription", "combined")).then().statusCode(200);
+        Response result = call(json, "DescribeAlarms", Map.of("AlarmNames", List.of(aggregate),
+                "AlarmTypes", List.of("CompositeAlarm")));
+        result.then().statusCode(200);
+        if (json) {
+            assertEquals(List.of(), result.jsonPath().getList("LogAlarms"));
+            assertEquals(1, result.jsonPath().getList("CompositeAlarms").size());
+            assertEquals("ALARM", result.jsonPath().getString("CompositeAlarms[0].StateValue"));
+            assertEquals("arn:aws-cn:cloudwatch:cn-north-1:000000000000:alarm:" + aggregate,
+                    result.jsonPath().getString("CompositeAlarms[0].AlarmArn"));
+            assertFalse(result.jsonPath().getBoolean("CompositeAlarms[0].ActionsEnabled"));
+        } else {
+            assertTrue(result.asString().contains("<LogAlarms>"));
+            assertEquals("ALARM", result.xmlPath().getString(
+                    "DescribeAlarmsResponse.DescribeAlarmsResult.CompositeAlarms.member.StateValue"));
+            assertTrue(result.asString().contains("arn:aws-cn:cloudwatch:cn-north-1:000000000000:alarm:" + aggregate));
+            assertEquals("false", result.xmlPath().getString(
+                    "DescribeAlarmsResponse.DescribeAlarmsResult.CompositeAlarms.member.ActionsEnabled"));
+        }
+        assertFalse(call(json, "DescribeAlarms", Map.of("AlarmNames", List.of(aggregate)))
+                .then().statusCode(200).extract().asString().contains(aggregate));
+        call(json, "SetAlarmState", Map.of("AlarmName", source, "StateValue", "OK",
+                "StateReason", "recovery")).then().statusCode(200);
+        Response recovered = call(json, "DescribeAlarms", Map.of("AlarmNames", List.of(aggregate),
+                "AlarmTypes", List.of("CompositeAlarm")));
+        assertEquals("OK", json ? recovered.jsonPath().getString("CompositeAlarms[0].StateValue")
+                : recovered.xmlPath().getString(
+                        "DescribeAlarmsResponse.DescribeAlarmsResult.CompositeAlarms.member.StateValue"));
+        for (String name : List.of(aggregate, source)) {
+            call(json, "DeleteAlarms", Map.of("AlarmNames", List.of(name))).then().statusCode(200);
+        }
+        Response empty = call(json, "DescribeAlarms", Map.of("AlarmNamePrefix", prefix,
+                "AlarmTypes", List.of("CompositeAlarm")));
+        assertFalse(empty.asString().contains(aggregate));
+        assertFalse(empty.asString().contains(source));
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void invalidRuleAndUnsupportedSuppressionCannotCreateAnAlarm(boolean json) throws Exception {
+        String name = "invalid-composite-" + Long.toString(System.nanoTime(), 36);
+        call(json, "PutCompositeAlarm", Map.of("AlarmName", name, "AlarmRule", "TRUE AND"))
+                .then().statusCode(400);
+        call(json, "PutCompositeAlarm", Map.of("AlarmName", name, "AlarmRule", "TRUE",
+                "ActionsSuppressor", "other")).then().statusCode(400);
+        assertFalse(call(json, "DescribeAlarms", Map.of("AlarmNames", List.of(name),
+                "AlarmTypes", List.of("CompositeAlarm"))).asString().contains(name));
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void alarmTypeValidationAcceptsLogAlarmsAndUsesSmithyErrors(boolean json) throws Exception {
+        call(json, "DescribeAlarms", Map.of("AlarmTypes", List.of("LogAlarm"))).then().statusCode(200);
+        Response error = call(json, "DescribeAlarms", Map.of("AlarmTypes", List.of("Unknown")));
+        error.then().statusCode(400);
+        assertEquals("1 validation error detected: Value '[Unknown]' at 'alarmTypes' failed to satisfy constraint: "
+                + "Member must satisfy constraint: [Member must satisfy enum value set: "
+                + "[CompositeAlarm, MetricAlarm, LogAlarm]]", errorMessage(json, error));
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void lengthErrorsUseSmithyMessages(boolean json) throws Exception {
+        for (String field : List.of("AlarmName", "AlarmRule")) {
+            int maximum = "AlarmName".equals(field) ? 255 : 10240;
+            String path = "AlarmName".equals(field) ? "alarmName" : "alarmRule";
+            for (String value : List.of("", "x".repeat(maximum + 1))) {
+                Map<String, String> request = new LinkedHashMap<>(Map.of("AlarmName", "length-validation", "AlarmRule", "TRUE"));
+                request.put(field, value);
+                Response error = call(json, "PutCompositeAlarm", request);
+                error.then().statusCode(400);
+                assertEquals("1 validation error detected: Value '" + value + "' at '" + path
+                        + "' failed to satisfy constraint: Member must have length "
+                        + (value.isEmpty() ? "greater than or equal to 1" : "less than or equal to " + maximum),
+                        errorMessage(json, error));
+            }
+        }
+    }
+
+    private String errorMessage(boolean json, Response response) {
+        return json ? response.jsonPath().getString("message") : response.xmlPath().getString("ErrorResponse.Error.Message");
+    }
+
+    private Response call(boolean json, String operation, Map<String, ?> body)
+            throws JsonProcessingException {
+        if ("PutCompositeAlarm".equals(operation)) {
+            String name = body.get("AlarmName").toString();
+            cleanup.register(() -> given().port(baseUri.getPort()).contentType("application/x-amz-json-1.0")
+                    .header("Authorization", SCOPE)
+                    .header("X-Amz-Target", "GraniteServiceVersion20100801.DeleteAlarms")
+                    .body(MAPPER.valueToTree(Map.of("AlarmNames", List.of(name))).toString())
+                    .post("/").then().statusCode(200));
+        }
+        if (json) {
+            return given().contentType("application/x-amz-json-1.0").header("Authorization", SCOPE)
+                    .header("X-Amz-Target", "GraniteServiceVersion20100801." + operation)
+                    .body(MAPPER.writeValueAsString(body)).post("/");
+        }
+        Map<String, Object> form = new LinkedHashMap<>();
+        form.put("Action", operation);
+        body.forEach((key, value) -> {
+            if (value instanceof List<?> values) {
+                for (int index = 0; index < values.size(); index++) {
+                    form.put(key + ".member." + (index + 1), values.get(index));
+                }
+            } else {
+                form.put(key, value);
+            }
+        });
+        return given().contentType("application/x-www-form-urlencoded").header("Authorization", SCOPE)
+                .formParams(form).post("/");
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CompositeAlarmServiceTest.java

@@ -0,0 +1,274 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.AwsException;
+import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.Alarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.CompositeAlarm;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.testing.ThreadStates;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class CompositeAlarmServiceTest {
+    private static final String REGION = "cn-north-1";
+    private final CloudWatchMetricsService service = new CloudWatchMetricsService(
+            new InMemoryStorage<>(), new InMemoryStorage<>(), new RegionResolver(REGION, "123456789012"));
+
+    @Test
+    void metricTransitionsPropagateThroughNamesArnsAndNestedComposites() {
+        MetricAlarm metric = new MetricAlarm();
+        metric.setAlarmName("latency");
+        service.putMetricAlarm(metric, REGION);
+        put("combined", "ALARM(\"" + metric.getAlarmArn() + "\") AND NOT ALARM(\"maintenance\")");
+        put("outer", "ALARM(combined) OR (TRUE AND FALSE)");
+        assertEquals("OK", state("outer"));
+        service.setAlarmState("latency", "ALARM", "high latency", null, REGION);
+        assertEquals("ALARM", state("combined"));
+        assertEquals("ALARM", state("outer"));
+        service.setAlarmState("latency", "OK", "recovered", null, REGION);
+        assertEquals("OK", state("outer"));
+        assertEquals(List.of(metric), service.describeAlarms(null, null, REGION));
+        assertTrue(service.describeCompositeAlarms(null, null, "us-east-1").isEmpty());
+        assertTrue(service.describeCompositeAlarms(List.of("combined"), null, REGION)
+                .getFirst().getAlarmArn().startsWith("arn:aws-cn:cloudwatch:"));
+    }
+
+    @Test
+    void updatesPreserveStateAndTagsUntilTheNextDependencyChange() {
+        put("source", "FALSE");
+        CompositeAlarm alarm = put("aggregate", "ALARM(source)");
+        service.tagResource(alarm.getAlarmArn(), Map.of("owner", "retained"), REGION);
+        service.setAlarmState("aggregate", "ALARM", "manual", null, REGION);
+        CompositeAlarm replacement = put("aggregate", "OK(source)");
+        assertEquals("ALARM", replacement.getStateValue());
+        assertEquals(Map.of("owner", "retained"), replacement.getTags());
+        service.setAlarmState("source", "ALARM", "dependency changed", null, REGION);
+        assertEquals("OK", state("aggregate"));
+    }
+
+    @Test
+    void unrelatedWritesPreserveManualCompositeStatesAndParentStates() {
+        put("source", "FALSE");
+        put("aggregate", "ALARM(source)");
+        put("parent", "ALARM(aggregate)");
+        service.setAlarmState("aggregate", "ALARM", "manual", null, REGION);
+        assertEquals("ALARM", state("parent"));
+        put("unrelated", "TRUE");
+        assertEquals("ALARM", state("aggregate"));
+        assertEquals("ALARM", state("parent"));
+        MetricAlarm metric = new MetricAlarm();
+        metric.setAlarmName("unrelated-metric");
+        service.putMetricAlarm(metric, REGION);
+        assertEquals("ALARM", state("aggregate"));
+        service.setAlarmState("unrelated-metric", "ALARM", "unrelated", null, REGION);
+        assertEquals("ALARM", state("aggregate"));
+        service.deleteAlarms(List.of("unrelated"), REGION);
+        assertEquals("ALARM", state("aggregate"));
+        assertEquals("ALARM", state("parent"));
+        service.setAlarmState("source", "ALARM", "child changed", null, REGION);
+        service.setAlarmState("source", "OK", "child recovered", null, REGION);
+        assertEquals("OK", state("aggregate"));
+        assertEquals("OK", state("parent"));
+    }
+
+    @Test
+    void unchangedChildStateDoesNotResetManuallySetParent() {
+        put("source", "FALSE");
+        put("child", "ALARM(source) AND FALSE");
+        put("parent", "ALARM(child)");
+        service.setAlarmState("parent", "ALARM", "manual", null, REGION);
+        service.setAlarmState("source", "ALARM", "child condition remains false", null, REGION);
+        assertEquals("OK", state("child"));
+        assertEquals("ALARM", state("parent"));
+    }
+
+    @Test
+    void creationAndDeletionReevaluateOnlyReferencingComposites() {
+        MetricAlarm metric = new MetricAlarm();
+        metric.setAlarmName("late");
+        put("watcher", "INSUFFICIENT_DATA(late)");
+        put("outer", "ALARM(watcher)");
+        assertEquals("OK", state("watcher"));
+        service.putMetricAlarm(metric, REGION);
+        assertEquals("ALARM", state("watcher"));
+        assertEquals("ALARM", state("outer"));
+        service.deleteAlarms(List.of("late"), REGION);
+        assertEquals("OK", state("watcher"));
+        assertEquals("OK", state("outer"));
+    }
+
+    @Test
+    void tagReadIsAnIndependentSnapshotOfTheOwnedAlarm() {
+        CompositeAlarm alarm = put("tag-snapshot", "TRUE");
+        service.tagResource(alarm.getAlarmArn(), Map.of("team", "old"), REGION);
+        Map<String, String> read = service.listTagsForResource(alarm.getAlarmArn(), REGION);
+        service.tagResource(alarm.getAlarmArn(), Map.of("team", "new"), REGION);
+        assertEquals(Map.of("team", "old"), read);
+        read.put("injected", "unowned");
+        assertEquals(Map.of("team", "new"), service.listTagsForResource(alarm.getAlarmArn(), REGION));
+    }
+
+    @Test
+    void cyclesStopEvaluationEvenBehindShortCircuitAndMustBeBrokenBeforeDelete() {
+        put("a", "FALSE AND ALARM(b)");
+        put("b", "ALARM(a)");
+        service.setAlarmState("b", "ALARM", "manual", null, REGION);
+        assertThrows(AwsException.class, () -> service.deleteAlarms(List.of("a"), REGION));
+        assertEquals(2, service.describeCompositeAlarms(null, null, REGION).size());
+        put("a", "FALSE");
+        service.deleteAlarms(List.of("b"), REGION);
+        assertEquals("OK", state("a"));
+        service.deleteAlarms(List.of("a"), REGION);
+        assertTrue(service.describeCompositeAlarms(null, null, REGION).isEmpty());
+    }
+
+    @Test
+    void manuallyChangingOneMemberDoesNotResumeACyclicDependency() {
+        put("a", "ALARM(b)");
+        put("b", "ALARM(a)");
+        assertEquals("OK", state("a"));
+        service.setAlarmState("b", "ALARM", "manual", null, REGION);
+        assertEquals("OK", state("a"));
+        put("b", "TRUE");
+        service.setAlarmState("b", "ALARM", "cycle broken", null, REGION);
+        assertEquals("ALARM", state("a"));
+    }
+
+    @Test
+    void parentQuotaTreatsAChildNameAndItsArnAsTheSameAlarm() {
+        CompositeAlarm child = put("child", "FALSE");
+        for (int index = 0; index < 150; index++) {
+            put("parent-" + index, "ALARM(child)");
+        }
+        assertThrows(AwsException.class, () -> put("overflow", "ALARM(\"" + child.getAlarmArn() + "\")"));
+        service.deleteAlarms(List.of("parent-0"), REGION);
+        assertEquals("OK", put("replacement", "ALARM(\"" + child.getAlarmArn() + "\")").getStateValue());
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"", "ALARM()", "TRUE junk", "TRUE AND", "ALARM(\"unterminated)",
+            "(TRUE", "ALARM(x) XOR FALSE", "ALARM(\"x\\z\")"})
+    void invalidRulesDoNotReplaceAnExistingAlarm(String invalid) {
+        put("saved", "TRUE");
+        assertThrows(AwsException.class, () -> put("saved", invalid));
+        assertEquals("TRUE", service.describeCompositeAlarms(List.of("saved"), null, REGION).getFirst().getAlarmRule());
+        assertEquals("ALARM", state("saved"));
+    }
+
+    @Test
+    void quotasAndAtomicCreatePreventExcessReferencesAndForeignNameReplacement() {
+        String many = IntStream.range(0, 101).mapToObj(i -> "ALARM(a" + i + ")")
+                .collect(Collectors.joining(" OR "));
+        assertThrows(AwsException.class, () -> put("too-many", many));
+        put("owned", "TRUE");
+        CompositeAlarm foreign = new CompositeAlarm();
+        foreign.setAlarmName("owned");
+        foreign.setAlarmRule("FALSE");
+        assertThrows(AwsException.class, () -> service.putCompositeAlarm(foreign, REGION, false));
+        assertEquals("ALARM", state("owned"));
+        assertThrows(AwsException.class, () -> put("deep", "(".repeat(250) + "TRUE" + ")".repeat(250)));
+        assertEquals("ALARM", put("many-not", "NOT ".repeat(502) + "TRUE").getStateValue());
+        assertThrows(AwsException.class, () -> put("elements", "FALSE OR ".repeat(500) + "TRUE"));
+    }
+
+    @Test
+    void mixedPersistenceLoadsLegacyMetricsAndRoundTripsComposites() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        TypeReference<Map<String, Alarm>> type = new TypeReference<>() {};
+        Map<String, Alarm> legacy = mapper.readValue(
+                "{\"legacy\":{\"alarmName\":\"old\",\"metricName\":\"CPU\",\"stateValue\":\"OK\"}}", type);
+        assertInstanceOf(MetricAlarm.class, legacy.get("legacy"));
+        CompositeAlarm alarm = put("new", "FALSE");
+        Map<String, Alarm> restored = mapper.readValue(mapper.writeValueAsBytes(Map.of("new", alarm)), type);
+        assertEquals("FALSE", assertInstanceOf(CompositeAlarm.class, restored.get("new")).getAlarmRule());
+        assertEquals("OK", restored.get("new").getStateValue());
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void concurrentTagChangesPreserveTheUpdatedDefinition(boolean remove) throws Exception {
+        CountDownLatch updating = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        InMemoryStorage<String, Alarm> storage = new InMemoryStorage<>() {
+            @Override
+            public void put(String key, Alarm value) {
+                if (value instanceof CompositeAlarm composite && "TRUE".equals(composite.getAlarmRule())) {
+                    updating.countDown();
+                    try {
+                        assertTrue(release.await(5, TimeUnit.SECONDS));
+                    } catch (InterruptedException error) {
+                        Thread.currentThread().interrupt();
+                        throw new AssertionError(error);
+                    }
+                }
+                super.put(key, value);
+            }
+        };
+        CloudWatchMetricsService concurrent = new CloudWatchMetricsService(new InMemoryStorage<>(), storage,
+                new RegionResolver(REGION, "123456789012"));
+        CompositeAlarm original = new CompositeAlarm();
+        original.setAlarmName("concurrent");
+        original.setAlarmRule("FALSE");
+        original.setTags(new HashMap<>(Map.of("team", "old")));
+        concurrent.putCompositeAlarm(original, REGION);
+        CompositeAlarm replacement = new CompositeAlarm();
+        replacement.setAlarmName("concurrent");
+        replacement.setAlarmRule("TRUE");
+        AtomicReference<Thread> tagThread = new AtomicReference<>();
+        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
+            Future<?> update = executor.submit(() -> concurrent.putCompositeAlarm(replacement, REGION));
+            Future<?> tags;
+            try {
+                assertTrue(updating.await(2, TimeUnit.SECONDS));
+                tags = executor.submit(() -> {
+                    tagThread.set(Thread.currentThread());
+                    if (remove) {
+                        concurrent.untagResource(original.getAlarmArn(), List.of("team"), REGION);
+                    } else {
+                        concurrent.tagResource(original.getAlarmArn(), Map.of("team", "new"), REGION);
+                    }
+                });
+                ThreadStates.awaitBlockedOrDone(tagThread, tags);
+                assertFalse(tags.isDone(), "tag mutation must share the definition update lock");
+            } finally {
+                release.countDown();
+            }
+            update.get(2, TimeUnit.SECONDS);
+            tags.get(2, TimeUnit.SECONDS);
+        }
+        CompositeAlarm result = concurrent.describeCompositeAlarms(List.of("concurrent"), null, REGION).getFirst();
+        assertEquals("TRUE", result.getAlarmRule());
+        assertEquals(remove ? Map.of() : Map.of("team", "new"), result.getTags());
+    }
+
+    private CompositeAlarm put(String name, String rule) {
+        CompositeAlarm alarm = new CompositeAlarm();
+        alarm.setAlarmName(name);
+        alarm.setAlarmRule(rule);
+        alarm.setActionsEnabled(true);
+        service.putCompositeAlarm(alarm, REGION);
+        return alarm;
+    }
+
+    private String state(String name) {
+        return service.describeCompositeAlarms(List.of(name), null, REGION).getFirst().getStateValue();
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metricstreams/CloudWatchMetricStreamsServiceTest.java

@@ -232,4 +232,29 @@ class CloudWatchMetricStreamsServiceTest {
         service.untagResource(arn, List.of("team"), REGION);
         assertFalse(service.listTagsForResource(arn, REGION).containsKey("team"));
     }
+
+    /**
+     * An ARN naming no stream is an error on all three operations, not a silent no-op. The ARN
+     * used here differs from a working one only in the stream name, so the outcome cannot be
+     * explained by the ARN failing to parse as a metric stream ARN.
+     */
+    @Test
+    void tagOperationsRejectAnArnThatNamesNoStream() {
+        String arn = service.putMetricStream(stream("ops"), REGION).getArn();
+        String ghost = arn.replace("metric-stream/ops", "metric-stream/nosuch");
+        assertTrue(CloudWatchMetricStreamsService.isMetricStreamArn(ghost));
+
+        for (Runnable call : List.<Runnable>of(
+                () -> service.tagResource(ghost, Map.of("env", "prod"), REGION),
+                () -> service.untagResource(ghost, List.of("env"), REGION),
+                () -> service.listTagsForResource(ghost, REGION))) {
+            AwsException e = assertThrows(AwsException.class, call::run);
+            assertEquals("ResourceNotFoundException", e.getErrorCode());
+            assertEquals(404, e.getHttpStatus());
+        }
+
+        // The failures wrote nothing: the real stream still carries only what it was given.
+        service.tagResource(arn, Map.of("team", "platform"), REGION);
+        assertEquals(Map.of("team", "platform"), service.listTagsForResource(arn, REGION));
+    }
 }

src/test/java/io/github/hectorvent/floci/testing/CfnStackWaits.java

@@ -0,0 +1,199 @@
+package io.github.hectorvent.floci.testing;
+
+import io.restassured.path.xml.XmlPath;
+import io.restassured.path.xml.element.Node;
+import io.restassured.response.Response;
+import io.restassured.specification.RequestSpecification;
+
+import java.util.List;
+import java.util.Set;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * DeleteStack answers before the stack is gone. The deletion runs on an executor, so a test
+ * that looks at the stack's resources right after the answer races it.
+ */
+public final class CfnStackWaits {
+
+    private CfnStackWaits() {
+    }
+
+    /** What DescribeStacks reported: the status and, when present, the StackStatusReason. */
+    public record StackState(String status, String reason) {
+    }
+
+    private static final Set<String> TERMINAL = Set.of(
+            "CREATE_COMPLETE", "CREATE_FAILED", "ROLLBACK_COMPLETE", "ROLLBACK_FAILED",
+            "UPDATE_COMPLETE", "UPDATE_FAILED", "UPDATE_ROLLBACK_COMPLETE", "UPDATE_ROLLBACK_FAILED",
+            "DELETE_COMPLETE", "DELETE_FAILED");
+
+    private static final Set<String> OPERATION_STARTS = Set.of("CREATE_IN_PROGRESS", "UPDATE_IN_PROGRESS");
+
+    /**
+     * Waits until DescribeStacks reports a terminal status for a create or update and returns
+     * it, so the caller decides what counts as success. A DescribeStacks that is not 200 (the
+     * stack does not exist) fails at once with the error body rather than polling the clock out;
+     * ten seconds without a terminal status fails too.
+     */
+    public static StackState awaitTerminal(String stackName) {
+        return awaitTerminal(stackName, null);
+    }
+
+    public static StackState awaitTerminal(String stackName, String authorization) {
+        long deadline = System.currentTimeMillis() + 10_000;
+        StackState last = new StackState(null, "");
+        while (System.currentTimeMillis() < deadline) {
+            RequestSpecification request = given()
+                .contentType("application/x-www-form-urlencoded")
+                .formParam("Action", "DescribeStacks")
+                .formParam("StackName", stackName);
+            if (authorization != null) {
+                request.header("Authorization", authorization);
+            }
+            Response response = request.when().post("/");
+            if (response.statusCode() != 200) {
+                fail("DescribeStacks " + stackName + " answered " + response.statusCode() + ": " + response.asString());
+            }
+            XmlPath xml = response.xmlPath();
+            String status = xml.getString("**.find { it.name() == 'StackStatus' }");
+            String reason = xml.getString("**.find { it.name() == 'StackStatusReason' }");
+            last = new StackState(status == null || status.isEmpty() ? null : status, reason == null ? "" : reason);
+            if (last.status() != null && TERMINAL.contains(last.status())) {
+                return last;
+            }
+            try {
+                Thread.sleep(50);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("Interrupted while waiting for stack " + stackName, e);
+            }
+        }
+        fail("Stack " + stackName + " reached no terminal status within ten seconds, last: " + last);
+        return last;
+    }
+
+    /**
+     * Waits until DescribeStacks reports {@code expectedStatus} and returns that state. Any other
+     * terminal status fails at once with its reason, as do a DescribeStacks that is not 200 and ten
+     * seconds without the expected status.
+     */
+    public static StackState awaitStatus(String stackName, String expectedStatus) {
+        long deadline = System.currentTimeMillis() + 10_000;
+        StackState last = new StackState(null, "");
+        while (System.currentTimeMillis() < deadline) {
+            Response response = given()
+                .contentType("application/x-www-form-urlencoded")
+                .formParam("Action", "DescribeStacks")
+                .formParam("StackName", stackName)
+                .when().post("/");
+            if (response.statusCode() != 200) {
+                fail("DescribeStacks " + stackName + " answered " + response.statusCode() + ": " + response.asString());
+            }
+            XmlPath xml = response.xmlPath();
+            String status = xml.getString("**.find { it.name() == 'StackStatus' }");
+            String reason = xml.getString("**.find { it.name() == 'StackStatusReason' }");
+            last = new StackState(status == null || status.isEmpty() ? null : status, reason == null ? "" : reason);
+            if (expectedStatus.equals(last.status())) {
+                return last;
+            }
+            if (last.status() != null && TERMINAL.contains(last.status())) {
+                fail("Stack " + stackName + " reached " + last.status() + " instead of " + expectedStatus
+                        + ": " + last.reason());
+            }
+            try {
+                Thread.sleep(50);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("Interrupted while waiting for stack " + stackName, e);
+            }
+        }
+        fail("Stack " + stackName + " did not reach " + expectedStatus + " within ten seconds, last: " + last);
+        return last;
+    }
+
+    /**
+     * The reason on the first resource event that failed a create or update in the stack's latest
+     * operation, the one its last stack-level CREATE_IN_PROGRESS or UPDATE_IN_PROGRESS began. The
+     * stack's own reason only names the failed resource, as on AWS, so the failure's message is read
+     * here. Fails when DescribeStackEvents is not 200 or no resource failed in that operation.
+     */
+    public static String failedResourceReason(String stackName) {
+        return failedResourceReason(stackName, null);
+    }
+
+    public static String failedResourceReason(String stackName, String authorization) {
+        RequestSpecification request = given()
+            .contentType("application/x-www-form-urlencoded")
+            .formParam("Action", "DescribeStackEvents")
+            .formParam("StackName", stackName);
+        if (authorization != null) {
+            request.header("Authorization", authorization);
+        }
+        Response response = request.when().post("/");
+        if (response.statusCode() != 200) {
+            fail("DescribeStackEvents " + stackName + " answered " + response.statusCode() + ": " + response.asString());
+        }
+        List<Node> newestFirst = response.xmlPath().getList("**.findAll { it.name() == 'member' }", Node.class);
+        String reason = null;
+        for (Node event : newestFirst) {
+            String status = text(event, "ResourceStatus");
+            if (text(event, "StackName").equals(text(event, "LogicalResourceId"))
+                    && text(event, "StackId").equals(text(event, "PhysicalResourceId"))) {
+                if (OPERATION_STARTS.contains(status)) {
+                    break;
+                }
+            } else if ("CREATE_FAILED".equals(status) || "UPDATE_FAILED".equals(status)) {
+                reason = text(event, "ResourceStatusReason");
+            }
+        }
+        if (reason == null) {
+            fail("Stack " + stackName + " has no failed resource event in its latest operation: " + response.asString());
+        }
+        return reason;
+    }
+
+    private static String text(Node event, String name) {
+        Node child = event.getNode(name);
+        return child == null ? "" : child.value();
+    }
+
+    /**
+     * Waits until DescribeStacks answers 400, which is how a missing stack is reported. Fails on
+     * DELETE_FAILED or after ten seconds.
+     */
+    public static void awaitStackDeleted(String stackName) {
+        awaitStackDeleted(stackName, null);
+    }
+
+    public static void awaitStackDeleted(String stackName, String authorization) {
+        long deadline = System.currentTimeMillis() + 10_000;
+        String body = "";
+        while (System.currentTimeMillis() < deadline) {
+            RequestSpecification request = given()
+                .contentType("application/x-www-form-urlencoded")
+                .formParam("Action", "DescribeStacks")
+                .formParam("StackName", stackName);
+            if (authorization != null) {
+                request.header("Authorization", authorization);
+            }
+            Response response = request.when().post("/");
+            body = response.asString();
+            assertThat(body, not(containsString("<StackStatus>DELETE_FAILED</StackStatus>")));
+            if (response.statusCode() == 400) {
+                return;
+            }
+            try {
+                Thread.sleep(50);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("Interrupted while waiting for stack " + stackName + " to be deleted", e);
+            }
+        }
+        fail("Stack " + stackName + " was not deleted within ten seconds: " + body);
+    }
+}

src/test/java/io/github/hectorvent/floci/testing/PartitionCleanup.java

@@ -0,0 +1,47 @@
+package io.github.hectorvent.floci.testing;
+
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * Deletes what a test created, in reverse order, after each test. The {@code @QuarkusTest}
+ * emulator is shared by every test class, so a resource left behind outlives the class that made
+ * it: one under {@code cn-north-1} shows up in every later cross-region listing (ResourceExplorer
+ * scans all resources), and an IAM user or an S3 bucket shows up in the next class to list either
+ * one unscoped. A test that leaves such a resource registers each create here.
+ *
+ * <pre>{@code
+ * @RegisterExtension
+ * final PartitionCleanup cleanup = new PartitionCleanup();
+ * ...
+ * createTable(region, name);
+ * cleanup.register(() -> deleteTable(region, name));
+ * }</pre>
+ *
+ * A failing cleanup never fails the test: it is reported on stderr so the leak is visible and
+ * the assertion that actually failed stays the headline.
+ */
+public final class PartitionCleanup implements AfterEachCallback {
+
+    private final Deque<Runnable> actions = new ArrayDeque<>();
+
+    public void register(Runnable action) {
+        actions.push(action);
+    }
+
+    @Override
+    public void afterEach(ExtensionContext context) {
+        while (!actions.isEmpty()) {
+            Runnable action = actions.pop();
+            try {
+                action.run();
+            } catch (RuntimeException | AssertionError e) {
+                System.err.println("PartitionCleanup: cleanup step failed after "
+                        + context.getDisplayName() + ": " + e);
+            }
+        }
+    }
+}

src/test/java/io/github/hectorvent/floci/testing/ThreadStates.java

@@ -0,0 +1,93 @@
+package io.github.hectorvent.floci.testing;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.Set;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.BooleanSupplier;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/** Waiting for a thread to reach a lock, for tests that force two requests to overlap. */
+public final class ThreadStates {
+
+    /** Every state a thread is in once it has stopped running: parked in any way, or ended. */
+    public static final Set<Thread.State> PARKED_OR_DONE = Collections.unmodifiableSet(EnumSet.of(
+            Thread.State.BLOCKED, Thread.State.WAITING, Thread.State.TIMED_WAITING, Thread.State.TERMINATED));
+
+    private ThreadStates() {}
+
+    /**
+     * Returns once {@code thread} is blocked entering a monitor or parked on a lock, or {@code task} has
+     * finished, whichever comes first. Code that serializes correctly leaves the second request blocked
+     * behind the first; code that does not lets it run to the end, which the caller's assertions
+     * then catch. Polls every 5 ms and gives up after five seconds rather than hang a broken test.
+     */
+    public static void awaitBlockedOrDone(AtomicReference<Thread> thread, Future<?> task)
+            throws InterruptedException {
+        awaitBlockedOrDone(thread::get, task::isDone, 5);
+    }
+
+    /**
+     * Returns once {@code thread} is blocked entering a monitor, parked on a lock, or is no longer alive.
+     * Same wait as the {@link Future} overload, for a test that owns the {@link Thread} directly; gives up
+     * after ten seconds and fails, since a thread still running unblocked by then is not overlapping the
+     * way the test needs.
+     */
+    public static void awaitBlockedOrDone(Thread thread) throws InterruptedException {
+        assertTrue(awaitBlockedOrDone(() -> thread, () -> !thread.isAlive(), 10),
+                "worker blocked or finished before timeout");
+    }
+
+    private static boolean awaitBlockedOrDone(Supplier<Thread> thread, BooleanSupplier done, long timeoutSeconds)
+            throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds);
+        while (System.nanoTime() < deadline && !done.getAsBoolean()) {
+            if (isBlocked(thread.get())) {
+                return true;
+            }
+            Thread.sleep(5);
+        }
+        return done.getAsBoolean() || isBlocked(thread.get());
+    }
+
+    private static boolean isBlocked(Thread thread) {
+        if (thread == null) {
+            return false;
+        }
+        Thread.State state = thread.getState();
+        return state == Thread.State.BLOCKED || state == Thread.State.WAITING;
+    }
+
+    /**
+     * Waits until {@code thread} is parked in any way (blocked on a monitor, waiting or timed
+     * waiting) or has ended, and returns the state it reached. Fails after ten seconds.
+     */
+    public static Thread.State awaitParkedOrDone(Thread thread) {
+        return awaitState(thread, PARKED_OR_DONE, Duration.ofSeconds(10), "the thread");
+    }
+
+    /**
+     * Waits until {@code thread} is in one of {@code states} and returns the one it reached, checking
+     * every millisecond. Fails naming {@code what} and the state it was left in once {@code timeout}
+     * passes, so a thread that never gets there fails the test instead of hanging it.
+     */
+    public static Thread.State awaitState(Thread thread, Set<Thread.State> states, Duration timeout, String what) {
+        long deadline = System.nanoTime() + timeout.toNanos();
+        Thread.State state = thread.getState();
+        while (!states.contains(state)) {
+            if (System.nanoTime() - deadline > 0) {
+                fail(what + " never reached " + states + ", it was " + state);
+            }
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+            state = thread.getState();
+        }
+        return state;
+    }
+}

src/test/resources/cloudformation/supported-resource-types.tsv

@@ -41,6 +41,7 @@ AWS::CloudFront::OriginRequestPolicy	CloudFrontCfnProvisioner
 AWS::CloudFront::ResponseHeadersPolicy	CloudFrontCfnProvisioner
 AWS::CloudTrail::Trail	CloudTrailCfnProvisioner
 AWS::CloudWatch::Alarm	CloudWatchCfnProvisioner
+AWS::CloudWatch::CompositeAlarm	CloudWatchCompositeAlarmCfnProvisioner
 AWS::CloudWatch::Dashboard	CloudWatchDashboardCfnProvisioner
 AWS::CodeBuild::Project	CodeBuildCfnProvisioner
 AWS::CodePipeline::CustomActionType	CodePipelineCfnProvisioner

tools/docs/cfn_resource_types.yaml

@@ -169,6 +169,9 @@ notes:
 # Trailing prose appended to a row after its type list, for behaviour that belongs to the service
 # rather than to one type.
 row_notes:
+  AWS::CloudWatch: >-
+    Composite alarms have backing definitions with name Ref and Arn GetAtt, rollback, and
+    replacement cleanup. Composite actions are stored only; suppression properties are rejected.
   AWS::Lambda: >-
     Inline `ZipFile` packages include the `cfn-response` (Node.js) / `cfnresponse` (Python) module
     AWS injects for that code path, so Solutions-style custom-resource handlers work.