apm-wo-analysis/tests/test_classify.py

"""Smoke test for the two-axis classifier.

Runs the DETERMINISTIC ``classify()`` (no Haiku, no network, no AWS) over the
canonical sample export and asserts the two-axis model keeps "Other" in the
single digits, plus a handful of known fixtures land in the right bucket. The
Haiku fallback is never exercised here.

Canonical fixture: ``~/Downloads/_documents/Sheet1-1.xlsx`` (347 rows, 13 cols).
If the file is absent the export-driven tests skip. Run with the repo venv:

    ./.venv/bin/python -m pytest tests/test_classify.py -s -q

CSV fixture (always present, runs in CI): ``tests/fixtures/sample_export.csv``.
"""

import collections
import csv
import re
import sys
from pathlib import Path

import pytest

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

import classify  # noqa: E402

# Column indices in the 13-column export.
COL_WO_STATUS = 6
COL_HOLD_REASON = 7
COL_LAST_COMMENT = 8

FIXTURE = Path.home() / "Downloads" / "_documents" / "Sheet1-1.xlsx"

# Committed CSV fixture — always present, no skipif.
CSV_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "sample_export.csv"

# Max acceptable deterministic "Other" share before any AI. CLAUDE.md: the
# two-axis model lands ~9% before Haiku; we hold the line at single digits.
MAX_OTHER_PCT = 10.0

# Threshold for the committed CSV fixture.  Measured deterministic Other% on
# the synthetic fixture: 7.41% (2 of 27 classified rows).  Threshold is set
# with headroom but still comfortably single-digit.
CSV_FIXTURE_MAX_OTHER_PCT = 9.0


# ---------------------------------------------------------------------------
# CSV loader helper (yields column-indexed tuples like openpyxl row values)
# ---------------------------------------------------------------------------


def _load_csv_rows(path: Path) -> list[tuple]:
    """Read a 13-column CSV export; return data rows as tuples (header skipped)."""
    with path.open(newline="") as fh:
        reader = csv.reader(fh)
        rows = list(reader)
    return [tuple(r) for r in rows[1:]]  # drop header row


def test_classification_constants_well_formed():
    assert "3rd Escalation" in classify.ESCALATION_CATEGORIES
    assert classify.HOLD_TO_CATEGORY["SCHEDULING"] == "Awaiting Scheduling"
    # Every escalation category is also action-needed.
    assert classify.ESCALATION_CATEGORIES <= classify.ACTION_NEEDED_CATEGORIES


def test_strip_html_unwraps_and_normalises():
    assert classify.strip_html(None) == ""
    assert classify.strip_html("   ") == ""
    assert classify.strip_html("<html>WO schedule confirmed with vendor</html>") == (
        "WO schedule confirmed with vendor"
    )
    # Nested tags, entities, smart quotes, and embedded URLs.
    raw = (
        "<html><div>1st attempt process for schedule confirmation. "
        "Vendor, please confirm ‘Schedule Start Date’ &amp; proceed. "
        "https://app.avetta.com/avt-cli/x</div></html>"
    )
    cleaned = classify.strip_html(raw)
    assert "<" not in cleaned and ">" not in cleaned
    assert "‘" not in cleaned and "’" not in cleaned  # smart quotes gone
    assert "'Schedule Start Date'" in cleaned
    assert "&amp;" not in cleaned and "&" in cleaned  # entity decoded
    assert "https://" not in cleaned  # URL reduced to a token


def test_known_intents():
    # Schedule confirmed (the dominant happy-path comment).
    cat, _ = classify.classify(
        "IP", "", "<html>WO schedule confirmed with vendor</html>"
    )
    assert cat == "Schedule Confirmed"

    # Cancelled via WO Status, even with a generic comment.
    cat, _ = classify.classify(
        "RCAN", "", "<html>WO Cancelled, created in error.</html>"
    )
    assert cat == "Cancelled"

    # 3rd-attempt escalation cadence.
    cat, _ = classify.classify(
        "H", "REPORT", "<html>3rd attempt process for schedule confirmation.</html>"
    )
    assert cat == "3rd Escalation"

    # Vendor no-show.
    cat, _ = classify.classify(
        "R", "", "<html>Site tech reported vendor was a no show for Friday.</html>"
    )
    assert cat == "Vendor No-Show"

    # Weekly cadence template.
    cat, _ = classify.classify(
        "IP",
        "",
        "<html>Weekly WO scheduled. Service reports required EOD Friday.</html>",
    )
    assert cat == "Weekly WO Scheduled"

    # Structured-only fallback: no comment intent fires, REPORT hold decides.
    cat, _ = classify.classify("IP", "REPORT", "<html></html>")
    assert cat == "Report / Docs Needed"


def test_mismatch_detection():
    # Comment claims completion while on a REPORT hold → mismatch surfaced.
    cat, mm = classify.classify(
        "IP", "REPORT", "<html>Vendor arrived and performed task.</html>"
    )
    assert cat == "Completed / Pending Close"
    assert mm is not None and "REPORT" in mm

    # Schedule confirmed while on a SCHEDULING hold → mismatch surfaced.
    cat, mm = classify.classify(
        "R", "SCHEDULING", "<html>WO schedule confirmed with vendor.</html>"
    )
    assert cat == "Schedule Confirmed"
    assert mm is not None

    # Clean case: no contradiction → no mismatch.
    _, mm = classify.classify(
        "IP", "", "<html>WO schedule confirmed with vendor</html>"
    )
    assert mm is None


