#!/usr/bin/env python3
"""PageHawk Signature Campaigns: the campaign engine.

One document + a recipient list + live e-sign state, rolled up into a console that answers
"where does this stand" against a TARGET, not a count.

This module owns everything the browser must not: the campaign record, the provider calls, the
API key, the bucket assignment, and the invariant that proves the rollup adds up. The surface
(campaigns.html) renders what this returns and nothing else.

Design rules, each one paid for by a real failure in the reference build (see
~/.claude/references/software-development/campaign-admin-dashboard.md):

  1. Buckets are ASSIGNED, never computed as a residual. Every recipient lands in exactly one
     bucket by a deterministic ladder, and the result is ASSERTED against the universe. The one
     permitted residual is "no record", and it is labelled as such in the UI.
  2. Vendor status is not truth. It is translated through our semantics, and the translation
     records what it translated FROM so the footnote can explain itself. An e-sign provider
     stamps a signing link "viewed" the moment the link is created, so our own record of having
     emailed the thing outranks the vendor's word.
  3. The entity is the signer of record; the natural person signs through it.
  4. Coverage is not progress. Sends are outreach. Only signatures move the bar.
  5. A roster goes stale silently. Any envelope the provider knows about that the roster does
     not is surfaced as a visible staleness warning, never dropped.
  6. The secret stays here. Nothing this module returns may carry a credential.

CLI (all commands print one JSON object on stdout):
    pagehawk_campaigns.py selftest
    pagehawk_campaigns.py demo   [--out FILE]
    pagehawk_campaigns.py new    --out FILE [--doc PDF] [--name NAME] [--target-mode value|count]
    pagehawk_campaigns.py status --file FILE
    pagehawk_campaigns.py refresh --file FILE
"""

import argparse
import copy
import json
import os
import re
import sys
import time
import uuid

try:
    from zoneinfo import ZoneInfo
    _ET = ZoneInfo("America/New_York")
except Exception:                                            # pragma: no cover - ancient python
    _ET = None

FORMAT_VERSION = 1
EASTERN_LABEL = "ET"

# ── Buckets ──────────────────────────────────────────────────────────────────────────────────
# DECLARED, in display order. Every recipient is assigned exactly one of these keys. `binds`
# marks the only bucket that counts toward the target: signatures bind, sends do not.
BUCKETS = [
    {"key": "signed",   "label": "Signed",    "desc": "binding",             "binds": True},
    {"key": "sent",     "label": "Sent",      "desc": "awaiting signature"},
    {"key": "prepared", "label": "Prepared",  "desc": "built, not yet sent"},
    {"key": "declined", "label": "Declined",  "desc": "will not sign"},
    {"key": "norec",    "label": "No record", "desc": "nothing created yet", "residual": True},
]
BUCKET_KEYS = [b["key"] for b in BUCKETS]

# Raw vendor states we know how to read, grouped by what they actually mean to us. Anything not
# listed falls through to the unknown path, which is loud rather than silent.
_VENDOR_SIGNED = {"completed", "signed", "executed", "complete", "finished"}
_VENDOR_DECLINED = {"declined", "rejected", "voided", "cancelled", "canceled"}
_VENDOR_OPEN = {"sent", "delivered", "viewed", "opened", "pending", "awaiting", "created", "draft"}

# Field names that may never appear in anything this module returns. Keys live server-side; a
# rollup that carries one is a defect, and `_scrub` + the selftest exist to make that provable.
_SECRET_HINTS = ("api_key", "apikey", "secret", "token", "authorization", "bearer", "password")


# ── Time ─────────────────────────────────────────────────────────────────────────────────────
def et_stamp(epoch=None):
    """A 12-hour Eastern timestamp, the only time format Jason reads."""
    epoch = time.time() if epoch is None else epoch
    if _ET is None:                                          # pragma: no cover
        return time.strftime("%b %-d, %-I:%M%p", time.localtime(epoch)).replace("AM", "am").replace("PM", "pm")
    import datetime
    d = datetime.datetime.fromtimestamp(epoch, _ET)
    hour = d.hour % 12 or 12
    return "%s %d, %d:%02d%s %s" % (d.strftime("%b"), d.day, hour, d.minute,
                                    "am" if d.hour < 12 else "pm", EASTERN_LABEL)


