feat: Python derived-field classifier with shadow telemetry (PO PR 2) (#106)

* feat: Python derived-field classifier with shadow telemetry for PO ingest

Port the site_code/trade/fiscal_year rules from EXTRACTION_PROMPT into a
pure, total derived_fields module applied in the shared enrich_parsed()
post-stage. Python fills gaps on both parse paths (the template path has
no LLM values, closing the derived-field gap opened by the PR #105
two-PR split) and never overwrites a non-null LLM value; on ai_fallback
a DerivedFieldAgreement EMF record per field shadows Python against the
LLM during the bake. Rules hardened against a full-corpus backtest
(3,422 real emails vs the LLM-written baseline): site_code 99.4% with
zero Python-wrong cases, fiscal_year 100%, trade 96.8% ex-deliberate.
Also: quantity/price now declared numeric in the prompt, and
derived_fields.py added to the po_stack bundling copy (deploy-time
ImportError otherwise).

* fix: security-review hardening — EMF value length clamp, aggregate trade CPU budget

sh-security-review (4 detectors + proof-or-kill verifier): PASS, 0
confirmed critical/high. Fixes the one confirmed low (unbounded
LLM-value str() into the DerivedFieldAgreement EMF log line, clamped to
64 chars) and adds the verifier-recommended defense-in-depth aggregate
character budget across line items in derive_trade (per-item caps alone
allowed ~10s full-core on a pathological direct-call input; unreachable
through the deployed handler but cheap to bound). Bundling cp list now
carries a warning comment (GPT-4.1 cross-review FIX).

README.md

@@ -17,7 +17,7 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed **determini
 2. SES (`INBOUND_MAIL` rule set) drops the raw MIME into `s3://po-ingest-emails-{AccountId}/inbound/`.
 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda.
 4. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for `amazon.coupahost.com` (see [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped.
-5. **Parse:** a pure, offline template parser (`template_parser.py`) tries the two known Coupa templates first — `coupa_new_po` (95.5% of traffic) and `coupa_cancellation` (2.9%) — behind a strict fail-closed validation gate. Only on a miss/invalid result does the Lambda fall back to the Claude-on-Bedrock AI extractor (`InvokeModel`), which extracts the identical structured-JSON contract (PO number, status, supplier, nested ship-to, line items). Numeric amounts are `Decimal` on both paths (the AI decode uses `parse_float=Decimal`; DynamoDB rejects floats). The derived classification fields (`site_code`, `trade`, `fiscal_year`) stay LLM-computed for now: the template parser leaves them `null`, and the shared `enrich_parsed()` post-stage (zip padding, `ship_to_raw`/`state` promotion, metadata) runs identically on both paths so downstream shapes are byte-identical.
+5. **Parse:** a pure, offline template parser (`template_parser.py`) tries the two known Coupa templates first — `coupa_new_po` (95.5% of traffic) and `coupa_cancellation` (2.9%) — behind a strict fail-closed validation gate. Only on a miss/invalid result does the Lambda fall back to the Claude-on-Bedrock AI extractor (`InvokeModel`), which extracts the identical structured-JSON contract (PO number, status, supplier, nested ship-to, line items). Numeric amounts are `Decimal` on both paths (the AI decode uses `parse_float=Decimal`; DynamoDB rejects floats). The derived classification fields (`site_code`, `trade`, `fiscal_year`) are computed by a deterministic Python classifier in the shared `enrich_parsed()` post-stage (zip padding, `ship_to_raw`/`state` promotion, metadata) which runs identically on both paths — Python-authoritative on the template path and an LLM-authoritative-with-Python-shadow bake on the AI-fallback path (see the **Derived fields** note below).
 6. Merge write to DynamoDB:
    - `new_po` — merge insert. Creates the PO, or backfills data into a pre-existing `Cancelled` skeleton left by an out-of-order cancellation (preserving the `Cancelled` status). No longer silently dropped when a record already exists.
    - `revision` — field-level merge (`update_item` SETs only the fields present in the revision). A revision that omits `line_items`/`supplier` no longer deletes them. Will not un-cancel a `Cancelled` PO.
@@ -28,6 +28,27 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed **determini
 
 **Deterministic template parser.** `template_parser.try_deterministic_parse()` classifies by exact subject regex, extracts the nested contract (`supplier{}`, `ship_to{}` — 8 keys, `line_items[]` — 10 keys per item), and returns a parsed result **only if** it passes a fail-closed validation gate: recursive exact key-set at every nesting level; `email_type` emitted **only** on the exact new-PO subject *and* a confirmed-safe `Status` (never defaulted — a cancellation misrouted as `new_po` would defeat the sticky-`Cancelled` guard); `po_number` shape + byte-equality with the subject, the body `PO ID`, the `Amazon Purchase Order #` heading, and the `orders/<id>` URL; duplicate-label anchor integrity (`Supplier`/`Shipping`/`Total` each appear twice, the first `Shipping` must be the literal `None` placeholder); money fidelity (every `Decimal` re-serializes byte-identically to its source token with a digit/comma border check — the thousands-separator-truncation kill switch — plus `sum(line amounts) == total`, proven against **both** `Total` blocks); bullet-metadata label discipline (closed label set, assigned by leading label, never ordinal — immune to the optional `Part Number` segment); USPS address shape on the raw pre-enrichment value; and rejection of any unparseable sentinel or residual `\r`/`\xa0` artifact. Multi-line-item (0.18%) and non-USD (0 observed) new-POs, comment emails, and anything else falls back to the AI extractor. **Data is never corrupted; only the fallback rate rises.** Every record emits one CloudWatch EMF metric (see below).
 
+**Derived fields (`site_code`, `trade`, `fiscal_year`).** These three are **classified**, not verbatim-extracted, so neither parse path emits them directly — instead a deterministic Python classifier (`derived_fields.derive_all()`) computes them inside the shared `enrich_parsed()` post-stage, which runs identically on both paths. `trade` is a closed label set (the same one the AI prompt enumerates): `Plumbing - PM`, `Plumbing - Reactive`, `Electrical`, `HVAC`, `Dock Doors`, `Doors`, `Signage`, `Carpentry`, `Fencing/Gates`, `Conveyance/MHE`, `Painting`, `Flooring`, `Janitorial`, `Fire/Life Safety`, `Landscaping/Yard`, `Roofing`, `Security/Locksmith`, `Snow Removal`, `PO Uplift`, `General Building - Emergency`, `General Building - Handyman`, `General Building - Project`, `General Building`. Authority differs by path:
+
+- **Template path** — the parser leaves all three `null`, so Python is authoritative: it fills them.
+- **AI-fallback path** — the LLM value stays **authoritative during the bake** (Python fills only a gap the LLM left `null`, never overwrites), and Python additionally runs in **shadow mode**: `enrich_parsed()` emits one `DerivedFieldAgreement` EMF record per field comparing the two.
+
+**`DerivedFieldAgreement` metric** — namespace `Seahaven/PoIngest`, metric name `DerivedFieldAgreement` (Unit Count, value 1), dimensioned **only** by `Field` × `Agreement` (cardinality fixed at 3 × 4). `Agreement` ∈ {`agree` (both non-null, equal after str-strip), `disagree` (both non-null, different), `llm_null_python_filled` (LLM null, Python supplied a value), `python_null` (LLM non-null, Python null)}; the record is skipped entirely when both are null. `po_number`, `PythonValue`, and `LlmValue` ride along as non-dimensioned Logs-Insights properties so a disagreement can be reviewed by example without inflating cardinality. Only the **AI-fallback** path emits these — the template path has no LLM value to shadow.
+
+Review disagreements during the bake with this CloudWatch Logs Insights query over `/aws/lambda/po-email-processor`:
+
+```
+fields @timestamp, Field, Agreement, LlmValue, PythonValue, po_number
+| filter ispresent(DerivedFieldAgreement)
+| filter Agreement = "disagree"
+| sort @timestamp desc
+| limit 200
+```
+
+Or aggregate overall agreement per field: `| filter ispresent(DerivedFieldAgreement) | stats count(*) by Field, Agreement`.
+
+> **Post-bake follow-up:** once agreement is acceptable, make Python authoritative on the AI-fallback path too (stop keeping the LLM value) and **drop the `site_code`/`trade`/`fiscal_year` rule sections from `EXTRACTION_PROMPT`** — the LLM stays authoritative on the fallback path only until then.
+
 **Lambdas** (`lambdas/po/`):
 | Function | Trigger | Purpose |
 |---|---|---|

cdk/po_stack.py

@@ -247,7 +247,12 @@ class PoIngestStack(Stack):
                         "-c",
                         "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
                         "-r requirements.txt -t /asset-output && "
-                        "cp handler.py ses_auth.py template_parser.py /asset-output/",
+                        # NOTE: every module handler.py imports as a sibling
+                        # MUST be listed here or the deploy ships a Lambda that
+                        # ImportErrors at runtime (bit us for template_parser
+                        # in PR #105 and nearly for derived_fields in PR #2).
+                        "cp handler.py ses_auth.py template_parser.py "
+                        "derived_fields.py /asset-output/",
                     ],
                 ),
             ),

docs/po-template-parser.md

@@ -1,6 +1,6 @@
 # PO Template-First Parser — Design, Investigation & Progress
 
-> **Status:** PR #1 implemented (extraction + gate + wiring + alarm + tests); PR #2 (derived-classifier factoring) not started · **Branch:** `feat/po-template-parser` (stacked on PR #99) · **Last updated:** 2026-07-16
+> **Status:** PR #1 implemented (extraction + gate + wiring + alarm + tests); PR #2 (derived-classifier factoring — shadow mode) implemented · **Branch:** `feat/po-template-parser` (stacked on PR #99) · **Last updated:** 2026-07-16
 >
 > Living document for making the Coupa **purchase-order** email parser *template-first with LLM fallback*, mirroring the work-order (WO) processor's PR #99. Captures the investigation, the data we gathered, the decisions made, the current scaffold, and everything still to do.
 
