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

@@ -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 `<qty> <UNIT> x <price>` line (e.g. `1.0 EACH x 55,206.00`); the Lines-block `<qty> 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/<id> == 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.
 
 ---

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
 
 

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<value>.*(?:\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<tok>[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))