def _stamp_obj(epoch=None):
    epoch = time.time() if epoch is None else epoch
    return {"epoch": int(epoch), "et": et_stamp(epoch)}


# ── Campaign record ──────────────────────────────────────────────────────────────────────────
def new_campaign(name="Untitled campaign", doc_path="", target_mode="value",
                 target_amount=0, currency="USD"):
    return {
        "phkcampaign": FORMAT_VERSION,
        "id": "camp_" + uuid.uuid4().hex[:12],
        "name": name,
        "created": _stamp_obj(),
        "document": {"path": doc_path, "name": os.path.basename(doc_path) if doc_path else ""},
        "target": {"mode": target_mode, "amount": target_amount, "currency": currency,
                   "label": "under signature" if target_mode == "value" else "signatures"},
        # Where state comes from. `fixtures` is offline and synthetic. `sealhawk` is the sending
        # engine (PageHawk owns the console, SealHawk owns sending, per Jason 2026-07-24); its
        # adapter is declared below and refuses to invent data until it is wired.
        "provider": {"kind": "fixtures", "campaign_ref": ""},
        "roster_synced": None,
        "recipients": [],
    }


def load_campaign(path):
    # Errors here reach Jason as a banner, so they say what happened in plain words rather than
    # handing an errno to the screen.
    if not os.path.exists(path):
        raise ValueError("There is no campaign file at %s." % path)
    try:
        with open(path, "r", encoding="utf-8") as fh:
            c = json.load(fh)
    except ValueError:
        raise ValueError("That file is not readable as a campaign. It may have been edited by hand "
                         "and left with a formatting error.")
    except OSError:
        raise ValueError("That campaign file could not be opened. Check that it has not been moved "
                         "or that its folder is still available.")
    if not isinstance(c, dict) or "recipients" not in c:
        raise ValueError("That file opened, but it is not a PageHawk campaign: it carries no "
                         "recipient list.")
    c.setdefault("phkcampaign", FORMAT_VERSION)
    c.setdefault("target", {"mode": "count", "amount": 0, "currency": "USD", "label": "signatures"})
    c.setdefault("provider", {"kind": "fixtures", "campaign_ref": ""})
    return c


def save_campaign(path, campaign):
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(_scrub(campaign), fh, indent=2, ensure_ascii=False)
    os.replace(tmp, path)
    return path


# ── Translation: vendor status is not truth ──────────────────────────────────────────────────
def translate(rec):
    """Map one recipient's raw record into OUR state.

    Returns (state, translated_from, why). `state` is one of the bucket keys.

    The load-bearing case: an e-sign provider marks an embedded signing link "viewed" the moment
    the link is created, so an envelope nobody ever emailed reads as counterparty engagement.
    Our own `emailed` record outranks the vendor's word, and the disagreement is recorded rather
    than smoothed over.
    """
    ev = rec.get("envelope") or {}
    raw = (ev.get("status") or "").strip().lower()
    if not ev or not raw:
        return "norec", "", ""
    if raw in _VENDOR_SIGNED:
        return "signed", raw, ""
    if raw in _VENDOR_DECLINED:
        return "declined", raw, ""
    emailed = bool(ev.get("emailed"))
    if emailed:
        # We DID send it, so the provider's word is not contradicted, only summarised. Every row
        # that shows a provider word still explains itself: a bare quoted status with no reason
        # reads as an unexplained disagreement.
        why = ""
        if raw in ("viewed", "opened"):
            why = ("we emailed it and the recipient opened it, which is real engagement. It still "
                   "reads Sent because only a signature binds")
        elif raw not in _VENDOR_OPEN:
            why = 'unrecognised provider status "%s"; treated as sent because we emailed it' % raw
        return "sent", raw, why
    why = ""
    if raw in ("viewed", "opened"):
        why = ('the provider stamps a signing link "%s" when the link is created, and we have '
               "no record of emailing it" % raw)
    elif raw not in _VENDOR_OPEN:
        why = 'unrecognised provider status "%s"; treated as built-but-unsent' % raw
    return "prepared", raw, why