@@ -208,6 +208,18 @@ Plus `missing_required_field` for the required labeled fields (per-item metadata
   - [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).
 
+### PR #2 implementation notes (2026-07-16)
+
+- **Shared classifier factoring.** `trade`/`site_code`/`fiscal_year` are computed by the pure, total `derived_fields.derive_all(parsed) -> {site_code, trade, fiscal_year}` and applied inside the shared `enrich_parsed()` post-stage, so both parse paths run identical classification.
+- **Shadow semantics — fill gaps, never overwrite.** For each field: if the incoming value is `null` and Python has a value, Python fills it (both paths); if a value is already present (only possible from the LLM on the ai_fallback path) it is **kept** — the LLM stays authoritative during the bake. The template path always arrives with all three `null` (the gate enforces the parser leaves them unset), so it is effectively Python-authoritative.
+- **EMF metric shape.** On the **ai_fallback path only**, one `DerivedFieldAgreement` record per field is emitted (namespace `Seahaven/PoIngest`, value 1, dims `[["Field","Agreement"]]`). `Agreement` ∈ `agree` | `disagree` | `llm_null_python_filled` | `python_null`; skipped when both values are `null`. `po_number`/`PythonValue`/`LlmValue` are non-dimensioned ride-alongs (cardinality fixed at Field × Agreement). The template path emits **no** agreement metric. The whole block is wrapped defensively so no classification/telemetry error can fail this S3-async invocation (an uncaught exception → retry storm → DLQ).
+- **`quantity`/`price` prompt tweak (deferred from PR #1) — done.** `EXTRACTION_PROMPT` now declares line-item `quantity`/`price` as `"number or null"` (was `"string or null"`); `parse_float=Decimal` already handles numerics and the `enrich_parsed` Decimal coercion stays as the safety net for a non-conforming model. The `site_code`/`fiscal_year`/`trade` **rule sections of the prompt are left intact** — the LLM stays authoritative on the fallback path until the post-bake follow-up.
+- **Backtest agreement stats (final, 2026-07-16).** Harness: every harvested real inbound email (3,422 objects; corpus is the rolling ~90-day S3 window) → template parse → `derive_all()` → per-field compare against the LLM-written values in `purchase-orders` (17,605-record scan). Population: 3,190 template-parsed new_po / 100 cancellations / 132 fallbacks (matches PR #1 triage). The `llm_null_python_filled` buckets (478/577/564 per field) are all records processed pre-May 2026 with no `data_source` attribute — an older prompt/writer era; Python filling those is strictly additive. On the modern-era comparable set:
+  - `site_code` — 2,695/2,712 **99.4%**, and **zero Python-wrong**: all 17 misses are LLM errors (16× PO-prefix `2D` stored as a site code, 1× `4101` garbage vs Python's correct `DRN5`).
+  - `fiscal_year` — 2,626/2,626 **100%**.
+  - `trade` — 2,188/2,613 raw (83.7%); **96.8% ex-deliberate**. The 425 mismatches decompose: 316 deliberate prompt-faithful `General Building - General Building Project` → `General Building - Project` (the LLM ignored the prompt's own BBM rule — biggest intentional behavior shift, ~10% of POs); 14 bollard → `Fencing/Gates` (explicit prompt keyword); 7 `General Building Technician` → Handyman (prompt rule); 16 LLM off-menu labels (`Electrical - Emergency`, `Plumbing - Project`, `General Building - Locksmith` — closed label set wins); 18 BBM plumbing rows where the LLM is internally inconsistent (the ladder follows the per-row LLM majority, e.g. Technician→PM 116:9, fixture-repair→Reactive ~140:2); ~54 residual free-text one-offs (2.1%).
+  - Corpus-driven rules beyond the prompt (recorded here as the authoritative spec deltas): site-code shapes for code-first dash names (`WPT2 - Amazon.com Services LLC`), leading token (`QDE1 (Co-located inside WDE1)`), mid-name token with digit (`Amazon Fresh UVA5 Non-Inv (Prime)`), free description token 4-5 chars w/ digit (`... aqt DSF7`), rightmost-token parens/ATTN resolution, ≥2-alphabetic-char token floor (kills `B187`, `2D`, `4101`), skip-list additions LLC/INC/CORP/LTD/ATTN; trade BBM plumbing 5-rung ladder and `conveyance` keyword.
+
 ---
 
 ## 9. TODO / Path Forward
@@ -224,8 +236,8 @@ Plus `missing_required_field` for the required labeled fields (per-item metadata
 - [x] `ruff check` + `ruff format` + `pytest` green locally (354 tests; CI must confirm).
 
 **PR #2 — derived-classifier factoring (follow-up)**
-- [ ] Move `trade`/`site_code`/`fiscal_year` into shared `enrich_parsed()`.
-- [ ] Run in **shadow mode**: compute Python values, log Python-vs-LLM disagreement via EMF, keep LLM authoritative during a bake period.
+- [x] Move `trade`/`site_code`/`fiscal_year` into shared `enrich_parsed()` (via `derived_fields.derive_all()`).
+- [x] Run in **shadow mode**: compute Python values, log Python-vs-LLM disagreement via the `DerivedFieldAgreement` EMF metric, keep LLM authoritative during a bake period.
 - [ ] Harden `site_code` (7-shape + skip-list, incl. multi-hop ATTN like `CBRE - RME - DLI6`) and `trade` against the full sample with per-shape fixtures.
 - [ ] Only after acceptable agreement: make Python authoritative and drop the fields from `EXTRACTION_PROMPT`.
 

lambdas/po/email_processor/derived_fields.py

@@ -0,0 +1,676 @@
+"""Deterministic derived-field classifiers for Coupa purchase order emails.
+
+Pure module: stdlib only -- no boto3, no network, no imports from handler or
+template_parser. Computes the three DERIVED_KEYS (site_code, fiscal_year, trade)
+that the template parser and the LLM path both leave for a shared post-stage to
+fill identically (see template_parser.py DERIVED_KEYS / enrich_parsed).
+
+This is a FAITHFUL v1 port of the English rules in the handler's
+EXTRACTION_PROMPT sections "## site_code extraction", "## fiscal_year" and
+"## trade classification". A corpus backtest will harden the thresholds later;
+this module intentionally invents no rule beyond those three sections.
+
+TOTALITY (hard requirement): this runs on the untrusted email path in an
+S3-async Lambda, where an uncaught exception means infinite retry -> DLQ ->
+silent data loss. Every public function therefore mirrors
+template_parser.try_deterministic_parse: it wraps its whole body in a broad
+try/except and NEVER raises -- a bad/hostile input yields None (or an all-None
+dict for derive_all), never a traceback. Inputs are treated as hostile: wrong
+types, None, non-dict line items and multi-megabyte strings are all tolerated.
+Scans are length-capped and every regex is linear (no nested quantifiers, no
+catastrophic backtracking).
+
+Public API:
+    derive_site_code(parsed)   -> str | None
+    derive_fiscal_year(parsed) -> str | None
+    derive_trade(parsed)       -> str | None
+    derive_all(parsed)         -> {"site_code":..., "trade":..., "fiscal_year":...}
+
+`parsed` is the post-extraction contract dict from either path. Relevant inputs:
+parsed["ship_to"]["name"], parsed["ship_to"]["attn"],
+parsed["line_items"][*]["description"|"amount"|"need_by"], parsed["order_date"].
+"""
+
+import re
+from decimal import Decimal, InvalidOperation
+
+# --- scan caps (hostile-input blast-radius limits) ---------------------------
+_MAX_ITEMS = 500  # line_items scanned at most
+_MAX_NAME = 4096  # ship_to name/attn chars scanned for a site code
+_MAX_DESC = 4096  # description chars scanned for a bracketed/standalone code
+_MAX_DATE = 1024  # date-string chars scanned for a year
+_MAX_TRADE = 100_000  # description chars scanned for trade keywords (per item)
+_MAX_TRADE_TOTAL = 200_000  # aggregate description chars classified per PO
+
+# ---------------------------------------------------------------------------
+# site_code
+# ---------------------------------------------------------------------------
+# Token shape: 3-5 chars, uppercase A-Z0-9, MUST start with a letter. Codes may
+# be all letters (e.g. KLAL) -- a digit is NOT required.
+_STRICT_CODE_RE = re.compile(r"[A-Z][A-Z0-9]{2,4}")
+# Same shape, but only as a standalone token (not embedded in a longer
+# alphanumeric run): '–WKY3' matches, 'Station' does not, 'DLI6X99' does not.
+_CODE_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9])[A-Z][A-Z0-9]{2,4}(?![A-Za-z0-9])")
+_PAREN_RE = re.compile(r"\(([^()\n]{1,40})\)")
+_BRACKET_CODE_RE = re.compile(r"\[([A-Z][A-Z0-9]{2,4})\]")
+# Description prefix: leading code then a dash/en-dash/em-dash separator.
+_PREFIX_CODE_RE = re.compile(r"\s*([A-Z][A-Z0-9]{2,4})\s*[-–—]")
+# Leading code-shaped token that opens the ship-to name and is delimited by a
+# space, dash or '(' -- e.g. 'WPT2 - Amazon...' / 'QDE1 (Co-located ...)'.
+_LEADING_CODE_RE = re.compile(r"([A-Z][A-Z0-9]{2,4})(?=[\s\-–—(])")
+# Free-token code in a description: standalone, length 4-5, letter-first.
+_FREE_CODE_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9])[A-Z][A-Z0-9]{3,4}(?![A-Za-z0-9])")
+_DASHES = "-–—"
+_DASH_SPLIT_RE = re.compile(r"[-–—]")
+
+# Never a site_code (from the prompt's "Not site codes" list). If the only
+# candidate for a shape is skip-listed, that shape yields None and evaluation
+# falls through to the next lower-priority shape.
+_SKIP = frozenset(
+    {
+        "RME",
+        "BBM",
+        "JLL",
+        "PARAG",
+        "ERIK",
+        "HVAC",
+        "LED",
+        "PVC",
+        "ADA",
+        "OSHA",
+        "EMR",
+        "BMS",
+        "DDC",
+        "MRO",
+        "NTE",
+        "EST",
+        # Corporate-suffix / addressing tokens -- never a site code.
+        "LLC",
+        "INC",
+        "CORP",
+        "LTD",
+        "ATTN",
+    }
+)
+
+
+def _is_valid_code(tok):
+    """A code-shaped token is a real site code only if it is not skip-listed AND
+    carries >=2 alphabetic characters.
+
+    Measured over all 1,063 distinct LLM-extracted codes: 999 have 3 letters, 7
+    have 4, 3 have 5, and NONE have fewer than 2. Sub-2-letter tokens (e.g.
+    'B187') are PO-prefix-style garbage, so every site-code shape rejects them.
+    """
+    return tok not in _SKIP and sum(c.isalpha() for c in tok) >= 2
+
+
+def _last_valid_code(text):
+    """Rightmost valid code-shaped token in `text`.
+
+    This is the multi-hop resolver: an ATTN like 'CBRE - RME - DLI6' yields the
+    rightmost code that passes _is_valid_code (DLI6), skipping the RME hop.
+    """
+    for tok in reversed(_CODE_TOKEN_RE.findall(text)):
+        if _is_valid_code(tok):
+            return tok
+    return None
+
+
+def _code_name_leading(name):
+    """Leading shape (highest priority): the name OPENS with a standalone
+    code-shaped token containing >=1 digit, delimited by space/dash/'(' --
+    'WPT2 - Amazon...' -> WPT2, 'QDE1 (Co-located inside WDE1)' -> QDE1.
+
+    The digit requirement + leading position outrank the parens shape, so a
+    'QDE1 (... WDE1)' resolves to the opener QDE1, not the host building WDE1.
+    """
+    if not isinstance(name, str):
+        return None
+    m = _LEADING_CODE_RE.match(name.strip()[:_MAX_NAME])
+    if not m:
+        return None
+    tok = m.group(1)
+    if _is_valid_code(tok) and any(c.isdigit() for c in tok):
+        return tok
+    return None
+
+
+def _code_name_parens(name):
+    """Shape 1: ship-to name in parentheses -- 'Services LLC (KLAL)' -> KLAL.
+
+    Uses the rightmost-non-skip resolver (same as ATTN) so parens content like
+    '(ATTN: Wagon Wheel DS Station -WKY3)' resolves to WKY3, not the skip-listed
+    ATTN hop.
+    """
+    if not isinstance(name, str):
+        return None
+    for content in _PAREN_RE.findall(name[:_MAX_NAME]):
+        code = _last_valid_code(content)
+        if code:
+            return code
+    return None
+
+
+def _code_name_after_dash(name):
+    """Shape 2: ship-to name with a dash.
+
+    Real ship-to names carry the code on EITHER side of the dash -- 'Amazon...
+    LLC - SNY5' (after) and 'WPT2 - Amazon... LLC' (before). First prefer any
+    segment that IS exactly a code-shaped token (leftmost such segment wins);
+    only if no whole segment is an exact code, fall back to scanning the tail
+    after the last dash (so a code-shaped word before the dash, e.g. 'LLC',
+    can never leak in when the real after-dash code is skip-listed).
+    """
+    if not isinstance(name, str):
+        return None
+    s = name[:_MAX_NAME]
+    idx = max(s.rfind(d) for d in _DASHES)
+    if idx == -1:
+        return None
+    for seg in _DASH_SPLIT_RE.split(s):
+        tok = seg.strip()
+        if _STRICT_CODE_RE.fullmatch(tok) and _is_valid_code(tok):
+            return tok
+    return _last_valid_code(s[idx + 1 :])
+
+
+def _code_name_midtoken(name):
+    """Mid-name shape (lower priority): a standalone uppercase code-shaped token
+    with >=1 digit anywhere in the name -- 'Amazon Fresh UVA5 Non-Inv (Prime)'
+    -> UVA5. The digit requirement means all-letter words (FRESH, ...) can never
+    match; mixed-case words never match the uppercase token shape at all.
+    """
+    if not isinstance(name, str):
+        return None
+    for tok in _CODE_TOKEN_RE.findall(name[:_MAX_NAME]):
+        if _is_valid_code(tok) and any(c.isdigit() for c in tok):
+            return tok
+    return None
+
+
+def _code_from_attn(attn):
+    """Shapes 3 & 4: ship-to ATTN line (with dash/en-dash, or direct).
+
+    Unified: the rightmost non-skip code-shaped token across the whole ATTN
+    value. Covers 'Wagon Wheel DS Station -WKY3' -> WKY3, 'HJX1' -> HJX1, and
+    the multi-hop 'CBRE - RME - DLI6' -> DLI6.
+    """
+    if not isinstance(attn, str):
+        return None
+    return _last_valid_code(attn[:_MAX_NAME])
+
+
+def _code_name_is(name):
+    """Shape 5: the ship-to name IS the code -- 'DBU2' -> DBU2."""
+    if not isinstance(name, str):
+        return None
+    n = name.strip()
+    if _STRICT_CODE_RE.fullmatch(n) and _is_valid_code(n):
+        return n
+    return None
+
+
+def _code_desc_prefix(desc):
+    """Shape 6: line-item description prefix -- 'DYO1 - ... - ...' -> DYO1."""
+    if not isinstance(desc, str):
+        return None
+    m = _PREFIX_CODE_RE.match(desc[:80])
+    if m and _is_valid_code(m.group(1)):
+        return m.group(1)
+    return None
+
+
+def _code_desc_bracket(desc):
+    """Shape 7: line-item description in brackets -- '[HMK4] ...' -> HMK4."""
+    if not isinstance(desc, str):
+        return None
+    for m in _BRACKET_CODE_RE.finditer(desc[:_MAX_DESC]):
+        if _is_valid_code(m.group(1)):
+            return m.group(1)
+    return None
+
+
+def _code_desc_freetoken(desc):
+    """Shape 8 (lowest priority): a standalone uppercase token, length 4-5,
+    letter-first, with >=1 digit, anywhere in a description -- 'Need Sea Haven
+    to pump out waste water aqt DSF7' -> DSF7. The length-4 floor (vs the 3-char
+    floor of other shapes) and the digit requirement keep this loose free-scan
+    from grabbing 3-letter uppercase words or all-letter tokens mid-sentence.
+    """
+    if not isinstance(desc, str):
+        return None
+    for tok in _FREE_CODE_TOKEN_RE.findall(desc[:_MAX_DESC]):
+        if _is_valid_code(tok) and any(c.isdigit() for c in tok):
+            return tok
+    return None
+
+
+def _iter_items(parsed):
+    """Yield up to _MAX_ITEMS dict line items from parsed; tolerant of junk."""
+    items = parsed.get("line_items")
+    if not isinstance(items, list):
+        return
+    for item in items[:_MAX_ITEMS]:
+        if isinstance(item, dict):
+            yield item
+
+
+def derive_site_code(parsed):
+    """Derive the Amazon facility site_code, or None. NEVER raises.
+
+    Shapes are tried in strict priority order. A skip-listed sole candidate for
+    a shape yields None for that shape and evaluation continues to the next:
+
+      leading  -- name opens with a digit-bearing code ('WPT2 - ...')
+      1 parens -- code inside '(...)'
+      2 dash   -- exact code-shaped segment either side of a dash
+      3/4 attn -- rightmost non-skip code in the ATTN line
+      5 name-is-- the name IS the code
+      midtoken -- a digit-bearing code anywhere in the name
+      6 prefix -- description leading code
+      7 bracket-- description '[CODE]'
+      8 free   -- a length 4-5 digit-bearing code anywhere in a description
+    """
+    try:
+        if not isinstance(parsed, dict):
+            return None
+        ship_to = parsed.get("ship_to")
+        if not isinstance(ship_to, dict):
+            ship_to = {}
+        name = ship_to.get("name")
+        attn = ship_to.get("attn")
+
+        for shape in (
+            _code_name_leading(name),  # leading (highest)
+            _code_name_parens(name),  # 1
+            _code_name_after_dash(name),  # 2
+            _code_from_attn(attn),  # 3 & 4
+            _code_name_is(name),  # 5
+            _code_name_midtoken(name),  # mid-name
+        ):
+            if shape:
+                return shape
+
+        for item in _iter_items(parsed):  # 6
+            code = _code_desc_prefix(item.get("description"))
+            if code:
+                return code
+        for item in _iter_items(parsed):  # 7
+            code = _code_desc_bracket(item.get("description"))
+            if code:
+                return code
+        for item in _iter_items(parsed):  # 8
+            code = _code_desc_freetoken(item.get("description"))
+            if code:
+                return code
+        return None
+    except Exception:  # noqa: BLE001 -- totality: never raise on the email path
+        return None
+
+
+# ---------------------------------------------------------------------------
+# fiscal_year
+# ---------------------------------------------------------------------------
+# A standalone 4-digit 20xx year. Word-bounded so it never matches digits buried
+# inside a longer run (e.g. the '2062' inside a PO id '18206023').
+_YEAR_RE = re.compile(r"\b(20\d{2})\b")
+# MM/DD/YY -> the trailing 2-digit year is expanded to 20YY.
+_SHORT_DATE_RE = re.compile(r"\b\d{1,2}/\d{1,2}/(\d{2})\b")
+
+
+def _year_from_date(value):
+    """Extract a 4-digit year from a date string: a literal 20xx, else a
+    MM/DD/YY whose YY expands to 20YY. Returns the 4-digit string or None."""
+    if not isinstance(value, str):
+        return None
+    s = value[:_MAX_DATE]
+    m = _YEAR_RE.search(s)
+    if m:
+        return m.group(1)
+    m = _SHORT_DATE_RE.search(s)
+    if m:
+        return "20" + m.group(1)
+    return None
+
+
+def _year_from_text(value):
+    """A standalone 20xx year anywhere in free text, or None."""
+    if not isinstance(value, str):
+        return None
+    m = _YEAR_RE.search(value[:_MAX_DESC])
+    return m.group(1) if m else None
+
+
+def derive_fiscal_year(parsed):
+    """Derive the 4-digit fiscal_year string, or None. NEVER raises.
+
+    Fallback order: (1) order_date year, (2) any line-item need_by year, (3) a
+    standalone 20xx year in any line-item description.
+    """
+    try:
+        if not isinstance(parsed, dict):
+            return None
+        year = _year_from_date(parsed.get("order_date"))
+        if year:
+            return year
+        for item in _iter_items(parsed):
+            year = _year_from_date(item.get("need_by"))
+            if year:
+                return year
+        for item in _iter_items(parsed):
+            year = _year_from_text(item.get("description"))
+            if year:
+                return year
+        return None
+    except Exception:  # noqa: BLE001 -- totality: never raise on the email path
+        return None
+
+
+# ---------------------------------------------------------------------------
+# trade
+# ---------------------------------------------------------------------------
+# Keyword matching interpretation (judgment call, documented): each keyword is
+# matched with a LEADING word boundary and no trailing boundary -- i.e. it hits
+# the keyword as a whole word OR as the prefix of a longer word. This makes
+# 'sign' match 'signage', 'dock door' match 'dock doors', and 'roof' match
+# 'roofing' (desired), while NOT matching a keyword buried mid-word ('ice' in
+# 'service'/'price', 'lock' in 'block'/'clock', 'gate' in 'mitigate') -- the
+# catastrophic false positives a bare substring test would produce. All matching
+# is done on a lower-cased, length-capped copy of the description.
+
+
+def _kw(*words):
+    """Compile a linear leading-boundary alternation of literal keywords."""
+    body = "|".join(re.escape(w) for w in words)
+    return re.compile(r"\b(?:" + body + r")")
+
+
+_PM_RE = _kw(
+    "plumbing pm",
+    "plumbing preventative",
+    "plumbing maintenance",
+    "plumbing - backflow",
+    "plumbing - water heater",
+)
+_PLUMBING_RE = _kw("plumbing")
+# BBM-structured 'Plumbing - <Fixture> - <Action> - BBM' row: 'plumbing' opening
+# the description immediately followed by a dash. Resolved by a measured ladder
+# (see _classify_desc) that matches the LLM baseline over the full corpus.
+_BBM_PLUMBING_RE = re.compile(r"plumbing\s*[-–—]")
+# Ladder rungs for a BBM plumbing row, checked in order:
+#   (a) emergency/reactive -> Reactive
+_BBM_PLUMBING_REACTIVE_WORD_RE = _kw("reactive", "emergency")
+#   (b) technician -> PM
+_BBM_PLUMBING_TECH_RE = _kw("technician")
+#   (c) water heater / backflow / water fountain -> PM (before (d), so a
+#       'water heater repair' hits PM here and not Reactive on 'repair')
+_BBM_PLUMBING_PM_RE = _kw("water heater", "backflow", "water fountain")
+#   (d) clog/unclog/leak/repair/sewer/drain -> Reactive ('clog'/'leak' prefixes
+#       also catch 'clogged'/'leaking')
+_BBM_PLUMBING_REACTIVE_RE = _kw("clog", "unclog", "leak", "repair", "sewer", "drain")
+#   (e) else (project, install, misc) -> PM
+# Plumbing-specific qualifiers -- these indicate Plumbing - Reactive on their
+# OWN, without the word 'plumbing' present ('CLOGGED PIT AUGER' -> Reactive).
+_PLUMBING_SPECIFIC_RE = _kw(
+    "clog",
+    "unclog",
+    "sewer",
+    "drain",
+    "grease trap",
+    "jetter",
+    "toilet",
+    "faucet",
+    "urinal",
+)
+# Generic reactive qualifiers -- these require the word 'plumbing' to be present
+# before they route to Plumbing - Reactive.
+_PLUMBING_GENERIC_RE = _kw(
+    "reactive",
+    "emergency",
+    "repair",
+    "leak",
+    "flood",
+    "water line",
+    "pipe",
+)
+_ELECTRICAL_RE = _kw(
+    "electrical",
+    "lighting",
+    "ballast",
+    "outlet",
+    "circuit",
+    "panel",
+    "generator",
+    "transformer",
+    "conduit",
+)
+_HVAC_RE = _kw(
+    "hvac",
+    "heating",
+    "cooling",
+    "air conditioning",
+    "rtu",
+    "ahu",
+    "vav",
+    "refrigerant",
+    "thermostat",
+    "ductwork",
+)
+_DOCK_DOORS_RE = _kw(
+    "dock door",
+    "dock leveler",
+    "dock plate",
+    "dock seal",
+    "dock bumper",
+)
+_DOORS_RE = _kw("door", "overhead door", "roll-up", "automatic door", "access door")
+_SIGNAGE_RE = _kw("sign", "banner", "wayfinding", "marquee", "directional")
+_CARPENTRY_RE = _kw("carpentry", "cabinet", "millwork", "trim", "shelving", "framing")
+_FENCE_RE = _kw("fence", "fencing", "bollard")
+_GATE_RE = _kw("gate")
+_CONVEYANCE_RE = _kw("conveyor", "conveyance", "mhe", "material handling", "sortation")
+_PAINTING_RE = _kw("paint", "painting", "primer", "coating", "touch-up")
+_FLOORING_RE = _kw("floor", "tile", "carpet", "epoxy", "polishing")
+_JANITORIAL_RE = _kw(
+    "janitorial", "cleaning", "custodial", "pressure wash", "power wash"
+)
+_FIRE_RE = _kw("fire", "sprinkler", "extinguisher", "fire alarm", "suppression")
+_LANDSCAPING_RE = _kw("landscape", "lawn", "tree", "yard", "mowing", "irrigation")
+_ROOFING_RE = _kw("roof", "roofing", "gutter", "downspout")
+# Security/Locksmith: 'lock' and 'key' match as EXACT whole words only (so
+# 'Locker' is not 'lock' and 'keyboard' is not 'key'); the multi-word terms keep
+# the leading-boundary/prefix behavior. The phrase 'key box' is stripped before
+# this test (see _classify_desc) so a Key Box fixture never triggers Locksmith.
+_SECURITY_RE = re.compile(r"\b(?:lock|key)\b|\b(?:access control|camera|security|cctv)")
+_SNOW_RE = _kw("snow", "ice", "salt", "de-ice", "plow")
+_EMERGENCY_RE = _kw("emergency")
+# General Building catch-all sub-shapes.
+_HANDYMAN_RE = re.compile(r"\bhandyman\b")
+_PROJECT_RE = re.compile(r"\bproject\b")
+
+TRADE_UPLIFT = "PO Uplift"
+TRADE_GENERAL_BUILDING = "General Building"
+
+# Simple (single-regex) trades, in strict priority order. The compound and
+# exception-bearing trades (Plumbing, Electrical, Fencing/Gates, the General
+# Building family, PO Uplift) are handled inline in _classify_desc.
+_SIMPLE_TRADES = (
+    (_HVAC_RE, "HVAC"),
+    (_DOCK_DOORS_RE, "Dock Doors"),
+    (_DOORS_RE, "Doors"),
+    (_SIGNAGE_RE, "Signage"),
+    (_CARPENTRY_RE, "Carpentry"),
+)
+# Simple trades that follow Fencing/Gates in the priority table. Security and
+# Snow are handled explicitly after this loop (Security needs the 'key box'
+# exclusion applied to its match target, so it can't share the generic loop).
+_SIMPLE_TRADES_TAIL = (
+    (_CONVEYANCE_RE, "Conveyance/MHE"),
+    (_PAINTING_RE, "Painting"),
+    (_FLOORING_RE, "Flooring"),
+    (_JANITORIAL_RE, "Janitorial"),
+    (_FIRE_RE, "Fire/Life Safety"),
+    (_LANDSCAPING_RE, "Landscaping/Yard"),
+    (_ROOFING_RE, "Roofing"),
+)
+
+
+def _classify_desc(desc):
+    """Classify a single non-empty description into one trade label.
+
+    Returns a label for any non-empty description -- 'General Building' is the
+    catch-all when nothing else matches. Callers must not pass empty/blank
+    descriptions (that case is handled upstream so it can return None).
+    """
+    low = desc[:_MAX_TRADE].lower()
+    stripped = low.strip()
+
+    # PO Uplift: exactly or primarily "PO Uplift".
+    if stripped == "po uplift" or stripped.startswith("po uplift"):
+        return TRADE_UPLIFT
+
+    # BBM-structured 'Plumbing - ...' row: resolved by a measured ladder tuned to
+    # the LLM baseline over the full corpus (first rung wins).
+    if _BBM_PLUMBING_RE.match(stripped):
+        if _BBM_PLUMBING_REACTIVE_WORD_RE.search(low):  # (a) emergency/reactive
+            return "Plumbing - Reactive"
+        if _BBM_PLUMBING_TECH_RE.search(low):  # (b) technician
+            return "Plumbing - PM"
+        if _BBM_PLUMBING_PM_RE.search(low):  # (c) heater/backflow/fountain
+            return "Plumbing - PM"
+        if _BBM_PLUMBING_REACTIVE_RE.search(low):  # (d) clog/leak/repair/...
+            return "Plumbing - Reactive"
+        return "Plumbing - PM"  # (e) project/install/misc
+    # Plumbing - PM (before Reactive) for non-BBM free text.
+    if _PM_RE.search(low):
+        return "Plumbing - PM"
+    # Plumbing - Reactive: a plumbing-specific qualifier ALONE (clog, drain,
+    # toilet, ...), or the word 'plumbing' PLUS a generic reactive qualifier.
+    if _PLUMBING_SPECIFIC_RE.search(low):
+        return "Plumbing - Reactive"
+    if _PLUMBING_RE.search(low) and _PLUMBING_GENERIC_RE.search(low):
+        return "Plumbing - Reactive"
+
+    # Electrical -- but electrical keywords do NOT match in a dock-door context.
+    if _ELECTRICAL_RE.search(low) and "dock door" not in low:
+        return "Electrical"
+
+    for regex, label in _SIMPLE_TRADES:
+        if regex.search(low):
+            return label
+
+    # Fencing/Gates: fence/fencing/bollard anywhere, or 'gate' but NOT 'dock
+    # gate' (the dock-gate occurrences are removed before the gate test).
+    if _FENCE_RE.search(low) or _GATE_RE.search(low.replace("dock gate", " ")):
+        return "Fencing/Gates"
+
+    for regex, label in _SIMPLE_TRADES_TAIL:
+        if regex.search(low):
+            return label
+
+    # Security/Locksmith: 'lock'/'key' as whole words, with 'key box' stripped
+    # first (mirrors the 'dock gate' exclusion) so a Key Box fixture is not one.
+    if _SECURITY_RE.search(low.replace("key box", " ")):
+        return "Security/Locksmith"
+    if _SNOW_RE.search(low):
+        return "Snow Removal"
+
+    # General Building family (catch-alls, in priority order).
+    if stripped.startswith("emer") or _EMERGENCY_RE.search(low):
+        return "General Building - Emergency"
+    if _HANDYMAN_RE.search(low) or "general building technician" in low:
+        return "General Building - Handyman"
+    if "general building project" in low or _PROJECT_RE.search(low):
+        return "General Building - Project"
+    # Keyword-less BBM 'General Building - <Fixture> - ...' rows (parking lot,
+    # fan, ceiling, television, locker, key box, ...) fall through here to the
+    # plain catch-all -- that matches the measured LLM majority for such rows.
+    return TRADE_GENERAL_BUILDING
+
+
+def _to_amount(value):
+    """Coerce a line amount (Decimal | str | int/float | junk) to a Decimal for
+    comparison. Anything unparseable becomes Decimal(0) -- defensive, never
+    raises, so an item with a bad amount simply cannot win the highest-value
+    tie-break."""
+    if isinstance(value, bool):
+        return Decimal(0)
+    if isinstance(value, Decimal):
+        return value if value.is_finite() else Decimal(0)
+    if isinstance(value, int):
+        return Decimal(value)
+    if isinstance(value, float):
+        try:
+            d = Decimal(str(value))
+            return d if d.is_finite() else Decimal(0)
+        except InvalidOperation:
+            return Decimal(0)
+    if isinstance(value, str):
+        try:
+            d = Decimal(value.replace(",", "").strip())
+            return d if d.is_finite() else Decimal(0)
+        except (InvalidOperation, ValueError):
+            return Decimal(0)
+    return Decimal(0)
+
+
+def derive_trade(parsed):
+    """Derive the PO's primary trade label, or None. NEVER raises.
+
+    Each line item with a usable description is classified. The PO trade is the
+    primary non-"PO Uplift" trade; when several distinct non-uplift trades are
+    present, the trade of the highest-`amount` non-uplift item wins. If every
+    described item is PO Uplift, returns "PO Uplift". With no line items or no
+    usable descriptions at all, returns None ("General Building" is only ever
+    returned for a description that matched nothing).
+    """
+    try:
+        if not isinstance(parsed, dict):
+            return None
+        classified = []  # (trade_label, amount_decimal)
+        # Aggregate CPU budget across ALL items: the per-item _MAX_TRADE cap
+        # alone still allows _MAX_ITEMS x _MAX_TRADE x ~25 regex passes
+        # (~10s full-core, minutes under the 256MB Lambda's CPU throttle ->
+        # timeout -> async retry -> DLQ). Real POs total well under 10k chars
+        # of description text; stop classifying once the budget is spent and
+        # decide from what was classified (sh-security-review
+        # PO-DERIVED-AVAIL-001 defense-in-depth).
+        budget = _MAX_TRADE_TOTAL
+        for item in _iter_items(parsed):
+            desc = item.get("description")
+            if not isinstance(desc, str) or not desc.strip():
+                continue
+            if budget <= 0:
+                break
+            desc = desc[:budget]
+            budget -= len(desc)
+            classified.append((_classify_desc(desc), _to_amount(item.get("amount"))))
+        if not classified:
+            return None
+
+        non_uplift = [(t, a) for (t, a) in classified if t != TRADE_UPLIFT]
+        if not non_uplift:
+            return TRADE_UPLIFT
+        if len({t for (t, _) in non_uplift}) == 1:
+            return non_uplift[0][0]
+        best_trade, best_amount = non_uplift[0]
+        for trade, amount in non_uplift[1:]:
+            if amount > best_amount:
+                best_trade, best_amount = trade, amount
+        return best_trade
+    except Exception:  # noqa: BLE001 -- totality: never raise on the email path
+        return None
+
+
+def derive_all(parsed):
+    """All three derived fields as a dict. NEVER raises: on any failure returns
+    an all-None dict so the caller always gets the three keys."""
+    try:
+        return {
+            "site_code": derive_site_code(parsed),
+            "trade": derive_trade(parsed),
+            "fiscal_year": derive_fiscal_year(parsed),
+        }
+    except Exception:  # noqa: BLE001 -- totality: never raise on the email path
+        return {"site_code": None, "trade": None, "fiscal_year": None}

lambdas/po/email_processor/handler.py

@@ -17,6 +17,7 @@ from decimal import Decimal, InvalidOperation
 from email import policy
 
 import boto3
+from derived_fields import derive_all
 from ses_auth import authenticate_inbound_email
 from template_parser import try_deterministic_parse
 
@@ -37,6 +38,18 @@ BEDROCK_MODEL_ID = os.environ.get(
 METRIC_NAMESPACE = "Seahaven/PoIngest"
 METRIC_NAME = "ParseOutcome"
 
+# Shadow telemetry for the Python-derived classifier bake. One EMF record per
+# derived field per email, emitted ONLY on the ai_fallback path (the template
+# path has no LLM value to compare against). Dimensioned by Field x Agreement
+# only -- PythonValue/LlmValue/po_number ride along as Logs-Insights-queryable
+# properties so the cardinality stays fixed at (3 fields x 4 categories).
+DERIVED_METRIC_NAME = "DerivedFieldAgreement"
+
+# The three classifier outputs derive_all() computes. Python fills these when
+# the extraction path left them null; on ai_fallback the LLM value (if any)
+# stays authoritative during the bake and Python only shadows it.
+DERIVED_FIELDS = ("site_code", "trade", "fiscal_year")
+
 # "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.
@@ -92,9 +105,9 @@ Analyze the following email and extract structured data. Return ONLY valid JSON
       "category": "string or null",
       "account_code": "string or null",
       "period": "string or null",
-      "quantity": "string or null",
+      "quantity": "number or null",
       "unit": "string or null",
-      "price": "string or null"
+      "price": "number or null"
     }
   ]
 }
