SSA full drift checker (template + staging headers)

docs/CHANGELOG-SSA.md

@@ -3,6 +3,31 @@
 Scoped changelog for the Site Safety Audit pipeline and its quality
 guards. Newest at top.
 
+## 2026-05-15
+
+### Added
+
+- **Full drift checker.** Two new checks beyond the minimal v1 CSV
+  header check:
+  - Every `PH_*` placeholder constant in `ssa_pipeline.py` must appear
+    as literal text in the SSA report template `.docx` (concatenated
+    `<w:t>` text across all `word/*.xml` parts, so tokens split across
+    formatting boundaries are still caught).
+  - Every `STAGING_HEADERS` entry must have a corresponding
+    `_STAGING_COL_WIDTHS` key. Subset semantics, not equality —
+    `_STAGING_COL_WIDTHS` legitimately carries forward-defined
+    gap-4/5/6 columns (phase, activity_ref, hrcw, swms_required, etc.)
+    that aren't yet in `STAGING_HEADERS`.
+
+### Changed
+
+- **Placeholder tokens centralised.** `build_ssa_report_docx` no longer
+  carries inline `{{SITE_ADDRESS}}` / `{{NARRATIVE_SUMMARY}}` /
+  `{{AUDIT_DATE}}` / `{{PREPARED_BY}}` strings in its `body_replacements`
+  and `footer_replacements` dicts. They're now module-level
+  `PH_*` constants plus an `SSA_REPORT_PLACEHOLDERS` tuple, which the
+  drift checker walks. Pure refactor — no behaviour change.
+
 ## 2026-05-14
 
 ### Added

pims/services/ssa_pipeline.py

@@ -77,6 +77,22 @@ def flag(self, reason: str) -> None:
 _REQUIRED_HEADER = ("timestamp", "observation", "filename")
 _TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}$")
 
+# Placeholder tokens substituted by build_ssa_report_docx. Centralised here so
+# the drift checker (tools/check_ssa_drift.py) can verify each token exists as
+# literal text in the SSA report template. Body and footer dicts both use the
+# same set of tokens; only the source value differs (long-form date in body
+# vs short-form date in footer for AUDIT_DATE).
+PH_SITE_ADDRESS = "{{SITE_ADDRESS}}"
+PH_NARRATIVE_SUMMARY = "{{NARRATIVE_SUMMARY}}"
+PH_AUDIT_DATE = "{{AUDIT_DATE}}"
+PH_PREPARED_BY = "{{PREPARED_BY}}"
+SSA_REPORT_PLACEHOLDERS: tuple[str, ...] = (
+    PH_SITE_ADDRESS,
+    PH_NARRATIVE_SUMMARY,
+    PH_AUDIT_DATE,
+    PH_PREPARED_BY,
+)
+
 
 def _decode_csv_bytes(raw: bytes) -> str:
     """Detect UTF-8 / UTF-8-SIG / CP1252 and return a unicode string.
@@ -2681,16 +2697,16 @@ def build_ssa_report_docx(
     audit_date_ordinal = _to_ordinal_date(audit_date_ddmmyyyy)
     audit_date_short = _to_short_date(audit_date_ddmmyyyy)
     body_replacements = {
-        "{{SITE_ADDRESS}}": full_site,
-        "{{NARRATIVE_SUMMARY}}": combined_narrative,
-        "{{AUDIT_DATE}}": audit_date_ordinal or audit_date_ddmmyyyy or "",
-        "{{PREPARED_BY}}": prepared_by or "",
+        PH_SITE_ADDRESS: full_site,
+        PH_NARRATIVE_SUMMARY: combined_narrative,
+        PH_AUDIT_DATE: audit_date_ordinal or audit_date_ddmmyyyy or "",
+        PH_PREPARED_BY: prepared_by or "",
     }
     footer_replacements = {
-        "{{SITE_ADDRESS}}": full_site,
-        "{{NARRATIVE_SUMMARY}}": combined_narrative,
-        "{{AUDIT_DATE}}": audit_date_short or audit_date_ddmmyyyy or "",
-        "{{PREPARED_BY}}": prepared_by or "",
+        PH_SITE_ADDRESS: full_site,
+        PH_NARRATIVE_SUMMARY: combined_narrative,
+        PH_AUDIT_DATE: audit_date_short or audit_date_ddmmyyyy or "",
+        PH_PREPARED_BY: prepared_by or "",
     }
     _replace_tokens_in_part(doc.part, body_replacements)
     # Split the combined narrative into the canonical two-paragraph

tests/test_ssa_drift_checker.py

@@ -5,28 +5,80 @@
 from pathlib import Path
 
 
-def test_drift_checker_clean_state_returns_zero(capsys, monkeypatch):
-    """Current main passes — no drift between paired constants."""
+def _import_checker():
+    """Import (or re-import) the drift checker module from tools/."""
     sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
     import check_ssa_drift  # noqa: E402
-    rc = check_ssa_drift.main()
+    return check_ssa_drift
+
+
+def test_drift_checker_clean_state_returns_zero(capsys):
+    """Current main passes — no drift across any of the wired checks."""
+    drift = _import_checker()
+    rc = drift.main()
     out = capsys.readouterr().out
-    assert rc == 0
+    assert rc == 0, out
     assert "[X]" in out
     assert "no drift" in out.lower()
 
 
 def test_drift_checker_flags_csv_header_mismatch(capsys, monkeypatch):
     """Forcing the preflight constant to diverge from ssa_pipeline's