# ── Rollup ───────────────────────────────────────────────────────────────────────────────────
def rollup(campaign, live_epoch=None):
    """Turn a campaign record into everything the console renders.

    Buckets are ASSIGNED here and the assignment is then ASSERTED against the universe, so a
    double-count or a dropped row cannot reach the screen quietly.
    """
    recips = campaign.get("recipients") or []
    target = campaign.get("target") or {}
    mode = target.get("mode") or "count"

    rows, assigned, seen = [], {k: [] for k in BUCKET_KEYS}, {}
    duplicates, unassigned = [], []

    for i, r in enumerate(recips):
        rid = str(r.get("id") or ("r%d" % (i + 1)))
        state, from_raw, why = translate(r)
        if state not in assigned:
            unassigned.append(rid)
            continue
        if rid in seen:
            duplicates.append(rid)                            # a real duplicate id in the roster
        seen[rid] = state
        assigned[state].append(rid)
        ev = r.get("envelope") or {}
        rows.append({
            "id": rid,
            "party": r.get("party") or "(unnamed counterparty)",
            "signs_through": r.get("signs_through") or "",
            "value": _num(r.get("value")),
            "bucket": state,
            "note": r.get("note") or "",
            "translated_from": from_raw if from_raw and from_raw != state else "",
            "translation_why": why,
            "events": _events(ev),
        })

    # THE INVARIANT, made visible. The reference build shipped a silent residual and counted a
    # party twice; Jason caught it by reading the table. Assert it, return it, and let the UI
    # render the result as a strip so the next occurrence announces itself.
    counts = {k: len(v) for k, v in assigned.items()}
    total_assigned = sum(counts.values())
    universe = len(recips)
    recon_ok = (total_assigned == universe) and not duplicates and not unassigned

    # Card counts and sums are derived from `rows`, which already carry the assignment, so the
    # cards can never drift from the table underneath them.
    bucket_out = [{
        "key": b["key"], "label": b["label"], "desc": b["desc"],
        "binds": bool(b.get("binds")), "residual": bool(b.get("residual")),
        "count": sum(1 for row in rows if row["bucket"] == b["key"]),
        "value": sum(row["value"] for row in rows if row["bucket"] == b["key"]),
    } for b in BUCKETS]

    signed_rows = [r for r in rows if r["bucket"] == "signed"]
    sent_rows = [r for r in rows if r["bucket"] == "sent"]

    if mode == "value":
        metric = sum(r["value"] for r in signed_rows)
        goal = _num(target.get("amount"))
        coverage_metric = sum(r["value"] for r in sent_rows) + metric
    else:
        metric = len(signed_rows)
        goal = int(_num(target.get("amount")) or universe)
        coverage_metric = len(sent_rows) + metric

    need = max(0, goal - metric)
    pct = (float(metric) / goal * 100.0) if goal else 0.0

    return {
        "ok": True,
        "campaign": {
            "id": campaign.get("id", ""),
            "name": campaign.get("name", ""),
            "document": campaign.get("document") or {},
            "target": {"mode": mode, "amount": goal, "currency": target.get("currency", "USD"),
                       "label": target.get("label", "")},
            "provider": (campaign.get("provider") or {}).get("kind", "fixtures"),
        },
        "live_as_of": _stamp_obj(live_epoch),
        "headline": {
            "mode": mode, "metric": metric, "target": goal, "need": need,
            "pct": round(min(100.0, pct), 1), "currency": target.get("currency", "USD"),
            "counts_only": "signatures",
        },
        # Named separately from the headline on purpose: coverage is outreach, not progress.
        "coverage": {"metric": coverage_metric, "count": len(sent_rows) + len(signed_rows),
                     "of": universe, "means": "outreach, not progress"},
        "buckets": bucket_out,
        "reconcile": {
            "ok": recon_ok,
            "buckets": len(BUCKETS),
            "assigned": total_assigned,
            "universe": universe,
            "duplicates": duplicates,
            "unassigned": unassigned,
        },
        "rows": rows,
        "staleness": campaign.get("staleness") or {"stale": False, "unknown": [], "message": ""},
        "roster_synced": campaign.get("roster_synced"),
    }