@@ -328,6 +341,65 @@ def _emit_parse_method_metric(method, template_id, reason_code, po_number):
     print(json.dumps(emf))
 
 
+def _derived_agreement(llm_value, python_value) -> str | None:
+    """Classify Python-vs-LLM agreement for one derived field.
+
+    Returns None when both values are None (nothing to compare -- the caller
+    then skips emission). Categories:
+      * ``agree``                 -- both non-None and equal after str-strip
+      * ``disagree``              -- both non-None but different
+      * ``llm_null_python_filled``-- LLM None, Python supplied a value
+      * ``python_null``           -- LLM non-None, Python None
+    """
+    if llm_value is None and python_value is None:
+        return None
+    if llm_value is None:
+        return "llm_null_python_filled"
+    if python_value is None:
+        return "python_null"
+    if str(llm_value).strip() == str(python_value).strip():
+        return "agree"
+    return "disagree"
+
+
+def _emit_derived_agreement_metric(field, llm_value, python_value, po_number):
+    """Emit one CloudWatch EMF line shadowing the Python-derived classifier
+    against the LLM value for a single derived field (ai_fallback path only).
+
+    Mirrors ``_emit_parse_method_metric``: zero-latency (no PutMetricData; the
+    role already has logs:PutLogEvents), Field x Agreement the only promoted
+    dimension set (cardinality 3x4). PythonValue/LlmValue/po_number ride along
+    as Logs-Insights-queryable properties so a disagreement can be reviewed by
+    example without inflating metric cardinality. No emission when both values
+    are None -- there is nothing to compare."""
+    agreement = _derived_agreement(llm_value, python_value)
+    if agreement is None:
+        return
+    emf = {
+        "_aws": {
+            "Timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
+            "CloudWatchMetrics": [
+                {
+                    "Namespace": METRIC_NAMESPACE,
+                    "Dimensions": [["Field", "Agreement"]],
+                    "Metrics": [{"Name": DERIVED_METRIC_NAME, "Unit": "Count"}],
+                }
+            ],
+        },
+        "Field": field,
+        "Agreement": agreement,
+        "po_number": po_number or "",
+        # Length-clamped: Python values are regex/enum-bounded by construction,
+        # but the LLM value is schema-unvalidated model output -- a hallucinated
+        # free-text field must not land unbounded in a 2-month log line
+        # (sh-security-review PO-DC-02, confirmed low).
+        "PythonValue": "" if python_value is None else str(python_value)[:64],
+        "LlmValue": "" if llm_value is None else str(llm_value)[:64],
+        DERIVED_METRIC_NAME: 1,
+    }
+    print(json.dumps(emf))
+
+
 def pad_zip(zip_code: str | None) -> str | None:
     if not zip_code:
         return zip_code