def test_classify_is_offline():
    """classify() must not import boto3 or reach the network."""
    import sys as _sys

    had_boto3 = "boto3" in _sys.modules
    classify.classify("IP", "REPORT", "<html>1st attempt process for report.</html>")
    # If boto3 wasn't already loaded, classify() must not have pulled it in.
    if not had_boto3:
        assert "boto3" not in _sys.modules


@pytest.mark.skipif(not FIXTURE.exists(), reason=f"sample export not found: {FIXTURE}")
def test_other_share_against_real_export(capsys):
    import openpyxl

    wb = openpyxl.load_workbook(FIXTURE, read_only=True, data_only=True)
    rows = list(wb.active.iter_rows(values_only=True))[1:]  # drop header

    dist = collections.Counter()
    mismatches = 0
    classified = 0
    blank = 0
    other_samples = []

    for row in rows:
        comment = row[COL_LAST_COMMENT]
        # Blank-comment rows are excluded from the classified total by design.
        if comment is None or str(comment).strip() == "":
            blank += 1
            continue
        classified += 1
        category, mismatch = classify.classify(
            row[COL_WO_STATUS], row[COL_HOLD_REASON], comment
        )
        dist[category] += 1
        if mismatch:
            mismatches += 1
        if category == "Other" and len(other_samples) < 20:
            other_samples.append(classify.strip_html(comment)[:90])

    other = dist["Other"]
    other_pct = other * 100.0 / classified if classified else 0.0

    with capsys.disabled():
        print(f"\n=== APM classifier smoke test: {FIXTURE.name} ===")
        print(
            f"rows={len(rows)} classified={classified} "
            f"blank-excluded={blank} (blank-comment rows excluded from the total)"
        )
        print("--- category distribution ---")
        for cat, count in dist.most_common():
            print(f"  {count:4d}  {count * 100.0 / classified:5.1f}%  {cat}")
        print(f"--- Other: {other} ({other_pct:.2f}%) ---")
        for sample in other_samples:
            print(f"    [Other] {sample}")
        print(f"--- mismatches flagged: {mismatches} ---")

    assert classified > 0
    assert other_pct <= MAX_OTHER_PCT, (
        f"deterministic Other {other_pct:.2f}% exceeds {MAX_OTHER_PCT}% — "
        "the two-axis ladder regressed"
    )
    # Two-axis model should comfortably beat the legacy ~17% Other.
    assert other_pct < 17.0


