Module 8: Capstone The Andes Cargo Genai Extractor

4. End-to-end walkthrough: the AI escalation path, mixed and declared

Next →

Description

This lesson walks through the right half of lesson 2's diagram — the escalation path — and, unlike lesson 3, it cannot be completed end to end with code run in its entirety, for the same reason every earlier module of this guide has already confirmed: Bedrock never gets invoked here. What this lesson does, with the exact same honesty as always, is run everything that really can be run, right up to the exact point of the bedrock:InvokeModel call, and from there, with a simulated response — a fixed dict, labeled without ambiguity — keep running the real custom guardrails. The result is a genuinely mixed walkthrough: ManifestParseFailed really does fire, twice; extract-shipment-manifest-fields really does run, all the way to building the exact request to Bedrock; and pre_invoke_checks.py/post_invoke_checks.py (M4) really do run, on the simulated response, with two different outcomes.

Connection to the module

This is the lesson where the handler.py that M5.6 signed with cosign — never showing its full contents — appears, for the first time in this guide, with its entire code. It reuses, without modifying a single line, scrub_pii() (M4.5) and validate_shipment_fields() (M4.6) — the same three-step pattern defense_in_depth_flow.py (M4.8) already tested, now wrapped in the real shape of a Lambda handler triggered by an EventBridge event.

Analogy: the human counter, right up to the exact point of picking up the phone

Go back to lesson 2's service-counter analogy: when the ATM can't resolve a transaction, the line moves toward a person at the counter. That person really does several things before resolving the case: reviews the document the customer brought, mentally drafts what they'll ask a specialist if escalating further turns out to be necessary, and even picks up the phone to dial the specialist's number. This lesson is exactly that journey, up to the exact moment the phone starts ringing — never up to the moment the specialist answers. What the person at the counter did before that call (reviewing the document, drafting the question) is completely real, verifiable, repeatable. What the specialist would answer is, in this lesson, a simulated response, written by hand in advance, so the rest of the procedure — what the person at the counter does after receiving a response, whatever it is — can also be demonstrated without actually needing the phone to connect.

Step 1 — ManifestParseFailed, really fired, twice

Two new manifests, neither used before in this guide, each with a different kind of parse failure — the same discipline of variety M7.4 already applied to its batch of 50:

#!/usr/bin/env python3
"""ai_escalation_path_walkthrough.py -- Module 8, lesson 4. Builds two REAL
ManifestParseFailed events -- process-shipment-manifest (Module 1, lesson 3)
would publish exactly these, byte for byte, given this exact input text.
Never uses random or datetime.now()."""

from __future__ import annotations

import json

# --- parse_manifest() / SHIPMENT_FIELDS_SCHEMA: unmodified, M1.3 / M4.6 ---


def parse_manifest(text: str) -> dict:
    fields = {}
    for line in text.strip().splitlines():
        if "=" in line:
            key, _, value = line.partition("=")
            fields[key.strip()] = value.strip()
    return fields


SHIPMENT_FIELDS_SCHEMA = (
    "shipmentId",
    "originCountry",
    "destinationCountry",
    "carrier",
    "weightKg",
)

# --- Two new manifests, neither used before in this guide -----------------

FREE_TEXT_4478 = (
    "Hey, quick one -- picking up 95kg from our Cali site tomorrow, "
    "heading to Guayaquil, same carrier as last time (AndesExpress). "
    "Can you confirm receipt? Ref AC-4478."
)

PARTIAL_4479 = "shipmentId=4479\noriginCountry=Chile\ncarrier=RutaSur\n"


