Migrate WO processor to Bedrock and fix comment_id collision

Switch the AI path from the Anthropic SDK to bedrock-runtime InvokeModel
on the inference profile us.anthropic.claude-haiku-4-5-20251001-v1:0
(BEDROCK_MODEL_ID env), so parsing no longer needs a provider API key or
Secrets Manager secret. The EXTRACTION_PROMPT and JSON contract are kept
byte-identical, so the AI-fallback output is unchanged. Try the new
deterministic template parser first and only call Bedrock on a
miss/invalid result.

Fix issue #23: the WorkOrderComments range key was
work_order_id#<comment_time>, so two emails on one WO with an identical
or absent comment time collided and overwrote each other. Derive a
12-hex suffix from the S3 object key alone -- deterministic, so an async
retry of the same object is byte-identical (idempotent) while distinct
emails get distinct keys -- and keep wall-clock now() out of the key
(literal 'nocomment' segment when comment_time is absent).

Also emit one CloudWatch EMF line per record (Seahaven/WorkorderIngest
ParseOutcome, dimensioned by ParseMethod/TemplateId) for parse-outcome
observability, replace the deprecated datetime.utcnow() with
datetime.now(timezone.utc), and drop the anthropic dependency.

Refs: #23

lambdas/wo/email_processor/handler.py

@@ -7,25 +7,34 @@ then writes the result to DynamoDB.
 """
 
 import email
+import hashlib
 import json
 import logging
 import os
 import re
-from datetime import datetime
+from datetime import datetime, timezone
 from email import policy
 
-import anthropic
 import boto3
 
+from template_parser import try_deterministic_parse
+
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
 s3 = boto3.client("s3")
 dynamodb = boto3.resource("dynamodb")
+bedrock = boto3.client("bedrock-runtime")
 
 WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders")
 COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments")
-ANTHROPIC_API_KEY_SECRET_ARN = os.environ.get("ANTHROPIC_API_KEY_SECRET_ARN")
+BEDROCK_MODEL_ID = os.environ.get(
+    "BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0"
+)
+
+# CloudWatch EMF namespace + metric for parse-outcome observability.
+METRIC_NAMESPACE = "Seahaven/WorkorderIngest"
+METRIC_NAME = "ParseOutcome"
 
 EXTRACTION_PROMPT = """\
 You are an email parser for a facilities maintenance work order system.
