feat(model-lifecycle): monitor Bedrock model lifecycle status, alert on change

app.py

@@ -50,6 +50,8 @@
         if config.get("export_retention_days")
         else None
     ),
+    # Bedrock model lifecycle alert recipients (optional; one address or a list)
+    model_lifecycle_notification_email=config.get("model_lifecycle_notification_email"),
     # Cognito SAML auth — only active when enable_saml_auth: true in config.yaml
     **({
         "cognito_domain_prefix": config.get("cognito_domain_prefix"),

cdk/main.py

@@ -10,6 +10,7 @@
 from .conversation_export import ConversationExport
 from .frontend import RagFrontend
 from .ingest import RagIngest
+from .model_lifecycle import ModelLifecycleMonitor
 from .waf import Waf
 
 
@@ -56,6 +57,8 @@ def __init__(
         export_url_expiry_days: int = 7,
         # None keeps every export indefinitely
         export_retention_days: int = None,
+        # Email(s) for Bedrock model lifecycle alerts - one address or a list
+        model_lifecycle_notification_email: str | list[str] = None,
         **kwargs,
     ) -> None:
         super().__init__(scope, construct_id, **kwargs)
@@ -155,3 +158,16 @@ def __init__(
             url_expiry_days=export_url_expiry_days,
             retain_exports_days=export_retention_days,
         )
+
+        # Weekly check that every Bedrock model in config.yaml is still ACTIVE,
+        # since AWS Health's deprecation notices need a support plan we don't have
+        ModelLifecycleMonitor(
+            self,
+            "ModelLifecycleMonitor",
+            chat_model=chat_model,
+            embedding_model=embedding_model,
+            video_text_model_id=video_text_model_id,
+            classifier_model=classifier_model,
+            document_filter_model=document_filter_model,
+            notification_email=model_lifecycle_notification_email,
+        )

cdk/model_lifecycle.py

@@ -0,0 +1,142 @@
+from aws_cdk import (
+    CfnOutput,
+    Duration,
+    Stack,
+    TimeZone,
+)
+from aws_cdk import (
+    aws_iam as iam,
+)
+from aws_cdk import (
+    aws_lambda as lambda_,
+)
+from aws_cdk import (
+    aws_scheduler as scheduler,
+)
+from aws_cdk import (
+    aws_scheduler_targets as scheduler_targets,
+)
+from aws_cdk import (
+    aws_sns as sns,
+)
+from aws_cdk import (
+    aws_sns_subscriptions as subscriptions,
+)
+from constructs import Construct
+
+STATUS_PARAM = "/abe/model-lifecycle/last-status"
+
+
+class ModelLifecycleMonitor(Construct):
+    """
+    Weekly check of Bedrock modelLifecycle.status for every model configured
+    in config.yaml, emailing an alert only when a status actually changes
+    (ACTIVE -> LEGACY, etc.) - AWS Health's DescribeEvents needs a
+    Business/Enterprise support plan we don't have, so this is a
+    support-plan-independent way to learn about deprecations. See
+    BEDROCK_MODEL_LIFECYCLE_MONITOR.md for the full design.
+    """
+
+    def __init__(
+        self,
+        scope: Construct,
+        construct_id: str,
+        chat_model: str,
+        embedding_model: str,
+        video_text_model_id: str,
+        classifier_model: str,
+        document_filter_model: str,
+        notification_email: str | list[str] = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(scope, construct_id, **kwargs)
+
+        #################################################################################
+        # NOTIFICATION TOPIC
+        #################################################################################
+        topic = sns.Topic(
+            self,
+            "ModelLifecycleTopic",
+            display_name="ABE model lifecycle alerts",
+        )
+        # config.yaml may give one address or a list of them. Each subscription
+        # has to be confirmed individually from its own inbox.
+        if isinstance(notification_email, str):
+            notification_email = [notification_email]
+        for address in dict.fromkeys(notification_email or []):
+            topic.add_subscription(subscriptions.EmailSubscription(address))
+
+        #################################################################################
+        # MONITOR LAMBDA
+        #################################################################################
+        monitor_lambda = lambda_.Function(
+            self,
+            "MonitorLambda",
+            function_name="abe-model-lifecycle-monitor",
+            runtime=lambda_.Runtime.PYTHON_3_13,
+            handler="monitor.handler",
+            code=lambda_.Code.from_asset("src/model_lifecycle"),
+            timeout=Duration.seconds(30),
+            environment={
+                "CHAT_MODEL_ID": chat_model,
+                "EMBEDDING_MODEL_ID": embedding_model,
+                "VIDEO_TEXT_MODEL_ID": video_text_model_id,
+                "CLASSIFIER_MODEL_ID": classifier_model,
+                "DOCUMENT_FILTER_MODEL_ID": document_filter_model,
+                "SNS_TOPIC_ARN": topic.topic_arn,
+                "STATUS_PARAM": STATUS_PARAM,
+            },
+        )
+        topic.grant_publish(monitor_lambda)
+        monitor_lambda.add_to_role_policy(
+            iam.PolicyStatement(
+                actions=["bedrock:GetFoundationModel", "bedrock:GetInferenceProfile"],
+                resources=[
+                    "arn:aws:bedrock:*::foundation-model/*",
+                    f"arn:aws:bedrock:*:{Stack.of(self).account}:inference-profile/*",
+                ],
+            )
+        )
+        monitor_lambda.add_to_role_policy(
+            iam.PolicyStatement(
+                actions=["ssm:GetParameter", "ssm:PutParameter"],
+                resources=[
+                    f"arn:aws:ssm:{Stack.of(self).region}:"
+                    f"{Stack.of(self).account}:parameter{STATUS_PARAM}"
+                ],
+            )
+        )
+
+        #################################################################################
+        # WEEKLY SCHEDULE
+        #################################################################################
+        # EventBridge Scheduler is timezone-aware, so 8am Eastern stays 8am
+        # Eastern across DST transitions. Legacy lead time is a minimum of 6
+        # months, so weekly (not daily) is plenty.
+        scheduler.Schedule(
+            self,
+            "WeeklyModelLifecycleSchedule",
+            schedule=scheduler.ScheduleExpression.cron(
+                minute="0",
+                hour="8",
+                week_day="MON",
+                time_zone=TimeZone.AMERICA_NEW_YORK,
+            ),
+            target=scheduler_targets.LambdaInvoke(
+                monitor_lambda, input=scheduler.ScheduleTargetInput.from_object({})
+            ),
+            description="Weekly Bedrock model lifecycle check (Mondays 8am ET)",
+        )
+
+        CfnOutput(
+            self,
+            "ModelLifecycleFunctionName",
+            value=monitor_lambda.function_name,
+            description="Invoke with an empty payload to check now instead of waiting for Monday",
+        )
+        CfnOutput(
+            self,
+            "ModelLifecycleTopicArn",
+            value=topic.topic_arn,
+            description="SNS topic for model lifecycle alerts (subscription must be confirmed)",
+        )

config.yaml.example

@@ -82,6 +82,17 @@ skip_existing_s3_files: true
 #   - programme-lead@example.edu
 # How long the emailed download link stays valid, in days (max 7).
 # export_url_expiry_days: 7
+
+# Who receives Bedrock model lifecycle alerts (ACTIVE -> LEGACY, etc.)
+# (optional). Each recipient confirms their own SNS subscription once, from
+# the email AWS sends them. Deliberately separate from notification_email so
+# adding names here doesn't fire a confirmation email at the content-sync
+# list. Unset means nobody is emailed. Accepts one address:
+# model_lifecycle_notification_email: you@example.edu
+# ...or several:
+# model_lifecycle_notification_email:
+#   - you@example.edu
+#   - teammate@example.edu
 # How long each generated file is kept in S3, in days. Unset keeps every export
 # indefinitely, so the console links in old emails never go stale.
 # export_retention_days: 90

src/model_lifecycle/monitor.py

@@ -0,0 +1,163 @@
+"""
+Bedrock model lifecycle monitor for ABE.
+
+Runs Mondays at 8:00 AM Eastern (EventBridge Scheduler). Each run:
+
+  1. resolves every configured model to its underlying foundation model
+     id(s) - bare ids and foundation-model ARNs go straight to
+     GetFoundationModel; inference-profile ARNs are resolved first via
+     GetInferenceProfile.
+  2. reads modelLifecycle.status for each (ACTIVE / LEGACY / ...; missing
+     means the model doesn't publish lifecycle data, e.g. some Marketplace
+     models - treated as UNKNOWN rather than a hard failure).
+  3. diffs against the last-seen status map in SSM Parameter Store.
+  4. on the very first run there is nothing to diff against, so the current
+     map is just stored as the baseline - otherwise every model would read
+     as "changed" (unknown -> ACTIVE) on day one, which is noise, not signal.
+  5. any other run: publishes one SNS message per model whose status
+     changed, then stores the new map.
+
+The API never returns the actual Legacy/EOL calendar dates (AWS only
+publishes those on the model-lifecycle docs page), so alerts link there for
+a human to check exact dates rather than guessing.
+"""
+import json
+import os
+
+import boto3
+
+bedrock = boto3.client("bedrock")
+ssm = boto3.client("ssm")
+sns = boto3.client("sns")
+
+SNS_TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]
+STATUS_PARAM = os.environ["STATUS_PARAM"]
+REGION = os.environ["AWS_REGION"]
+
+MODELS = {
+    "chat": os.environ["CHAT_MODEL_ID"],
+    "embedding": os.environ["EMBEDDING_MODEL_ID"],
+    "video_ingest": os.environ["VIDEO_TEXT_MODEL_ID"],
+    "classifier": os.environ["CLASSIFIER_MODEL_ID"],
+    "document_filter": os.environ["DOCUMENT_FILTER_MODEL_ID"],
+}
+
+DOCS_URL = "https://docs.aws.amazon.com/bedrock/latest/userguide/model-lifecycle.html"
+
+
+def _resolve_foundation_model_id(identifier):
+    """Bare model ids and foundation-model ARNs go straight through;
+    inference-profile ARNs resolve to an underlying foundation model ARN
+    first, since GetFoundationModel rejects inference-profile ARNs.
+
+    A profile's models[] list mixes a region-less ARN (arn:...bedrock:::...)
+    with one qualified for each supported region - GetFoundationModel
+    rejects an ARN whose region doesn't match the client's, so the
+    region-qualified entry has to be picked explicitly rather than just
+    taking models[0]."""
+    if ":inference-profile/" not in identifier:
+        return identifier
+    profile = bedrock.get_inference_profile(inferenceProfileIdentifier=identifier)
+    model_arns = [model["modelArn"] for model in profile["models"]]
+    for arn in model_arns:
+        if f":{REGION}::" in arn:
+            return arn
+    return model_arns[0]
+
+
+def _lifecycle_status(model_identifier):
+    details = bedrock.get_foundation_model(modelIdentifier=model_identifier)[
+        "modelDetails"
+    ]
+    return details.get("modelLifecycle", {}).get("status", "UNKNOWN")
+
+
+def check_all():
+    statuses = {}
+    for key, identifier in MODELS.items():
+        foundation_id = _resolve_foundation_model_id(identifier)
+        statuses[key] = _lifecycle_status(foundation_id)
+    return statuses
+
+
+def load_last_statuses():
+    try:
+        value = ssm.get_parameter(Name=STATUS_PARAM)["Parameter"]["Value"]
+        return json.loads(value)
+    except ssm.exceptions.ParameterNotFound:
+        return {}
+    except (ValueError, TypeError) as e:
+        # A corrupt status map would silently hide a real status change, so
+        # fail loudly instead of treating it as "nothing stored yet"
+        raise RuntimeError(f"Could not read status map {STATUS_PARAM}: {e}") from e
+
+
+def save_statuses(statuses):
+    ssm.put_parameter(
+        Name=STATUS_PARAM,
+        Value=json.dumps(statuses),
+        Type="String",
+        Overwrite=True,
+        Description="Last-seen Bedrock modelLifecycle.status per config.yaml model key",
+    )
+
+
+# What each status means for someone deciding whether action is needed, and
+# whether config.yaml has to change before an exact date even shows up.
+# LEGACY's minimum is documented by AWS (see DOCS_URL) - not observed via
+# API - so it reads as a lower bound, not a guess.
+_STATUS_MEANINGS = {
+    "LEGACY": (
+        "AWS keeps a model in LEGACY for at least 6 months before fully "
+        "retiring it (EOL), but the exact retirement date is only on the "
+        "docs page below, not in the API. Plan to move this config.yaml "
+        "key to a newer model before then."
+    ),
+    "EOL": (
+        "This model has reached end-of-life. Bedrock may already be "
+        "rejecting requests to it - config.yaml needs a replacement model "
+        "for this key now, not just eventually."
+    ),
+    "ACTIVE": (
+        "This model is back on standard support - no action needed."
+    ),
+}
+_DEFAULT_MEANING = (
+    "Bedrock reported this status without a recognized meaning here - "
+    "check the docs page below."
+)
+
+
+def _publish_change(config_key, model_id, old_status, new_status):
+    subject = f"ABE model lifecycle: {config_key} is now {new_status}"[:100]
+    meaning = _STATUS_MEANINGS.get(new_status, _DEFAULT_MEANING)
+    body = (
+        f"You're getting this email because the \"{config_key}\" model ABE "
+        f"uses just changed Bedrock lifecycle status: {old_status} -> {new_status}.\n\n"
+        f"{meaning}\n\n"
+        f"config.yaml model key: {config_key}\n"
+        f"Model: {model_id}\n\n"
+        "The Bedrock API does not report exact Legacy/EOL calendar dates - "
+        "check the docs page for those:\n"
+        f"{DOCS_URL}"
+    )
+    sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=subject, Message=body)
+
+
+def handler(event, context):
+    current = check_all()
+    last = load_last_statuses()
+
+    if not last:
+        save_statuses(current)
+        return {"initialized": True, "statuses": current}
+
+    changed = {
+        key: (last.get(key), status)
+        for key, status in current.items()
+        if last.get(key) != status
+    }
+    for key, (old_status, new_status) in changed.items():
+        _publish_change(key, MODELS[key], old_status, new_status)
+    save_statuses(current)
+    return {"changed": list(changed.keys()), "statuses": current}