def _events(ev):
    """The progressive-disclosure detail row: what happened to this envelope, when, in ET."""
    out = []
    for key, label in (("created", "Document created"), ("emailed", "Emailed"),
                       ("viewed", "Opened by recipient"), ("signed", "Signed"),
                       ("declined", "Declined"), ("superseded", "Superseded")):
        v = ev.get(key)
        if not v:
            continue
        out.append({"label": label, "when": et_stamp(v) if isinstance(v, (int, float)) else str(v)})
    return out


def _num(v):
    try:
        return float(v) if isinstance(v, float) else int(v or 0)
    except (TypeError, ValueError):
        return 0


# ── Providers ────────────────────────────────────────────────────────────────────────────────
# A provider answers exactly one question: what does the source of truth say about this
# campaign's envelopes right now. It never renders, never assigns buckets, and never returns a
# credential. Adding a provider means adding one function here.

def provider_fetch(campaign):
    """Return {envelopes: {recipient_id: {...}}, unknown: [refs], source: str}."""
    kind = ((campaign.get("provider") or {}).get("kind") or "fixtures").lower()
    if kind == "fixtures":
        return _fetch_fixtures(campaign)
    if kind == "sealhawk":
        return _fetch_sealhawk(campaign)
    raise ValueError("Unknown campaign provider: %s" % kind)


def _fetch_fixtures(campaign):
    """Offline, synthetic, deterministic. The roster already carries its own envelopes, so a
    refresh simply re-reads them. This is what the app ships with and what the tests run on."""
    env = {}
    for r in campaign.get("recipients") or []:
        if r.get("envelope"):
            env[str(r.get("id"))] = r["envelope"]
    return {"envelopes": env, "unknown": [], "source": "fixtures (synthetic, offline)"}


def _fetch_sealhawk(campaign):
    """SealHawk is the sending engine; PageHawk owns the console (Jason 2026-07-24).

    Deliberately NOT implemented against a live endpoint in this build: the SealHawk contract
    library lane owns that repo, and the shared campaign record is going into the epiphani-account
    D1 rather than being invented here. What this function guarantees today is the boundary: the
    key is read server-side, it never enters a return value, and an unwired provider says so out
    loud instead of fabricating state.
    """
    key = _read_secret("SEALHAWK_API_KEY", "sealhawk-api-key")
    if not key:
        raise RuntimeError("SealHawk is not connected yet on this Mac. The console is running on "
                           "synthetic fixtures until the shared campaign record is wired.")
    raise RuntimeError("The SealHawk adapter is not wired yet. Refusing to invent campaign state.")


def _read_secret(env_name, keychain_item):
    """Server-side only. Env first, then the login Keychain. The value is never returned to the
    surface, never logged, and never written into the campaign file."""
    v = os.environ.get(env_name, "").strip()
    if v:
        return v
    try:
        import subprocess
        out = subprocess.run(["/usr/bin/security", "find-generic-password", "-s", keychain_item, "-w"],
                             capture_output=True, text=True, timeout=5)
        return out.stdout.strip() if out.returncode == 0 else ""
    except Exception:
        return ""


def refresh(campaign, fetch=None):
    """Pull live state, fold it into the roster, and make staleness visible.

    An envelope the provider knows about that the roster does not is the silent-staleness bug
    from the reference architecture. It is surfaced here rather than dropped.

    `fetch` is an injectable provider seam. It exists so the selftest can exercise THIS function
    against a source of truth that disagrees with the roster, rather than re-implementing the
    comparison in the test and proving nothing.
    """
    res = (fetch or provider_fetch)(campaign)
    envelopes = res.get("envelopes") or {}
    known = {str(r.get("id")) for r in campaign.get("recipients") or []}
    for r in campaign.get("recipients") or []:
        rid = str(r.get("id"))
        if rid in envelopes:
            r["envelope"] = envelopes[rid]
    unknown = [k for k in envelopes if k not in known] + list(res.get("unknown") or [])
    campaign["roster_synced"] = _stamp_obj()
    campaign["staleness"] = {
        "stale": bool(unknown),
        "unknown": unknown,
        "message": ("%d envelope%s at the source of truth %s no counterparty in this roster. "
                    "Add them, or this console is under-reporting."
                    % (len(unknown), "" if len(unknown) == 1 else "s",
                       "has" if len(unknown) == 1 else "have")) if unknown else "",
    }
    return campaign