@@ -65,17 +74,6 @@ Rules:
 """
 
 
-def get_anthropic_client() -> anthropic.Anthropic:
-    """Create Anthropic client, fetching API key from Secrets Manager if configured."""
-    if ANTHROPIC_API_KEY_SECRET_ARN:
-        secrets = boto3.client("secretsmanager")
-        secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
-        api_key = secret["SecretString"]
-        return anthropic.Anthropic(api_key=api_key)
-    # Fall back to ANTHROPIC_API_KEY env var (for local testing)
-    return anthropic.Anthropic()
-
-
 def parse_raw_email(raw_bytes: bytes) -> dict:
     """Parse a raw email into subject, sender, body text."""
     msg = email.message_from_bytes(raw_bytes, policy=policy.default)
@@ -108,10 +106,8 @@ def parse_raw_email(raw_bytes: bytes) -> dict:
     }
 
 
-def extract_with_claude(email_data: dict) -> dict:
-    """Send parsed email to Claude for structured extraction."""
-    client = get_anthropic_client()
-
+def extract_with_bedrock(email_data: dict) -> dict:
+    """Send parsed email to Claude on Bedrock for structured extraction."""
     email_text = (
         f"Subject: {email_data['subject']}\n"
         f"From: {email_data['sender']}\n"
@@ -122,18 +118,22 @@ def extract_with_claude(email_data: dict) -> dict:
         f"{email_data['body']}"
     )
 
-    response = client.messages.create(
-        model="claude-haiku-4-5-20251001",
-        max_tokens=1024,
-        messages=[
+    resp = bedrock.invoke_model(
+        modelId=BEDROCK_MODEL_ID,
+        body=json.dumps(
             {
-                "role": "user",
-                "content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}",
+                "anthropic_version": "bedrock-2023-05-31",
+                "max_tokens": 1024,
+                "messages": [
+                    {
+                        "role": "user",
+                        "content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}",
+                    }
+                ],
             }
-        ],
+        ),
     )
-
-    response_text = response.content[0].text
+    response_text = json.loads(resp["body"].read())["content"][0]["text"]
 
     # Extract JSON from response (handle markdown code blocks)
     json_match = re.search(r"```(?:json)?\s*(.*?)```", response_text, re.DOTALL)
@@ -143,11 +143,37 @@ def extract_with_claude(email_data: dict) -> dict:
     return json.loads(response_text.strip())
 
 
+def emit_parse_metric(method, template_id, reason_code, work_order_id):
+    """Emit one CloudWatch EMF line recording the parse outcome.
+
+    Zero-latency (no PutMetricData API call): the extraction path is async and
+    the role already has logs:PutLogEvents. ParseMethod/TemplateId are the only
+    promoted (dimensioned) fields to keep cardinality low; ReasonCode and
+    work_order_id ride along as Logs-Insights-queryable properties."""
+    emf = {
+        "_aws": {
+            "CloudWatchMetrics": [
+                {
+                    "Namespace": METRIC_NAMESPACE,
+                    "Dimensions": [["ParseMethod", "TemplateId"]],
+                    "Metrics": [{"Name": METRIC_NAME, "Unit": "Count"}],
+                }
+            ],
+        },
+        "ParseMethod": method,
+        "TemplateId": template_id or "unknown",
+        "ReasonCode": reason_code or "ok",
+        "work_order_id": work_order_id or "",
+        METRIC_NAME: 1,
+    }
+    print(json.dumps(emf))
+
+
 def save_work_order(parsed: dict, s3_key: str):
     """Create or update a work order in DynamoDB."""
     table = dynamodb.Table(WORK_ORDERS_TABLE)
     work_order_id = parsed["work_order_id"]
-    now = datetime.utcnow().isoformat()
+    now = datetime.now(timezone.utc).isoformat()
 
     # Build update expression dynamically from non-null fields
     field_map = {
@@ -210,14 +236,27 @@ def save_work_order(parsed: dict, s3_key: str):
     logger.info(f"Saved work order {work_order_id}")
 
 
-def save_event(parsed: dict, s3_key: str):
-    """Save an event to the events table. Every email creates an event entry."""
+def save_event(parsed: dict, s3_key: str, object_key: str):
+    """Save an event to the events table. Every email creates an event entry.
+
+    Issue #23: the range key (attr name stays ``comment_id``) must be unique per
+    source email AND identical across Lambda async retries of the same S3 object.
+    The uniqueness suffix is a deterministic hash of the S3 object key alone (not
+    the s3:// URI, so it is stable across a bucket rename), and wall-clock now()
+    is kept OUT of the key. When comment_time is absent we use the literal
+    'nocomment' segment rather than now() -- otherwise each retry would produce a
+    different key and duplicate the row. Two distinct emails on the same WO map
+    to distinct object keys -> distinct rows."""
     table = dynamodb.Table(COMMENTS_TABLE)
 
     work_order_id = parsed["work_order_id"]
     email_type = parsed.get("email_type", "unknown")
-    event_time = parsed.get("comment_time") or datetime.utcnow().isoformat()
-    event_id = f"{work_order_id}#{event_time}"
+    key_suffix = hashlib.sha256(object_key.encode("utf-8")).hexdigest()[:12]
+    comment_time = parsed.get("comment_time")
+    time_part = comment_time if comment_time else "nocomment"
+    event_id = f"{work_order_id}#{time_part}#{key_suffix}"
+    # created_at is display-only; may fall back to now() without affecting the key.
+    display_time = comment_time or datetime.now(timezone.utc).isoformat()
 
     item = {
         "work_order_id": work_order_id,
@@ -225,9 +264,9 @@ def save_event(parsed: dict, s3_key: str):
         "record_type": email_type,
         "commenter": parsed.get("commenter") or "",
         "text": parsed.get("comment_text") or "",
-        "created_at": event_time,
+        "created_at": display_time,
         "source_email_s3_key": s3_key,
-        "ingested_at": datetime.utcnow().isoformat(),
+        "ingested_at": datetime.now(timezone.utc).isoformat(),
     }
 
     table.put_item(Item=item)
@@ -250,12 +289,19 @@ def handler(event, context):
         email_data = parse_raw_email(raw_email)
         logger.info(f"Subject: {email_data['subject']}")
 
-        # Extract structured data with Claude
-        parsed = extract_with_claude(email_data)
+        # Deterministic template parse first; fall back to the AI extractor only
+        # on a miss or an invalid (fail-closed) result.
+        parsed, method, template_id, reason = try_deterministic_parse(email_data)
+        if parsed is None:
+            parsed = extract_with_bedrock(email_data)
+            method = "ai_fallback"
         logger.info(
-            f"Parsed: type={parsed.get('email_type')}, wo={parsed.get('work_order_id')}"
+            f"Parsed ({method}/{template_id}/{reason}): "
+            f"type={parsed.get('email_type')}, wo={parsed.get('work_order_id')}"
         )
 
+        emit_parse_metric(method, template_id, reason, parsed.get("work_order_id"))
+
         if not parsed.get("work_order_id"):
             logger.warning(f"No work order ID found in email, skipping: {key}")
             continue
@@ -265,7 +311,8 @@ def handler(event, context):
         # Always upsert the work order with any new info
         save_work_order(parsed, s3_key)
 
-        # Save every email as an event for history tracking
-        save_event(parsed, s3_key)
+        # Save every email as an event for history tracking. The raw object key
+        # (not the s3:// URI) drives the retry-idempotent comment_id suffix.
+        save_event(parsed, s3_key, key)
 
     return {"statusCode": 200, "body": "OK"}

lambdas/wo/email_processor/requirements.txt

@@ -1,2 +1 @@
-anthropic>=0.116.0
 boto3>=1.43.47