From 3418c7c819ecc6aacadc4ed979c8240a094715aa Mon Sep 17 00:00:00 2001
From: Adam Moussa <adam@seahavenind.com>
Date: Thu, 16 Jul 2026 15:14:58 -0400
Subject: [PATCH] Wire template-first parse into PO handler with EMF metric

Run try_deterministic_parse ahead of the Bedrock extractor and
fall back only on a miss/invalid (fail-closed) result. The shared
enrich_parsed post-stage and the save_cancellation/save_revision/
save_new_po routing are untouched, so both paths write identical
DynamoDB shapes and the po-ingest-site-extractor stream contract
is preserved.

Each record emits one ParseMethod EMF line (Seahaven/PoIngest/
ParseOutcome, dimension sets [ParseMethod] and
[ParseMethod,TemplateId], ReasonCode/po_number ride-alongs)
mirroring the WO idiom. The metric fires before the Bedrock call
so a Bedrock-side error still records the ai_fallback outcome.

Refs: #99
---
 lambdas/po/email_processor/handler.py | 65 +++++++++++++++++++++++++--
 1 file changed, 62 insertions(+), 3 deletions(-)

diff --git a/lambdas/po/email_processor/handler.py b/lambdas/po/email_processor/handler.py
index c8a85f8..5fa766e 100644
--- a/lambdas/po/email_processor/handler.py
+++ b/lambdas/po/email_processor/handler.py
@@ -2,7 +2,8 @@
 PO email processor Lambda.
 
 Triggered by S3 events when SES delivers a Coupa PO email.
-Parses the raw email, sends it to Claude for structured extraction,
+Parses the raw email, tries the deterministic template parser first, falls
+back to Claude on Bedrock for structured extraction on a miss/invalid result,
 then writes the result to the purchase-orders DynamoDB table.
 """
 
@@ -17,6 +18,7 @@ from email import policy
 
 import boto3
 from ses_auth import authenticate_inbound_email
+from template_parser import try_deterministic_parse
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -30,6 +32,11 @@ BEDROCK_MODEL_ID = os.environ.get(
     "BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0"
 )
 
+# CloudWatch EMF namespace/metric for the parse-outcome metric (the PO
+# fallback-rate alarm in cdk/po_stack.py reads the ["ParseMethod"] series).
+METRIC_NAMESPACE = "Seahaven/PoIngest"
+METRIC_NAME = "ParseOutcome"
+
 # "Cancelled" is a sticky, authoritative status: once a PO reaches it, a later
 # new_po/revision may enrich other fields but must never move it back to a
 # non-cancelled status.
@@ -285,6 +292,42 @@ def extract_with_claude(email_data: dict) -> dict:
     return json.loads(response_text.strip(), parse_float=Decimal)
 
 
+def _emit_parse_method_metric(method, template_id, reason_code, po_number):
+    """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
+    po_number ride along as Logs-Insights-queryable properties.
+
+    Two dimension sets are published: ["ParseMethod"] (aggregated across all
+    template ids -- the series the fallback-rate alarm queries) AND
+    ["ParseMethod", "TemplateId"] (per-template breakdown for Logs Insights /
+    dashboards). CloudWatch materializes only the exact dimension sets listed
+    here and does NOT auto-aggregate, so the alarm's single-dimension query
+    would receive no data unless ["ParseMethod"] is emitted explicitly."""
+    emf = {
+        "_aws": {
+            # EMF requires Timestamp (epoch ms); without it CloudWatch may not
+            # extract the metric datapoint from the log event.
+            "Timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
+            "CloudWatchMetrics": [
+                {
+                    "Namespace": METRIC_NAMESPACE,
+                    "Dimensions": [["ParseMethod"], ["ParseMethod", "TemplateId"]],
+                    "Metrics": [{"Name": METRIC_NAME, "Unit": "Count"}],
+                }
+            ],
+        },
+        "ParseMethod": method,
+        "TemplateId": template_id or "unknown",
+        "ReasonCode": reason_code or "ok",
+        "po_number": po_number or "",
+        METRIC_NAME: 1,
+    }
+    print(json.dumps(emf))
+
+
 def pad_zip(zip_code: str | None) -> str | None:
     if not zip_code:
         return zip_code
@@ -477,9 +520,25 @@ def handler(event, context):
         email_data = parse_raw_email(raw_email)
         logger.info(f"Subject: {email_data['subject']}")
 
-        parsed = extract_with_claude(email_data)
+        # Deterministic template parse first; fall back to the Bedrock AI
+        # extractor only on a miss or an invalid (fail-closed) result. The
+        # metric is emitted before the Bedrock call, so a Bedrock-side error
+        # still records the ai_fallback outcome (the invocation then errors
+        # into the errors alarm / DLQ as before).
+        parsed, parse_method, template_id, parse_reason = try_deterministic_parse(
+            email_data
+        )
+        _emit_parse_method_metric(
+            parse_method,
+            template_id,
+            parse_reason,
+            parsed.get("po_number") if parsed else None,
+        )
+        if parsed is None:
+            parsed = extract_with_claude(email_data)
         logger.info(
-            f"Parsed: type={parsed.get('email_type')}, po={parsed.get('po_number')}"
+            f"Parsed ({parse_method}/{template_id}/{parse_reason}): "
+            f"type={parsed.get('email_type')}, po={parsed.get('po_number')}"
         )
 
         if not parsed.get("po_number"):