# ── Output hygiene ───────────────────────────────────────────────────────────────────────────
def _scrub(obj):
    """Strip anything credential-shaped on the way out. Belt and braces: no code path should put
    a key here, and the selftest proves that removing this would be caught."""
    if isinstance(obj, dict):
        return {k: _scrub(v) for k, v in obj.items()
                if not any(h in str(k).lower() for h in _SECRET_HINTS)}
    if isinstance(obj, list):
        return [_scrub(v) for v in obj]
    return obj


def emit(obj):
    sys.stdout.write(json.dumps(_scrub(obj), ensure_ascii=False))
    sys.stdout.write("\n")


# ── Demo fixtures ────────────────────────────────────────────────────────────────────────────
# Obviously fake by construction: no real counterparty, person, email, or amount. Every entity
# is invented, and the emails all land on example.invalid, which cannot receive mail by RFC.
_DEMO = [
    ("Northwind Provisioning LLC",  "Dana Alvarez-Quist",  185000, "completed", 1, 1, "", ""),
    ("Tessellate Freight Co.",      "Marcus Oyelaran",     142500, "completed", 1, 1, "", ""),
    ("Bright Meridian Holdings",    "Priya Raghunathan",   120000, "completed", 1, 1,
     "Signed the superseded v2; countersigned v3 the same day.", ""),
    ("Calderon Analytics Group",    "Iva Calderon",         98000, "completed", 1, 1, "", ""),
    ("Halyard Systems Inc.",        "Tobias Fenn",          76500, "completed", 1, 1, "", ""),
    ("Quarry Lane Partners",        "Simone Adeyemi",       64000, "completed", 1, 1, "", ""),
    ("Vantage Orchard Trust",       "Ruth Kowalczyk",      210000, "viewed",    1, 1,
     "Third follow-up sent; counsel is reviewing.", ""),
    ("Ferro and Blaine Industrial", "Aurelio Ferro",       155000, "sent",      1, 1, "", ""),
    ("Sable Ridge Logistics",       "Nneka Obi-Sturgess",   88000, "sent",      1, 1,
     "Asked for a 30-day payment term; open question.", ""),
    ("Kestrel Point Media",         "Devon Marchetti",      54000, "delivered", 1, 1, "", ""),
    ("Ambleside Chemical Ltd.",     "Hana Vosburgh",        47500, "sent",      1, 1, "", ""),
    ("Torrance Grain Cooperative",  "Bill Ostrowski",       39000, "sent",      1, 1, "", ""),
    # The translation cases: created but never emailed. A vendor calls these "viewed"; we do not.
    ("Winnow and Marsh Advisory",   "Odalys Prieto",       132000, "viewed",    1, 0,
     "Entity name confirmed; ready to send.", ""),
    ("Pellard Manufacturing",       "Gustav Pellard",       97000, "viewed",    1, 0,
     "Waiting on their signatory designation.", ""),
    ("Alder Creek Utilities",       "Marisol Devereaux",    71000, "created",   1, 0, "", ""),
    ("Stonebridge Import Co.",      "Yusuf Barrantes",      52000, "created",   1, 0, "", ""),
    ("Lyle Harbour Services",       "Constance Ngo",        66000, "declined",  1, 1,
     "Declined: renewing with an incumbent through 2027.", ""),
    ("Redmond Fixture Works",       "Alastair Whitby",      31000, "declined",  1, 1,
     "Declined pending a legal-entity merger.", ""),
    ("Cobalt Vine Beverages",       "Renata Silvestri",    118000, "",          0, 0, "", ""),
    ("Highfield Rail Group",        "Emeka Danforth",       94000, "",          0, 0, "", ""),
    ("Ptarmigan Outdoor Co.",       "Lucia Brennhaug",      58000, "",          0, 0, "", ""),
    ("Mercer Row Textiles",         "Ingrid Sallowby",      45000, "",          0, 0, "", ""),
    ("Ossining Valve Corp.",        "Ramon Esquivel",       36000, "",          0, 0, "", ""),
    ("Delacroix Paper Mills",       "Beatriz Ndiaye",       28000, "",          0, 0, "", ""),
]