def build_manifest_parse_failed_event(manifest_key: str, raw_text: str) -> dict:
    """Exactly the event shape Module 1, lesson 3 already established --
    reused here without any change, applied to two new manifests."""
    parsed = parse_manifest(raw_text)
    missing = sorted(set(SHIPMENT_FIELDS_SCHEMA) - set(parsed.keys()))
    reason = "no key=value pairs found" if not parsed else f"missing required field(s): {', '.join(missing)}"
    detail = {"manifestKey": manifest_key, "rawText": raw_text, "reason": reason}
    return {
        "Source": "andescargo.shipments",
        "DetailType": "Manifest Parse Failed",
        "Detail": json.dumps(detail),
        "EventBusName": "andes-cargo-events",
    }


if __name__ == "__main__":
    event_4478 = build_manifest_parse_failed_event(
        "manifests/year=2026/month=08/shipment-4478-manifest.txt", FREE_TEXT_4478
    )
    event_4479 = build_manifest_parse_failed_event(
        "manifests/year=2026/month=08/shipment-4479-manifest.txt", PARTIAL_4479
    )
    print(json.dumps(event_4478, indent=2))
    print()
    print(json.dumps(event_4479, indent=2))
python3 ai_escalation_path_walkthrough.py

What to expect (literal — really run; parse_manifest() with no change from M1.3):

{
  "Source": "andescargo.shipments",
  "DetailType": "Manifest Parse Failed",
  "Detail": "{\"manifestKey\": \"manifests/year=2026/month=08/shipment-4478-manifest.txt\", \"rawText\": \"Hey, quick one -- picking up 95kg from our Cali site tomorrow, heading to Guayaquil, same carrier as last time (AndesExpress). Can you confirm receipt? Ref AC-4478.\", \"reason\": \"no key=value pairs found\"}",
  "EventBusName": "andes-cargo-events"
}

{
  "Source": "andescargo.shipments",
  "DetailType": "Manifest Parse Failed",
  "Detail": "{\"manifestKey\": \"manifests/year=2026/month=08/shipment-4479-manifest.txt\", \"rawText\": \"shipmentId=4479\\noriginCountry=Chile\\ncarrier=RutaSur\\n\", \"reason\": \"missing required field(s): destinationCountry, weightKg\"}",
  "EventBusName": "andes-cargo-events"
}

Two events, two different reasons: 4478 has no = at all in its text ("no key=value pairs found"); 4479 does have three of the five fields, but is missing two ("missing required field(s): destinationCountry, weightKg"). Both would really publish to the andes-cargo-events bus — this is exactly the code process-shipment-manifest runs, with no difference whatsoever.

Step 2 — handler.py, the complete code of extract-shipment-manifest-fields

M5.6 signed functions/extract-shipment-manifest-fields/handler.py with cosign, without showing its content beyond "the minimal handler, not polished." Here, for the first time in this guide, its complete code:

#!/usr/bin/env python3
"""functions/extract-shipment-manifest-fields/handler.py -- Module 8, lesson 4
of genai-on-aws-production-guide. The Lambda handler BedrockManifestExtractorRole
(Module 3, lesson 4) executes as, triggered by ManifestParseFailed (Module 1,
lesson 3). Deliberately minimal, never a polished prompt (Module 1, lesson 4
-- the boundary this guide holds without exception).

Wires pre_invoke_checks.scrub_pii() and post_invoke_checks.validate_shipment_fields()
(Module 4, lessons 5-6) -- REAL, unmodified, tested code -- around the exact
point where bedrock-runtime.invoke_model() would be called. That call is
built in full (build_invoke_request()) but NEVER executed anywhere in this
guide -- Bedrock is "Included in Plans: Ultimate" only (Module 3, lesson 6;
Module 4, lesson 7). This is the same three-step order
guardrails/defense_in_depth_flow.py (Module 4, lesson 8) already tested in
isolation, now wired to a real EventBridge event shape.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "guardrails"))

from post_invoke_checks import validate_shipment_fields  # noqa: E402
from pre_invoke_checks import scrub_pii  # noqa: E402

GUARDRAIL_ID = "andes-cargo-manifest-extractor-guardrail"  # bedrock.tf, Module 3/4
GUARDRAIL_VERSION = "DRAFT"
MODEL_ID = "amazon.nova-lite-v1:0"  # GENAI-COST-PROFILE.md, Module 2, section 2

PROMPT_TEMPLATE = (
    "Extract shipment fields from the manifest text below. Return a JSON "
    "object with exactly these keys: shipmentId, originCountry, "
    "destinationCountry, carrier, weightKg. If a field is not present in "
    "the text, omit the key rather than guessing.\n\nManifest text:\n{text}"
)


def build_invoke_request(redacted_text: str) -> dict:
    """The exact request bedrock-runtime.invoke_model() would receive --
    REAL, executed code. Constructing this dict never touches a network."""
    return {
        "modelId": MODEL_ID,
        "guardrailIdentifier": GUARDRAIL_ID,
        "guardrailVersion": GUARDRAIL_VERSION,
        "body": json.dumps({"inputText": PROMPT_TEMPLATE.format(text=redacted_text)}),
    }


def handle_manifest_parse_failed(event: dict, representative_model_response: dict) -> dict:
    """1. scrub_pii() over rawText -- REAL, always.
    2. build_invoke_request() -- REAL, constructs the exact call.
       >>> response = bedrock_runtime.invoke_model(**request)  <-- the line
       this guide never executes. representative_model_response, injected
       explicitly by the caller, stands in for `response` from here on.
    3. validate_shipment_fields() over that response -- REAL, always.
    4. The write-to-Shipments decision: only if step 3 passed.
    """
    detail = json.loads(event["Detail"])
    raw_text = detail["rawText"]

    scrub_result = scrub_pii(raw_text)
    request = build_invoke_request(scrub_result.redacted_text)

    # response = bedrock_runtime.invoke_model(**request)  # NEVER executed --
    # see Module 3, lesson 6 and Module 4, lesson 7 for why.
    candidate = representative_model_response

    validation = validate_shipment_fields(candidate)

    return {
        "manifestKey": detail["manifestKey"],
        "requestModelId": request["modelId"],
        "requestGuardrailId": request["guardrailIdentifier"],
        "requestBodyBytes": len(request["body"]),
        "piiFoundInInput": scrub_result.found_pii,
        "schemaValid": validation.is_valid,
        "schemaErrors": validation.errors,
        "wouldWriteToShipments": validation.is_valid,
    }


def format_outcome(outcome: dict) -> str:
    lines = [
        f"manifestKey:        {outcome['manifestKey']}",
        f"requestModelId:     {outcome['requestModelId']}",
        f"requestGuardrailId: {outcome['requestGuardrailId']}",
        f"pre_invoke_checks:  {'PII FOUND (redacted)' if outcome['piiFoundInInput'] else 'CLEAN'}",
        f"post_invoke_checks: {'PASS' if outcome['schemaValid'] else 'FAIL'}",
    ]
    for err in outcome["schemaErrors"]:
        lines.append(f"  - {err}")
    decision = "WRITE to Shipments" if outcome["wouldWriteToShipments"] else "DO NOT WRITE to Shipments"
    lines.append(f"DECISION: {decision}")
    return "\n".join(lines)

Notice the commented-out line inside handle_manifest_parse_failed(): # response = bedrock_runtime.invoke_model(**request) — the exact point, marked right in the code itself, where this guide stops. Everything before that line is real; everything after it uses representative_model_response, never the result of that commented-out call.

Step 3 — Running the complete walkthrough, two scenarios

if __name__ == "__main__":
    # Scenario A: shipment 4478, COMPLETE representative response.
    outcome_a = handle_manifest_parse_failed(
        event_4478,
        representative_model_response={
            "shipmentId": "4478",
            "originCountry": "Colombia",
            "destinationCountry": "Ecuador",
            "carrier": "AndesExpress",
            "weightKg": "95",
        },
    )
    print("=== Scenario A: shipment 4478, complete representative response ===")
    print(format_outcome(outcome_a))
    print()

    # Scenario B: shipment 4479, representative response WITHOUT weightKg --
    # the model extracted what the text DID have, but couldn't invent the
    # weight because the original manifest didn't have it either.
    outcome_b = handle_manifest_parse_failed(
        event_4479,
        representative_model_response={
            "shipmentId": "4479",
            "originCountry": "Chile",
            "destinationCountry": "Peru",
            "carrier": "RutaSur",
        },
    )
    print("=== Scenario B: shipment 4479, incomplete representative response ===")
    print(format_outcome(outcome_b))
python3 handler.py

What to expect — manifestKey/requestModelId/requestGuardrailId/pre_invoke_checks literal (real code, executed); post_invoke_checks/DECISION literal over the representative response (the check itself is real; the candidate it evaluates is a fixed dict, labeled as such in this lesson, never the output of an invocation):

=== Scenario A: shipment 4478, complete representative response ===
manifestKey:        manifests/year=2026/month=08/shipment-4478-manifest.txt
requestModelId:     amazon.nova-lite-v1:0
requestGuardrailId: andes-cargo-manifest-extractor-guardrail
pre_invoke_checks:  CLEAN
post_invoke_checks: PASS
DECISION: WRITE to Shipments

=== Scenario B: shipment 4479, incomplete representative response ===
manifestKey:        manifests/year=2026/month=08/shipment-4479-manifest.txt
requestModelId:     amazon.nova-lite-v1:0
requestGuardrailId: andes-cargo-manifest-extractor-guardrail
pre_invoke_checks:  CLEAN
post_invoke_checks: FAIL
  - missing required field(s): weightKg
DECISION: DO NOT WRITE to Shipments

Two different outcomes, over the same real code: Scenario A passes because the representative response — built, deliberately, to match what the real manifest actually described (95kg, Cali to Guayaquil, AndesExpress) — has all five fields. Scenario B fails because the representative response honestly reflects that 4479's original manifest never mentioned a weight — neither the deterministic parser nor a reasonable extraction could have invented a data point the text never had. post_invoke_checks.py rejects it, exactly as designed, regardless of whether the reason for the missing field is "the model didn't extract it" or "the data was never in the text."

Step 4 — pytest, confirming the complete handler.py

"""test_handler.py -- Module 8, lesson 4. Exercises handle_manifest_parse_failed()
end to end, with fixed events and fixed representative responses. No random,
no datetime.now()."""

from ai_escalation_path_walkthrough import event_4478, event_4479
from handler import build_invoke_request, handle_manifest_parse_failed


def test_build_invoke_request_uses_nova_lite():
    request = build_invoke_request("clean text")
    assert request["modelId"] == "amazon.nova-lite-v1:0"
    assert request["guardrailIdentifier"] == "andes-cargo-manifest-extractor-guardrail"


def test_scenario_a_complete_response_writes_to_shipments():
    outcome = handle_manifest_parse_failed(
        event_4478,
        {"shipmentId": "4478", "originCountry": "Colombia", "destinationCountry": "Ecuador",
         "carrier": "AndesExpress", "weightKg": "95"},
    )
    assert outcome["wouldWriteToShipments"] is True
    assert outcome["schemaErrors"] == ()


def test_scenario_b_incomplete_response_blocks_the_write():
    outcome = handle_manifest_parse_failed(
        event_4479,
        {"shipmentId": "4479", "originCountry": "Chile", "destinationCountry": "Peru", "carrier": "RutaSur"},
    )
    assert outcome["wouldWriteToShipments"] is False
    assert "missing required field(s): weightKg" in outcome["schemaErrors"][0]


def test_request_never_contains_a_response_field():
    """The request this handler builds has no key related to a response --
    confirms, structurally, that build_invoke_request() only ever
    constructs an outbound request, never simulates an inbound one."""
    request = build_invoke_request("some text")
    assert "response" not in request
    assert set(request.keys()) == {"modelId", "guardrailIdentifier", "guardrailVersion", "body"}
pytest test_handler.py -v -p no:randomly

What to expect (literal — really run):

============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.1.1, pluggy-1.6.0
collected 4 items

test_handler.py::test_build_invoke_request_uses_nova_lite PASSED         [ 25%]
test_handler.py::test_scenario_a_complete_response_writes_to_shipments PASSED [ 50%]
test_handler.py::test_scenario_b_incomplete_response_blocks_the_write PASSED [ 75%]
test_handler.py::test_request_never_contains_a_response_field PASSED     [100%]

============================== 4 passed in 0.01s ===============================

Step 5 — The exact point, said once more, plainly

   WHAT THIS LESSON REALLY EXECUTES              FROM HERE ON, REPRESENTATIVE

   parse_manifest() (Step 1)                       bedrock_runtime.invoke_model(
   build_manifest_parse_failed_event() (Step 1)      **request)  -- NEVER
   scrub_pii() (Step 2, inside the handler)          uncommented, anywhere
   build_invoke_request() (Step 2)                   in this guide
   validate_shipment_fields() (Step 2)
   the WRITE / DO NOT WRITE decision (Step 2)       representative_model_response
                                                     -- a fixed dict, written
   Verified with pytest (Step 4)                     by hand, labeled as such
                                                     right in the parameter name

No part of this lesson invokes a Bedrock model. The reason, already confirmed twice in this guide with different evidence: M3.6 confirmed, live, that tflocal apply on the guardrail stops because Bedrock is "Included in Plans: Ultimate" — a LocalStack tier above the free one this entire lab runs on; M4.7 confirmed, with ApplyGuardrail's real schema, that even if the resource existed, confirming it really blocks something would require a real invocation, not just its existence. This lesson doesn't repeat those investigations — it inherits them, and builds around them as much real code as possible.

Common mistakes

Uncommenting the line bedrock_runtime.invoke_model(**request) "just to see what happens" (curiosity breaking this guide's strictest rule). What happens: someone, following this lesson on their own machine with a real AWS account and Bedrock access, uncomments the line to really test it. How to spot it: if your copy of handler.py has that line active. How to fix it: nothing stops you, on your own account, from doing this — but doing it takes you outside the exact scope this guide declared since M1.2: "does NOT run any real model inference." If you do it, your result is no longer "this guide executed" — it's "this guide extended by you, with your own account" — an honest distinction worth keeping, the same one M3.6 already asked for apply.

Presenting Scenario A's DECISION: WRITE to Shipments as if the record were already written to the real table (confusing the decision with the execution). What happens: someone, showing this lesson in an interview, says "the system wrote shipment 4478 to Shipments." How to spot it: if your description of this lesson doesn't distinguish between "decided it would write" and "really wrote." How to fix it: handle_manifest_parse_failed() decides whether it would write, based on validation.is_valid — it never calls, anywhere in its code, write_shipment_record() or any DynamoDB operation. The correct phrasing: "the handler determined, with real guardrails, that this candidate had the correct shape to be written" — not "the system completed the write."

Assuming this lesson's two representative responses (Scenario A complete, Scenario B incomplete) were chosen at random, with no relation to each manifest's original text (losing the example's narrative coherence). What happens: someone, replicating this pattern, builds a representative response with no logical connection to the input manifest. How to spot it: if your representative candidate invents an origin country the original text didn't even mention. How to fix it: every representative response in this lesson reflects, precisely, what the original manifest actually said — Scenario B, specifically, doesn't invent a weightKg because the real manifest (4479) never mentioned one; an honest representative extraction never "improves" the original text, it only reorganizes into JSON what the text actually contained, the exact same honesty discipline that holds up every representativeModelResponse in evals/fixtures/sample_manifests.json (M7.6).

Exercises

Exercise 1 — Build, yourself, a third scenario with a new manifest (for example, 4480) and a representative response that includes an unexpected field, such as confidenceScore. Predict the result before running it.

See solution

post_invoke_checks.validate_shipment_fields() would flag unexpected_fields = ("confidenceScore",), with is_valid = False — exactly the same case M4.6 (test_unexpected_extra_field_fails) and M7.6 (fixture 4476) already tested separately. pre_invoke_checks would keep reporting based on 4480's input text, unrelated to this field — the final decision would be DO NOT WRITE to Shipments, with the exact error listed. This exercise confirms that handler.py, by reusing validate_shipment_fields() with no change, automatically inherits every rule that function already enforces, with no need to duplicate any logic.

Exercise 2 — Explain why build_invoke_request() includes guardrailIdentifier and guardrailVersion in the request, instead of letting Bedrock use a "default" guardrail. What would happen if those two fields were omitted?

See solution

Bedrock has no concept of a "default" guardrail for an account — every invocation explicitly decides whether it references a guardrail or not. M4.4, Gap 4 already explained this precisely: a managed guardrail only protects an invocation that actually references it through these two parameters; if build_invoke_request() omitted them, the resulting request would invoke the model with none of the six active policies, with no visible error flagging it — exactly the operational risk that same lesson named as Gap 4. Declaring them here, explicitly, in the handler's code, is the direct mitigation of that risk.

Exercise 3 — Predict what would change in this lesson's Step 3 "What to expect" if, someday, this guide ran against a real AWS account with Bedrock enabled, and handler.py's commented-out line were activated. Which fields of the outcome would stay identical, and which would change?

See solution

manifestKey, requestModelId, requestGuardrailId, and piiFoundInInput would stay identical — they depend exclusively on parse_manifest(), scrub_pii(), and build_invoke_request(), none of which needs a real invocation. schemaValid, schemaErrors, and wouldWriteToShipments could change, because they'd depend on the model's real response instead of the representative response fixed by hand in this lesson — Nova Lite might, for example, extract Scenario B's weightKg if the model infers a reasonable value from context (something this guide can never confirm without the real invocation). This exercise pins down, precisely, exactly which part of this lesson's handler.py is agnostic to whether Bedrock exists or not, and which part depends entirely on the real response.

Summary and next step

This lesson walked through the escalation path end to end, showing handler.py's complete contents for the first time: ManifestParseFailed really fired, twice, with two different failure reasons; pre_invoke_checks.py/post_invoke_checks.py really ran, over representative responses labeled without ambiguity; and the real, executed construction of the exact request to bedrock:InvokeModel — with the line that would execute it explicitly commented out, right in the code itself. You confirmed the complete flow with four pytest cases, including a test that the built request never contains any response field.

Before moving on you should be able to: point at handler.py's exact line and explain where the real part ends and the representative part begins; build your own third scenario with a new manifest and a new representative response; and defend, without hesitation, why "decided it would write" is not the same as "wrote."

Lesson 5 gathers, in a single document, every piece that stayed representative across this module's eight lessons AND across the seven earlier modules — this entire guide's complete honesty ledger, with the exact technical reason for each row.

Resources

  1. This same course, Module 1, lesson 3 — the origin of the ManifestParseFailed event, rebuilt twice in this lesson's Step 1.
  2. This same course, Module 4, lessons 5, 6, and 8 — the origin of scrub_pii(), validate_shipment_fields(), and the three-step pattern handler.py reuses with no change.
  3. This same course, Module 3, lesson 6, and Module 4, lesson 7 — the source of the two findings ("Ultimate-only," ApplyGuardrail's schema) cited in this lesson's Step 5.
  4. AWS Docs — InvokeModel — official reference for the parameters build_invoke_request() constructs, including guardrailIdentifier/guardrailVersion.
← Previous3. End-to-end walkthrough: the deterministic, cheap path, executedNext →5. What this guide left representative