@@ -337,8 +409,13 @@ def pad_zip(zip_code: str | None) -> str | None:
     return zip_code
 
 
-def enrich_parsed(parsed: dict, s3_key: str, email_subject: str):
-    """Add metadata and promote nested fields to top level."""
+def enrich_parsed(parsed: dict, s3_key: str, email_subject: str, *, parse_method: str):
+    """Add metadata and promote nested fields to top level.
+
+    ``parse_method`` ("template" | "ai_fallback") selects the derived-field
+    shadow behavior below: agreement telemetry is emitted only on ai_fallback,
+    where an LLM value exists to compare the Python classifier against.
+    """
     now = datetime.now(timezone.utc).isoformat()
     parsed["raw_s3_key"] = s3_key
     parsed["processed_at"] = now
@@ -378,6 +455,37 @@ def enrich_parsed(parsed: dict, s3_key: str, email_subject: str):
                 # but a bare JSON int would slip through as Python int; coerce
                 # so both paths emit one canonical Decimal type (cross-review FIX).
                 item[field] = Decimal(str(value))
+
+    # Derived-field classification (site_code, trade, fiscal_year). Python
+    # derivation FILLS GAPS on BOTH paths but NEVER OVERWRITES: an LLM-supplied
+    # value (only possible on the ai_fallback path) stays authoritative during
+    # the bake period. On ai_fallback we additionally emit one shadow EMF record
+    # per field comparing the Python value to the LLM value, so agreement can be
+    # measured before Python becomes authoritative and the rules are dropped
+    # from EXTRACTION_PROMPT (a post-bake follow-up).
+    #
+    # The whole block is wrapped defensively: derive_all() is total and pure,
+    # but this is an S3-async Lambda where any uncaught exception means a retry
+    # storm -> DLQ, so no classification/telemetry error may ever fail the
+    # invocation.
+    try:
+        python_vals = derive_all(parsed)
+        for field in DERIVED_FIELDS:
+            llm_value = parsed.get(field)
+            python_value = python_vals.get(field)
+            if llm_value is None and python_value is not None:
+                # Python fills the gap on both paths.
+                parsed[field] = python_value
+            # else: a non-None LLM value (ai_fallback only) is kept as-is.
+            if parse_method == "ai_fallback":
+                _emit_derived_agreement_metric(
+                    field, llm_value, python_value, parsed.get("po_number")
+                )
+    except Exception:  # noqa: BLE001 - telemetry must never fail the invocation
+        logger.exception(
+            "derived-field classification/telemetry failed; continuing without it"
+        )
+
     return parsed
 
 