def demo_campaign():
    base = int(time.time()) - 9 * 86400
    c = new_campaign(name="Master Services Agreement: 2026 refresh",
                     target_mode="value", target_amount=1500000)
    c["document"] = {"path": "", "name": "MSA-2026-refresh.pdf"}
    c["provider"] = {"kind": "fixtures", "campaign_ref": "demo"}
    c["synthetic"] = True
    for i, (party, person, value, status, created, emailed, note, _x) in enumerate(_DEMO):
        rid = "r%02d" % (i + 1)
        rec = {
            "id": rid, "party": party, "signs_through": person,
            "email": re.sub(r"[^a-z]+", ".", person.lower()).strip(".") + "@example.invalid",
            "value": value, "note": note,
        }
        if created:
            ev = {"status": status, "created": base + i * 3600}
            if emailed:
                ev["emailed"] = base + 86400 + i * 3600
            if status == "completed":
                ev["signed"] = base + 5 * 86400 + i * 3600
            if status == "declined":
                ev["declined"] = base + 6 * 86400 + i * 3600
            rec["envelope"] = ev
        c["recipients"].append(rec)
    c = refresh(c)
    return c


# ── Selftest ─────────────────────────────────────────────────────────────────────────────────
def selftest():
    """Every check here has a NEGATIVE CONTROL: a deliberately broken input that MUST fail.
    A test that cannot fail proves nothing, and this rollup exists because a silent miscount
    shipped once already."""
    checks = []

    def ck(name, passed, detail=""):
        checks.append({"check": name, "pass": bool(passed), "detail": detail})

    c = demo_campaign()
    r = rollup(c)

    ck("demo reconciles", r["reconcile"]["ok"],
       "assigned %d of %d" % (r["reconcile"]["assigned"], r["reconcile"]["universe"]))
    ck("every row has exactly one bucket",
       len(r["rows"]) == r["reconcile"]["universe"] and
       all(row["bucket"] in BUCKET_KEYS for row in r["rows"]))
    ck("bucket counts sum to the universe",
       sum(b["count"] for b in r["buckets"]) == len(c["recipients"]))

    # NEGATIVE CONTROL 1: break the reconcile on purpose. If this passes, the invariant is
    # decorative and the whole strip is theatre.
    broken = copy.deepcopy(c)
    broken["recipients"].append(dict(broken["recipients"][0]))   # same id twice
    rb = rollup(broken)
    ck("NEGATIVE CONTROL: a duplicated recipient FAILS the reconcile",
       (not rb["reconcile"]["ok"]) and rb["reconcile"]["duplicates"],
       "duplicates=%s" % rb["reconcile"]["duplicates"])

    # NEGATIVE CONTROL 2: the vendor's word must not win. A created-but-unsent envelope the
    # provider calls "viewed" must read Prepared, never Sent, and never Signed.
    vend = {"id": "x1", "party": "Test Entity LLC", "signs_through": "Test Person",
            "value": 1, "envelope": {"status": "viewed", "created": 1}}
    state, from_raw, why = translate(vend)
    ck('vendor "viewed" with no send maps to Prepared', state == "prepared" and from_raw == "viewed",
       "why: " + why)
    vend2 = dict(vend); vend2["envelope"] = {"status": "viewed", "created": 1, "emailed": 2}
    ck('the SAME vendor word maps to Sent once we emailed it', translate(vend2)[0] == "sent")

    # NEGATIVE CONTROL 3: only signatures move the bar. Flip a signed row to sent and the
    # headline must DROP; if it does not, the bar is counting coverage.
    only_sig = copy.deepcopy(c)
    for rec in only_sig["recipients"]:
        if (rec.get("envelope") or {}).get("status") == "completed":
            rec["envelope"]["status"] = "sent"
            rec["envelope"].pop("signed", None)
            break
    ck("NEGATIVE CONTROL: moving one row out of Signed LOWERS the headline",
       rollup(only_sig)["headline"]["metric"] < r["headline"]["metric"])

    # NEGATIVE CONTROL 4: staleness must be able to fire, and it must fire through the REAL
    # refresh() rather than through logic re-typed in the test. Drop two counterparties from the
    # roster while the source of truth still knows about them.
    stale = copy.deepcopy(c)
    stale["recipients"] = stale["recipients"][:-2]
    def _disagreeing_source(_camp):
        return {"envelopes": {x["id"]: x.get("envelope") or {"status": "sent", "emailed": 1}
                              for x in c["recipients"]}, "unknown": [], "source": "test"}
    stale = refresh(stale, fetch=_disagreeing_source)
    ck("NEGATIVE CONTROL: an envelope missing from the roster fires the staleness warning",
       stale["staleness"]["stale"] and len(stale["staleness"]["unknown"]) == 2,
       "unknown=%s" % stale["staleness"]["unknown"])
    ck("the staleness warning reaches the rollup the surface renders",
       rollup(stale)["staleness"]["stale"] is True)
    ck("a healthy roster is NOT stale", r["staleness"]["stale"] is False)

    # Count-mode target (the second campaign shape).
    cnt = copy.deepcopy(c)
    cnt["target"] = {"mode": "count", "amount": 20, "currency": "USD", "label": "signatures"}
    rc = rollup(cnt)
    ck("count-mode target counts signatures, not dollars",
       rc["headline"]["metric"] == sum(1 for x in r["rows"] if x["bucket"] == "signed")
       and rc["headline"]["target"] == 20)

    # Secrets never leave.
    leak = {"provider": {"kind": "sealhawk", "api_key": "sk-should-never-appear"}}
    ck("NEGATIVE CONTROL: a credential in the record is scrubbed from output",
       "sk-should-never-appear" not in json.dumps(_scrub(leak)))
    ck("no credential-shaped key in a real rollup",
       not any(h in json.dumps(r).lower() for h in ("api_key", "\"secret\"", "bearer ")))

    # Fixtures are obviously fake.
    ck("fixtures are obviously synthetic",
       all("@example.invalid" in (x.get("email") or "") for x in c["recipients"]))

    # 12-hour Eastern, never 24-hour.
    s = et_stamp(1784000000)
    ck("stamp is 12-hour Eastern", s.endswith("ET") and ("am" in s or "pm" in s), s)

    passed = sum(1 for x in checks if x["pass"])
    return {"ok": passed == len(checks), "passed": passed, "total": len(checks), "checks": checks}


