clanker-saddle/test/bundles/push.py

"""The push bundle: Web Push over `Spry.Push` semantics (see
test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: push — real Web Push over VAPID, Spry.Push underneath. The store
# (spry_push_subscriptions), the RFC 8291/8292 crypto and the delivery POST
# are Spry's (inline 2-attempt/250ms retry, 401/403/404/410 sweeping); the
# payloads, audiences, notify-me prefs and the bell UX stay the app's.
# Subscription CRUD + validation, the block/lifecycle dispatch matrix
# (global vs opted-in), sweeping, inline retry, the send_test probe and
# protocol-level payload verification (RFC 8291 decryption + ES256 JWT)
# against the fake push service.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    def subscription_count(endpoint):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from spry_push_subscriptions where endpoint = ?",
                (endpoint,)).fetchone()[0]
        finally:
            conn.close()

    print("== service worker asset ==")
    status, sw = http("GET", "/sw.js", timeout=15)
    check(status == 200 and "addEventListener('push'" in sw
          and "notificationclick" in sw and "clients.openWindow" in sw,
          "the service worker serves at the origin root with push + click routing")

    print("== VAPID keys from the \"spry\" config section ==")
    # Spry owns the keys: `spry keys` writes the "spry" section the app
    # reads — the app no longer generates or persists key material.
    config = json.load(open(os.path.join(WORK, "web-config.json")))
    spry = config.get("spry") or {}
    check(bool(spry.get("vapid_public_key")) and bool(spry.get("vapid_private_key")),
          "the VAPID keypair lives in web-config.json's spry section")
    check("vapid" not in config,
          "the dead boot-time vapid section is gone")
    vapid_private_key = ec.derive_private_key(
        int.from_bytes(b64url_dec(spry["vapid_private_key"]), "big"), ec.SECP256R1())
    derived_public = vapid_private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    check(b64url_dec(spry["vapid_public_key"]) == derived_public,
          "the configured public key matches the private scalar (a valid P-256 pair)")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_vapid_key") == spry["vapid_public_key"],
          "the dashboard exposes the spry section's VAPID key to subscribers")
    check(dash.get("push_subscribed") is False and bool(dash.get("push_subscribe")),
          "the bell starts unsubscribed and carries the subscribe action")

    print("== subscription CRUD + auth validation ==")
    browser = ec.generate_private_key(ec.SECP256R1())
    browser_public = browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    browser_auth = os.urandom(16)
    subscribe_fields = {"p256dh": b64url(browser_public), "auth": b64url(browser_auth)}
    valid_endpoint = "http://127.0.0.1:1/x"
    check("absolute http(s)" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": "ftp://example.com/push", **subscribe_fields})) or ""),
        "a non-http endpoint is rejected")
    check("p256dh" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": b64url(os.urandom(64)),
         "auth": subscribe_fields["auth"]})) or ""),
        "a malformed p256dh point is rejected")
    check("auth" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(os.urandom(15))})) or ""),
        "a malformed auth secret is rejected")

    push = FakePushService()
    push.serve()
    endpoint_main = push.url("/main")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, **subscribe_fields})
    conn = db()
    rows = conn.execute("select user_id, p256dh, auth from spry_push_subscriptions"
                        " where endpoint = ?", (endpoint_main,)).fetchall()
    admin_id = conn.execute("select id from spry_users where username = 'admin'"
                            ).fetchone()[0]
    stale = conn.execute("select count(*) from sqlite_master where name ="
                         " 'push_subscriptions'").fetchone()[0]
    conn.close()
    check(len(rows) == 1 and rows[0][0] == admin_id,
          "a valid subscription stores against the signed-in user")
    check(stale == 0, "the old push_subscriptions table is gone (fresh database)")
    # Re-subscribing the same endpoint replaces (never duplicates) — the
    # endpoint is Spry's subscription identity.
    rotated_auth = os.urandom(16)
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(rotated_auth)})
    conn = db()
    rows = conn.execute("select auth from spry_push_subscriptions where endpoint = ?",
                        (endpoint_main,)).fetchall()
    conn.close()
    check(len(rows) == 1 and rows[0][0] == b64url(rotated_auth),
          "re-subscribing an endpoint replaces its keys in place")

    print("== notify-me pref persistence ==")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push pref task", "priority": "5"})
    pref_task = task_id_of("Push pref task")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?"
                       " and user_id = ?", (pref_task, admin_id)).fetchone()[0] == 1,
          "Notify me persists a task_notify_prefs row for the user")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is True,
          "the task slot reflects notify_me after the toggle")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?",
                       (pref_task,)).fetchone()[0] == 0,
          "Don't notify me removes the pref row")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is False,
          "the task slot reflects the opt-out")

    print("== dispatch matrix — global blocks vs opted-in lifecycle ==")
    # Two "browsers" for admin (/main, /gone) and one for another user
    # (id 999, rows seeded directly — push targeting is per user).
    endpoint_gone = push.url("/gone")
    push.status_for["/gone"] = 410
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_gone, **subscribe_fields})
    other_browser = ec.generate_private_key(ec.SECP256R1())
    other_public = other_browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    other_auth = os.urandom(16)
    endpoint_other = push.url("/other")
    conn = db()
    conn.execute("insert into spry_push_subscriptions (user_id, endpoint, p256dh,"
                 " auth, created) values (999, ?, ?, ?,"
                 " strftime('%s','now'))",
                 (endpoint_other, b64url(other_public), b64url(other_auth)))
    conn.commit()
    conn.close()

    # Task 1: admin opted in via the UI toggle; the other user is not.
    # Baselines are captured BEFORE the enqueue: the first-start push
    # fires at the allocation, the instant the dispatch lands.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push lifecycle task", "priority": "5"})
    task1 = task_id_of("Push lifecycle task")
    act(auth, task_state(auth, task1)["toggle_notify"], {})
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    gone_before = len(push.hits("/gone"))
    enqueue_task(auth, task1)
    check(wait_for(lambda: fake.running_task == task1 or False, timeout=30,
                   what="task 1 assignment") is not None,
          "task 1 dispatches to the fake daemon")
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="first-start push to the opted-in admin") is not None,
          "first start notifies the opted-in user's subscription")
    check(wait_for(lambda: len(push.hits("/gone")) > gone_before, timeout=15,
                   what="first-start push to the gone endpoint") is not None,
          "first start reaches every subscription of the opted-in user")
    check(quiet_window(quiet_seconds(),
                       lambda: len(push.hits("/other")) == other_before,
                       "no first-start push to the not-opted-in user"),
          "first start never reaches users without the notify pref")
    check(wait_task_done(auth, task1, timeout=60) is not None,
          "task 1 completes")
    _, done_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                      browser_public, rotated_auth)
    check(done_payload["title"] == "Push lifecycle task — completed"
          and done_payload["url"] == "/tasks/%d" % task1,
          "completion notifies the opted-in user (decrypted payload)")
    check(len(push.hits("/other")) == other_before,
          "completion stays with the opted-in user only")
    check(wait_for(lambda: subscription_count(endpoint_gone) == 0, timeout=15,
                   what="410 endpoint swept") is not None,
          "a 410 Gone endpoint is swept from spry_push_subscriptions")

    # Block events go to EVERY push-subscribed user globally. A live
    # (non-terminal) host task with an open session on the fake carries
    # the block through the results push — task1 is done and terminal
    # pushes are consumed silently, so the host is its own fixture row.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Push block host', 'working', 5,"
                 " strftime('%s','now'))")
    push_host = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (push_host, fake_id))
    push_host_session = conn.execute(
        "select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()

    def raise_global_block(question):
        return saddle_api("POST", "/api/saddle/results", fake_token, {
            "session_id": push_host_session, "state": "blocked",
            "block": {"question": question, "type": "question"}})

    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    raise_global_block("Global question for everyone?")
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="global block push") is not None,
          "a raised block notifies every push-subscribed user globally")
    block_hit = push.hits("/other")[-1]
    _, payload = decrypt_webpush(block_hit, other_browser, other_public, other_auth)
    check(payload["title"] == "Push block host — question raised"
          and payload["body"] == "Global question for everyone?"
          and payload["url"] == "/tasks/%d" % push_host,
          "the block payload carries task title + kind, question and click URL (%s)"
          % payload)
    envelope = block_hit["body"]
    record_size = int.from_bytes(envelope[16:20], "big")
    # Spry/Locksmith's aes128gcm envelope declares the fixed 4096 max
    # record size (RFC 8291 §4.2) — the record itself is payload +
    # delimiter + 16-byte tag, as long as the ciphertext.
    ciphertext_len = len(envelope) - (21 + 65)
    check(record_size == 4096 and ciphertext_len > 16,
          "the RFC 8291 envelope declares Spry's fixed 4096 record size"
          " over a well-formed record")
    claims = verify_vapid(block_hit, spry["vapid_public_key"], spry["vapid_private_key"])
    check(claims.get("aud") == "http://127.0.0.1:%d" % push.port
          and claims.get("sub", "").startswith("mailto:")
          and claims.get("exp", 0) > time.time(),
          "the VAPID JWT verifies: aud = the endpoint origin, mailto sub, future exp")
    check(block_hit["headers"].get("ttl") == "60"
          and block_hit["headers"].get("content-type") == "application/octet-stream",
          "the push POST carries TTL and the encrypted content type")
    # RFC 8291 section 4: the aes128gcm body MUST announce its encoding
    # — push services answer HTTP 400 without it (the test-notification
    # 400 regression).
    check(block_hit["headers"].get("content-encoding") == "aes128gcm",
          "the push POST carries Content-Encoding: aes128gcm (RFC 8291)")
    check(block_hit["headers"].get("urgency") == "normal"
          and block_hit["headers"].get("authorization", "")
          .startswith("vapid t="),
          "the push POST carries Spry's Urgency and the VAPID Authorization")

    # Task 2: only the OTHER user opted in — start + failure reach them
    # alone. Baseline again before the enqueue (first start fires at
    # the allocation).
    fake.fail_next = True
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push failure task", "priority": "5"})
    task2 = task_id_of("Push failure task")
    conn = db()
    conn.execute("insert into task_notify_prefs (task_id, user_id) values (?, 999)",
                 (task2,))
    conn.commit()
    conn.close()
    main_before2 = len(push.hits("/main"))
    other_before2 = len(push.hits("/other"))
    enqueue_task(auth, task2)
    check(wait_for(lambda: fake.running_task == task2 or False, timeout=30,
                   what="task 2 assignment") is not None,
          "task 2 dispatches (scripted to fail)")
    check(wait_for(lambda: task_state(auth, task2)["status"] == "failed"
                   or False, timeout=30, what="task 2 failed") is not None,
          "task 2 fails")
    check(wait_for(lambda: len(push.hits("/other")) >= other_before2 + 2,
                   timeout=15, what="opted-in user's started+failed pushes") is not None,
          "the opted-in user gets first start AND failure for the task")
    started_hit = [hit for hit in push.hits("/other")[other_before2:]][0]
    _, started_payload = decrypt_webpush(started_hit, other_browser,
                                         other_public, other_auth)
    check(started_payload["title"] == "Push failure task — started",
          "the first-start payload names task + kind (%s)" % started_payload)
    _, failure_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                         other_public, other_auth)
    check(failure_payload["title"] == "Push failure task — failed"
          and failure_payload["url"] == "/tasks/%d" % task2,
          "the failure payload names the task and links its page")
    check(len(push.hits("/main")) == main_before2,
          "an agent failure never reaches users without the pref")

    print("== inline retry (Spry's bounded delivery) ==")
    # Spry's sender retries INLINE: two attempts, 250ms apart, then
    # classifies. One transient 503 therefore lands on the second
    # attempt moments later — no dispatcher tick, no queue row.
    endpoint_flaky = push.url("/flaky")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_flaky, **subscribe_fields})
    push.fail_next["/flaky"] = 1
    flaky_before = len(push.hits("/flaky"))
    raise_global_block("Flaky delivery question?")
    check(wait_for(lambda: len(push.hits("/flaky")) >= flaky_before + 2,
                   timeout=20, what="flaky endpoint retried") is not None,
          "a 503 send retries inline (Spry: 2 attempts) and lands")
    flaky_hits = push.hits("/flaky")
    if len(flaky_hits) >= flaky_before + 2:
        first, second = flaky_hits[-2], flaky_hits[-1]
        gap = second["at"] - first["at"]
        check(0.05 <= gap <= 5.0,
              "the retry follows the first attempt closely (Spry: 250ms) (%.2fs)"
              % gap)
        _, flaky_payload = decrypt_webpush(second, browser, browser_public,
                                           browser_auth)
        check(flaky_payload["body"] == "Flaky delivery question?",
              "the retried delivery is a fresh, complete encryption (%s)"
              % flaky_payload)
    else:
        check(False, "the inline retry fixture did not land")

    print("== a hard-failing endpoint is RETRYABLE, never swept ==")
    # RFC 8030 §7 semantics (Spry): a persistent 503 exhausts the inline
    # attempts and classifies RETRYABLE — the notification is NOT
    # delivered, but the subscription row SURVIVES (only 401/403/404/410
    # sweep). No give-up log line and no queue table anymore.
    endpoint_dead = push.url("/dead")
    push.status_for["/dead"] = 503
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_dead, **subscribe_fields})
    dead_before = len(push.hits("/dead"))
    raise_global_block("Dead endpoint question?")
    check(wait_for(lambda: len(push.hits("/dead")) >= dead_before + 2,
                   timeout=20, what="dead endpoint attempts") is not None,
          "a 503 endpoint is POSTed twice (both inline attempts)")
    check(len(push.hits("/dead")) <= dead_before + 2,
          "exactly two attempts — Spry's ATTEMPTS bound, no dispatcher-tick retries")
    check(subscription_count(endpoint_dead) == 1,
          "a retryable endpoint is NOT swept (the row survives for the next send)")

    print("== plan 12: goal notifications — blocks/success/failure always ==")
    # Goal blocks, completion approval and failure push to EVERY
    # subscribed user (the mirror on the goal-run task stays quiet).
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Push goal", "brief": "Notifies everyone.",
        "branch_name": "goal/push"})
    conn = db()
    row = conn.execute("select id from goals where title = 'Push goal'").fetchone()
    conn.close()
    push_goal = int(row[0])
    fake.goal_script = [("plan", "# Push plan\n\n1. Ring the bells")]
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    act(auth, goal_state(push_goal)["start"], {})
    check(wait_for(lambda: goal_state(push_goal)["status"] == "plan_review" or False,
                   timeout=60, what="push goal enters plan review") is not None,
          "the push goal reaches plan review")
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="goal plan-review push") is not None,
          "a goal plan review notifies every subscribed user globally")
    _, plan_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                      other_public, other_auth)
    check("Push goal" in plan_payload["title"] and "plan" in plan_payload["title"]
          and plan_payload["url"] == "/goals/%d" % push_goal,
          "the goal block payload carries the goal title + click URL (%s)"
          % plan_payload)
    main_before = len(push.hits("/main"))
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(push_goal)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    check(wait_for(lambda: goal_state(push_goal)["status"] == "active" or False,
                   timeout=30, what="push goal active") is not None,
          "the push goal activates")
    check(quiet_window(3, lambda: len(push.hits("/main")) == main_before,
                       "no push for an internal resume"),
          "the goal agent's own park/resume does not notify (internal)")

    # `saddle failure` in goal context fails the goal — global push.
    fake.goal_script = [("fail", "the harness collapsed")]
    other_before2 = len(push.hits("/other"))
    main_before2 = len(push.hits("/main"))
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, completed_at, created_at) values (1, ?, 'task',"
                 " 'Warm', 'done', 5,"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (push_goal,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: goal_state(push_goal)["status"] == "failed" or False,
                   timeout=60, what="goal failure lands") is not None,
          "the goal agent's failure fails the goal")
    check(wait_for(lambda: len(push.hits("/other")) > other_before2
                   and len(push.hits("/main")) > main_before2, timeout=15,
                   what="goal failure push") is not None,
          "a goal failure notifies every subscribed user globally")
    _, fail_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                      other_public, other_auth)
    check("Push goal" in fail_payload["title"] and "failed" in fail_payload["title"]
          and fail_payload["url"] == "/goals/%d" % push_goal,
          "the goal failure payload carries the goal title + click URL (%s)"
          % fail_payload)

    # Completion approval pushes "complete" globally (success always).
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Push done goal", "brief": "Completes.",
        "branch_name": "goal/push-done"})
    conn = db()
    row = conn.execute("select id from goals where title = 'Push done goal'").fetchone()
    conn.close()
    done_goal = int(row[0])
    fake.goal_script = [("plan", "# Done plan")]
    act(auth, goal_state(done_goal)["start"], {})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "plan_review" or False,
                   timeout=60, what="done goal plan review") is not None,
          "the done goal reaches plan review")
    fake.goal_script = [("done", "# Completion\n\nShipped.")]
    act(auth, goal_state(done_goal)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "completion_review"
                   or False, timeout=60, what="done goal completion review") is not None,
          "the approved done goal reaches completion review")
    main_before = len(push.hits("/main"))
    act(auth, goal_state(done_goal)["completion_pending_block"]["answer_action"],
        {"answer": "Review approved. Continue with the task."})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "complete" or False,
                   timeout=30, what="done goal completes") is not None,
          "the done goal completes")
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="goal success push") is not None,
          "a goal completion notifies every subscribed user globally")
    _, done_goal_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                           browser_public, rotated_auth)
    check("Push done goal" in done_goal_payload["title"]
          and "complete" in done_goal_payload["title"]
          and done_goal_payload["url"] == "/goals/%d" % done_goal,
          "the goal success payload carries the goal title + click URL (%s)"
          % done_goal_payload)

    print("== unsubscribe removes the subscription ==")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_subscribed") is True,
          "the bell shows subscribed while endpoints exist")
    act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_main})
    check(subscription_count(endpoint_main) == 0,
          "unsubscribing removes the endpoint's row")

    # ------------------------------------------------------------------
    # Per-browser bell + truthful delivery + test notification
    # (the notifications-QoL task): the user-global flag lied to a
    # second browser, refused deliveries were silently "delivered",
    # and there was no end-to-end probe.
    # ------------------------------------------------------------------

    def fingerprints_of(auth_handle):
        state = slot_public(auth_handle, "/", "dashboard")
        return state.get("push_endpoints") or []

    def fingerprint_of(endpoint):
        return hashlib.sha256(endpoint.encode()).hexdigest()[:16]

    print("== per-browser bell: fingerprint state shape ==")
    # This section owns exactly the subscriptions it asserts on: drop
    # the dispatch fixtures left on admin's account first.
    for fixture_endpoint in (endpoint_flaky, endpoint_dead):
        act(auth, dash["push_unsubscribe"], {"endpoint": fixture_endpoint})
    # Fresh admin subscriptions on TWO fake endpoints = two browsers.
    push2 = FakePushService()
    push2.serve()
    endpoint_a = push2.url("/a")
    endpoint_b = push2.url("/b")
    browser_a = ec.generate_private_key(ec.SECP256R1())
    browser_a_public = browser_a.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    browser_a_auth = os.urandom(16)
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_a, "p256dh": b64url(browser_a_public),
         "auth": b64url(browser_a_auth)})
    browser_b = ec.generate_private_key(ec.SECP256R1())
    browser_b_public = browser_b.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    browser_b_auth = os.urandom(16)
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_b, "p256dh": b64url(browser_b_public),
         "auth": b64url(browser_b_auth)})
    dash = slot_public(auth, "/", "dashboard")
    prints = dash.get("push_endpoints") or []
    check(isinstance(prints, list) and len(prints) == 2,
          "two subscriptions => both listed for the user (%s)" % prints)
    check(fingerprint_of(endpoint_a) in prints
          and fingerprint_of(endpoint_b) in prints,
          "each entry is the sha256/16-hex fingerprint of its endpoint")
    check(endpoint_a not in json.dumps(dash),
          "the raw endpoint URL never leaves the server")
    check(bool(dash.get("push_test")) and bool(dash.get("push_subscribe")),
          "the dashboard carries the test + subscribe actions")

    print("== per-browser bell: second browser still offered Enable ==")
    # The old user-global flag showed "Turn off" in EVERY browser once
    # any subscription existed; the per-browser client check keys on
    # the fingerprint list instead. A third endpoint the server does
    # NOT know = an unsubscribed browser of the same user: the state
    # that drives the client verdict must not contain its fingerprint.
    endpoint_c = push2.url("/c")
    check(fingerprint_of(endpoint_c) not in fingerprints_of(auth),
          "an unknown browser's fingerprint is absent — the bell shows Enable there")
    # ...while the subscribed browsers still see themselves present.
    check(fingerprint_of(endpoint_a) in fingerprints_of(auth)
          and fingerprint_of(endpoint_b) in fingerprints_of(auth),
          "the two subscribed browsers each see themselves in the list")

    print("== test notification: real end-to-end round-trip ==")
    a_before = len(push2.hits("/a"))
    b_before = len(push2.hits("/b"))
    result = act(auth, dash["push_test"], {})
    message = action_notify(result) or ""
    check(wait_for(lambda: len(push2.hits("/a")) > a_before
                   and len(push2.hits("/b")) > b_before, timeout=15,
                   what="test notification POSTs") is not None,
          "push_test fires a REAL push to both of the user's endpoints")
    test_hit = push2.hits("/a")[-1]
    _, test_payload = decrypt_webpush(test_hit, browser_a,
                                      browser_a_public, browser_a_auth)
    check(test_payload["title"] == "Test notification"
          and test_payload["url"] == "/",
          "the test payload decrypts with title + url (goes to the dashboard)"
          " (%s)" % test_payload)
    check(test_hit["headers"].get("content-encoding") == "aes128gcm"
          and test_hit["headers"].get("ttl") == "60",
          "the test-notification POST carries Content-Encoding: aes128gcm"
          " + TTL (the HTTP 400 fix)")
    check("2 browser" in message,
          "the action response reports both deliveries (%s)" % message)
    set_types = [d.get("type") for d in result if d.get("type") == "set"]
    check(len(set_types) >= 1, "the action also refreshes the dashboard slot")

    print("== test notification: a dead endpoint prunes + reports ==")
    # Spry sweeps 401/403 exactly like 404/410 (RFC 8030: invalid
    # credentials for THAT subscription can never heal) — the probe's
    # DeliveryOutcome[] maps that to `pruned` (NOT failed: nothing
    # retryable happened), the row goes away and the toast still
    # separates the two outcomes.
    push2.status_for["/b"] = 401
    b_before = len(push2.hits("/b"))
    result = act(auth, dash["push_test"], {})
    message = action_notify(result) or ""
    check(len(push2.hits("/b")) > b_before, "the refused endpoint was still POSTed")
    check("1 browser" in message and "0 failed" in message,
          "a refused endpoint reports the delivery + the sweep as not-failed (%s)"
          % message)
    check(subscription_count(endpoint_b) == 0,
          "a 401 endpoint's row is swept (Spry: UNAUTHORIZED sweeps like GONE)")
    check(wait_for(lambda: len(fingerprints_of(auth)) == 1, timeout=15,
                   what="bell state drops the swept endpoint") is not None,
          "the refreshed bell state drops the swept endpoint")

    print("== truthful delivery: retryable failures never count as delivered ==")
    # A persistently 503ing endpoint stays RETRYABLE in Spry's model:
    # both inline attempts fail, the row SURVIVES and the send_test
    # probe surfaces the honest `failed` count in its toast.
    endpoint_bad = push2.url("/bad")
    push2.status_for["/bad"] = 503
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_bad, "p256dh": b64url(browser_b_public),
         "auth": b64url(browser_b_auth)})
    bad_before = len(push2.hits("/bad"))
    result = act(auth, dash["push_test"], {})
    message = action_notify(result) or ""
    check(len(push2.hits("/bad")) == bad_before + 2,
          "the retryable endpoint got exactly two inline attempts (%d)"
          % (len(push2.hits("/bad")) - bad_before))
    check("1 browser" in message and "failed" in message,
          "the probe reports the delivery that landed + the failure (%s)" % message)
    check(subscription_count(endpoint_bad) == 1,
          "a retryable endpoint is NOT swept — it may recover")

    print("== truthful delivery: 410 pruning regression ==")
    push2.status_for["/bad"] = 410
    bad_before = len(push2.hits("/bad"))
    raise_global_block("Prune after refusal question?")
    check(wait_for(lambda: len(push2.hits("/bad")) > bad_before, timeout=15,
                   what="POST to the now-gone endpoint") is not None,
          "the event reaches the endpoint before its 410")
    check(wait_for(lambda: subscription_count(endpoint_bad) == 0, timeout=15,
                   what="410 endpoint swept after retryable episode") is not None,
          "a 410 endpoint is swept even after an earlier retryable episode")
    check(wait_for(lambda: len(fingerprints_of(auth)) == 1, timeout=15,
                   what="bell state drops the pruned endpoint") is not None,
          "the refreshed bell state drops the pruned endpoint")

    print("== per-browser bell: markup + client verdict wiring ==")
    status, home_page = http("GET", "/", timeout=15)
    check(status == 200 and "data-push-test" in home_page
          and "data-bell-on" in home_page and "data-bell-off" in home_page,
          "the bell card ships the per-browser buttons + test button")
    # Goal 41 task 3: the fingerprint engine moved from home.html's
    # inline script into app.js (window.initPushBell) so the dashboard's
    # rail strip and /profile's settings card share it; the home page
    # wires it and still reads the dashboard slot's endpoint list.
    app_js = ""
    try:
        with open(os.path.join(REPO, "src", "pages", "app.js"), "r",
                  errors="replace") as handle:
            app_js = handle.read()
    except OSError:
        pass
    check("initPushBell" in home_page and "stateType" in home_page,
          "the home page wires the shared bell engine on the dashboard slot")
    check("crypto.subtle.digest" in app_js
          and "endpoint_fingerprint" in app_js,
          "the shared engine fingerprints this browser's own endpoint")

    print("== cleanup: unsubscribe the probe fixtures ==")
    act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_a})
    check(subscription_count(endpoint_a) == 0,
          "unsubscribing removes the probe's endpoint row")