procurement-ingest/lambdas/po/email_processor/handler.py

"""
PO email processor Lambda.

Triggered by S3 events when SES delivers a Coupa PO email.
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.

The concerns are split across flat sibling modules (all bundled into the same
Lambda asset, so bare-name imports resolve):
  * extraction.py  -- extract_with_claude + the Bedrock client
  * enrichment.py  -- enrich_parsed + pad_zip + derived-field classification
  * telemetry.py   -- the EMF ParseMethod / derived-agreement emit wrappers
  * persistence.py -- the purchase-orders merge-writes (save_*)
  * prompts.py     -- EXTRACTION_PROMPT (re-exported below for tests)
This module keeps the S3 event loop, fail-closed sender auth, and email_type
routing.
"""

import logging

import boto3
from email_parsing import parse_raw_email
from enrichment import enrich_parsed
from extraction import extract_with_claude
from persistence import save_cancellation, save_new_po, save_revision

# EXTRACTION_PROMPT is re-exported so handler.EXTRACTION_PROMPT still resolves
# for the tests that dereference it as a module attribute.
from prompts import EXTRACTION_PROMPT  # noqa: F401
from ses_auth import authenticate_inbound_email
from telemetry import _emit_parse_method_metric
from template_parser import try_deterministic_parse, validate_ai_fallback

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Lazily-built, cached S3 client. Public name ``s3`` is preserved so the
# monkeypatch attribute is unchanged; building at first CALL (not import) keeps
# the moto-before-handler invariant and honors any patched fake.
s3 = None


def _get_s3():
    global s3
    if s3 is None:
        s3 = boto3.client("s3")
    return s3


def handler(event, context):
    """Lambda entry point. Triggered by S3 ObjectCreated events."""
    # Direct-invoke healthcheck (post-deploy smoke). This MUST be the very first
    # thing handler() does -- before any S3 fetch, before ses_auth, before the
    # Records loop -- so it (a) creates no accept path for mail (real mail is an
    # S3 ObjectCreated event whose top-level keys AWS controls; email content
    # can never set a top-level "healthcheck" key), and (b) emits no EMF metric
    # and no log line that could match the sender_auth_rejected metric-filter
    # pattern, so two deploys in ~30 min never page that alarm.
    if isinstance(event, dict) and event.get("healthcheck") is True:
        return {"healthcheck": "ok"}

    for record in event.get("Records", []):
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]
        s3_key = f"s3://{bucket}/{key}"

        logger.info(f"Processing email: {s3_key}")

        response = _get_s3().get_object(Bucket=bucket, Key=key)
        raw_email = response["Body"].read()

        # Fail-closed sender authentication (INFRA-107): only mail with an
        # SES-stamped dkim=pass verdict for an allowlisted domain may create
        # or update purchase orders. Rejected mail is logged and skipped
        # without erroring the invocation (no retries / DLQ spam).
        if not authenticate_inbound_email(raw_email, s3_key):
            continue

        email_data = parse_raw_email(raw_email)
        logger.info(f"Subject: {email_data['subject']}")

        # 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)
            # Fail-closed gate on the raw model output (#104 parity): runs
            # BEFORE enrich_parsed and BEFORE any dispatch/save. INTENTIONAL
            # DOUBLE-COUNT: ParseMethod=ai_fallback was already emitted above,
            # BEFORE the Bedrock call (deliberate -- a Bedrock-side error must
            # still record the outcome), so a rejected email produces BOTH an
            # ai_fallback and an ai_fallback_rejected datapoint. The po_stack
            # fallback-rate alarm therefore EXCLUDES the rejected series from
            # its rate math (fb already counts these emails once); see
            # cdk/po_stack.py and README.
            ok, val_reason, normalized = validate_ai_fallback(parsed)
            if not ok:
                logger.warning(
                    f"AI-fallback output rejected by validation gate "
                    f"({val_reason}); skipping: {key}"
                )
                _emit_parse_method_metric(
                    "ai_fallback_rejected",
                    template_id,
                    val_reason,
                    str(parsed.get("po_number"))[:64]
                    if isinstance(parsed, dict) and parsed.get("po_number")
                    else None,
                )
                # Skip, never raise: attacker-controlled input must not churn
                # the retry/DLQ path.
                continue
            parsed = normalized
        logger.info(
            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"):
            logger.warning(f"No PO number found in email, skipping: {key}")
            continue

        parsed = enrich_parsed(
            parsed, s3_key, email_data["subject"], parse_method=parse_method
        )

        email_type = parsed.get("email_type")
        if email_type == "cancellation":
            save_cancellation(parsed)
        elif email_type == "revision":
            save_revision(parsed)
        else:
            save_new_po(parsed)

    return {"statusCode": 200, "body": "OK"}