@@ -570,7 +678,9 @@ def handler(event, context):
             logger.warning(f"No PO number found in email, skipping: {key}")
             continue
 
-        parsed = enrich_parsed(parsed, s3_key, email_data["subject"])
+        parsed = enrich_parsed(
+            parsed, s3_key, email_data["subject"], parse_method=parse_method
+        )
 
         email_type = parsed.get("email_type")
         if email_type == "cancellation":

lambdas/po/email_processor/tests/_po_parser_support.py

@@ -69,6 +69,9 @@ def load_po_handler():
     siblings = {
         "template_parser": load_template_parser(),
         "ses_auth": _load_module("ses_auth.py", f"{_HANDLER_NAME}__ses_auth"),
+        "derived_fields": _load_module(
+            "derived_fields.py", f"{_HANDLER_NAME}__derived_fields"
+        ),
     }
     saved = {name: sys.modules.get(name) for name in siblings}
     sys.modules.update(siblings)

lambdas/po/email_processor/tests/test_po_bedrock_fallback.py

@@ -205,11 +205,18 @@ def test_two_path_parity_through_enrich_and_save(fake_dynamo):
     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."""
+    parser-derived golden verbatim. EXTRACTION_PROMPT now declares quantity and
+    price as numbers, but the shared enrich_parsed Decimal coercion remains a
+    safety net for a non-conforming model that still returns them as strings --
+    which is exactly the case this test feeds. The shared enrich_parsed must
+    converge both to Decimal (DynamoDB Number) or the two paths write different
+    attribute types for the same email.
+
+    Derived classification (site_code/trade/fiscal_year) now runs in the shared
+    enrich_parsed on BOTH paths, but with different authority: the template path
+    takes the Python value; the ai_fallback path KEEPS the LLM value (the bake).
+    Those fields are normalized out of the stream-parity comparison below, which
+    guards the quantity/price type convergence + non-derived shape parity."""
     stem = NEW_PO_STEMS[0]
     email_data = load_email("new-po", stem)
     s3_key = "s3://po-ingest-emails-x/inbound/parity"
@@ -230,20 +237,22 @@ def test_two_path_parity_through_enrich_and_save(fake_dynamo):
         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).
+    # A prompt-obedient LLM supplies the derived classification on the fallback
+    # path; enrich_parsed KEEPS this LLM value (authoritative during the bake).
     llm_parsed["site_code"] = "DYO1"
 
     enriched_a = po_handler.enrich_parsed(
-        template_parsed, s3_key, email_data["subject"]
+        template_parsed, s3_key, email_data["subject"], parse_method="template"
+    )
+    enriched_b = po_handler.enrich_parsed(
+        llm_parsed, s3_key, email_data["subject"], parse_method="ai_fallback"
     )
-    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
 
-    # enrich_parsed passes the LLM-derived site_code through untouched...
-    assert enriched_a["site_code"] is None
+    # The ai_fallback path keeps the LLM-supplied site_code (not overwritten by
+    # the Python classifier during the bake).
     assert enriched_b["site_code"] == "DYO1"
-    # ...and converges quantity/price to Decimal on the prompt-shaped side.
+    # enrich_parsed converges quantity/price to Decimal on the prompt-shaped side.
     from decimal import Decimal
 
     item_a = enriched_a["line_items"][0]
@@ -252,15 +261,19 @@ def test_two_path_parity_through_enrich_and_save(fake_dynamo):
         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.
+    # Everything except the derived fields (which now diverge by design -- Python
+    # on the template path, LLM on the fallback path) 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
+    # logic (derived fields normalized on BOTH sides so the comparison isolates
+    # the quantity/price attribute types, not the deliberate derived divergence).
+    for key in template_parser.DERIVED_KEYS:
+        enriched_a[key] = None
+        enriched_b[key] = 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
@@ -272,17 +285,18 @@ def test_enrich_parsed_pads_short_zip_identically_for_llm_path():
     pad_zip then repairs them inside the SHARED enrich_parsed, so whichever
     path produced the dict, downstream sees the same padded zip."""
     parsed = {"ship_to": {"address": "x", "state": "MA", "zip": "2149"}}
-    po_handler.enrich_parsed(parsed, "s3://b/k", "subj")
+    po_handler.enrich_parsed(parsed, "s3://b/k", "subj", parse_method="template")
     assert parsed["ship_to"]["zip"] == "02149"
     assert parsed["ship_to_raw"] == "x"
     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."""
