mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Adam Moussa
bb4d0710e8
Record the provider switch to the Bedrock inference profile (no Anthropic API key or Secrets Manager secret, with the retired secrets flagged for manual deletion), the WO deterministic-template-first + AI-fallback flow, the new ParseOutcome EMF metric and template-fallback-rate alarm, the issue #23 comment_id format change, the +00:00 aware-UTC timestamp shift, and offline test instructions. Refs: #23
235 lines
19 KiB
Markdown
235 lines
19 KiB
Markdown
# Procurement Ingest
|
|
|
|
|
|
|
|
|
|
|
|
Unified email ingestion pipelines for Amazon procurement data. Two independent pipelines — purchase orders (Coupa) and work orders (APM/Hexagon EAM) — share a single repo and CDK app but deploy as separate CloudFormation stacks.
|
|
|
|
## Pipelines
|
|
|
|
### Purchase Orders (`po-ingest` stack)
|
|
|
|
Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude Haiku 4.5 **on Amazon Bedrock**, and written to the `purchase-orders` DynamoDB table. PO parsing is fully AI (no deterministic parser).
|
|
|
|
**Flow:**
|
|
1. Coupa sends a PO email (new, revision, or cancellation).
|
|
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. Claude (via Bedrock `InvokeModel`) extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). Numeric amounts are decoded with `parse_float=Decimal` (DynamoDB rejects floats).
|
|
5. Conditional write to DynamoDB:
|
|
- `new_po` — idempotent insert (no-op if PO exists)
|
|
- `revision` — unconditional overwrite
|
|
- `cancellation` — marks existing row `Cancelled`
|
|
6. DynamoDB Streams feeds downstream consumers:
|
|
- **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync
|
|
- **Site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into `verified-sites` table
|
|
|
|
**Lambdas** (`lambdas/po/`):
|
|
| Function | Trigger | Purpose |
|
|
|---|---|---|
|
|
| `po-email-processor` | S3 ObjectCreated | Claude extraction + DynamoDB write |
|
|
| `po-ingest-site-extractor` | DynamoDB Streams | Site code/address extraction -> `verified-sites` |
|
|
| `po-web-ui` | Manual invoke | HTML dashboard (public Function URL removed 2026-06-08, INFRA-74) |
|
|
|
|
**Tables:**
|
|
- `purchase-orders` (PK: `po_number`, Streams: NEW_IMAGE) — shared with seahaven-slack-bot (read-only; see Shared Resources)
|
|
- `verified-sites` (PK: `siteCode`) — ~1,100 unique Amazon facility sites (`by-state` GSI removed 2026-06-03, audit M-20)
|
|
- `pending-site-review` (PK: `po_number`) — unresolvable POs for manual Payee Central verification
|
|
|
|
### Work Orders (`WorkorderIngestStack` stack)
|
|
|
|
Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received at `apm@int.seahaven.com`, parsed **deterministic-template-first with a Claude-on-Bedrock fallback**, and written to the `WorkOrders` DynamoDB table.
|
|
|
|
**Flow:**
|
|
1. Hexagon EAM sends email notifications (new assignments, comments, updates, cancellations) to `amazon@seahavenind.com`.
|
|
2. Gmail filter forwards APM emails to `apm@int.seahaven.com` (SES).
|
|
3. SES drops the raw MIME into `s3://workorder-ingest-emails-{AccountId}/inbound/`.
|
|
4. S3 triggers the `workorder-email-processor` Lambda.
|
|
5. **Parse:** a pure, offline template parser (`template_parser.py`) tries the two known Hexagon templates first, 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. Both paths emit the identical structured-JSON contract (work order ID, site code, severity, priority, dates, assigned technician).
|
|
6. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`.
|
|
|
|
**Deterministic template parser.** ~93.6% of WO traffic is the plain-text "AMAZON UPDATE WO DETAILS \<id\>" comment template (T1) and ~6.4% is the HTML "AMAZON assign Work Order \<id\> on building \<SITE\>" assignment template (T2). `template_parser.try_deterministic_parse()` classifies by subject, extracts the contract fields, and returns a parsed result **only if** it passes a strict validation gate (exact contract-key set; `work_order_id` matches the subject and is all-digits; the T1 `Work Order: <id>` double-space is literally present; `email_type` matches the template; `site_code` shape; per-type required fields; and a label-bleed guard so a value that over-ran into another field fails). Anything that fails — the rare update/cancellation shapes, Hexagon template drift, or an extractor exception — falls back to the AI extractor. **Data is never corrupted; only the fallback rate rises.** Every record emits one CloudWatch EMF metric (see below).
|
|
|
|
**Lambdas** (`lambdas/wo/`):
|
|
| Function | Trigger | Purpose |
|
|
|---|---|---|
|
|
| `workorder-email-processor` | S3 ObjectCreated | Claude extraction + DynamoDB write |
|
|
| `workorder-web-ui` | Manual invoke | HTML dashboard (public Function URL removed 2026-06-08, INFRA-74) |
|
|
|
|
**Tables:**
|
|
- `WorkOrders` (PK: `work_order_id`) — `site-code-index` and `status-index` GSIs removed 2026-06-03 (audit M-20)
|
|
- `WorkOrderComments` (PK: `work_order_id`, SK: `comment_id`) — see the `comment_id` format note below
|
|
|
|
## Architecture
|
|
|
|
**IaC:** AWS CDK (Python), two stacks in one app, region `us-east-1`.
|
|
|
|
All Lambdas: Python 3.12, ARM64, 60-day log retention.
|
|
|
|
**LLM provider — Amazon Bedrock.** Both email processors call Claude Haiku 4.5 through the Bedrock inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0` (`bedrock-runtime` `InvokeModel`), configured via the `BEDROCK_MODEL_ID` env var. There is **no Anthropic API key and no Secrets Manager secret** any more. Each processor role is granted `bedrock:InvokeModel` + `bedrock:InvokeModelWithResponseStream` on **both** the inference-profile ARN **and** the per-region foundation-model ARNs for `us-east-1` / `us-east-2` / `us-west-2` (empty-account foundation-model ARNs) — the `us.*` profile can route cross-region, so a profile-only grant would `AccessDenied` at runtime under load.
|
|
|
|
> **Retired secrets (manual cleanup outstanding):** the former secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` had `RemovalPolicy.RETAIN`, so removing them from the CDK stacks **orphans** them rather than deleting them. Delete both by hand post-deploy and revoke the stored keys at the provider.
|
|
|
|
**SES:** Both stacks add rules to the shared `INBOUND_MAIL` receipt rule set on `int.seahaven.com`.
|
|
|
|
**Failure handling (INFRA-41):** Each email-processor is async-invoked (S3 → Lambda). Both have a CDK-managed SQS dead-letter queue (`dead_letter_queue=`, 14-day retention, SSL-enforced) so a failed parse is captured rather than silently dropped after Lambda's retries.
|
|
|
|
### CloudWatch alarms
|
|
|
|
Every alarm is **ALARM-only** (no OK action), sends to the shared `site-alerts` SNS topic (imported once per stack via `Topic.from_topic_arn`), and uses `TreatMissingData.NOT_BREACHING`.
|
|
|
|
**Lambda alarms** (`AWS/Lambda`, `FunctionName` dimension):
|
|
|
|
| Alarm | Functions | Metric / config |
|
|
|---|---|---|
|
|
| `<fn>-errors` | `po-email-processor`, `po-ingest-site-extractor`, `workorder-email-processor` | `Errors` Sum, 5 min, `> 0`, eval 1 |
|
|
| `<fn>-throttles` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-email-processor` | `Throttles` Sum, 5 min, `> 0`, eval 1 |
|
|
| `<fn>-duration` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui` (p99); `workorder-email-processor` (p95) | `Duration` percentile, 5 min, `>= 45000` ms (75% of the 60s timeout), eval 3 / datapoints 2 |
|
|
|
|
The `<fn>-duration` and `<fn>-throttles` alarms for `po-email-processor` and `workorder-email-processor` supersede the orphaned, CLI-created `Lambda-Duration-*` / `Lambda-Throttles-*` alarms (deleted post-deploy).
|
|
|
|
**DLQ alarms** (`AWS/SQS`): `po-email-processor-dlq-messages` and `workorder-email-processor-dlq-messages` fire when any message is visible on an email-processor DLQ (`ApproximateNumberOfMessagesVisible` Maximum, 5 min, `> 0`, eval 1) — a message there means an email was dropped after Lambda exhausted its async retries.
|
|
|
|
**Parse-outcome metric + fallback-rate alarm (workorder-ingest):** the WO processor writes one CloudWatch **EMF** line per email to namespace `Seahaven/WorkorderIngest`, metric `ParseOutcome` (Unit Count, value 1), dimensioned by `ParseMethod` (`template` | `ai_fallback`) and `TemplateId` (`update_plaintext` | `assign_html` | `unknown`). Non-dimension EMF properties `ReasonCode` and `work_order_id` are queryable in Logs Insights but not promoted to metrics (kept low-cardinality). EMF is used instead of `PutMetricData` so there is no extra sync call / latency / IAM grant on the async hot path (the role already has `logs:PutLogEvents`). The alarm `workorder-email-processor-template-fallback-rate` fires when the AI-fallback share of parses exceeds **15%** sustained (a `MathExpression` with `FILL(...,0)` and a ≥10-sample volume floor over 15-minute periods, eval 3 / datapoints 2) — catching Hexagon template-drift coverage collapse while the volume floor + `FILL` prevent low-volume false pages / `INSUFFICIENT_DATA`. ALARM-only `SnsAction` to `site-alerts`, no OK action, `NOT_BREACHING`. The 15-minute period is a deliberate deviation from the 5-minute house style to accumulate a stable denominator at the low ~760/day volume.
|
|
|
|
**DynamoDB alarms** (`AWS/DynamoDB`): each owned table gets `<table>-throttles` (`ThrottledRequests`) and `<table>-system-errors` (`SystemErrors`). These metrics emit only at the `TableName` + `Operation` dimension set, so each alarm is a `Sum` math expression across the operations the table uses (Get/BatchGet/Query/Scan/Put/Update/Delete/BatchWrite). Tables covered: `purchase-orders`, `verified-sites`, `pending-site-review` (po-ingest); `WorkOrders`, `WorkOrderComments` (workorder-ingest).
|
|
|
|
## Shared Resources
|
|
|
|
### `purchase-orders` table (owned here)
|
|
|
|
The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack** (defined in `cdk/po_stack.py` with `RemovalPolicy.RETAIN`, `StreamViewType.NEW_IMAGE`, and SSE-KMS encryption with the shared customer-managed CMK `alias/seahaven-dynamodb`, INFRA-95 / M-3). The `po-email-processor` Lambda is the authoritative writer — it performs the conditional inserts, revision overwrites, and cancellation updates described above.
|
|
|
|
**Consumers (read-only):**
|
|
|
|
| Repo | How it reads | Purpose |
|
|
|---|---|---|
|
|
| `seahaven-slack-bot` | `po-sync` (DynamoDB Streams + daily scan) and `wo-po-lookup` | Daily KB sync + Bedrock agent PO lookups |
|
|
|
|
The consumer imports the table via `Table.fromTableName(...)` and is granted read-only access (`grantReadData`); it does not own or define it.
|
|
|
|
**Schema-coordination rule:** Any change to the `purchase-orders` schema (partition key, item shape, attribute names, streams view type) must be coordinated with `seahaven-slack-bot`. The owner here ships the change; the consumer must be updated in lockstep so its readers do not break. Treat schema changes as a cross-repo migration, not a local edit.
|
|
|
|
**Known exception (INFRA-51):** `amazon-po-parser` currently writes directly to `purchase-orders` outside this stack (backfill/enrichment scripts). This second writer is being folded into the `po-ingest` pipeline so this stack is the sole writer; until INFRA-51 closes, coordinate any schema change with `amazon-po-parser` as well.
|
|
|
|
### `WorkOrders` and `WorkOrderComments` tables (owned here)
|
|
|
|
Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.py`, `RemovalPolicy.RETAIN`):
|
|
|
|
- `WorkOrders` — PK `work_order_id` (S).
|
|
- `WorkOrderComments` — PK `work_order_id` (S), SK `comment_id` (S).
|
|
|
|
> **`comment_id` format change (issue #23).** The `WorkOrderComments` range key is now
|
|
> `work_order_id#<comment_time|nocomment>#<sha256(s3_object_key)[:12]>`
|
|
> (e.g. `11144580730#2026-04-27T23:51:48#a1b2c3d4e5f6`, or `…#nocomment#…` when the source
|
|
> email carries no comment time). Previously it was `work_order_id#<timestamp>`, where two
|
|
> emails on the same WO with an identical/absent comment time collided and overwrote each other.
|
|
> The 12-hex suffix is derived from the **S3 object key alone** — deterministic, so a Lambda
|
|
> async **retry** of the same object produces a byte-identical key (idempotent, no duplicate row),
|
|
> while two distinct emails on the same WO get distinct keys. Wall-clock `now()` is kept **out** of
|
|
> the key. Consumers that split on `#` and read index `[0]`/`[1]` are unaffected; anything that
|
|
> treated "everything after the first `#`" as a bare timestamp now also captures the hash segment.
|
|
|
|
> **Timestamp format shift.** All stored ISO timestamps (`created_at`, `updated_at`, `ingested_at`
|
|
> on WO; `processed_at`, `cancelled_at` on PO) moved from naive `datetime.utcnow().isoformat()`
|
|
> to timezone-aware `datetime.now(timezone.utc).isoformat()`, so they now carry a `+00:00` suffix
|
|
> (e.g. `2026-07-15T12:00:00+00:00`). Downstream parsers that assumed a naive/no-offset string
|
|
> must accept the offset.
|
|
|
|
Both currently use default DynamoDB encryption — they are **not** yet on the shared customer-managed CMK (`alias/seahaven-dynamodb`, INFRA-95 / M-3); that migration is tracked in INFRA-6. The `workorder-email-processor` role already holds a pre-emptive encrypt/decrypt grant on the CMK so the migration won't break it.
|
|
|
|
**Consumer (read-only) — data contract:** `seahaven-slack-bot` imports both tables via `Table.fromTableName(...)` (`grantReadData`) and reads them from two Lambdas: `workorder-sync` (daily full-table scan into the Bedrock knowledge base) and `wo-po-lookup` (the Bedrock agent's direct WO lookup action group). The bot depends on the PK/SK schema above and these attributes: on `WorkOrders` — `description`, `wo_status`, `customer`, `site_code`, `building`, `address`, `severity`, `priority`, `assigned_to`, `date_reported`, `scheduled_start`, `due_date`, `updated_at`; on `WorkOrderComments` — `created_at` (used to sort comments), `commenter`, `text`. Any change to table name, key schema, these attribute names, or the encryption configuration (e.g. the INFRA-6 CMK migration) must be coordinated with `seahaven-slack-bot` before it ships, or the Bedrock agent breaks at runtime (not at deploy — the tables are imported by name, so there is no compile-time link).
|
|
|
|
### `verified-sites` table (owned here)
|
|
|
|
Owned by this repo's `po-ingest` stack (`cdk/po_stack.py`). PK `siteCode` (S); default DynamoDB encryption (NOT the shared CMK).
|
|
|
|
**Consumer (read-only) — data contract:** `seahaven-slack-bot`'s `wo-po-lookup` Lambda imports this table via `Table.fromTableName(...)` for the Bedrock agent's `lookup_site` action. It does point lookups by `siteCode` and reads `address`, `fullAddress`, `city`, `state`, `zip`, `latitude`, `longitude`, `notes`. Coordinate any change to the table name, key schema, or these attribute names with `seahaven-slack-bot`.
|
|
|
|
> **GSI drift (INFRA-138):** the `by-state` GSI was removed here on 2026-06-03 (audit M-20, "0 reads in 30d"), but `seahaven-slack-bot` still queries `IndexName: 'by-state'` for its state-listing path, so that path fails at runtime today. Restoring the GSI or removing the consumer's state path needs to be reconciled cross-repo. This is the kind of silent owner-side lifecycle change this data-contract note exists to prevent.
|
|
|
|
## Documentation
|
|
|
|
The canonical map of Sea Haven's AWS infrastructure lives in Confluence. This project's `po-ingest` and `WorkorderIngestStack` stacks are represented there as Mermaid subgraphs.
|
|
|
|
- **[AWS Architecture Map](https://seahaven.atlassian.net/wiki/spaces/IT/pages/1540098)** (Confluence, IT space, page 1540098)
|
|
|
|
## CI/CD
|
|
|
|
GitHub Actions with reusable workflows from `Sea-Haven-Industries/.github` (all pinned to a commit SHA of `main`):
|
|
- **CI** (`ci.yaml`, PR to `main`): linting + `cdk synth` via `ci-python-sam.yaml`
|
|
- **CD** (`deploy.yaml`, push to `main`): CDK deploy via `cd-cdk.yaml` (OIDC auth)
|
|
- Plus dependency review and PR labeler workflows on every PR
|
|
|
|
Branch protection on `main` — all changes through PR.
|
|
|
|
## Setup
|
|
|
|
1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1`
|
|
2. Ensure the Bedrock inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0` is enabled in `us-east-1` (it is; the CDK grants cover cross-region routing to `us-east-2`/`us-west-2`). No API key or secret to set — the processors authenticate to Bedrock via their IAM roles.
|
|
3. Deploy both stacks:
|
|
```bash
|
|
cd cdk
|
|
pip install -r requirements.txt
|
|
cdk deploy --all
|
|
```
|
|
4. **Post-deploy cleanup (one-time):** the retired `RETAIN`-policy secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` are orphaned by this deploy, not deleted. Remove them and revoke the keys at the provider:
|
|
```bash
|
|
aws secretsmanager delete-secret --secret-id po-ingest/anthropic-api-key --force-delete-without-recovery
|
|
aws secretsmanager delete-secret --secret-id workorder-ingest/anthropic-api-key --force-delete-without-recovery
|
|
```
|
|
5. Dashboards: `po-web-ui` and `workorder-web-ui` have no public endpoint (the Function URLs were removed 2026-06-08, INFRA-74). Invoke them manually and render the returned HTML, e.g. `aws lambda invoke --function-name po-web-ui /tmp/out.json`.
|
|
|
|
## Tests
|
|
|
|
Offline unit tests (no AWS, no network) run via pytest from the repo root:
|
|
|
|
```bash
|
|
pip install pytest
|
|
pytest
|
|
```
|
|
|
|
Coverage:
|
|
- `tests/` — shared handler tests (`parse_raw_email`, `pad_zip`).
|
|
- `lambdas/wo/email_processor/tests/` — the deterministic WO parser suite: golden-file tests over 55 real scrubbed `.eml` fixtures (`test_parser.py`), fail-closed validation-gate rules and adversarial/injection cases (`test_validation_gate.py`), the issue #23 `comment_id` idempotency invariants (`test_comment_id.py`), and the Bedrock-fallback dispatch/EMF-metric behavior with a mocked `invoke_model` (`test_bedrock_fallback.py`). Golden JSON lives under `tests/fixtures/expected/`.
|
|
|
|
Both roots are discovered by `pytest.ini` (`testpaths`).
|
|
|
|
## Scripts
|
|
|
|
**Reprocess PO emails** (re-run parser against all emails still in S3):
|
|
```bash
|
|
python scripts/reprocess.py # dry-run
|
|
python scripts/reprocess.py --execute # invoke po-email-processor for each
|
|
```
|
|
|
|
**Backfill verified sites** (one-time scan of historical POs):
|
|
```bash
|
|
python scripts/backfill_sites.py
|
|
```
|
|
|
|
## Directory Structure
|
|
|
|
```
|
|
cdk/
|
|
app.py # Two stacks: po-ingest + WorkorderIngestStack
|
|
po_stack.py # Purchase order pipeline resources
|
|
wo_stack.py # Work order pipeline resources
|
|
lambdas/
|
|
po/ # PO pipeline Lambdas
|
|
email_processor/ # handler.py (Bedrock, fully AI)
|
|
site_extractor/
|
|
web_ui/
|
|
wo/ # WO pipeline Lambdas
|
|
email_processor/
|
|
handler.py # template-first + Bedrock fallback, EMF metric, #23 comment_id
|
|
template_parser.py # pure deterministic parser + fail-closed validation gate
|
|
tests/ # golden-file + validation-gate + comment_id + fallback tests
|
|
web_ui/
|
|
scripts/
|
|
reprocess.py
|
|
backfill_sites.py
|
|
tests/ # shared handler tests (discovered with the WO suite)
|
|
test_local.py # Parse sample emails through Bedrock locally (no AWS mutation)
|
|
```
|