# ── CLI ──────────────────────────────────────────────────────────────────────────────────────
def main(argv=None):
    ap = argparse.ArgumentParser(description="PageHawk Signature Campaigns engine")
    sub = ap.add_subparsers(dest="cmd")

    sub.add_parser("selftest")
    p = sub.add_parser("demo"); p.add_argument("--out", default="")
    p = sub.add_parser("new")
    p.add_argument("--out", required=True); p.add_argument("--doc", default="")
    p.add_argument("--name", default="Untitled campaign")
    p.add_argument("--target-mode", default="value", choices=["value", "count"])
    p.add_argument("--target-amount", type=float, default=0)
    p = sub.add_parser("status"); p.add_argument("--file", required=True)
    p = sub.add_parser("refresh"); p.add_argument("--file", required=True)

    a = ap.parse_args(argv)
    try:
        if a.cmd == "selftest":
            r = selftest(); emit(r); return 0 if r["ok"] else 1
        if a.cmd == "demo":
            c = demo_campaign()
            if a.out:
                save_campaign(a.out, c)
            emit({"ok": True, "file": a.out, "rollup": rollup(c)}); return 0
        if a.cmd == "new":
            c = new_campaign(a.name, a.doc, a.target_mode, a.target_amount)
            save_campaign(a.out, c)
            emit({"ok": True, "file": a.out, "rollup": rollup(c)}); return 0
        if a.cmd == "status":
            emit({"ok": True, "file": a.file, "rollup": rollup(load_campaign(a.file))}); return 0
        if a.cmd == "refresh":
            c = refresh(load_campaign(a.file))
            save_campaign(a.file, c)
            emit({"ok": True, "file": a.file, "rollup": rollup(c)}); return 0
        ap.print_help(sys.stderr); return 2
    except Exception as e:
        emit({"ok": False, "error": str(e)})
        return 1


if __name__ == "__main__":
    sys.exit(main())
