feat(notify): Teams DM bell beside failure emails (T1)

CLAUDE.md

@@ -239,6 +239,41 @@ file, then `tar xvf` extracts. Both steps log stdout+stderr.
 
 ---
 
+## Failure notifications (Teams DM bell)
+
+Every `[FAILED]` / `[STALE]` / `[EXCLUSION-ABORT]` still sends its durable audit
+email via `send_alert_email()` (smtplib → smtp.purdue.edu). In addition, `notify_bell()`
+fires a **Teams alert** beside that email — the T1 "backup didn't happen / needs you
+NOW" attention signal (email alone gets tuned out). The card title is prefixed
+`ACT NOW:`.
+
+- **Delivered as a 1:1 Teams DM, no @mention.** The card carries no `<at>` / no
+  `msteams.entities`: a DM already notifies, and a Power-Automate flow-bot chat post
+  does **not** resolve `<at>` entities (they render as literal text). This mirrors
+  florasense-tools PR #106's `--dm` mode. **DM-vs-channel is decided entirely by the
+  Power Automate flow behind `EARLY_TEAMS_WEBHOOK`**, not by the payload — switching to
+  a DM is a Power-Automate web-UI change, done separately from this code.
+- **Stdlib-only** (`urllib.request`, imports inside the function body per invariant #1)
+  — no new deps, and reuses the Adaptive-Card shape proven in
+  `florasense-tools/fortress/reship_watch_alert.py`.
+- **ENV-GATED + DEFAULT-OFF.** A silent no-op (no POST) unless `EARLY_TEAMS_WEBHOOK`
+  is set, so it stays dormant until wired into the run environment (Negishi cron /
+  `archive_local.sbatch`). No mention env vars are needed (or read).
+- **Never breaks a run.** Only HTTP `{200, 202}` count as delivered; any error
+  (absent/failed webhook, network) degrades to email-only with a logged reason and
+  **never raises**. The webhook is a **secret** (carries a `sig`) — never printed/logged.
+- **Two copies, kept in sync.** The module-level `notify_bell()`/`build_teams_card()`
+  serve the LOCAL orchestration sites; `send_to_fortress` carries a self-contained
+  **inline twin** (closes over `write_log` only, imports inside) for the one alert that
+  can fire on a compute node under `--globus` — the exclusion companion-guard abort,
+  which continues the run and so never reaches the local Step-2 wrapper. All other
+  verification `[FAILED]`s raise → that wrapper alerts + bells.
+
+Not to be confused with the FloraSense `reship_watch_alert.py` deployment glue (a
+separate, lower-tier alert for the sustained reship-watch / Phase-3 trigger).
+
+---
+
 ## Authentication (`--globus` only)
 
 Uses the personal Globus identity (native app auth):

archive.py

@@ -195,6 +195,92 @@ def send_alert_email(emails, subject, body):
         print(f"  WARNING: Could not send alert email: {e}")
 
 
+def build_teams_card(title, text):
+    """
+    Build a Teams "message" Adaptive Card payload — the shape proven in
+    florasense-tools/fortress/reship_watch_alert.py (build_teams_card). Pure: no
+    network, no env, no imports — trivially unit-testable and safe to inline anywhere.
+
+    NO @mention. Fortress T1 alerts are delivered as a 1:1 Teams DM (Jarrod's routing
+    decision, 2026-07-11), and a DM already notifies the recipient. A Power-Automate
+    flow-bot chat post does NOT resolve "<at>...</at>" mention entities — they render
+    as literal text — so a DM card must carry no mention block (this mirrors
+    florasense-tools PR #106's --dm mode). Whether the webhook posts a DM or a channel
+    message is decided entirely by the Power Automate flow behind $EARLY_TEAMS_WEBHOOK,
+    not by this payload.
+    """
+    return {"type": "message", "attachments": [{
+        "contentType": "application/vnd.microsoft.card.adaptive",
+        "content": {
+            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+            "type": "AdaptiveCard",
+            "version": "1.4",
+            "body": [
+                {"type": "TextBlock", "text": title, "weight": "Bolder",
+                 "size": "Medium", "wrap": True},
+                {"type": "TextBlock", "text": text, "wrap": True},
+            ],
+        },
+    }]}
+
+
+def notify_bell(subject, body):
+    """
+    Fire a Teams alert (a 1:1 DM) BESIDE the durable audit email — the T1 "backup
+    didn't happen / needs you NOW" attention signal that email alone no longer
+    delivers (routine mail gets tuned out). The card title is prefixed "ACT NOW:" so
+    the one alert surface reads as the top triage tier.
+
+    ENV-GATED + DEFAULT-OFF: a silent no-op (returns False, no POST) unless
+    $EARLY_TEAMS_WEBHOOK is set, so this stays dormant until the webhook is wired into
+    the run environment (e.g. Negishi cron / archive_local.sbatch). No @mention is sent
+    (see build_teams_card — a DM already notifies); the DM-vs-channel destination is
+    set by the Power Automate flow behind the webhook, not here.
+
+    Stdlib-only (urllib) and NEVER raises: a missing or failing webhook must never
+    break an archive run, so every error path degrades to email-only (the caller has
+    already sent the email) with a logged reason. Only HTTP {200, 202} count as
+    delivered.
+
+    The webhook is a SECRET (it carries a `sig`) — it is NEVER printed or logged.
+
+    Imports are inside the body per engine invariant #1 (Compute functions are
+    serialized with dill and cannot see outer-scope imports). This module-level copy
+    runs in the LOCAL orchestration; send_to_fortress carries a self-contained inline
+    twin for the one alert that can fire on a compute node under --globus (the
+    exclusion companion-guard abort) — keep the two in sync.
+
+    Returns True iff the bell was delivered (HTTP 200/202), else False.
+    """
+    import os
+    import json
+    import urllib.request
+    import urllib.error
+
+    url = os.environ.get("EARLY_TEAMS_WEBHOOK")
+    if not url:
+        return False  # default-off: bell not wired on this host — the email is the record
+
+    title = subject if subject.startswith("ACT NOW:") else f"ACT NOW: {subject}"
+    try:
+        payload = json.dumps(build_teams_card(title, body)).encode()
+        req = urllib.request.Request(url, data=payload, method="POST")
+        req.add_header("Content-Type", "application/json")
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            status = getattr(resp, "status", None) or resp.getcode()
+    except urllib.error.HTTPError as e:
+        status = e.code
+    except Exception as e:
+        # Network/URL error. NOTE: never interpolate `url` here — it is a secret.
+        print(f"  WARNING: Teams bell not delivered ({type(e).__name__}); email still sent.")
+        return False
+
+    if status in (200, 202):
+        return True
+    print(f"  WARNING: Teams bell not delivered (HTTP {status}); email still sent.")
+    return False
+
+
 def get_compute_client(emails):
     """
     Authenticate using your personal Globus identity (native app auth).
@@ -221,6 +307,7 @@ def get_compute_client(emails):
             "[FAILED] Fortress archive: authentication error",
             msg
         )
+        notify_bell("[FAILED] Fortress archive: authentication error", msg)
         sys.exit(1)
 
 
@@ -2129,6 +2216,49 @@ def send_email(subject, body):
         except Exception as e:
             write_log(f"Failed to send email: {e}", level="WARNING")
 
+    def notify_bell(subject, body):
+        """
+        Inline, Compute-safe twin of the module-level notify_bell(): a T1 Teams DM
+        BESIDE the audit email. This one can run on a compute node under --globus, so
+        it references no module/outer names and imports inside (invariant #1) — it only
+        closes over write_log, exactly like send_email does. Keep in sync with the
+        module-level notify_bell()/build_teams_card().
+
+        Env-gated + default-off on $EARLY_TEAMS_WEBHOOK (a SECRET — never printed); NO
+        @mention (a DM already notifies; a flow-bot chat post renders <at> as literal
+        text); only HTTP {200,202} = delivered; NEVER raises (a failed/absent webhook
+        must never break the run).
+        """
+        import os
+        import json
+        import urllib.request
+        import urllib.error
+        url = os.environ.get("EARLY_TEAMS_WEBHOOK")
+        if not url:
+            return
+        title = subject if subject.startswith("ACT NOW:") else "ACT NOW: " + subject
+        payload = json.dumps({"type": "message", "attachments": [{
+            "contentType": "application/vnd.microsoft.card.adaptive",
+            "content": {
+                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                "type": "AdaptiveCard", "version": "1.4",
+                "body": [
+                    {"type": "TextBlock", "text": title, "weight": "Bolder",
+                     "size": "Medium", "wrap": True},
+                    {"type": "TextBlock", "text": body, "wrap": True},
+                ]}}]}).encode()
+        try:
+            req = urllib.request.Request(url, data=payload, method="POST")
+            req.add_header("Content-Type", "application/json")
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                status = getattr(resp, "status", None) or resp.getcode()
+            if status not in (200, 202):
+                write_log(f"Teams bell not delivered (HTTP {status}); email still sent.",
+                          level="WARNING")
+        except Exception as e:  # never printed with the URL (it's a secret)
+            write_log(f"Teams bell not delivered ({type(e).__name__}); email still sent.",
+                      level="WARNING")
+
     write_log(f"Starting archive workflow")
     write_log(f"Project: {project_name}")
     write_log(f"Source folder (Depot): {source_folder}")
@@ -2188,6 +2318,16 @@ def send_email(subject, body):
             f"Log (TXT):  {txt_path}\n"
             f"Log (JSON): {json_path}"
         )
+        # Bell beside the email: unlike the verification [FAILED] paths (which raise
+        # → the local Step 2 wrapper alerts + bells), this abort CONTINUES the run, so
+        # it never reaches that wrapper — fire the T1 bell here or it stays email-only.
+        notify_bell(
+            f"[EXCLUSION-ABORT] Fortress archive: {project_name}",
+            f"Exclusion companion guard tripped for {project_name}: exclusion was "
+            f"SUPPRESSED and EVERYTHING backed up (no data lost), but the exclude "
+            f"spec / classification needs review.\n\n{detail}\n\n"
+            f"Log (JSON): {json_path}"
+        )
 
     subprocess.run(f'hsi "mkdir -p {fortress_base_dir}"',
                    shell=True, capture_output=True)
@@ -2680,6 +2820,7 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
         print(f"\n  ERROR: {msg}")
         send_alert_email(emails,
                          f"[FAILED] Fortress archive: size_routing + --globus ({project_name})", msg)
+        notify_bell(f"[FAILED] Fortress archive: size_routing + --globus ({project_name})", msg)
         sys.exit(1)
 
     # `run` abstracts how a worker function is executed: in-process by default, or
@@ -2710,6 +2851,7 @@ def run(fn, *fn_args):
                    "Add the Globus Compute endpoint UUID, or run without --globus (local mode).")
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress archive: missing endpoint_id ({project_name})", msg)
+            notify_bell(f"[FAILED] Fortress archive: missing endpoint_id ({project_name})", msg)
             sys.exit(1)
 
         print("#" * 64)
@@ -2747,6 +2889,7 @@ def run(fn, *fn_args):
             )
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress archive: endpoint unreachable ({project_name})", msg)
+            notify_bell(f"[FAILED] Fortress archive: endpoint unreachable ({project_name})", msg)
             sys.exit(1)
         remote_major = int(remote_version_str.split('.')[0])
         remote_minor = int(remote_version_str.split('.')[1])
@@ -2767,6 +2910,7 @@ def run(fn, *fn_args):
             )
             print(f"\n  WARNING: {msg}")
             send_alert_email(emails, "[FAILED] Fortress archive: Python version mismatch", msg)
+            notify_bell("[FAILED] Fortress archive: Python version mismatch", msg)
             response = input("\n  Continue anyway? (yes/no): ").strip().lower()
             if response != 'yes':
                 print("Aborted.")
@@ -2805,13 +2949,15 @@ def run(fn, *fn_args):
                    f"Source: {source_folder}\nPattern: {file_pattern}\n\nError:\n{e}")
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress routing enumerate: {project_name}", msg)
+            notify_bell(f"[FAILED] Fortress routing enumerate: {project_name}", msg)
             sys.exit(1)
 
         if not survivors:
             msg = (f"Size-routing found no files to archive for {project_name} "
                    f"(pattern {file_pattern!r} under {source_folder}, after exclusion).")
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress routing: nothing to archive ({project_name})", msg)
+            notify_bell(f"[FAILED] Fortress routing: nothing to archive ({project_name})", msg)
             sys.exit(1)
 
         def ship_one(obj, stem):
@@ -2831,6 +2977,7 @@ def ship_one(obj, stem):
                    f"re-running resumes (unchanged objects skip).\n\nError:\n{e}")
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress routing ship: {project_name}", msg)
+            notify_bell(f"[FAILED] Fortress routing ship: {project_name}", msg)
             sys.exit(1)
 
         n_ship, n_skip = len(summary["shipped"]), len(summary["skipped"])
@@ -2947,6 +3094,8 @@ def ship_one(obj, stem):
                 print(f"\n  ERROR: could not enumerate source to validate resume: {e}")
                 send_alert_email(emails, f"[FAILED] Fortress archive resume check: {project_name}",
                                  f"source enumeration failed for {project_name}:\n{e}")
+                notify_bell(f"[FAILED] Fortress archive resume check: {project_name}",
+                            f"source enumeration failed for {project_name}:\n{e}")
                 sys.exit(1)
 
         # Staleness guard: an existing zip is only safe to reuse if it still reflects
@@ -2991,6 +3140,7 @@ def ship_one(obj, stem):
                 )
                 print(f"\n  ERROR: {msg}")
                 send_alert_email(emails, f"[STALE] Fortress archive resume blocked: {project_name}", msg)
+                notify_bell(f"[STALE] Fortress archive resume blocked: {project_name}", msg)
                 sys.exit(1)
             print(f"  [OK] Existing zip matches live source ({src_count} files)")
         print(f"  [OK] {len(members)} file(s) verified from existing zip")
@@ -3009,6 +3159,7 @@ def ship_one(obj, stem):
             )
             print(f"\n  ERROR: {msg}")
             send_alert_email(emails, f"[FAILED] Fortress archive step 1: {project_name}", msg)
+            notify_bell(f"[FAILED] Fortress archive step 1: {project_name}", msg)
             sys.exit(1)
         print(f"  [OK] {len(members)} file(s) matched and zipped")
     print(f"  [OK] Zip:          {zip_path}")
@@ -3053,6 +3204,7 @@ def ship_one(obj, stem):
         )
         print(f"\n  ERROR: {msg}")
         send_alert_email(emails, f"[FAILED] Fortress archive step 2: {project_name}", msg)
+        notify_bell(f"[FAILED] Fortress archive step 2: {project_name}", msg)
         sys.exit(1)
     print(f"  [OK] htar_large transfer complete")
     print(f"  [OK] Round-trip retrieval and MD5 verification passed")