"""The push bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: push — plan 11 task E: real Web Push over VAPID. VAPID
# persistence, the service worker, subscription CRUD + validation,
# notify-me prefs, the block/lifecycle dispatch matrix (global vs
# opted-in), 410 pruning, retry-next-tick delivery 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 push_subscriptions where endpoint = ?",
                (endpoint,)).fetchone()[0]
        finally:
            conn.close()

    print("== task E: 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("== task E: VAPID keypair generation + persistence ==")
    config = json.load(open(os.path.join(WORK, "web-config.json")))
    vapid = config.get("vapid") or {}
    check(bool(vapid.get("public_key")) and bool(vapid.get("private_key")),
          "the VAPID keypair persists into web-config.json's vapid section")
    vapid_private_key = ec.derive_private_key(
        int.from_bytes(b64url_dec(vapid["private_key"]), "big"), ec.SECP256R1())
    derived_public = vapid_private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    check(b64url_dec(vapid["public_key"]) == derived_public,
          "the persisted public key matches the private scalar (a valid P-256 pair)")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_vapid_key") == vapid["public_key"],
          "the dashboard exposes the same 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("== task E: 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 push_subscriptions"
                        " where endpoint = ?", (endpoint_main,)).fetchall()
    admin_id = conn.execute("select id from spry_users where username = 'admin'"
                            ).fetchone()[0]
    conn.close()
    check(len(rows) == 1 and rows[0][0] == admin_id,
          "a valid subscription stores against the signed-in user")
    # Re-subscribing the same endpoint replaces (never duplicates).
    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 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("== task E: 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("== task E: 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 push_subscriptions (user_id, endpoint, p256dh, auth,"
                 " created_at) 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")
    task1_session = fake.session_id
    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 pruned") is not None,
          "a 410 Gone endpoint is pruned from push_subscriptions")

    # Block events go to EVERY push-subscribed user globally. The
    # poll-era /update raise is gone: 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)
    ciphertext_len = len(block_hit["body"]) - (21 + 65)
    envelope = block_hit["body"]
    record_size = int.from_bytes(envelope[16:20], "big")
    # rs = plaintext + delimiter (1) + tag (16) == ciphertext-with-tag len.
    check(record_size == ciphertext_len,
          "the RFC 8291 envelope record size covers payload + delimiter + tag")
    claims = verify_vapid(block_hit, vapid["public_key"], 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") == "high"
          and block_hit["headers"].get("authorization", "")
          .startswith("vapid t="),
          "the push POST carries 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"),
          "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("== task E: retry next tick (best-effort delivery) ==")
    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 failed send retries on the next dispatcher tick and lands")

    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).
    push_goal = None
    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("== task E: 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 ==")
    # The retry fixture left admin subscribed to /flaky; drop it so
    # this section owns exactly the subscriptions it asserts on.
    act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_flaky})
    # 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: failure surfaces the real reason ==")
    push2.status_for["/b"] = 401
    b_before = len(push2.hits("/b"))
    result = act(auth, dash["push_test"], {})
    message = action_notify(result) or ""
    check("reached 1 browser" in message and "HTTP 401" in message,
          "a refused endpoint reports reached/failed + the status (%s)" % message)
    check(len(push2.hits("/b")) > b_before, "the refused endpoint was still POSTed")

    print("== truthful delivery: 4xx is refused, retried, never delivered ==")
    # A refused (401) endpoint must NOT count as delivered; the queue
    # retries on later ticks and — after MAX_ATTEMPTS — gives up WITH a
    # warning instead of silently claiming success. The subscription
    # row survives (a VAPID key rotation may heal it).
    row_before = subscription_count(endpoint_b)
    b_before = len(push2.hits("/b"))
    raise_global_block("Refused delivery question?")
    check(wait_for(lambda: len(push2.hits("/b")) > b_before, timeout=15,
                   what="first attempt at the refused endpoint") is not None,
          "the event still reaches the refused endpoint")
    def give_up_in_log():
        try:
            with open(os.path.join(WORK, "app.log")) as handle:
                return "gave up after %d attempts" % 5 in handle.read()
        except IOError:
            return False
    check(wait_for(give_up_in_log, timeout=30, what="give-up warning in app.log")
          is not None,
          "after MAX_ATTEMPTS the give-up warning lands in the log (never silent)")
    check(subscription_count(endpoint_b) == row_before,
          "a refused endpoint is NOT pruned (it may recover)")
    check(len(fingerprints_of(auth)) == 2,
          "the bell state still lists both endpoints after give-up")

    print("== truthful delivery: 404/410 pruning regression ==")
    push2.status_for["/b"] = 410
    b_before = len(push2.hits("/b"))
    raise_global_block("Prune after refusal question?")
    check(wait_for(lambda: len(push2.hits("/b")) > b_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_b) == 0, timeout=15,
                   what="410 endpoint pruned after refusal episode") is not None,
          "a 410 endpoint is pruned even after an earlier refusal 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")