-    must trip the drift check."""
-    sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
-    import check_ssa_drift  # noqa: E402
+    must trip the CSV header check."""
+    drift = _import_checker()
     monkeypatch.setattr(
-        check_ssa_drift, "_EXPECTED_CSV_HEADERS",
+        drift, "_EXPECTED_CSV_HEADERS",
         ("timestamp", "observation", "DIVERGED"),
     )
-    rc = check_ssa_drift.main()
+    rc = drift.main()
     out = capsys.readouterr().out
     assert rc == 1
     assert "DRIFT" in out
     assert "DIVERGED" in out
+
+
+def test_drift_checker_flags_missing_placeholder_in_template(
+    capsys, monkeypatch,
+):
+    """Adding a fake PH_* that doesn't exist in the template must trip
+    the placeholder check."""
+    drift = _import_checker()
+    monkeypatch.setattr(
+        drift, "SSA_REPORT_PLACEHOLDERS",
+        ("{{SITE_ADDRESS}}", "{{NEVER_IN_TEMPLATE}}"),
+    )
+    rc = drift.main()
+    out = capsys.readouterr().out
+    assert rc == 1
+    assert "DRIFT" in out
+    assert "NEVER_IN_TEMPLATE" in out
+
+
+def test_drift_checker_flags_staging_header_without_width(
+    capsys, monkeypatch,
+):
+    """A STAGING_HEADERS entry with no corresponding width entry must
+    trip the staging-width check."""
+    drift = _import_checker()
+    monkeypatch.setattr(
+        drift, "STAGING_HEADERS",
+        drift.STAGING_HEADERS + ("phantom_column",),
+    )
+    rc = drift.main()
+    out = capsys.readouterr().out
+    assert rc == 1
+    assert "DRIFT" in out
+    assert "phantom_column" in out
+
+
+def test_drift_checker_tolerates_extra_width_entries(capsys):
+    """_STAGING_COL_WIDTHS legitimately carries forward-defined gap-4/5/6
+    columns that aren't yet in STAGING_HEADERS. The subset semantics must
+    NOT flag this as drift."""
+    drift = _import_checker()
+    rc = drift.main()
+    out = capsys.readouterr().out
+    # Today's main has gap-4/5/6 width entries (phase, hrcw, swms_*)
+    # absent from STAGING_HEADERS. If this test fails, the check
+    # tightened to equality and is now too strict.
+    assert rc == 0, out

tools/check_ssa_drift.py

@@ -4,35 +4,62 @@
 
     python tools/check_ssa_drift.py
 