+    """EXTRACTION_PROMPT now declares quantity/price as numbers, but the shared
+    enrich_parsed Decimal coercion is retained as a SAFETY NET: a model that
+    still returns them as strings is coerced 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 = {
@@ -291,7 +305,7 @@ def test_enrich_parsed_coerces_prompt_string_quantity_price():
             {"quantity": "N/A", "price": None},
         ]
     }
-    po_handler.enrich_parsed(parsed, "s3://b/k", "subj")
+    po_handler.enrich_parsed(parsed, "s3://b/k", "subj", parse_method="template")
     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

lambdas/po/email_processor/tests/test_po_derived_fields.py

@@ -0,0 +1,605 @@
+"""Unit tests for the deterministic derived-field classifiers.
+
+derived_fields is a pure stdlib module with no sibling imports, so it is loaded
+directly by file path under a unique module name (mirroring the module-collision
+guard in _po_parser_support): a bare ``import derived_fields`` could collide with
+another test root in the same pytest session.
+"""
+
+import importlib.util
+import os
+import time
+from decimal import Decimal
+
+import pytest
+
+_HERE = os.path.dirname(__file__)
+_MODULE_DIR = os.path.abspath(os.path.join(_HERE, ".."))
+
+
+def _load_derived_fields():
+    name = "po_email_processor__derived_fields"
+    spec = importlib.util.spec_from_file_location(
+        name, os.path.join(_MODULE_DIR, "derived_fields.py")
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+df = _load_derived_fields()
+
+
+def _ship(name=None, attn=None):
+    return {"ship_to": {"name": name, "attn": attn}}
+
+
+def _items(*descs_amounts):
+    """Build parsed with line items. Each arg is desc or (desc, amount)."""
+    items = []
+    for entry in descs_amounts:
+        if isinstance(entry, tuple):
+            desc, amount = entry
+        else:
+            desc, amount = entry, "1.00"
+        items.append({"description": desc, "amount": amount})
+    return {"line_items": items}
+
+
+# ===========================================================================
+# site_code -- shapes 1-7
+# ===========================================================================
+def test_site_code_shape1_name_parens():
+    assert df.derive_site_code(_ship(name="Amazon.com Services LLC (KLAL)")) == "KLAL"
+
+
+def test_site_code_shape2_name_after_dash():
+    assert df.derive_site_code(_ship(name="Amazon.com Services LLC - SNY5")) == "SNY5"
+
+
+def test_site_code_shape3_attn_en_dash():
+    parsed = _ship(attn="Wagon Wheel DS Station –WKY3")
+    assert df.derive_site_code(parsed) == "WKY3"
+
+
+def test_site_code_shape4_attn_direct():
+    assert df.derive_site_code(_ship(attn="HJX1")) == "HJX1"
+
+
+def test_site_code_shape5_name_is_code():
+    assert df.derive_site_code(_ship(name="DBU2")) == "DBU2"
+
+
+def test_site_code_shape6_desc_prefix():
+    parsed = _items("DYO1 - Sea Haven Ind - Plumbing Repairs")
+    assert df.derive_site_code(parsed) == "DYO1"
+
+
+def test_site_code_shape7_desc_bracket():
+    parsed = _items("[HMK4] Assemble 3 Wire Security Cages")
+    assert df.derive_site_code(parsed) == "HMK4"
+
+
+def test_site_code_all_letters_code():
+    # Codes may be all letters (no digit required).
+    assert df.derive_site_code(_ship(name="Amazon (KLAL)")) == "KLAL"
+
+
+def test_site_code_leading_token_dash():
+    # Code-first dash layout: opener wins.
+    assert df.derive_site_code(_ship(name="WPT2 - Amazon.com Services LLC")) == "WPT2"
+
+
+def test_site_code_leading_token_parens_beats_paren_content():
+    # Leading digit-bearing opener outranks the parens shape: QDE1, not the host
+    # building WDE1 mentioned inside the parens.
+    assert df.derive_site_code(_ship(name="QDE1 (Co-located inside WDE1)")) == "QDE1"
+
+
+def test_site_code_leading_requires_digit():
+    # A leading all-letter code does NOT trigger the leading shape; it falls to
+    # the parens shape, which resolves the parenthesized code.
+    assert df.derive_site_code(_ship(name="KLAL (SNY5)")) == "SNY5"
+
+
+def test_site_code_dash_after_side():
+    # Documented after-dash layout still works (exact segment after the dash).
+    assert df.derive_site_code(_ship(name="Amazon.com Services LLC - SNY5")) == "SNY5"
+
+
+def test_site_code_parens_attn_hop_resolves_to_code():
+    # Parens content uses the rightmost-non-skip resolver: skip ATTN, take WKY3.
+    parsed = _ship(name="(ATTN: Wagon Wheel DS Station –WKY3)")
+    assert df.derive_site_code(parsed) == "WKY3"
+
+
+def test_site_code_midname_token_with_digit():
+    # A digit-bearing code mid-name is found; all-letter words never match.
+    assert (
+        df.derive_site_code(_ship(name="Amazon Fresh UVA5 Non-Inv (Prime)")) == "UVA5"
+    )
+
+
+def test_site_code_desc_freetoken():
+    # Free-token shape: a length 4-5 digit-bearing code anywhere in a desc.
+    parsed = _items("Need Sea Haven to pump out waste water aqt DSF7")
+    assert df.derive_site_code(parsed) == "DSF7"
+
+
+def test_site_code_desc_freetoken_requires_len4_and_digit():
+    # 3-char uppercase tokens and all-letter tokens do NOT match the free shape.
+    assert df.derive_site_code(_items("call the ABC crew for HELLO work")) is None
+
+
+def test_site_code_two_letter_minimum_po_prefix_rejected():
+    # 'B187' is a 1-letter PO-prefix-style token: the leading shape rejects it
+    # (needs >=2 letters), and the parens shape yields the real code POR3.
+    assert df.derive_site_code(_ship(name="B187 (POR3)")) == "POR3"
+
+
+@pytest.mark.parametrize("code", ["KLAL", "DYO1", "USF4", "QDE1"])
+def test_site_code_two_letter_minimum_accepts_real_codes(code):
+    # Legitimate codes (>=2 alphabetic chars) are still accepted as the name.
+    assert df.derive_site_code(_ship(name=code)) == code
+
+
+@pytest.mark.parametrize("bad", ["B187", "X23", "N456"])
+def test_site_code_one_letter_token_never_accepted(bad):
+    # A code-shaped token with fewer than 2 alphabetic chars is never a code,
+    # across every shape (name-is, ATTN, parens, bracket).
+    assert df.derive_site_code(_ship(name=bad)) is None
+    assert df.derive_site_code(_ship(attn=bad)) is None
+    assert df.derive_site_code(_ship(name=f"Amazon LLC ({bad})")) is None
+    assert df.derive_site_code(_items(f"[{bad}] work")) is None
+
+
+def test_site_code_multi_hop_attn():
+    # "CBRE - RME - DLI6" -> rightmost non-skip code -> DLI6 (RME hop skipped).
+    assert df.derive_site_code(_ship(attn="CBRE - RME - DLI6")) == "DLI6"
+
+
+def test_site_code_priority_parens_beats_dash():
+    parsed = _ship(name="Amazon (KLAL) - SNY5")
+    assert df.derive_site_code(parsed) == "KLAL"
+
+
+def test_site_code_attn_beats_name_is_code():
+    parsed = {"ship_to": {"name": "DBU2", "attn": "HJX1"}}
+    assert df.derive_site_code(parsed) == "HJX1"
+
+
+def test_site_code_prefix_beats_bracket():
+    parsed = {
+        "line_items": [
+            {"description": "Assemble [HMK4] cages"},
+            {"description": "DYO1 - repairs"},
+        ]
+    }
+    # Shape 6 (prefix) is scanned across all items before shape 7 (bracket).
+    assert df.derive_site_code(parsed) == "DYO1"
+
+
+@pytest.mark.parametrize(
+    "token",
+    [
+        "RME",
+        "BBM",
+        "JLL",
+        "PARAG",
+        "ERIK",
+        "HVAC",
+        "LED",
+        "PVC",
+        "ADA",
+        "OSHA",
+        "EMR",
+        "BMS",
+        "DDC",
+        "MRO",
+        "NTE",
+        "EST",
+        "LLC",
+        "INC",
+        "CORP",
+        "LTD",
+        "ATTN",
+    ],
+)
+def test_site_code_skip_list_sole_candidate_is_none(token):
+    # Skip-listed token as the sole candidate for every shape -> None.
+    assert df.derive_site_code(_ship(name=token)) is None
+    assert df.derive_site_code(_ship(name=f"Amazon LLC ({token})")) is None
+    assert df.derive_site_code(_ship(name=f"Amazon LLC - {token}")) is None
+    assert df.derive_site_code(_ship(attn=token)) is None
+    assert df.derive_site_code(_items(f"[{token}] work")) is None
+    assert df.derive_site_code(_items(f"{token} - work")) is None
+
+
+def test_site_code_skip_then_lower_priority_shape_wins():
+    # ATTN skip-listed only, but a line item carries a real prefix code.
+    parsed = {
+        "ship_to": {"name": None, "attn": "RME"},
+        "line_items": [{"description": "DYO1 - Plumbing Repairs"}],
+    }
+    assert df.derive_site_code(parsed) == "DYO1"
+
+
+def test_site_code_lowercase_rejected():
+    assert df.derive_site_code(_ship(name="Amazon (klal)")) is None
+
+
+def test_site_code_no_dash_no_leak_of_llc():
+    # 'LLC' is code-shaped but must not be extracted from a plain name.
+    assert df.derive_site_code(_ship(name="Amazon.com Services LLC")) is None
+
+
+def test_site_code_after_dash_skip_does_not_leak_prefix_token():
+    # 'LLC' before the dash must not leak when the after-dash code is skip-listed.
+    assert df.derive_site_code(_ship(name="Amazon.com Services LLC - RME")) is None
+
+
+@pytest.mark.parametrize("garbage", ["AB", "A1", "TOOLONG6", "1BCD", "", "amazon"])
+def test_site_code_garbage_shapes_rejected(garbage):
+    # Too short, digit-first, too long, lowercase -> not a code.
+    assert df.derive_site_code(_ship(name=garbage)) is None
+
+
+def test_site_code_none_when_nothing():
+    assert df.derive_site_code({"ship_to": {"name": "Some Warehouse"}}) is None
+
+
+# ===========================================================================
+# trade
+# ===========================================================================
+@pytest.mark.parametrize(
+    "desc,expected",
+    [
+        ("Plumbing PM Quarterly", "Plumbing - PM"),
+        ("Plumbing - Backflow test", "Plumbing - PM"),
+        ("Plumbing leak repair", "Plumbing - Reactive"),
+        ("Plumbing unclog toilet", "Plumbing - Reactive"),
+        ("Lighting ballast replacement", "Electrical"),
+        ("Circuit panel upgrade", "Electrical"),
+        ("RTU refrigerant recharge", "HVAC"),
+        ("Air conditioning ductwork", "HVAC"),
+        ("Dock door leveler repair", "Dock Doors"),
+        ("Dock seal replacement", "Dock Doors"),
+        ("Overhead door spring", "Doors"),
+        ("Automatic door operator", "Doors"),
+        ("Wayfinding sign install", "Signage"),
+        ("Cabinet millwork build", "Carpentry"),
+        ("Perimeter fence repair", "Fencing/Gates"),
+        ("Bollard install", "Fencing/Gates"),
+        ("Conveyor belt repair", "Conveyance/MHE"),
+        ("Interior painting touch-up", "Painting"),
+        ("Tile flooring replacement", "Flooring"),
+        ("Custodial cleaning service", "Janitorial"),
+        ("Fire sprinkler inspection", "Fire/Life Safety"),
+        ("Lawn mowing service", "Landscaping/Yard"),
+        ("Roof gutter repair", "Roofing"),
+        ("Access control camera install", "Security/Locksmith"),
+        ("Snow plowing and salt", "Snow Removal"),
+        (
+            "General Building - General Building Technician",
+            "General Building - Handyman",
+        ),
+        ("General Building - General Building Project", "General Building - Project"),
+        # Conveyance keyword (BBM taxonomy) routes to Conveyance/MHE, not the
+        # General Building - Emergency / General Building catch-alls.
+        ("Conveyance - Emergency - Opex CAR", "Conveyance/MHE"),
+        ("Conveyance - Conveyance Technician - BBM", "Conveyance/MHE"),
+        # BBM-structured 'Plumbing - ...' rows resolve via the measured ladder:
+        # (a) emergency/reactive -> Reactive, (b) technician -> PM, (c) water
+        # heater/backflow/fountain -> PM, (d) clog/leak/repair/... -> Reactive,
+        # (e) else -> PM.
+        ("Plumbing - Plumbing Technician - BBM", "Plumbing - PM"),  # (b)
+        ("Plumbing - Emergency Callout - BBM", "Plumbing - Reactive"),  # (a)
+        ("Plumbing - Toilet - Toilet Repair - BBM", "Plumbing - Reactive"),  # (d)
+        ("Plumbing - Sink - Sink Clogged - BBM", "Plumbing - Reactive"),  # (d)
+        # (c) beats (d): water heater/fountain 'repair' stays PM.
+        (
+            "Plumbing - Water Heater - Water Heater Repair - BBM",
+            "Plumbing - PM",
+        ),
+        (
+            "Plumbing - Water Fountain - Water Fountain Repair - BBM",
+            "Plumbing - PM",
+        ),
+        ("Plumbing - Plumbing Project - Medium", "Plumbing - PM"),  # (e)
+        # Plumbing-specific qualifiers indicate Reactive even without 'plumbing'.
+        ("EPO FOR SEA HAVEN (CLOGGED PIT AUGER)", "Plumbing - Reactive"),
+        ("Unclog the floor drain in restroom", "Plumbing - Reactive"),
+        # General Building - Project widening: a standalone 'project' segment.
+        ("General Building - Project - BBM", "General Building - Project"),
+        ("Ops Driven Defect - Project - Large", "General Building - Project"),
+        # Handyman widening: only a standalone 'handyman' word (5a) and the
+        # 'General Building Technician' prompt row -> Handyman.
+        ("Handyman services for the facility", "General Building - Handyman"),
+        # Keyword-less BBM 'General Building - <Fixture> - ...' rows land on the
+        # plain catch-all (measured LLM majority; fix 5b reverted).
+        ("General Building - Locker - Install - BBM", "General Building"),
+        ("General Building - Key Box - Repair - BBM", "General Building"),
+        ("General Building - Fan - Fan Install - BBM", "General Building"),
+        ("General Building - Bike Rack - Removal - BBM", "General Building"),
+        # Bollard stays Fencing/Gates (prompt rule; unchanged).
+        ("General Building - Bollard - Install - BBM", "Fencing/Gates"),
+        ("Facility maintenance misc work", "General Building"),
+    ],
+)
+def test_trade_single_keyword(desc, expected):
+    assert df.derive_trade(_items(desc)) == expected
+
+
+def test_trade_pm_beats_reactive():
+    # "plumbing pm" (priority 1) outranks the "leak" reactive qualifier.
+    assert df.derive_trade(_items("Plumbing PM inspection incl leak check")) == (
+        "Plumbing - PM"
+    )
+
+
+def test_trade_dock_door_suppresses_electrical():
+    # Electrical keyword present, but a dock-door context routes to Dock Doors.
+    parsed = _items("Dock door electrical panel repair")
+    assert df.derive_trade(parsed) == "Dock Doors"
+
+
+def test_trade_dock_door_beats_doors():
+    assert df.derive_trade(_items("Dock door seal replacement")) == "Dock Doors"
+
+
+def test_trade_dock_gate_not_fencing():
+    parsed = _items("Dock gate maintenance")
+    assert df.derive_trade(parsed) != "Fencing/Gates"
+    assert df.derive_trade(parsed) == "General Building"
+
+
+def test_trade_locker_is_not_locksmith():
+    # 'Locker' must NOT match the whole word 'lock' -> not Security/Locksmith.
+    parsed = _items("Locker replacement in break room")
+    assert df.derive_trade(parsed) != "Security/Locksmith"
+    assert df.derive_trade(parsed) == "General Building"
+
+
+def test_trade_key_box_is_not_locksmith():
+    # 'Key Box' is stripped before the Security test -> not Security/Locksmith.
+    parsed = _items("Key Box installation at dock office")
+    assert df.derive_trade(parsed) != "Security/Locksmith"
+
+
+def test_trade_lock_whole_word_still_locksmith():
+    # A real whole-word 'lock' still routes to Security/Locksmith.
+    assert df.derive_trade(_items("Broken lock needs replacement")) == (
+        "Security/Locksmith"
+    )
+
+
+def test_trade_emergency_prefix():
+    assert (
+        df.derive_trade(_items("EMER site response")) == "General Building - Emergency"
+    )
+
+
+def test_trade_emergency_word():
+    parsed = _items("Emergency board-up of facility")
+    assert df.derive_trade(parsed) == "General Building - Emergency"
+
+
+def test_trade_uplift_only():
+    parsed = _items("PO Uplift", "PO Uplift adjustment")
+    assert df.derive_trade(parsed) == "PO Uplift"
+
+
+def test_trade_uplift_plus_single_trade():
+    # One non-uplift trade + uplift -> the non-uplift trade is primary.
+    parsed = _items(("PO Uplift", "10.00"), ("Plumbing leak repair", "500.00"))
+    assert df.derive_trade(parsed) == "Plumbing - Reactive"
+
+
+def test_trade_mixed_highest_value_wins():
+    parsed = _items(
+        ("PO Uplift", "9999.00"),  # excluded from being primary
+        ("Plumbing leak repair", "500.00"),
+        ("Lighting circuit repair", "800.00"),
+    )
+    assert df.derive_trade(parsed) == "Electrical"
+
+
+def test_trade_mixed_decimal_amounts():
+    parsed = {
+        "line_items": [
+            {"description": "Roof gutter repair", "amount": Decimal("120.00")},
+            {"description": "Fire sprinkler inspection", "amount": Decimal("900.00")},
+        ]
+    }
+    assert df.derive_trade(parsed) == "Fire/Life Safety"
+
+
+def test_trade_no_line_items_is_none():
+    assert df.derive_trade({"line_items": []}) is None
+    assert df.derive_trade({}) is None
+
+
+def test_trade_no_descriptions_is_none():
+    assert df.derive_trade({"line_items": [{"amount": "10.00"}]}) is None
+    assert df.derive_trade({"line_items": [{"description": "   "}]}) is None
+
+
+def test_trade_general_building_only_with_a_description():
+    # A described-but-unmatched item -> General Building (the catch-all).
+    assert df.derive_trade(_items("Misc facility task")) == "General Building"
+
+
+# ===========================================================================
+# fiscal_year
+# ===========================================================================
+def test_fiscal_year_order_date_full_year():
+    assert df.derive_fiscal_year({"order_date": "06/15/2025"}) == "2025"
+
+
+def test_fiscal_year_order_date_iso():
+    assert df.derive_fiscal_year({"order_date": "2024-11-02"}) == "2024"
+
+
+def test_fiscal_year_order_date_yy_expansion():
+    assert df.derive_fiscal_year({"order_date": "06/15/25"}) == "2025"
+
+
+def test_fiscal_year_need_by_fallback():
+    parsed = {
+        "order_date": None,
+        "line_items": [{"need_by": "2024-03-01"}],
+    }
+    assert df.derive_fiscal_year(parsed) == "2024"
+
+
+def test_fiscal_year_need_by_yy_expansion():
+    parsed = {"order_date": None, "line_items": [{"need_by": "3/1/23"}]}
+    assert df.derive_fiscal_year(parsed) == "2023"
+
+
+def test_fiscal_year_description_fallback():
+    parsed = {
+        "order_date": None,
+        "line_items": [{"need_by": None, "description": "HVB2 - 2025 - Plumbing PM"}],
+    }
+    assert df.derive_fiscal_year(parsed) == "2025"
+
+
+def test_fiscal_year_order_date_precedence():
+    parsed = {
+        "order_date": "2022-01-01",
+        "line_items": [{"need_by": "2025-01-01", "description": "2030 work"}],
+    }
+    assert df.derive_fiscal_year(parsed) == "2022"
+
+
+def test_fiscal_year_none_when_nothing():
+    parsed = {"order_date": "no date here", "line_items": [{"description": "work"}]}
+    assert df.derive_fiscal_year(parsed) is None
+
+
+def test_fiscal_year_ignores_digits_inside_po_id():
+    # '18206023' must not yield a bogus year via an embedded '2062' run.
+    assert df.derive_fiscal_year({"order_date": "PO 2D-18206023"}) is None
+
+
+# ===========================================================================
+# derive_all
+# ===========================================================================
+def test_derive_all_shape_and_values():
+    parsed = {
+        "ship_to": {"name": "Amazon.com Services LLC (KLAL)"},
+        "order_date": "06/15/2025",
+        "line_items": [{"description": "Plumbing leak repair", "amount": "500.00"}],
+    }
+    assert df.derive_all(parsed) == {
+        "site_code": "KLAL",
+        "trade": "Plumbing - Reactive",
+        "fiscal_year": "2025",
+    }
+
+
+def test_derive_all_keys_always_present():
+    result = df.derive_all(None)
+    assert set(result) == {"site_code", "trade", "fiscal_year"}
+    assert result == {"site_code": None, "trade": None, "fiscal_year": None}
+
+
+# ===========================================================================
+# Totality fuzz -- nothing raises, everything returns None-ish, fast
+# ===========================================================================
+_HOSTILE = [
+    None,
+    {},
+    "not a dict",
+    123,
+    [],
+    {"ship_to": "garbage-string"},
+    {"ship_to": None},
+    {"ship_to": ["a", "b"]},
+    {"ship_to": {"name": 123, "attn": ["x"]}},
+    {"line_items": None},
+    {"line_items": {"a": 1}},
+    {"line_items": "string"},
+    {"line_items": ["foo", "bar", 42, None]},
+    {"line_items": [None, 1, {"description": None}, {"description": 5}]},
+    {"order_date": 12345},
+    {"order_date": ["2025"]},
+    {"line_items": [{"description": "x", "amount": object()}]},
+    {"line_items": [{"description": "PO Uplift", "amount": "not-a-number"}]},
+]
+
+
+@pytest.mark.parametrize("parsed", _HOSTILE)
+def test_totality_no_raise(parsed):
+    # Every public function tolerates hostile input without raising.
+    assert df.derive_site_code(parsed) is None or isinstance(
+        df.derive_site_code(parsed), str
+    )
+    assert df.derive_trade(parsed) is None or isinstance(df.derive_trade(parsed), str)
+    assert df.derive_fiscal_year(parsed) is None or isinstance(
+        df.derive_fiscal_year(parsed), str
+    )
+    result = df.derive_all(parsed)
+    assert set(result) == {"site_code", "trade", "fiscal_year"}
+
+
+def test_totality_huge_strings_fast():
+    huge = "x" * 1_000_000
+    parsed = {
+        "ship_to": {"name": huge, "attn": huge},
+        "order_date": huge,
+        "line_items": [{"description": huge, "amount": "1.00"}],
+    }
+    start = time.time()
+    site = df.derive_site_code(parsed)
+    trade = df.derive_trade(parsed)
+    year = df.derive_fiscal_year(parsed)
+    elapsed = time.time() - start
+    assert elapsed < 2.0
+    assert site is None
+    # A million 'x' is a non-empty description that matches nothing.
+    assert trade == "General Building"
+    assert year is None
+
+
+def test_totality_huge_bracket_code_scan_capped():
+    # A code past the scan cap must not be found (and must not hang).
+    desc = ("." * 10_000) + "[HMK4] work"
+    start = time.time()
+    result = df.derive_site_code(_items(desc))
+    assert time.time() - start < 2.0
+    assert result is None
+
+
+class TestAggregateTradeBudget:
+    """derive_trade must bound AGGREGATE work across items, not just per-item."""
+
+    def test_adversarial_many_large_items_completes_fast(self):
+        import time
+
+        parsed = {
+            "line_items": [
+                {"description": ("air " * 25000)[:100000], "amount": "1"}
+                for _ in range(500)
+            ]
+        }
+        start = time.perf_counter()
+        result = df.derive_trade(parsed)
+        elapsed = time.perf_counter() - start
+        # Budget-capped: only ~2 of the 500 items are classified; the rest are
+        # skipped. Must complete in well under a second even on slow hardware.
+        assert elapsed < 2.0
+        assert result == "General Building"
+
+    def test_budget_never_degrades_realistic_multi_item_pos(self):
+        # Real multi-line POs total well under the budget: every item is
+        # classified and the highest-value non-uplift rule still applies.
+        parsed = {
+            "line_items": [
+                {"description": "PO Uplift", "amount": "9000"},
+                {"description": "dock door repair", "amount": "100"},
+                {"description": "electrical panel replacement", "amount": "500"},
+            ]
+        }
+        assert df.derive_trade(parsed) == "Electrical"

lambdas/po/email_processor/tests/test_po_derived_wiring.py

@@ -0,0 +1,221 @@
+"""Wiring tests for the Python derived-field classifier in enrich_parsed.
+
+These cover the SHADOW semantics only -- how the handler fills / keeps
+site_code/trade/fiscal_year and emits the DerivedFieldAgreement EMF metric --
+NOT the derive_all classifier itself (that is owned by
+test_po_derived_fields.py). derive_all is exercised through its public seam on
+the handler module (``po_handler.derive_all``), monkeypatched to drive each
+agreement category deterministically, plus one real-integration smoke.
+"""
+
+import json
+
+from _po_parser_support import (
+    NEW_PO_STEMS,
+    load_email,
+    po_handler,
+    template_parser,
+)
+
+
+def _emf_lines(capsys):
+    """Parse the EMF records enrich_parsed printed to stdout, keyed by Field."""
+    out = capsys.readouterr().out
+    records = [json.loads(ln) for ln in out.splitlines() if ln.strip()]
+    return {rec["Field"]: rec for rec in records}
+
+
+def _patch_derive_all(monkeypatch, **vals):
+    result = {"site_code": None, "trade": None, "fiscal_year": None}
+    result.update(vals)
+    monkeypatch.setattr(po_handler, "derive_all", lambda parsed: result)
+
+
+def _parsed(**overrides):
+    base = {
+        "po_number": "2D-70000001",
+        "site_code": None,
+        "trade": None,
+        "fiscal_year": None,
+    }
+    base.update(overrides)
+    return base
+
+
+# --- template path -----------------------------------------------------------
+
+
+def test_template_path_fills_none_fields_and_emits_no_emf(monkeypatch, capsys):
+    """On the template path every field arrives None (the gate proves the parser
+    leaves them unset); Python fills all three and NO agreement EMF is emitted
+    (there is no LLM value to shadow)."""
+    _patch_derive_all(
+        monkeypatch,
+        site_code="HMK4",
+        trade="Plumbing - Reactive",
+        fiscal_year="2025",
+    )
+    parsed = _parsed()
+    out = po_handler.enrich_parsed(parsed, "s3://b/k", "subj", parse_method="template")
+    assert out["site_code"] == "HMK4"
+    assert out["trade"] == "Plumbing - Reactive"
+    assert out["fiscal_year"] == "2025"
+    assert _emf_lines(capsys) == {}  # no agreement EMF on the template path
+
+
+# --- ai_fallback path: fill / keep + agreement categories --------------------
+
+
+def test_ai_fallback_keeps_llm_value_on_disagree(monkeypatch, capsys):
+    """LLM value stays authoritative during the bake even when Python disagrees;
+    the shadow EMF records 'disagree' carrying BOTH values as properties."""
+    _patch_derive_all(monkeypatch, site_code="HMK4")
+    parsed = _parsed(site_code="DYO1")
+    out = po_handler.enrich_parsed(
+        parsed, "s3://b/k", "subj", parse_method="ai_fallback"
+    )
+    assert out["site_code"] == "DYO1"  # LLM survives, not overwritten
+
+    rec = _emf_lines(capsys)["site_code"]
+    assert rec["Agreement"] == "disagree"
+    assert rec["LlmValue"] == "DYO1"
+    assert rec["PythonValue"] == "HMK4"
+    assert rec["po_number"] == "2D-70000001"
+    assert rec[po_handler.DERIVED_METRIC_NAME] == 1
+    metric = rec["_aws"]["CloudWatchMetrics"][0]
+    assert metric["Namespace"] == "Seahaven/PoIngest"
+    assert metric["Dimensions"] == [["Field", "Agreement"]]
+    assert metric["Metrics"] == [
+        {"Name": po_handler.DERIVED_METRIC_NAME, "Unit": "Count"}
+    ]
+
+
+def test_ai_fallback_agree_after_strip(monkeypatch, capsys):
+    """Equality is str-strip normalized; equal values record 'agree' and the LLM
+    value is kept verbatim (Python's whitespace-variant does not overwrite)."""
+    _patch_derive_all(monkeypatch, site_code=" DYO1 ")
+    parsed = _parsed(site_code="DYO1")
+    out = po_handler.enrich_parsed(
+        parsed, "s3://b/k", "subj", parse_method="ai_fallback"
+    )
+    assert out["site_code"] == "DYO1"
+    assert _emf_lines(capsys)["site_code"]["Agreement"] == "agree"
+
+
+def test_ai_fallback_llm_null_python_filled(monkeypatch, capsys):
+    """LLM None + Python non-None: Python fills the gap and the record is
+    categorized 'llm_null_python_filled'."""
+    _patch_derive_all(monkeypatch, trade="HVAC")
+    parsed = _parsed()
+    out = po_handler.enrich_parsed(
+        parsed, "s3://b/k", "subj", parse_method="ai_fallback"
+    )
+    assert out["trade"] == "HVAC"  # gap filled
+
+    rec = _emf_lines(capsys)["trade"]
+    assert rec["Agreement"] == "llm_null_python_filled"
+    assert rec["LlmValue"] == ""
+    assert rec["PythonValue"] == "HVAC"
+
+
+def test_ai_fallback_python_null(monkeypatch, capsys):
+    """LLM non-None + Python None: LLM value kept, record categorized
+    'python_null'."""
+    _patch_derive_all(monkeypatch)  # all None
+    parsed = _parsed(fiscal_year="2025")
+    out = po_handler.enrich_parsed(
+        parsed, "s3://b/k", "subj", parse_method="ai_fallback"
+    )
+    assert out["fiscal_year"] == "2025"
+
+    rec = _emf_lines(capsys)["fiscal_year"]
+    assert rec["Agreement"] == "python_null"
+    assert rec["PythonValue"] == ""
+    assert rec["LlmValue"] == "2025"
+
+
+def test_ai_fallback_both_none_emits_nothing(monkeypatch, capsys):
+    """Both None: nothing to compare, so NO EMF record is emitted for that
+    field (cardinality is not spent on empty comparisons)."""
+    _patch_derive_all(monkeypatch)  # all None
+    parsed = _parsed()  # all None
+    po_handler.enrich_parsed(parsed, "s3://b/k", "subj", parse_method="ai_fallback")
+    assert _emf_lines(capsys) == {}
+
+
+def test_ai_fallback_emits_one_record_per_comparable_field(monkeypatch, capsys):
+    """Cardinality check: exactly one record per field that has something to
+    compare -- here two fields, the both-None third is skipped."""
+    _patch_derive_all(monkeypatch, site_code="HMK4", trade="HVAC")
+    parsed = _parsed(site_code="DYO1")  # fiscal_year both-None -> skipped
+    po_handler.enrich_parsed(parsed, "s3://b/k", "subj", parse_method="ai_fallback")
+    recs = _emf_lines(capsys)
+    assert set(recs) == {"site_code", "trade"}
+    assert recs["site_code"]["Agreement"] == "disagree"
+    assert recs["trade"]["Agreement"] == "llm_null_python_filled"
+
+
+# --- totality: telemetry/classification never fails the invocation -----------
+
+
+def test_derive_all_exception_cannot_propagate(monkeypatch):
+    """derive_all is total, but the whole block is defensively wrapped anyway:
+    a raising classifier must NOT propagate (this S3-async Lambda would retry ->
+    DLQ). enrich_parsed still returns, earlier enrichment (zip padding) is
+    intact, and the derived fields are left untouched."""
+
+    def boom(parsed):
+        raise RuntimeError("classifier blew up")
+
+    monkeypatch.setattr(po_handler, "derive_all", boom)
+    parsed = _parsed(ship_to={"address": "x", "state": "MA", "zip": "2149"})
+    out = po_handler.enrich_parsed(
+        parsed, "s3://b/k", "subj", parse_method="ai_fallback"
+    )
+    assert out is parsed  # returned normally, no exception
+    assert out["ship_to"]["zip"] == "02149"  # pre-derive enrichment applied
+    assert out["site_code"] is None  # derived block bailed cleanly
+
+
+def test_emit_derived_agreement_metric_skips_when_both_none(capsys):
+    """The helper itself is the both-None gate: no line is printed."""
+    po_handler._emit_derived_agreement_metric("site_code", None, None, "2D-1")
+    assert capsys.readouterr().out == ""
+
+
+# --- EXTRACTION_PROMPT numeric declaration -----------------------------------
+
+
+def test_extraction_prompt_declares_quantity_price_numeric():
+    """The deferred PR #1 prompt tweak: line-item quantity/price are declared
+    numeric, not 'string or null'. The enrich_parsed Decimal coercion stays as a
+    safety net (test_enrich_parsed_coerces_prompt_string_quantity_price)."""
+    prompt = po_handler.EXTRACTION_PROMPT
+    assert '"quantity": "number or null"' in prompt
+    assert '"price": "number or null"' in prompt
+    assert '"quantity": "string or null"' not in prompt
+    assert '"price": "string or null"' not in prompt
+
+
+# --- real derive_all integration smoke ---------------------------------------
+
+
+def test_template_path_real_derive_all_is_wired(monkeypatch):
+    """End-to-end against the real derive_all: a template-parsed new_po (derived
+    fields None) runs through enrich_parsed and each derived field ends up
+    consistent with what the real classifier yields for the final dict -- proving
+    the classifier is actually invoked and its output applied (no monkeypatch)."""
+    email_data = load_email("new-po", NEW_PO_STEMS[0])
+    parsed, method, _t, reason = template_parser.try_deterministic_parse(email_data)
+    assert (method, reason) == ("template", "ok")
+    for key in template_parser.DERIVED_KEYS:
+        assert parsed[key] is None  # parser leaves them unset
+
+    po_handler.enrich_parsed(
+        parsed, "s3://b/k", email_data["subject"], parse_method="template"
+    )
+    # derive_all is pure and the fill does not feed back into its inputs, so the
+    # filled values must equal a recomputation over the enriched dict.
+    recomputed = po_handler.derive_all(parsed)
+    for key in template_parser.DERIVED_KEYS:
+        assert parsed[key] == recomputed[key]

tests/conftest.py

@@ -27,7 +27,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
 # be bound to the right pipeline's file around each handler exec -- relying on
 # sys.path ordering (or on whatever a previously collected suite left in
 # sys.modules) silently binds a handler to the OTHER pipeline's sibling.
-_SIBLING_MODULES = ("ses_auth", "template_parser")
+_SIBLING_MODULES = ("ses_auth", "template_parser", "derived_fields")
 
 
 def _load_module(path, module_name):