@pytest.mark.skipif(not FIXTURE.exists(), reason=f"sample export not found: {FIXTURE}")
def test_known_fixture_rows_in_export():
    """Anchor on real rows found in the canonical export."""
    import openpyxl

    wb = openpyxl.load_workbook(FIXTURE, read_only=True, data_only=True)
    rows = list(wb.active.iter_rows(values_only=True))[1:]

    by_status = collections.defaultdict(list)
    schedule_confirmed_row = None
    report_completion_mismatch = None

    for row in rows:
        comment = row[COL_LAST_COMMENT]
        if comment is None or str(comment).strip() == "":
            continue
        text = classify.strip_html(comment)
        by_status[row[COL_WO_STATUS]].append(row)
        if (
            schedule_confirmed_row is None
            and "schedule confirmed with vendor" in text.lower()
            and not (row[COL_HOLD_REASON] or "").strip()
        ):
            schedule_confirmed_row = row
        if (
            report_completion_mismatch is None
            and (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT"
            and "performed task" in text.lower()
        ):
            report_completion_mismatch = row

    # A "WO schedule confirmed with vendor" row → Schedule Confirmed.
    assert schedule_confirmed_row is not None, "fixture lacks a schedule-confirmed row"
    cat, _ = classify.classify(
        schedule_confirmed_row[COL_WO_STATUS],
        schedule_confirmed_row[COL_HOLD_REASON],
        schedule_confirmed_row[COL_LAST_COMMENT],
    )
    assert cat == "Schedule Confirmed"

    # Any RCAN row → Cancelled.
    assert "RCAN" in by_status, "fixture lacks an RCAN row"
    rcan = by_status["RCAN"][0]
    cat, _ = classify.classify(
        rcan[COL_WO_STATUS], rcan[COL_HOLD_REASON], rcan[COL_LAST_COMMENT]
    )
    assert cat == "Cancelled"

    # A REPORT-hold row whose comment claims completion → mismatch non-None.
    if report_completion_mismatch is not None:
        _, mm = classify.classify(
            report_completion_mismatch[COL_WO_STATUS],
            report_completion_mismatch[COL_HOLD_REASON],
            report_completion_mismatch[COL_LAST_COMMENT],
        )
        assert mm is not None


# ---------------------------------------------------------------------------
# CSV fixture tests — always run (no skipif), so the quality gate fires in CI.
# ---------------------------------------------------------------------------


def test_other_share_against_csv_fixture(capsys):
    """Classification quality gate against the committed synthetic CSV fixture.

    Measured deterministic Other%: 7.41% (2/27).  Threshold: 9.0%.
    This test runs unconditionally in CI.
    """
    rows = _load_csv_rows(CSV_FIXTURE)

    dist: collections.Counter = collections.Counter()
    mismatches = 0
    classified = 0
    blank = 0
    other_samples: list[str] = []

    for row in rows:
        comment = row[COL_LAST_COMMENT] if len(row) > COL_LAST_COMMENT else ""
        # Blank-comment rows are excluded from the classified total by design.
        if comment is None or str(comment).strip() == "":
            blank += 1
            continue
        classified += 1
        category, mismatch = classify.classify(
            row[COL_WO_STATUS], row[COL_HOLD_REASON], comment
        )
        dist[category] += 1
        if mismatch:
            mismatches += 1
        if category == "Other" and len(other_samples) < 20:
            other_samples.append(classify.strip_html(comment)[:90])

    other = dist["Other"]
    other_pct = other * 100.0 / classified if classified else 0.0

    with capsys.disabled():
        print(f"\n=== APM classifier smoke test (CSV fixture): {CSV_FIXTURE.name} ===")
        print(
            f"rows={len(rows)} classified={classified} "
            f"blank-excluded={blank} (blank-comment rows excluded from the total)"
        )
        print("--- category distribution ---")
        for cat, count in dist.most_common():
            print(f"  {count:4d}  {count * 100.0 / classified:5.1f}%  {cat}")
        print(f"--- Other: {other} ({other_pct:.2f}%) ---")
        for sample in other_samples:
            print(f"    [Other] {sample}")
        print(f"--- mismatches flagged: {mismatches} ---")

    assert classified > 0, "CSV fixture produced no classified rows"
    assert other_pct <= CSV_FIXTURE_MAX_OTHER_PCT, (
        f"deterministic Other {other_pct:.2f}% exceeds {CSV_FIXTURE_MAX_OTHER_PCT}% — "
        "the two-axis ladder regressed against the committed fixture"
    )
    # Fixture must exercise mismatch detection (WO-1012: REPORT hold + performed task).
    assert mismatches >= 1, "CSV fixture should contain at least one mismatch row"


def test_known_rows_in_csv_fixture():
    """Anchor checks on the synthetic CSV fixture — runs unconditionally in CI."""
    rows = _load_csv_rows(CSV_FIXTURE)

    by_status: collections.defaultdict = collections.defaultdict(list)
    schedule_confirmed_row = None
    report_completion_mismatch = None
    third_esc_rows: list[tuple] = []
    cancelled_rows: list[tuple] = []
    structured_report_rows: list[tuple] = []
    structured_scheduling_rows: list[tuple] = []

    for row in rows:
        comment = row[COL_LAST_COMMENT] if len(row) > COL_LAST_COMMENT else ""
        if comment is None or str(comment).strip() == "":
            continue
        text = classify.strip_html(comment)
        by_status[row[COL_WO_STATUS]].append(row)

        if (
            schedule_confirmed_row is None
            and "schedule confirmed with vendor" in text.lower()
            and not (row[COL_HOLD_REASON] or "").strip()
        ):
            schedule_confirmed_row = row
        if (
            report_completion_mismatch is None
            and (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT"
            and "performed task" in text.lower()
        ):
            report_completion_mismatch = row
        if re.search(r"\b3rd\b.*\battempt\b", text.lower()):
            third_esc_rows.append(row)
        if row[COL_WO_STATUS] == "RCAN":
            cancelled_rows.append(row)
        # Structured-only: HTML-wrapped empty comment (strips to "") with REPORT hold.
        if (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT" and text == "":
            structured_report_rows.append(row)
        if (row[COL_HOLD_REASON] or "").strip().upper() == "SCHEDULING" and text == "":
            structured_scheduling_rows.append(row)

    # "WO schedule confirmed with vendor" → Schedule Confirmed.
    assert schedule_confirmed_row is not None, "fixture lacks a schedule-confirmed row"
    cat, _ = classify.classify(
        schedule_confirmed_row[COL_WO_STATUS],
        schedule_confirmed_row[COL_HOLD_REASON],
        schedule_confirmed_row[COL_LAST_COMMENT],
    )
    assert cat == "Schedule Confirmed", f"expected Schedule Confirmed, got {cat!r}"

    # RCAN rows → Cancelled.
    assert cancelled_rows, "fixture lacks an RCAN row"
    cat, _ = classify.classify(
        cancelled_rows[0][COL_WO_STATUS],
        cancelled_rows[0][COL_HOLD_REASON],
        cancelled_rows[0][COL_LAST_COMMENT],
    )
    assert cat == "Cancelled", f"expected Cancelled, got {cat!r}"

    # 3rd Escalation rows are present and classify correctly.
    assert third_esc_rows, "fixture lacks a 3rd-escalation row"
    cat, _ = classify.classify(
        third_esc_rows[0][COL_WO_STATUS],
        third_esc_rows[0][COL_HOLD_REASON],
        third_esc_rows[0][COL_LAST_COMMENT],
    )
    assert cat == "3rd Escalation", f"expected 3rd Escalation, got {cat!r}"

    # Structured-only REPORT rows → Report / Docs Needed.
    assert structured_report_rows, "fixture lacks a structured-only REPORT hold row"
    cat, _ = classify.classify(
        structured_report_rows[0][COL_WO_STATUS],
        structured_report_rows[0][COL_HOLD_REASON],
        structured_report_rows[0][COL_LAST_COMMENT],
    )
    assert cat == "Report / Docs Needed", f"expected Report / Docs Needed, got {cat!r}"

    # Structured-only SCHEDULING rows → Awaiting Scheduling.
    assert structured_scheduling_rows, "fixture lacks a structured-only SCHEDULING row"
    cat, _ = classify.classify(
        structured_scheduling_rows[0][COL_WO_STATUS],
        structured_scheduling_rows[0][COL_HOLD_REASON],
        structured_scheduling_rows[0][COL_LAST_COMMENT],
    )
    assert cat == "Awaiting Scheduling", f"expected Awaiting Scheduling, got {cat!r}"

    # REPORT-hold row with completion comment → mismatch surfaced.
    assert report_completion_mismatch is not None, (
        "fixture lacks a REPORT-hold row with a completion comment (mismatch case)"
    )
    _, mm = classify.classify(
        report_completion_mismatch[COL_WO_STATUS],
        report_completion_mismatch[COL_HOLD_REASON],
        report_completion_mismatch[COL_LAST_COMMENT],
    )
    assert mm is not None, "expected a mismatch reason for WO-1012 but got None"
    assert "REPORT" in mm, f"mismatch reason should mention REPORT hold: {mm!r}"
-												Scaffold apm-wo-analysis repository

Stand up the Phase 0 CDK scaffold for the daily APM work-order
analysis pipeline: two-stack CDK app (pipeline + grafana), classifier
and slack-post Lambda packages, dashboards-as-code, the local
drop-folder uploader, and a classifier smoke-test placeholder.

Wire CI/CD to the org reusable workflows: ci.yaml -> ci-python-sam
(ruff + cdk synth) and deploy.yaml -> cd-cdk (OIDC, cdk deploy --all).
Pin aws-cdk-lib==2.253.1; Lambdas target Python 3.12 / arm64.

Rewrite .gitignore to the org Python-CDK standard so the source-of-
truth files (CLAUDE.md, docs/, .claude/agents) are tracked while build
artifacts (.venv, cdk.out, caches) stay ignored.

Domain logic, stack resources, and dashboards are stubbed and filled
in across Phases 1-5 (docs/BUILD.md). cdk synth is green for both
stacks; ruff check/format pass.

											
										
										
											2026-05-28 16:10:20 -04:00
+								"""Smoke test for the two-axis classifier.
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
+								Runs the DETERMINISTIC ``classify()`` (no Haiku, no network, no AWS) over the
 								canonical sample export and asserts the two-axis model keeps "Other" in the
 								single digits, plus a handful of known fixtures land in the right bucket. The
 								Haiku fallback is never exercised here.
 								Canonical fixture: ``~/Downloads/_documents/Sheet1-1.xlsx`` (347 rows, 13 cols).
 								If the file is absent the export-driven tests skip. Run with the repo venv:
 								    ./.venv/bin/python -m pytest tests/test_classify.py -s -q
-												Add synthetic export fixture and run classification quality gate in CI

											
										
										
											2026-05-29 14:30:44 -04:00
 								CSV fixture (always present, runs in CI): ``tests/fixtures/sample_export.csv``.
-												Scaffold apm-wo-analysis repository

Stand up the Phase 0 CDK scaffold for the daily APM work-order
analysis pipeline: two-stack CDK app (pipeline + grafana), classifier
and slack-post Lambda packages, dashboards-as-code, the local
drop-folder uploader, and a classifier smoke-test placeholder.

Wire CI/CD to the org reusable workflows: ci.yaml -> ci-python-sam
(ruff + cdk synth) and deploy.yaml -> cd-cdk (OIDC, cdk deploy --all).
Pin aws-cdk-lib==2.253.1; Lambdas target Python 3.12 / arm64.

Rewrite .gitignore to the org Python-CDK standard so the source-of-
truth files (CLAUDE.md, docs/, .claude/agents) are tracked while build
artifacts (.venv, cdk.out, caches) stay ignored.

Domain logic, stack resources, and dashboards are stubbed and filled
in across Phases 1-5 (docs/BUILD.md). cdk synth is green for both
stacks; ruff check/format pass.

											
										
										
											2026-05-28 16:10:20 -04:00
+								"""
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
+								import collections
-												Add synthetic export fixture and run classification quality gate in CI

											
										
										
											2026-05-29 14:30:44 -04:00
+								import csv
 								import re
-												Scaffold apm-wo-analysis repository

Stand up the Phase 0 CDK scaffold for the daily APM work-order
analysis pipeline: two-stack CDK app (pipeline + grafana), classifier
and slack-post Lambda packages, dashboards-as-code, the local
drop-folder uploader, and a classifier smoke-test placeholder.

Wire CI/CD to the org reusable workflows: ci.yaml -> ci-python-sam
(ruff + cdk synth) and deploy.yaml -> cd-cdk (OIDC, cdk deploy --all).
Pin aws-cdk-lib==2.253.1; Lambdas target Python 3.12 / arm64.

Rewrite .gitignore to the org Python-CDK standard so the source-of-
truth files (CLAUDE.md, docs/, .claude/agents) are tracked while build
artifacts (.venv, cdk.out, caches) stay ignored.

Domain logic, stack resources, and dashboards are stubbed and filled
in across Phases 1-5 (docs/BUILD.md). cdk synth is green for both
stacks; ruff check/format pass.

											
										
										
											2026-05-28 16:10:20 -04:00
+								import sys
 								from pathlib import Path
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
+								import pytest
-												Scaffold apm-wo-analysis repository

Stand up the Phase 0 CDK scaffold for the daily APM work-order
analysis pipeline: two-stack CDK app (pipeline + grafana), classifier
and slack-post Lambda packages, dashboards-as-code, the local
drop-folder uploader, and a classifier smoke-test placeholder.

Wire CI/CD to the org reusable workflows: ci.yaml -> ci-python-sam
(ruff + cdk synth) and deploy.yaml -> cd-cdk (OIDC, cdk deploy --all).
Pin aws-cdk-lib==2.253.1; Lambdas target Python 3.12 / arm64.

Rewrite .gitignore to the org Python-CDK standard so the source-of-
truth files (CLAUDE.md, docs/, .claude/agents) are tracked while build
artifacts (.venv, cdk.out, caches) stay ignored.

Domain logic, stack resources, and dashboards are stubbed and filled
in across Phases 1-5 (docs/BUILD.md). cdk synth is green for both
stacks; ruff check/format pass.

											
										
										
											2026-05-28 16:10:20 -04:00
+								sys.path.insert(
 , str(Path(__file__).resolve().parent.parent / "lambdas" / "classifier")
 								)
 								import classify  # noqa: E402
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
+								# Column indices in the 13-column export.
 								COL_WO_STATUS = 6
 								COL_HOLD_REASON = 7
 								COL_LAST_COMMENT = 8
 								FIXTURE = Path.home() / "Downloads" / "_documents" / "Sheet1-1.xlsx"
-												Add synthetic export fixture and run classification quality gate in CI

											
										
										
											2026-05-29 14:30:44 -04:00
+								# Committed CSV fixture — always present, no skipif.
 								CSV_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "sample_export.csv"
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
+								# Max acceptable deterministic "Other" share before any AI. CLAUDE.md: the
 								# two-axis model lands ~9% before Haiku; we hold the line at single digits.
 								MAX_OTHER_PCT = 10.0
-												Add synthetic export fixture and run classification quality gate in CI

											
										
										
											2026-05-29 14:30:44 -04:00
+								# Threshold for the committed CSV fixture.  Measured deterministic Other% on
 								# the synthetic fixture: 7.41% (2 of 27 classified rows).  Threshold is set
 								# with headroom but still comfortably single-digit.
 								CSV_FIXTURE_MAX_OTHER_PCT = 9.0
 								# ---------------------------------------------------------------------------
 								# CSV loader helper (yields column-indexed tuples like openpyxl row values)
 								# ---------------------------------------------------------------------------
 								def _load_csv_rows(path: Path) -> list[tuple]:
 								    """Read a 13-column CSV export; return data rows as tuples (header skipped)."""
 								    with path.open(newline="") as fh:
 								        reader = csv.reader(fh)
 								        rows = list(reader)
 								    return [tuple(r) for r in rows[1:]]  # drop header row
-												Scaffold apm-wo-analysis repository

Stand up the Phase 0 CDK scaffold for the daily APM work-order
analysis pipeline: two-stack CDK app (pipeline + grafana), classifier
and slack-post Lambda packages, dashboards-as-code, the local
drop-folder uploader, and a classifier smoke-test placeholder.

Wire CI/CD to the org reusable workflows: ci.yaml -> ci-python-sam
(ruff + cdk synth) and deploy.yaml -> cd-cdk (OIDC, cdk deploy --all).
Pin aws-cdk-lib==2.253.1; Lambdas target Python 3.12 / arm64.

Rewrite .gitignore to the org Python-CDK standard so the source-of-
truth files (CLAUDE.md, docs/, .claude/agents) are tracked while build
artifacts (.venv, cdk.out, caches) stay ignored.

Domain logic, stack resources, and dashboards are stubbed and filled
in across Phases 1-5 (docs/BUILD.md). cdk synth is green for both
stacks; ruff check/format pass.

											
										
										
											2026-05-28 16:10:20 -04:00
 								def test_classification_constants_well_formed():
 								    assert "3rd Escalation" in classify.ESCALATION_CATEGORIES
 								    assert classify.HOLD_TO_CATEGORY["SCHEDULING"] == "Awaiting Scheduling"
 								    # Every escalation category is also action-needed.
 								    assert classify.ESCALATION_CATEGORIES <= classify.ACTION_NEEDED_CATEGORIES
-												Add two-axis classifier Lambda and CDK wiring (Phase 2)

Implement the core classification engine and wire it into the pipeline stack.

classify.py: two-axis classifier — HTML-strip, comment-intent regex buckets
(escalations → status inquiry, most-specific first), Hold Reason / WO Status
structured state, comment-vs-state mismatch detector, and a Claude Haiku
fallback (Secrets Manager key) reserved for ambiguous free-text. Exports
ESCALATION_CATEGORIES / ACTION_NEEDED_CATEGORIES.

handler.py: S3-triggered handler — parse xlsx/csv, classify each non-blank
row, write a per-WO Parquet snapshot to analytics/dt=YYYY-MM-DD/ (registers
the Glue partition via awswrangler) and a summary.json for slack-post (Phase 4).

pipeline_stack.py: Glue database, ARM64 Python 3.12 classifier Lambda
(Docker-bundled deps), S3 raw/ notification (.xlsx/.csv), and least-privilege
IAM (read raw/, read-write analytics/, scoped Glue catalog, read Anthropic key).

Smoke-tested against the real export: 347 rows, "Other" at 5.2% (target ~9%),
18 mismatches flagged. 7/7 unit + smoke tests pass; cdk synth green.

											
										
										
											2026-05-28 17:09:52 -04:00
 								def test_strip_html_unwraps_and_normalises():
 								    assert classify.strip_html(None) == ""
 								    assert classify.strip_html("   ") == ""
 								    assert classify.strip_html("<html>WO schedule confirmed with vendor</html>") == (
 								        "WO schedule confirmed with vendor"
 								    )
 								    # Nested tags, entities, smart quotes, and embedded URLs.
 								    raw = (
 								        "<html><div>1st attempt process for schedule confirmation. "
 								        "Vendor, please confirm ‘Schedule Start Date’ &amp; proceed. "
 								        "https://app.avetta.com/avt-cli/x</div></html>"
 								    )
 								    cleaned = classify.strip_html(raw)
 								    assert "<" not in cleaned and ">" not in cleaned
 								    assert "‘" not in cleaned and "’" not in cleaned  # smart quotes gone
 								    assert "'Schedule Start Date'" in cleaned
 								    assert "&amp;" not in cleaned and "&" in cleaned  # entity decoded
 								    assert "https://" not in cleaned  # URL reduced to a token
 								def test_known_intents():
 								    # Schedule confirmed (the dominant happy-path comment).
 								    cat, _ = classify.classify(
 								        "IP", "", "<html>WO schedule confirmed with vendor</html>"
 								    )
 								    assert cat == "Schedule Confirmed"
 								    # Cancelled via WO Status, even with a generic comment.
 								    cat, _ = classify.classify(
 								        "RCAN", "", "<html>WO Cancelled, created in error.</html>"
 								    )
 								    assert cat == "Cancelled"
 								    # 3rd-attempt escalation cadence.
 								    cat, _ = classify.classify(
 								        "H", "REPORT", "<html>3rd attempt process for schedule confirmation.</html>"
 								    )
 								    assert cat == "3rd Escalation"
 								    # Vendor no-show.
 								    cat, _ = classify.classify(
 								        "R", "", "<html>Site tech reported vendor was a no show for Friday.</html>"
 								    )
 								    assert cat == "Vendor No-Show"
 								    # Weekly cadence template.
 								    cat, _ = classify.classify(
 								        "IP",
 								        "",
 								        "<html>Weekly WO scheduled. Service reports required EOD Friday.</html>",
 								    )
 								    assert cat == "Weekly WO Scheduled"
 								    # Structured-only fallback: no comment intent fires, REPORT hold decides.
 								    cat, _ = classify.classify("IP", "REPORT", "<html></html>")
 								    assert cat == "Report / Docs Needed"
 								def test_mismatch_detection():
 								    # Comment claims completion while on a REPORT hold → mismatch surfaced.
 								    cat, mm = classify.classify(
 								        "IP", "REPORT", "<html>Vendor arrived and performed task.</html>"
 								    )
 								    assert cat == "Completed / Pending Close"
 								    assert mm is not None and "REPORT" in mm
 								    # Schedule confirmed while on a SCHEDULING hold → mismatch surfaced.
 								    cat, mm = classify.classify(
 								        "R", "SCHEDULING", "<html>WO schedule confirmed with vendor.</html>"
 								    )
 								    assert cat == "Schedule Confirmed"
 								    assert mm is not None
 								    # Clean case: no contradiction → no mismatch.
 								    _, mm = classify.classify(
 								        "IP", "", "<html>WO schedule confirmed with vendor</html>"
 								    )
 								    assert mm is None
 								def test_classify_is_offline():
 								    """classify() must not import boto3 or reach the network."""
 								    import sys as _sys
 								    had_boto3 = "boto3" in _sys.modules
 								    classify.classify("IP", "REPORT", "<html>1st attempt process for report.</html>")
 								    # If boto3 wasn't already loaded, classify() must not have pulled it in.
 								    if not had_boto3:
 								        assert "boto3" not in _sys.modules
 								@pytest.mark.skipif(not FIXTURE.exists(), reason=f"sample export not found: {FIXTURE}")
 								def test_other_share_against_real_export(capsys):
 								    import openpyxl
 								    wb = openpyxl.load_workbook(FIXTURE, read_only=True, data_only=True)
 								    rows = list(wb.active.iter_rows(values_only=True))[1:]  # drop header
 								    dist = collections.Counter()
 								    mismatches = 0
 								    classified = 0
 								    blank = 0
 								    other_samples = []
 								    for row in rows:
 								        comment = row[COL_LAST_COMMENT]
 								        # Blank-comment rows are excluded from the classified total by design.
 								        if comment is None or str(comment).strip() == "":
 								            blank += 1
 								            continue
 								        classified += 1
 								        category, mismatch = classify.classify(
 								            row[COL_WO_STATUS], row[COL_HOLD_REASON], comment
 								        )
 								        dist[category] += 1
 								        if mismatch:
 								            mismatches += 1
 								        if category == "Other" and len(other_samples) < 20:
 								            other_samples.append(classify.strip_html(comment)[:90])
 								    other = dist["Other"]
 								    other_pct = other * 100.0 / classified if classified else 0.0
 								    with capsys.disabled():
 								        print(f"\n=== APM classifier smoke test: {FIXTURE.name} ===")
 								        print(
 								            f"rows={len(rows)} classified={classified} "
 								            f"blank-excluded={blank} (blank-comment rows excluded from the total)"
 								        )
 								        print("--- category distribution ---")
 								        for cat, count in dist.most_common():
 								            print(f"  {count:4d}  {count * 100.0 / classified:5.1f}%  {cat}")
 								        print(f"--- Other: {other} ({other_pct:.2f}%) ---")
 								        for sample in other_samples:
 								            print(f"    [Other] {sample}")
 								        print(f"--- mismatches flagged: {mismatches} ---")
 								    assert classified > 0
 								    assert other_pct <= MAX_OTHER_PCT, (
 								        f"deterministic Other {other_pct:.2f}% exceeds {MAX_OTHER_PCT}% — "
 								        "the two-axis ladder regressed"
 								    )
 								    # Two-axis model should comfortably beat the legacy ~17% Other.
 								    assert other_pct < 17.0
 								@pytest.mark.skipif(not FIXTURE.exists(), reason=f"sample export not found: {FIXTURE}")
 								def test_known_fixture_rows_in_export():
 								    """Anchor on real rows found in the canonical export."""
 								    import openpyxl
 								    wb = openpyxl.load_workbook(FIXTURE, read_only=True, data_only=True)
 								    rows = list(wb.active.iter_rows(values_only=True))[1:]
 								    by_status = collections.defaultdict(list)
 								    schedule_confirmed_row = None
 								    report_completion_mismatch = None
 								    for row in rows:
 								        comment = row[COL_LAST_COMMENT]
 								        if comment is None or str(comment).strip() == "":
 								            continue
 								        text = classify.strip_html(comment)
 								        by_status[row[COL_WO_STATUS]].append(row)
 								        if (
 								            schedule_confirmed_row is None
 								            and "schedule confirmed with vendor" in text.lower()
 								            and not (row[COL_HOLD_REASON] or "").strip()
 								        ):
 								            schedule_confirmed_row = row
 								        if (
 								            report_completion_mismatch is None
 								            and (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT"
 								            and "performed task" in text.lower()
 								        ):
 								            report_completion_mismatch = row
 								    # A "WO schedule confirmed with vendor" row → Schedule Confirmed.
 								    assert schedule_confirmed_row is not None, "fixture lacks a schedule-confirmed row"
 								    cat, _ = classify.classify(
 								        schedule_confirmed_row[COL_WO_STATUS],
 								        schedule_confirmed_row[COL_HOLD_REASON],
 								        schedule_confirmed_row[COL_LAST_COMMENT],
 								    )
 								    assert cat == "Schedule Confirmed"
 								    # Any RCAN row → Cancelled.
 								    assert "RCAN" in by_status, "fixture lacks an RCAN row"
 								    rcan = by_status["RCAN"][0]
 								    cat, _ = classify.classify(
 								        rcan[COL_WO_STATUS], rcan[COL_HOLD_REASON], rcan[COL_LAST_COMMENT]
 								    )
 								    assert cat == "Cancelled"
 								    # A REPORT-hold row whose comment claims completion → mismatch non-None.
 								    if report_completion_mismatch is not None:
 								        _, mm = classify.classify(
 								            report_completion_mismatch[COL_WO_STATUS],
 								            report_completion_mismatch[COL_HOLD_REASON],
 								            report_completion_mismatch[COL_LAST_COMMENT],
 								        )
 								        assert mm is not None
-												Add synthetic export fixture and run classification quality gate in CI

											
										
										
											2026-05-29 14:30:44 -04:00
 								# ---------------------------------------------------------------------------
 								# CSV fixture tests — always run (no skipif), so the quality gate fires in CI.
 								# ---------------------------------------------------------------------------
 								def test_other_share_against_csv_fixture(capsys):
 								    """Classification quality gate against the committed synthetic CSV fixture.
 								    Measured deterministic Other%: 7.41% (2/27).  Threshold: 9.0%.
 								    This test runs unconditionally in CI.
 								    """
 								    rows = _load_csv_rows(CSV_FIXTURE)
 								    dist: collections.Counter = collections.Counter()
 								    mismatches = 0
 								    classified = 0
 								    blank = 0
 								    other_samples: list[str] = []
 								    for row in rows:
 								        comment = row[COL_LAST_COMMENT] if len(row) > COL_LAST_COMMENT else ""
 								        # Blank-comment rows are excluded from the classified total by design.
 								        if comment is None or str(comment).strip() == "":
 								            blank += 1
 								            continue
 								        classified += 1
 								        category, mismatch = classify.classify(
 								            row[COL_WO_STATUS], row[COL_HOLD_REASON], comment
 								        )
 								        dist[category] += 1
 								        if mismatch:
 								            mismatches += 1
 								        if category == "Other" and len(other_samples) < 20:
 								            other_samples.append(classify.strip_html(comment)[:90])
 								    other = dist["Other"]
 								    other_pct = other * 100.0 / classified if classified else 0.0
 								    with capsys.disabled():
 								        print(f"\n=== APM classifier smoke test (CSV fixture): {CSV_FIXTURE.name} ===")
 								        print(
 								            f"rows={len(rows)} classified={classified} "
 								            f"blank-excluded={blank} (blank-comment rows excluded from the total)"
 								        )
 								        print("--- category distribution ---")
 								        for cat, count in dist.most_common():
 								            print(f"  {count:4d}  {count * 100.0 / classified:5.1f}%  {cat}")
 								        print(f"--- Other: {other} ({other_pct:.2f}%) ---")
 								        for sample in other_samples:
 								            print(f"    [Other] {sample}")
 								        print(f"--- mismatches flagged: {mismatches} ---")
 								    assert classified > 0, "CSV fixture produced no classified rows"
 								    assert other_pct <= CSV_FIXTURE_MAX_OTHER_PCT, (
 								        f"deterministic Other {other_pct:.2f}% exceeds {CSV_FIXTURE_MAX_OTHER_PCT}% — "
 								        "the two-axis ladder regressed against the committed fixture"
 								    )
 								    # Fixture must exercise mismatch detection (WO-1012: REPORT hold + performed task).
 								    assert mismatches >= 1, "CSV fixture should contain at least one mismatch row"
 								def test_known_rows_in_csv_fixture():
 								    """Anchor checks on the synthetic CSV fixture — runs unconditionally in CI."""
 								    rows = _load_csv_rows(CSV_FIXTURE)
 								    by_status: collections.defaultdict = collections.defaultdict(list)
 								    schedule_confirmed_row = None
 								    report_completion_mismatch = None
 								    third_esc_rows: list[tuple] = []
 								    cancelled_rows: list[tuple] = []
 								    structured_report_rows: list[tuple] = []
 								    structured_scheduling_rows: list[tuple] = []
 								    for row in rows:
 								        comment = row[COL_LAST_COMMENT] if len(row) > COL_LAST_COMMENT else ""
 								        if comment is None or str(comment).strip() == "":
 								            continue
 								        text = classify.strip_html(comment)
 								        by_status[row[COL_WO_STATUS]].append(row)
 								        if (
 								            schedule_confirmed_row is None
 								            and "schedule confirmed with vendor" in text.lower()
 								            and not (row[COL_HOLD_REASON] or "").strip()
 								        ):
 								            schedule_confirmed_row = row
 								        if (
 								            report_completion_mismatch is None
 								            and (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT"
 								            and "performed task" in text.lower()
 								        ):
 								            report_completion_mismatch = row
 								        if re.search(r"\b3rd\b.*\battempt\b", text.lower()):
 								            third_esc_rows.append(row)
 								        if row[COL_WO_STATUS] == "RCAN":
 								            cancelled_rows.append(row)
 								        # Structured-only: HTML-wrapped empty comment (strips to "") with REPORT hold.
 								        if (row[COL_HOLD_REASON] or "").strip().upper() == "REPORT" and text == "":
 								            structured_report_rows.append(row)
 								        if (row[COL_HOLD_REASON] or "").strip().upper() == "SCHEDULING" and text == "":
 								            structured_scheduling_rows.append(row)
 								    # "WO schedule confirmed with vendor" → Schedule Confirmed.
 								    assert schedule_confirmed_row is not None, "fixture lacks a schedule-confirmed row"
 								    cat, _ = classify.classify(
 								        schedule_confirmed_row[COL_WO_STATUS],
 								        schedule_confirmed_row[COL_HOLD_REASON],
 								        schedule_confirmed_row[COL_LAST_COMMENT],
 								    )
 								    assert cat == "Schedule Confirmed", f"expected Schedule Confirmed, got {cat!r}"
 								    # RCAN rows → Cancelled.
 								    assert cancelled_rows, "fixture lacks an RCAN row"
 								    cat, _ = classify.classify(
 								        cancelled_rows[0][COL_WO_STATUS],
 								        cancelled_rows[0][COL_HOLD_REASON],
 								        cancelled_rows[0][COL_LAST_COMMENT],
 								    )
 								    assert cat == "Cancelled", f"expected Cancelled, got {cat!r}"
 								    # 3rd Escalation rows are present and classify correctly.
 								    assert third_esc_rows, "fixture lacks a 3rd-escalation row"
 								    cat, _ = classify.classify(
 								        third_esc_rows[0][COL_WO_STATUS],
 								        third_esc_rows[0][COL_HOLD_REASON],
 								        third_esc_rows[0][COL_LAST_COMMENT],
 								    )
 								    assert cat == "3rd Escalation", f"expected 3rd Escalation, got {cat!r}"
 								    # Structured-only REPORT rows → Report / Docs Needed.
 								    assert structured_report_rows, "fixture lacks a structured-only REPORT hold row"
 								    cat, _ = classify.classify(
 								        structured_report_rows[0][COL_WO_STATUS],
 								        structured_report_rows[0][COL_HOLD_REASON],
 								        structured_report_rows[0][COL_LAST_COMMENT],
 								    )
 								    assert cat == "Report / Docs Needed", f"expected Report / Docs Needed, got {cat!r}"
 								    # Structured-only SCHEDULING rows → Awaiting Scheduling.
 								    assert structured_scheduling_rows, "fixture lacks a structured-only SCHEDULING row"
 								    cat, _ = classify.classify(
 								        structured_scheduling_rows[0][COL_WO_STATUS],
 								        structured_scheduling_rows[0][COL_HOLD_REASON],
 								        structured_scheduling_rows[0][COL_LAST_COMMENT],
 								    )
 								    assert cat == "Awaiting Scheduling", f"expected Awaiting Scheduling, got {cat!r}"
 								    # REPORT-hold row with completion comment → mismatch surfaced.
 								    assert report_completion_mismatch is not None, (
 								        "fixture lacks a REPORT-hold row with a completion comment (mismatch case)"
 								    )
 								    _, mm = classify.classify(
 								        report_completion_mismatch[COL_WO_STATUS],
 								        report_completion_mismatch[COL_HOLD_REASON],
 								        report_completion_mismatch[COL_LAST_COMMENT],
 								    )
 								    assert mm is not None, "expected a mismatch reason for WO-1012 but got None"
 								    assert "REPORT" in mm, f"mismatch reason should mention REPORT hold: {mm!r}"