-Drift is the silent divergence between paired constants in different
-files. Today this checker covers one case — the Evidence_Master.csv
-header tuple, which is duplicated in:
-
-  - pims/services/ssa_pipeline.py   ``_REQUIRED_HEADER``
-  - pims/services/ssa_quality/preflight.py  ``_EXPECTED_CSV_HEADERS``
-
-Both must stay aligned; the preflight check would silently start
-accepting CSVs the pipeline can't parse if they drift.
+Drift is the silent divergence between paired facts in different files.
+A constant in one module, a literal in a template, a header tuple in a
+docs file — they all encode the same truth, and if one moves while the
+others don't, the build keeps working but produces wrong output. Word's
+recovery prompt tolerates it; downstream consumers do not.
+
+Checks today:
+
+  1. ``Evidence_Master.csv`` header tuple — duplicated in
+     ``ssa_pipeline._REQUIRED_HEADER`` and
+     ``preflight._EXPECTED_CSV_HEADERS``. Must stay aligned or preflight
+     silently accepts CSVs the pipeline can't parse.
+  2. SSA report template placeholders — every ``PH_*`` constant in
+     ``ssa_pipeline.py`` must appear as literal text somewhere in the
+     report template ``.docx``. Catches "renamed PH_FOO in code but
+     forgot the template" (or vice versa).
+  3. Staging xlsx header widths — every entry in
+     ``STAGING_HEADERS`` must have a corresponding entry in
+     ``_STAGING_COL_WIDTHS``. Subset (not equality): extra width entries
+     are tolerated (gap-4/5/6 forward-defined columns); missing entries
+     are not.
 
 Future checks can be added by appending to ``_CHECKS`` — each is a
 ``(label, callable)`` returning ``None`` on pass or a string detail on
 fail.
 """
 from __future__ import annotations
 
+import re
 import sys
+import zipfile
 from pathlib import Path
+from typing import Callable, Optional
 
 # Add repo root to sys.path so this script can be run from anywhere.
 _REPO_ROOT = Path(__file__).resolve().parent.parent
 if str(_REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(_REPO_ROOT))
 
-from pims.services.ssa_pipeline import _REQUIRED_HEADER
+from pims.services.ssa_pipeline import (
+    _REQUIRED_HEADER,
+    _STAGING_COL_WIDTHS,
+    SSA_REPORT_PLACEHOLDERS,
+    SSA_REPORT_TEMPLATE,
+    STAGING_HEADERS,
+)
 from pims.services.ssa_quality.preflight import _EXPECTED_CSV_HEADERS
 
 
-def _check_csv_headers() -> str | None:
+# Matches ``<w:t ...>text</w:t>`` runs so we can extract Word's visible
+# text content from word/*.xml parts (placeholders may be split across
+# multiple <w:t> runs when formatting boundaries land mid-token).
+_W_T_RE = re.compile(r"<w:t[^>]*>([^<]*)</w:t>")
+
+
+def _check_csv_headers() -> Optional[str]:
     if _REQUIRED_HEADER == _EXPECTED_CSV_HEADERS:
         return None
     return (
@@ -41,9 +68,49 @@ def _check_csv_headers() -> str | None:
     )
 
 
-_CHECKS: tuple[tuple[str, callable], ...] = (
+def _check_template_placeholders() -> Optional[str]:
+    """Every PH_* must appear as literal text in the report template."""
+    if not SSA_REPORT_TEMPLATE.is_file():
+        return f"template missing: {SSA_REPORT_TEMPLATE}"
+    # Concatenate <w:t> text across every word/*.xml part — covers
+    # body, headers, footers, cover-page text boxes. Tag-strip via
+    # findall then join, which collapses split-across-runs tokens.
+    parts: list[str] = []
+    with zipfile.ZipFile(SSA_REPORT_TEMPLATE) as z:
+        for name in z.namelist():
+            if name.startswith("word/") and name.endswith(".xml"):
+                xml = z.read(name).decode("utf-8", errors="ignore")
+                parts.extend(_W_T_RE.findall(xml))
+    haystack = "".join(parts)
+    missing = [ph for ph in SSA_REPORT_PLACEHOLDERS if ph not in haystack]
+    if missing:
+        return f"placeholder(s) not found in template: {missing}"
+    return None
+
+
+def _check_staging_header_widths() -> Optional[str]:
+    """Every STAGING_HEADERS entry must have an _STAGING_COL_WIDTHS key.
+
+    Subset check, not equality — _STAGING_COL_WIDTHS legitimately
+    carries forward-defined gap-4/5/6 columns (phase, activity_ref,
+    hrcw, swms_required, etc.) that aren't yet in STAGING_HEADERS.
+    """
+    missing = [h for h in STAGING_HEADERS if h not in _STAGING_COL_WIDTHS]
+    if missing:
+        return (
+            f"STAGING_HEADERS entries missing _STAGING_COL_WIDTHS entry: "
+            f"{missing}"
+        )
+    return None
+
+
+_CHECKS: tuple[tuple[str, Callable[[], Optional[str]]], ...] = (
     ("Evidence_Master.csv header tuple matches across modules",
      _check_csv_headers),
+    ("SSA report template carries every PH_* placeholder",
+     _check_template_placeholders),
+    ("Every STAGING_HEADERS entry has a width definition",
+     _check_staging_header_widths),
 )