From 3dfe15b4b12cc7093fe414b423d6a54e16125785 Mon Sep 17 00:00:00 2001 From: Adam Moussa Date: Thu, 16 Jul 2026 16:03:43 -0400 Subject: [PATCH] Converge quantity/price to Decimal on both paths EXTRACTION_PROMPT declares quantity and price as JSON strings, so a prompt-obedient Bedrock response stores DynamoDB Strings where the template parser stores Numbers -- divergent attribute types for the same email on the purchase-orders stream. Coerce numeric strings to Decimal in the shared enrich_parsed post-stage (thousands-separator safe; non-numeric strings kept verbatim) so both paths converge; prompt rewording itself remains PR #2 scope. The two-path parity test was circular -- it replayed the parser- derived golden as 'the LLM output', so it could never see the type divergence. It now feeds a prompt-shaped payload (string quantity/ price, LLM-filled site_code) through enrich_parsed and save_new_po, and the fixture-hygiene test now asserts the scrubbed transport-token header classes so fixture regressions are caught. --- docs/po-template-parser.md | 3 +- lambdas/po/email_processor/handler.py | 22 +++- .../tests/test_po_bedrock_fallback.py | 109 +++++++++++++++++- 3 files changed, 128 insertions(+), 6 deletions(-) diff --git a/docs/po-template-parser.md b/docs/po-template-parser.md index 98cbc06..0796793 100644 --- a/docs/po-template-parser.md +++ b/docs/po-template-parser.md @@ -204,7 +204,7 @@ Plus `missing_required_field` for the required labeled fields (per-item metadata - [x] `ParseMethod` EMF metric set (`Seahaven/PoIngest`/`ParseOutcome`, dims `[["ParseMethod"],["ParseMethod","TemplateId"]]`, `ReasonCode`/`po_number` ride-alongs), emitted before the Bedrock call. - [x] Fallback-rate alarm in `cdk/po_stack.py`, retuned for ~57/day (6h periods · ≥8 volume floor · >20% · 2-of-4) — `npx cdk synth po-ingest` green. - [x] `cdk/po_stack.py` bundling `cp` now includes `template_parser.py` (was a deploy-time ImportError waiting to happen). - - [x] Offline test suite `lambdas/po/email_processor/tests/` (132 tests): goldens for all 25 positive fixtures, every §6 reason code covered, dual line-ending identity, two-path enrich/save parity, fixture hygiene (ses_auth + leak-sweep markers), Bedrock dispatch/EMF. Full-root pytest: 354 green. + - [x] Offline test suite `lambdas/po/email_processor/tests/` (136 tests): goldens for all 25 positive fixtures, every §6 reason code covered, dual line-ending identity, two-path enrich/save parity, fixture hygiene (ses_auth + leak-sweep markers), Bedrock dispatch/EMF. Full-root pytest: 358 green. - [x] Committed sanitized `.eml` fixture corpus (header + body scrub, per-file digit cipher, narrow `.gitignore` exception scoped to `lambdas/po/email_processor/tests/fixtures/`). - [x] README updated (PO flow, parser section, alarm numbers + justification, tests, repo layout). @@ -244,6 +244,7 @@ Plus `missing_required_field` for the required labeled fields (per-item metadata - **`ship_to.street`/`ship_to.address` join convention:** multi-line segments joined with `'\n'`; `address` = the lines from `ship_to.name` through `United States` inclusive (this is what `enrich_parsed` promotes to `ship_to_raw` on the site-extractor stream). - **`quantity`/`unit`/`price` source:** the Items-summary ` x ` line (e.g. `1.0 EACH x 55,206.00`); the Lines-block ` EA` line is gate evidence only (V13 numeric cross-check). Emails without a summary line (e.g. new-po-14) leave all three `null` — nullable by contract. - **V10 URL-id sweep outcome:** all 20 harvested new_po emails satisfy `orders/ == po_number` digit suffix — the strict V10 check stays live (no demotion needed); a corpus-sweep test locks it. +- **Canonical `quantity`/`price` type = Decimal (DynamoDB Number), converged in the shared `enrich_parsed`:** `EXTRACTION_PROMPT` declares both fields as JSON strings, so a prompt-obedient Bedrock response arrives as `str` where the template path emits `Decimal`. The shared post-stage coerces numeric strings (thousands-separator-safe) to `Decimal` so both paths write the same attribute type to the purchase-orders table stream; non-numeric strings are left verbatim. Prompt rewording itself stays PR #2 scope. The two-path parity test feeds a prompt-shaped payload (string quantity/price, LLM-filled `site_code`) — never the parser-derived golden verbatim — so it cannot be circular. - **Second-pass header scrub (post-review):** the first-pass harvest scrub left the real SES `Feedback-ID` sender-identity hash on every Coupa fixture and, on the two non-Coupa fixtures, an embedded second SES block's `X-Ses-Receipt`, the Exchange cross-tenant UPN ciphertext, Gmail ARC `fh=` / `X-Gm-*` tokens, and related opaque routing blobs. All replaced with same-shape `ScrubbedFixture` placeholders (byte-safe, CRLF preserved); the fixture-hygiene test now asserts these token classes are scrubbed in every header block so regressions are caught. --- diff --git a/lambdas/po/email_processor/handler.py b/lambdas/po/email_processor/handler.py index 5fa766e..58d8fda 100644 --- a/lambdas/po/email_processor/handler.py +++ b/lambdas/po/email_processor/handler.py @@ -13,7 +13,7 @@ import logging import os import re from datetime import datetime, timezone -from decimal import Decimal +from decimal import Decimal, InvalidOperation from email import policy import boto3 @@ -353,6 +353,26 @@ def enrich_parsed(parsed: dict, s3_key: str, email_subject: str): if ship_to.get("zip"): ship_to["zip"] = pad_zip(ship_to["zip"]) + + # Canonical numeric type for quantity/price across BOTH parse paths: the + # template parser emits Decimal (DynamoDB Number) while EXTRACTION_PROMPT + # asks the LLM for these two fields as JSON strings (which the Bedrock + # json.loads leaves as str -> DynamoDB String). Coercing here -- in the + # SHARED post-stage -- converges the attribute type to Number for + # equivalent parsed dicts, preserving the two-path parity contract on the + # purchase-orders table stream. Prompt rewording itself is PR #2 scope. + # A non-numeric string is left verbatim (still stored, as a String) -- + # dropping it would lose LLM-extracted evidence. + for item in parsed.get("line_items") or []: + if not isinstance(item, dict): + continue + for field in ("quantity", "price"): + value = item.get(field) + if isinstance(value, str): + try: + item[field] = Decimal(value.replace(",", "").strip()) + except InvalidOperation: + pass return parsed diff --git a/lambdas/po/email_processor/tests/test_po_bedrock_fallback.py b/lambdas/po/email_processor/tests/test_po_bedrock_fallback.py index c046d31..8e01be4 100644 --- a/lambdas/po/email_processor/tests/test_po_bedrock_fallback.py +++ b/lambdas/po/email_processor/tests/test_po_bedrock_fallback.py @@ -5,6 +5,7 @@ shapes for both paths, and the save_* routing is untouched.""" import copy import json +import re import pytest @@ -201,7 +202,14 @@ def test_two_path_parity_through_enrich_and_save(fake_dynamo): LLM-shaped dict for the same email must produce byte-identical DynamoDB items after the shared enrich_parsed post-stage and save_new_po's None-dropping logic (the po-ingest-site-extractor reads site_code/ship_to - off the table stream).""" + off the table stream). + + NON-CIRCULAR: the LLM side is a PROMPT-SHAPED payload, not the + parser-derived golden verbatim -- EXTRACTION_PROMPT declares quantity and + price as JSON strings, so a prompt-obedient Bedrock response arrives with + str where the template path has Decimal. The shared enrich_parsed must + converge both to Decimal (DynamoDB Number) or the two paths write + different attribute types for the same email.""" stem = NEW_PO_STEMS[0] email_data = load_email("new-po", stem) s3_key = "s3://po-ingest-emails-x/inbound/parity" @@ -212,17 +220,47 @@ def test_two_path_parity_through_enrich_and_save(fake_dynamo): email_data ) assert reason == "ok" - # The golden IS the equivalent LLM output: same contract, loaded through - # the same parse_float=Decimal decode the Bedrock path uses. llm_parsed = copy.deepcopy(load_golden(stem)) + # Reshape to what a prompt-obedient LLM actually returns: quantity/price + # as strings ('"quantity": "string or null"'), price in the email's own + # thousands-separated rendering. + for item in llm_parsed["line_items"]: + if item["quantity"] is not None: + item["quantity"] = str(item["quantity"]) + if item["price"] is not None: + item["price"] = format(item["price"], ",.2f") + assert isinstance(llm_parsed["line_items"][0]["quantity"], str) + # Derived classification (site_code/trade/fiscal_year) stays LLM-only in + # PR #1: the LLM fills it, the template parser leaves it None (doc §3.6). + llm_parsed["site_code"] = "DYO1" enriched_a = po_handler.enrich_parsed( template_parsed, s3_key, email_data["subject"] ) enriched_b = po_handler.enrich_parsed(llm_parsed, s3_key, email_data["subject"]) enriched_b["processed_at"] = enriched_a["processed_at"] # only wall-clock differs - assert enriched_a == enriched_b + # enrich_parsed passes the LLM-derived site_code through untouched... + assert enriched_a["site_code"] is None + assert enriched_b["site_code"] == "DYO1" + # ...and converges quantity/price to Decimal on the prompt-shaped side. + from decimal import Decimal + + item_a = enriched_a["line_items"][0] + item_b = enriched_b["line_items"][0] + if item_a["quantity"] is not None: + assert isinstance(item_a["quantity"], Decimal) + assert isinstance(item_b["quantity"], Decimal) + assert isinstance(item_b["price"], Decimal) + # Everything except the deliberately-LLM-only derived fields is identical. + derived = set(template_parser.DERIVED_KEYS) + parity_a = {k: v for k, v in enriched_a.items() if k not in derived} + parity_b = {k: v for k, v in enriched_b.items() if k not in derived} + assert parity_a == parity_b + + # Like-for-like DynamoDB item shapes through save_new_po's None-dropping + # logic (derived fields normalized so the comparison isolates types). + enriched_b["site_code"] = None po_handler.save_new_po(enriched_a) po_handler.save_new_po(enriched_b) update_a, update_b = fake_dynamo.tables[po_handler.PO_TABLE].updates @@ -240,6 +278,64 @@ def test_enrich_parsed_pads_short_zip_identically_for_llm_path(): assert parsed["state"] == "MA" +def test_enrich_parsed_coerces_prompt_string_quantity_price(): + """EXTRACTION_PROMPT asks for quantity/price as strings; the shared + enrich_parsed coerces them to Decimal so the LLM path writes the same + DynamoDB attribute type (Number) as the template path. Non-numeric + strings are left verbatim rather than dropped.""" + from decimal import Decimal + + parsed = { + "line_items": [ + {"quantity": "1.0", "price": "55,206.00"}, + {"quantity": "N/A", "price": None}, + ] + } + po_handler.enrich_parsed(parsed, "s3://b/k", "subj") + assert parsed["line_items"][0]["quantity"] == Decimal("1.0") + assert parsed["line_items"][0]["price"] == Decimal("55206.00") + assert parsed["line_items"][1]["quantity"] == "N/A" # left verbatim + assert parsed["line_items"][1]["price"] is None + + +# Opaque-token header classes the harvest scrub must have replaced with +# same-shape ScrubbedFixture placeholders -- in EVERY header block of EVERY +# fixture, including embedded/second SES blocks and forwarded-mail headers +# (the PR #99 lesson: the first WO harvest scrubbed only the primary block). +_TOKEN_HEADER_RE = re.compile( + rb"^(?:X-Ses-Receipt|X-SES-RECEIPT|Feedback-ID|X-Gm-Gg|X-Gm-Message-State" + rb"|X-Gm-Features|X-Ms-Exchange-Crosstenant-Userprincipalname" + rb"|X-Forwarded-Encrypted|X-Exchange-Routingpolicychecked" + rb"|X-Cse-Connectionguid|X-Cse-Msgguid|X-Received):" + rb"(?P.*(?:\r?\n[ \t].*)*)", + re.IGNORECASE | re.MULTILINE, +) +# Long base64-ish runs inside those header values (and in ARC fh= tags) must +# carry the scrub marker; anything else is a real transport/identity token. +_TOKEN_RUN_RE = re.compile(rb"[A-Za-z0-9+/=_-]{24,}") +_ARC_FH_RE = re.compile(rb"fh=(?P[A-Za-z0-9+/=]+);") + + +def _unfold_header_value(value): + """Unfold a (possibly RFC2047-chunked) header value so placeholder runs + that the scrub split across folded encoded-words reassemble before the + token-run scan (otherwise a chunk boundary can split the marker itself).""" + value = re.sub(rb"\r?\n[ \t]+", b"", value) # unfold continuations + value = re.sub(rb"\?=\s*=\?utf-8\?q\?", b"", value, flags=re.I) # join words + value = re.sub(rb"=\?utf-8\?q\?", b"", value, flags=re.I) + value = value.replace(b"?=", b"") + return value + + +def _assert_token_headers_scrubbed(raw, where): + for header in _TOKEN_HEADER_RE.finditer(raw): + value = _unfold_header_value(header.group("value")) + for run in _TOKEN_RUN_RE.findall(value): + assert b"ScrubbedFixture" in run, (where, run[:60]) + for m in _ARC_FH_RE.finditer(raw): + assert b"ScrubbedFixture" in m.group("tok"), (where, m.group("tok")[:60]) + + def test_fixture_hygiene_ses_auth_and_scrub_markers(): """Every committed Coupa .eml must pass the fail-closed sender-auth gate (dkim=pass for amazon.coupahost.com) exactly like production mail, and @@ -271,6 +367,7 @@ def test_fixture_hygiene_ses_auth_and_scrub_markers(): # Leak-sweep markers: the harvest scrub replaced DKIM b=/bh= values and # SES receipt tokens with same-shape placeholders. assert b"ScrubbedFixture" in raw, (subdir, stem) + _assert_token_headers_scrubbed(raw, (subdir, stem)) # Non-Coupa fixtures must NOT authenticate (they are rejected before the # parser in production; committed here for parser-level fallback coverage). for stem in [s for s in AI_FALLBACK_STEMS if s.startswith("non-coupa")]: @@ -279,3 +376,7 @@ def test_fixture_hygiene_ses_auth_and_scrub_markers(): raw, {"amazon.coupahost.com"} ) assert not accepted, stem + # Non-Coupa fixtures carry forwarded-mail / embedded-SES header blocks + # (Exchange UPN ciphertext, Gmail ARC fh= / X-Gm-* tokens, a second + # X-Ses-Receipt): those token classes must be scrubbed too. + _assert_token_headers_scrubbed(raw, ("ai-fallback", stem))