import asyncio
import sqlite3
import threading
import time
from urllib.parse import parse_qs, urlsplit

import httpx
import pytest
from fastapi import Request

from src import main
from src.session_store import SessionStoreError
from src.views import FRONTEND_BUILD


@pytest.fixture
def access_control(monkeypatch, tmp_path):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
    monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
    monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security-events.sqlite3"))
    monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
    monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
    monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")


async def fresh_grant(client, action: str, target: str) -> str:
    response = await client.post(
        "/api/v1/fresh-authorization",
        json={
            "access_token": "correct horse battery staple",
            "action": action,
            "target": target,
        },
        headers={
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        },
    )
    assert response.status_code == 201
    return response.json()["grant"]


@pytest.mark.anyio
async def test_passkey_enrollment_options_require_fresh_authorization_and_are_one_time(
    access_control,
):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Timmy's phone",
            },
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        missing = await client.post("/api/v1/passkeys/registration/options", headers=headers)
        grant = await fresh_grant(client, "enroll_passkey", "current_device")
        created = await client.post(
            "/api/v1/passkeys/registration/options",
            headers={**headers, "X-Step-Up-Grant": grant},
        )
        replayed = await client.post(
            "/api/v1/passkeys/registration/options",
            headers={**headers, "X-Step-Up-Grant": grant},
        )

    assert missing.status_code == 428
    assert missing.json()["detail"]["action"] == "enroll_passkey"
    assert created.status_code == 201
    assert created.headers["cache-control"] == "no-store"
    options = created.json()
    assert options["rp"] == {"id": "test", "name": "Stackchain Dashboard"}
    assert options["user"]["name"] == "stackchain-operator"
    assert options["authenticatorSelection"]["userVerification"] == "required"
    assert options["challenge"]
    assert replayed.status_code == 428


@pytest.mark.anyio
async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
    access_control, monkeypatch
):
    class VerifiedRegistration:
        credential_id = b"phone-credential"
        credential_public_key = b"credential-public-key"
        sign_count = 0

    class VerifiedAuthentication:
        new_sign_count = 1

    monkeypatch.setattr(
        main.passkeys,
        "verify_registration",
        lambda **_kwargs: VerifiedRegistration(),
        raising=False,
    )
    monkeypatch.setattr(
        main.passkeys,
        "verify_authentication",
        lambda **_kwargs: VerifiedAuthentication(),
        raising=False,
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as bootstrap:
        await bootstrap.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        csrf_headers = {
            "Origin": "https://test",
            "X-CSRF-Token": bootstrap.cookies["stackchain_csrf"],
        }
        grant = await fresh_grant(bootstrap, "enroll_passkey", "current_device")
        options = await bootstrap.post(
            "/api/v1/passkeys/registration/options",
            headers={**csrf_headers, "X-Step-Up-Grant": grant},
        )
        enrolled = await bootstrap.post(
            "/api/v1/passkeys/registration/verify",
            json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
            headers=csrf_headers,
        )

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as returning:
        sign_in_options = await returning.post("/api/v1/passkeys/authentication/options")
        signed_in = await returning.post(
            "/api/v1/passkeys/authentication/verify",
            json={
                "challenge": sign_in_options.json()["challenge"],
                "credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
                "device_label": "Phone",
                "action": "sign_in",
                "target": "dashboard",
            },
        )
        activity = await returning.get("/api/v1/security-events")

    assert enrolled.status_code == 201
    assert enrolled.json() == {"enrolled": True}
    assert sign_in_options.status_code == 200
    assert sign_in_options.json()["allowCredentials"][0]["id"] == "cGhvbmUtY3JlZGVudGlhbA"
    assert signed_in.status_code == 200
    assert signed_in.json() == {"authenticated": True, "method": "passkey"}
    assert activity.json()["events"][0] == {
        "id": activity.json()["events"][0]["id"],
        "kind": "sign_in",
        "method": "passkey",
        "device_label": "Phone",
        "target": "dashboard",
        "created_at": activity.json()["events"][0]["created_at"],
        "status": "completed",
    }
    assert activity.json()["events"][1] == {
        "id": activity.json()["events"][1]["id"],
        "kind": "passkey_enrolled",
        "method": "passkey",
        "device_label": "Phone",
        "target": "passkey",
        "created_at": activity.json()["events"][1]["created_at"],
        "status": "completed",
    }
    assert "stackchain_session=" in signed_in.headers["set-cookie"]
    assert "correct horse battery staple" not in signed_in.text


@pytest.mark.anyio
async def test_stale_nonzero_passkey_counter_denies_sign_in_and_records_anomaly(
    access_control, monkeypatch
):
    class StaleAuthentication:
        new_sign_count = 4

    monkeypatch.setattr(
        main.passkeys,
        "verify_authentication",
        lambda **_kwargs: StaleAuthentication(),
    )
    await asyncio.to_thread(
        main._passkey_store().register,
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=4,
        device_label="Phone",
        management_id="phone-management-id",
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        options = await client.post("/api/v1/passkeys/authentication/options")
        denied = await client.post(
            "/api/v1/passkeys/authentication/verify",
            json={
                "challenge": options.json()["challenge"],
                "credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
                "device_label": "Phone",
                "action": "sign_in",
                "target": "dashboard",
            },
        )

    events = main._security_event_store().list(limit=10).events
    assert denied.status_code == 401
    assert "stackchain_session=" not in denied.headers.get("set-cookie", "")
    assert [event.kind for event in events] == ["passkey_counter_anomaly"]
    assert events[0].device_label == "Phone"
    assert events[0].target == "sign_in:dashboard"


@pytest.mark.anyio
async def test_stale_nonzero_passkey_counter_denies_fresh_authorization(
    access_control, monkeypatch
):
    class StaleAuthentication:
        new_sign_count = 4

    monkeypatch.setattr(
        main.passkeys,
        "verify_authentication",
        lambda **_kwargs: StaleAuthentication(),
    )
    await asyncio.to_thread(
        main._passkey_store().register,
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=4,
        device_label="Phone",
        management_id="phone-management-id",
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        options = await client.post(
            "/api/v1/passkeys/authorization/options",
            json={"action": "merge_pull", "target": "stackchain/dashboard#42"},
            headers=headers,
        )
        denied = await client.post(
            "/api/v1/passkeys/authorization/verify",
            json={
                "challenge": options.json()["challenge"],
                "credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
                "action": "merge_pull",
                "target": "stackchain/dashboard#42",
            },
            headers=headers,
        )

    events = main._security_event_store().list(limit=10).events
    assert options.status_code == 200
    assert denied.status_code == 401
    assert "grant" not in denied.json()
    assert events[0].kind == "passkey_counter_anomaly"
    assert events[0].target == "merge_pull:stackchain/dashboard#42"


@pytest.mark.anyio
async def test_passkey_options_are_source_limited_before_challenge_generation(
    access_control, monkeypatch
):
    monkeypatch.setenv("STACKCHAIN_PASSKEY_OPTIONS_MAX_ATTEMPTS", "2")
    await asyncio.to_thread(
        main._passkey_store().register,
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management-id",
    )
    first_source = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
    other_source = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))

    async with httpx.AsyncClient(transport=first_source, base_url="https://test") as client:
        first = await client.post("/api/v1/passkeys/authentication/options")
        second = await client.post("/api/v1/passkeys/authentication/options")
        blocked = await client.post("/api/v1/passkeys/authentication/options")
    async with httpx.AsyncClient(transport=other_source, base_url="https://test") as client:
        available = await client.post("/api/v1/passkeys/authentication/options")

    assert first.status_code == second.status_code == available.status_code == 200
    assert blocked.status_code == 429
    assert blocked.json() == {"detail": "Too many passkey sign-in attempts"}
    assert blocked.headers["retry-after"] == "60"
    with sqlite3.connect(main._passkey_store().path) as connection:
        assert connection.execute(
            "SELECT COUNT(*) FROM passkey_challenges WHERE purpose = 'authentication'"
        ).fetchone() == (3,)


@pytest.mark.anyio
async def test_failed_passkey_verification_consumes_the_shared_sign_in_budget(
    access_control,
):
    await asyncio.to_thread(
        main._passkey_store().register,
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management-id",
    )
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        failures = []
        for _attempt in range(3):
            options = await client.post("/api/v1/passkeys/authentication/options")
            failures.append(
                await client.post(
                    "/api/v1/passkeys/authentication/verify",
                    json={
                        "challenge": options.json()["challenge"],
                        "credential": {"id": "d3JvbmctY3JlZGVudGlhbA"},
                        "device_label": "Phone",
                        "action": "sign_in",
                        "target": "dashboard",
                    },
                )
            )
        blocked = await client.post("/api/v1/passkeys/authentication/options")

    assert [response.status_code for response in failures] == [401, 401, 401]
    assert blocked.status_code == 429
    assert blocked.headers["retry-after"] == "60"


@pytest.mark.anyio
async def test_successful_passkey_sign_in_clears_prior_source_failures(
    access_control, monkeypatch
):
    class VerifiedAuthentication:
        new_sign_count = 1

    monkeypatch.setattr(
        main.passkeys,
        "verify_authentication",
        lambda **_kwargs: VerifiedAuthentication(),
    )
    _signed, _managed_session = await asyncio.to_thread(
        main.dashboard_auth.issue_session,
        device_label="Phone",
        management_id="phone-management-id",
    )
    await asyncio.to_thread(
        main._passkey_store().register,
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management-id",
    )
    source = "203.0.113.10"
    attempts = main._login_attempt_store()
    await asyncio.to_thread(attempts.record_failure, source)
    await asyncio.to_thread(attempts.record_failure, source)
    transport = httpx.ASGITransport(app=main.app, client=(source, 1234))

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        options = await client.post("/api/v1/passkeys/authentication/options")
        signed_in = await client.post(
            "/api/v1/passkeys/authentication/verify",
            json={
                "challenge": options.json()["challenge"],
                "credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
                "device_label": "Phone",
                "action": "sign_in",
                "target": "dashboard",
            },
        )

    assert signed_in.status_code == 200, signed_in.text
    with sqlite3.connect(attempts.path) as connection:
        assert connection.execute(
            "SELECT COUNT(*) FROM login_attempts WHERE source_hash = ?",
            (attempts._digest(source),),
        ).fetchone() == (0,)


@pytest.mark.anyio
async def test_passkey_fresh_authorization_is_exact_target_bound_and_single_use(
    access_control, monkeypatch
):
    class VerifiedRegistration:
        credential_id = b"phone-credential"
        credential_public_key = b"credential-public-key"
        sign_count = 0

    class VerifiedAuthentication:
        new_sign_count = 1

    monkeypatch.setattr(
        main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
    )
    monkeypatch.setattr(
        main.passkeys, "verify_authentication", lambda **_kwargs: VerifiedAuthentication()
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        enrollment_grant = await fresh_grant(client, "enroll_passkey", "current_device")
        registration = await client.post(
            "/api/v1/passkeys/registration/options",
            headers={**headers, "X-Step-Up-Grant": enrollment_grant},
        )
        await client.post(
            "/api/v1/passkeys/registration/verify",
            json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
            headers=headers,
        )
        options = await client.post(
            "/api/v1/passkeys/authorization/options",
            json={"action": "close_issue", "target": "stackchain/api#7"},
            headers=headers,
        )
        review_options = await client.post(
            "/api/v1/passkeys/authorization/options",
            json={
                "action": "submit_pull_review",
                "target": "stackchain/api#7@abc123:approve",
            },
            headers=headers,
        )
        ceremony = {
            "challenge": options.json()["challenge"],
            "credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
            "action": "close_issue",
            "target": "stackchain/api#7",
        }
        authorized = await client.post(
            "/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
        )
        replayed = await client.post(
            "/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
        )
        wrong_target = await client.post(
            "/api/v1/passkeys/authorization/verify",
            json={**ceremony, "target": "stackchain/api#8"},
            headers=headers,
        )

    assert options.status_code == 200
    assert review_options.status_code == 200
    assert review_options.json()["challenge"]
    assert authorized.status_code == 201
    assert authorized.json()["grant"]
    assert authorized.json()["expires_in"] == 90
    assert replayed.status_code == 409
    assert wrong_target.status_code == 409


@pytest.mark.anyio
async def test_revoking_an_enrolled_device_also_revokes_its_passkey(
    access_control, monkeypatch
):
    class VerifiedRegistration:
        credential_id = b"phone-credential"
        credential_public_key = b"credential-public-key"
        sign_count = 0

    monkeypatch.setattr(
        main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
    )
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        phone_headers = {
            "Origin": "https://test",
            "X-CSRF-Token": phone.cookies["stackchain_csrf"],
        }
        enrollment_grant = await fresh_grant(phone, "enroll_passkey", "current_device")
        registration = await phone.post(
            "/api/v1/passkeys/registration/options",
            headers={**phone_headers, "X-Step-Up-Grant": enrollment_grant},
        )
        await phone.post(
            "/api/v1/passkeys/registration/verify",
            json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
            headers=phone_headers,
        )
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
        phone_device = next(item for item in devices if item["device_label"] == "Phone")
        grant = await fresh_grant(laptop, "revoke_device", phone_device["management_id"])
        revoked = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        passkey_options = await laptop.post("/api/v1/passkeys/authentication/options")

    assert revoked.status_code == 200
    assert passkey_options.status_code == 404


@pytest.mark.anyio
async def test_device_revocation_failure_preserves_its_passkey_and_session(
    access_control,
):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        phone_device = next(
            item
            for item in (await laptop.get("/api/v1/sessions")).json()["devices"]
            if item["device_label"] == "Phone"
        )
        store = main._passkey_store()
        await asyncio.to_thread(
            store.register,
            credential_id=b"phone-credential",
            public_key=b"phone-public-key",
            sign_count=0,
            device_label="Phone",
            management_id=phone_device["management_id"],
        )
        grant = await fresh_grant(
            laptop, "revoke_device", phone_device["management_id"]
        )
        with sqlite3.connect(store.path) as connection:
            connection.execute(
                "CREATE TRIGGER block_phone_delete BEFORE DELETE ON active_sessions "
                f"WHEN OLD.management_id = '{phone_device['management_id']}' "
                "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
            )

        response = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_session = await phone.get("/api/v1/session")
        passkeys = await laptop.get("/api/v1/passkeys")

    assert response.status_code == 503
    assert phone_session.status_code == 200
    assert phone_session.json()["authenticated"] is True
    assert [item["device_label"] for item in passkeys.json()["passkeys"]] == ["Phone"]


@pytest.mark.anyio
async def test_orphaned_passkey_is_listed_safely_and_can_be_selectively_revoked(
    access_control, monkeypatch
):
    class VerifiedRegistration:
        credential_id = b"old-phone-credential"
        credential_public_key = b"old-phone-public-key"
        sign_count = 0

    monkeypatch.setattr(
        main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as phone:
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Old phone"},
        )
        phone_headers = {
            "Origin": "https://test",
            "X-CSRF-Token": phone.cookies["stackchain_csrf"],
        }
        enrollment_grant = await fresh_grant(phone, "enroll_passkey", "current_device")
        registration = await phone.post(
            "/api/v1/passkeys/registration/options",
            headers={**phone_headers, "X-Step-Up-Grant": enrollment_grant},
        )
        enrolled = await phone.post(
            "/api/v1/passkeys/registration/verify",
            json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
            headers=phone_headers,
        )
        await phone.delete("/api/v1/session", headers=phone_headers)

    store = main._passkey_store()
    await asyncio.to_thread(
        store.register,
        credential_id=b"backup-credential",
        public_key=b"backup-public-key",
        sign_count=0,
        device_label="Backup key",
        management_id="backup-management-id",
    )

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as laptop:
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        listed = await laptop.get("/api/v1/passkeys")
        old_phone = next(
            item for item in listed.json()["passkeys"] if item["device_label"] == "Old phone"
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
        }
        missing_grant = await laptop.delete(
            f"/api/v1/passkeys/{old_phone['management_id']}", headers=headers
        )
        grant = await fresh_grant(laptop, "revoke_passkey", old_phone["management_id"])
        removed = await laptop.delete(
            f"/api/v1/passkeys/{old_phone['management_id']}",
            headers={**headers, "X-Step-Up-Grant": grant},
        )
        remaining = await laptop.get("/api/v1/passkeys")
        sign_in_options = await laptop.post("/api/v1/passkeys/authentication/options")

    assert enrolled.status_code == 201
    assert listed.status_code == 200
    assert listed.headers["cache-control"] == "no-store"
    assert set(old_phone) == {
        "management_id", "device_label", "created_at", "active", "current"
    }
    assert old_phone["active"] is False
    assert old_phone["current"] is False
    assert b"old-phone-credential" not in listed.content
    assert b"old-phone-public-key" not in listed.content
    assert missing_grant.status_code == 428
    assert missing_grant.json()["detail"] == {
        "detail": "Fresh authorization required",
        "code": "step_up_required",
        "action": "revoke_passkey",
        "target": old_phone["management_id"],
    }
    assert removed.status_code == 200
    assert removed.json() == {
        "revoked": True,
        "current_session": False,
        "session_revoked": False,
    }
    assert [item["device_label"] for item in remaining.json()["passkeys"]] == ["Backup key"]
    assert sign_in_options.status_code == 200
    assert sign_in_options.json()["allowCredentials"] == [
        {"id": "YmFja3VwLWNyZWRlbnRpYWw", "type": "public-key"}
    ]


@pytest.mark.anyio
async def test_removing_a_remote_passkey_also_revokes_its_active_session(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
        phone_device = next(item for item in devices if item["device_label"] == "Phone")
        await asyncio.to_thread(
            main._passkey_store().register,
            credential_id=b"phone-credential",
            public_key=b"phone-public-key",
            sign_count=0,
            device_label="Phone",
            management_id=phone_device["management_id"],
        )
        grant = await fresh_grant(
            laptop, "revoke_passkey", phone_device["management_id"]
        )
        removed = await laptop.delete(
            f"/api/v1/passkeys/{phone_device['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_status = await phone.get("/api/v1/session")
        laptop_status = await laptop.get("/api/v1/session")

    assert removed.json() == {
        "revoked": True,
        "current_session": False,
        "session_revoked": True,
    }
    assert phone_status.status_code == 401
    assert phone_status.json()["code"] == "session_revoked"
    assert laptop_status.status_code == 200


@pytest.mark.anyio
async def test_removing_the_current_passkey_keeps_the_current_session_active(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        current = next(
            item for item in (await client.get("/api/v1/sessions")).json()["devices"]
            if item["current"]
        )
        await asyncio.to_thread(
            main._passkey_store().register,
            credential_id=b"current-credential",
            public_key=b"current-public-key",
            sign_count=0,
            device_label="Laptop",
            management_id=current["management_id"],
        )
        grant = await fresh_grant(client, "revoke_passkey", current["management_id"])
        removed = await client.delete(
            f"/api/v1/passkeys/{current['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        still_active = await client.get("/api/v1/session")
        activity = await client.get("/api/v1/security-events")

    assert removed.json() == {
        "revoked": True,
        "current_session": True,
        "session_revoked": False,
    }
    assert still_active.status_code == 200
    assert activity.json()["events"][0]["kind"] == "passkey_revoked"
    assert activity.json()["events"][0]["status"] == "completed"


@pytest.mark.anyio
async def test_default_operator_mode_fails_closed_before_gitea_when_secrets_are_missing(monkeypatch):
    monkeypatch.delenv("STACKCHAIN_DASHBOARD_AUTH_MODE", raising=False)
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        health = await client.get("/healthz")
        responses = [
            await client.get("/readyz"),
            await client.get("/login"),
            await client.get("/"),
            await client.get("/api/v1/context"),
        ]

    assert health.status_code == 200
    assert [response.status_code for response in responses] == [503, 503, 503, 503]
    assert all(response.headers["cache-control"] == "no-store" for response in responses)
    assert all(response.json() == {
        "detail": "Dashboard authentication is not configured"
    } for response in responses)
    assert called is False


@pytest.mark.anyio
async def test_operator_mode_rejects_reused_or_incomplete_secrets(monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "same-secret-with-at-least-thirty-two-characters")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "same-secret-with-at-least-thirty-two-characters")
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        reused = await client.get("/login")
        monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
        incomplete = await client.get("/login")

    assert reused.status_code == 503
    assert incomplete.status_code == 503
    assert "same-secret" not in reused.text


@pytest.mark.anyio
async def test_insecure_local_mode_is_restricted_to_loopback(monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "insecure-local")
    local_transport = httpx.ASGITransport(app=main.app, client=("127.0.0.1", 1234))
    remote_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))
    async with httpx.AsyncClient(transport=local_transport, base_url="http://test") as client:
        local = await client.get("/login")
    async with httpx.AsyncClient(transport=remote_transport, base_url="http://test") as client:
        remote = await client.get("/login")

    assert local.status_code == 200
    assert remote.status_code == 403
    assert remote.json() == {"detail": "Insecure local mode requires a loopback client"}


@pytest.mark.anyio
async def test_anonymous_private_request_is_rejected_before_gitea(access_control, monkeypatch):
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get("/api/v1/context")

    assert response.status_code == 401
    assert response.json() == {"detail": "Authentication required"}
    assert response.headers["cache-control"] == "no-store"
    assert called is False


@pytest.mark.anyio
async def test_anonymous_share_target_redirect_preserves_only_bounded_capture_fields(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get(
            "/",
            params={
                "title": "Production crash",
                "text": "Steps from the mobile app",
                "url": "https://example.com/incidents/42",
                "next": "https://evil.example/steal",
            },
        )

    assert response.status_code == 303
    login_query = parse_qs(urlsplit(response.headers["location"]).query)
    assert login_query == {
        "continue": [
            "./?title=Production+crash&text=Steps+from+the+mobile+app&url="
            "https%3A%2F%2Fexample.com%2Fincidents%2F42"
        ]
    }
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_anonymous_app_shortcut_preserves_only_a_known_launch_action(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        valid = await client.get("/", params={"launch": "agenda", "next": "https://evil.example"})
        invalid = await client.get("/", params={"launch": "delete-account"})

    assert valid.status_code == 303
    assert parse_qs(urlsplit(valid.headers["location"]).query) == {
        "continue": ["./?launch=agenda"]
    }
    assert invalid.status_code == 303
    assert invalid.headers["location"] == "login"


@pytest.mark.anyio
async def test_anonymous_search_preview_preserves_only_bounded_canonical_continuation(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        valid = await client.get(
            "/", params={"search": "release blocker", "preview": "issue:stackchain/api:42"}
        )
        malformed = await client.get(
            "/", params={"search": "release blocker", "preview": "issue:stackchain/api:0"}
        )
        duplicate = await client.get(
            "/?search=release&search=other&preview=issue%3Astackchain%2Fapi%3A42"
        )
        oversized = await client.get(
            "/", params={"search": "release", "preview": f"issue:{'x' * 201}/api:42"}
        )

    assert parse_qs(urlsplit(valid.headers["location"]).query) == {
        "continue": [
            "./?search=release+blocker&preview=issue%3Astackchain%2Fapi%3A42"
        ]
    }
    assert malformed.headers["location"] == "login"
    assert duplicate.headers["location"] == "login"
    assert oversized.headers["location"] == "login"


@pytest.mark.anyio
async def test_anonymous_search_preview_preserves_only_valid_search_scope(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        valid = await client.get("/", params={
            "search": "release blocker", "preview": "pull:stackchain/api:42",
            "search_kind": "pull", "search_state": "open",
            "search_repository": "stackchain/api",
        })
        invalid = await client.get("/", params={
            "search": "release blocker", "preview": "pull:stackchain/api:42",
            "search_kind": "script", "search_state": "secret",
        })

    valid_continue = parse_qs(urlsplit(valid.headers["location"]).query)["continue"][0]
    invalid_continue = parse_qs(urlsplit(invalid.headers["location"]).query)["continue"][0]
    assert valid_continue.endswith(
        "search=release+blocker&preview=pull%3Astackchain%2Fapi%3A42&"
        "search_kind=pull&search_state=open&search_repository=stackchain%2Fapi"
    )
    assert invalid_continue == (
        "./?search=release+blocker&preview=pull%3Astackchain%2Fapi%3A42"
    )


@pytest.mark.anyio
async def test_anonymous_shared_screenshot_preserves_bounded_sign_in_continuation(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        valid = await client.get(
            "/",
            params={
                "launch": "new",
                "shared": "image",
                "title": "Broken checkout",
                "text": "Steps to reproduce",
            },
        )
        wrong_launch = await client.get(
            "/", params={"launch": "agenda", "shared": "image"}
        )
        wrong_marker = await client.get(
            "/", params={"launch": "new", "shared": "document"}
        )
        duplicate_marker = await client.get(
            "/?launch=new&shared=image&shared=image"
        )

    assert valid.status_code == 303
    assert parse_qs(urlsplit(valid.headers["location"]).query) == {
        "continue": [
            "./?title=Broken+checkout&text=Steps+to+reproduce&launch=new&shared=image"
        ]
    }
    assert wrong_launch.headers["location"] == "login"
    assert wrong_marker.headers["location"] == "login"
    assert duplicate_marker.headers["location"] == "login"


@pytest.mark.anyio
async def test_oversized_share_target_is_not_carried_through_login(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get(
            "/", params={"title": "x" * 201, "text": "keep me"}
        )

    assert response.status_code == 303
    assert response.headers["location"] == "login"


@pytest.mark.anyio
async def test_sign_in_creates_secure_session_without_echoing_access_token(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert response.status_code == 200
    assert response.json() == {"authenticated": True}
    cookies = response.headers.get_list("set-cookie")
    assert any("stackchain_session=" in value and "HttpOnly" in value and "Secure" in value and "SameSite=strict" in value for value in cookies)
    assert any("stackchain_csrf=" in value and "Secure" in value and "SameSite=strict" in value and "HttpOnly" not in value for value in cookies)
    assert "correct horse battery staple" not in response.text
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_sign_in_throttles_repeated_failures_with_retry_guidance(access_control):
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        failures = [
            await client.post("/api/v1/session", json={"access_token": "wrong"})
            for _ in range(3)
        ]
        blocked = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert [response.status_code for response in failures] == [401, 401, 401]
    assert blocked.status_code == 429
    assert blocked.json() == {"detail": "Too many sign-in attempts"}
    assert blocked.headers["retry-after"].isdigit()
    assert blocked.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_security_activity_surfaces_failed_and_blocked_token_sign_ins(access_control):
    attacker_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
    operator_transport = httpx.ASGITransport(app=main.app, client=("198.51.100.4", 1234))
    async with httpx.AsyncClient(
        transport=attacker_transport, base_url="https://test"
    ) as attacker:
        failures = [
            await attacker.post(
                "/api/v1/session",
                json={"access_token": "wrong", "device_label": "Unknown"},
            )
            for _ in range(4)
        ]
    async with httpx.AsyncClient(
        transport=operator_transport, base_url="https://test"
    ) as operator:
        signed_in = await operator.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Phone",
            },
        )
        activity = await operator.get("/api/v1/security-events")

    assert [response.status_code for response in failures] == [401, 401, 401, 429]
    assert signed_in.status_code == 200
    assert activity.status_code == 200
    assert activity.headers["cache-control"] == "no-store"
    alerts = activity.json()["authentication_alerts"]
    assert len(alerts) == 1
    assert alerts[0]["method"] == "token"
    assert alerts[0]["failed_count"] == 3
    assert alerts[0]["blocked_count"] == 1
    assert alerts[0]["first_at"] <= alerts[0]["last_at"]


@pytest.mark.anyio
async def test_successful_sign_in_clears_prior_failures(access_control):
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        for _ in range(2):
            await client.post("/api/v1/session", json={"access_token": "wrong"})
        success = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        after_success = [
            await client.post("/api/v1/session", json={"access_token": "wrong"})
            for _ in range(3)
        ]

    assert success.status_code == 200
    assert [response.status_code for response in after_success] == [401, 401, 401]


@pytest.mark.anyio
async def test_authenticated_get_reaches_private_api(access_control, monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy", "full_name": "", "email": ""}

    async def empty():
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        signed_in = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.get("/api/v1/context")

    assert signed_in.status_code == 200
    assert response.status_code == 200
    assert response.json()["user"]["login"] == "timmy"


@pytest.mark.anyio
async def test_session_status_reuses_the_middleware_validation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        original_store = main.dashboard_auth._session_store()
        lookups = 0

        class CountingStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                nonlocal lookups
                lookups += 1
                return original_store.status(
                    session_id,
                    expires_at,
                    idle_timeout_seconds=idle_timeout_seconds,
                )

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
        response = await client.get("/api/v1/session")

    assert response.status_code == 200
    payload = response.json()
    assert payload["authenticated"] is True
    assert isinstance(payload["expires_at"], int)
    assert payload["expires_at"] > int(time.time())
    assert response.headers["cache-control"] == "no-store"
    assert lookups == 1


@pytest.mark.anyio
async def test_authenticated_session_status_exposes_csrf_proof_and_offline_lease(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.get("/api/v1/session")

    payload = response.json()
    assert response.status_code == 200
    assert payload["authenticated"] is True
    assert payload["csrf_token"] == client.cookies["stackchain_csrf"]
    assert payload["expires_at"] > int(time.time())
    assert payload["idle_expires_at"] > int(time.time())
    assert "correct horse battery staple" not in response.text


@pytest.mark.anyio
async def test_session_status_reports_server_idle_deadline_without_extending_it(
    access_control, monkeypatch
):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "900")
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        expected_last_active = int(time.time()) - 120
        with sqlite3.connect(main.dashboard_auth._session_store().path) as connection:
            connection.execute(
                "UPDATE active_sessions SET last_active_at = ?", (expected_last_active,)
            )

        first = await client.get("/api/v1/session")
        second = await client.get("/api/v1/session")
        with sqlite3.connect(main.dashboard_auth._session_store().path) as connection:
            stored_last_active = connection.execute(
                "SELECT last_active_at FROM active_sessions"
            ).fetchone()[0]

    assert first.status_code == 200
    assert first.json()["idle_expires_at"] == expected_last_active + 900
    assert second.json()["idle_expires_at"] == expected_last_active + 900
    assert stored_last_active == expected_last_active


@pytest.mark.anyio
async def test_anonymous_session_status_discloses_no_offline_lease(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.get("/api/v1/session")

    assert response.status_code == 401
    assert response.json() == {"detail": "Authentication required"}
    assert "expires_at" not in response.text
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_idle_session_is_rejected_with_distinct_api_and_page_recovery(access_control, monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "900")
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        with sqlite3.connect(main.dashboard_auth._session_store().path) as connection:
            connection.execute(
                "UPDATE active_sessions SET last_active_at = ?", (int(time.time()) - 900,)
            )

        api_response = await client.get("/api/v1/session")
        page_response = await client.get("/")

    assert api_response.status_code == 401
    assert api_response.json() == {
        "detail": "Authentication required",
        "code": "session_idle",
    }
    assert page_response.status_code == 303
    assert page_response.headers["location"] == "login?reason=session-idle"


@pytest.mark.anyio
async def test_activity_heartbeat_extends_the_server_idle_deadline(access_control, monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "900")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_TTL_SECONDS", "100")
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        heartbeat = await client.post(
            "/api/v1/session/activity",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )
        status = await client.get("/api/v1/session")

    assert heartbeat.status_code == 200
    assert heartbeat.json()["active"] is True
    assert heartbeat.json()["idle_expires_at"] == status.json()["expires_at"]


@pytest.mark.anyio
async def test_merge_requires_single_use_fresh_authorization_bound_to_exact_target(
    access_control, monkeypatch
):
    merge_calls = []

    async def assigned(repository, number):
        return True

    async def merge(repository, number, expected_head_sha):
        merge_calls.append((repository, number, expected_head_sha))
        return {"number": number, "merged": True, "state": "closed"}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf_headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        missing = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=csrf_headers,
        )
        authorized = await client.post(
            "/api/v1/fresh-authorization",
            json={
                "access_token": "correct horse battery staple",
                "action": "merge_pull",
                "target": "stackchain/api#7",
            },
            headers=csrf_headers,
        )
        grant_headers = {**csrf_headers, "X-Step-Up-Grant": authorized.json()["grant"]}
        merged = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=grant_headers,
        )
        replayed = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/merge",
            json={"expected_head_sha": "abc123"},
            headers=grant_headers,
        )

    assert missing.status_code == 428
    assert missing.json() == {
        "detail": {
            "detail": "Fresh authorization required",
            "code": "step_up_required",
            "action": "merge_pull",
            "target": "stackchain/api#7",
        }
    }
    assert authorized.status_code == 201
    assert authorized.json()["expires_in"] == 90
    assert merged.status_code == 200
    assert replayed.status_code == 428
    assert merge_calls == [("stackchain/api", 7, "abc123")]


@pytest.mark.anyio
async def test_other_high_impact_routes_require_fresh_authorization_before_mutation(
    access_control, monkeypatch
):
    close_calls = []

    async def assigned(repository, number):
        return True

    async def close(repository, number):
        close_calls.append((repository, number))
        return {"number": number, "state": "closed"}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        await laptop.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
        )
        remote = next(
            item
            for item in (await phone.get("/api/v1/sessions")).json()["devices"]
            if not item["current"]
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": phone.cookies["stackchain_csrf"],
        }
        closed = await phone.patch(
            "/api/v1/repos/stackchain/api/issues/7/close", headers=headers
        )
        revoked = await phone.delete(
            f"/api/v1/sessions/{remote['management_id']}", headers=headers
        )
        revoked_all = await phone.delete("/api/v1/sessions", headers=headers)
        laptop_still_active = await laptop.get("/api/v1/session")

    assert [closed.status_code, revoked.status_code, revoked_all.status_code] == [428, 428, 428]
    assert [closed.json()["detail"]["action"], revoked.json()["detail"]["action"], revoked_all.json()["detail"]["action"]] == [
        "close_issue", "revoke_device", "revoke_all_sessions"
    ]
    assert close_calls == []
    assert laptop_still_active.status_code == 200


@pytest.mark.anyio
async def test_consequential_review_requires_decision_and_head_bound_fresh_authorization(
    access_control, monkeypatch
):
    calls = []

    async def requested(repository, number):
        calls.append(("requested", repository, number))
        return True

    async def submit(repository, number, head, decision, body):
        calls.append(("submit", repository, number, head, decision, body))
        return {"id": 91, "state": "APPROVED"}

    monkeypatch.setattr(main, "is_requested_review", requested)
    monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
    transport = httpx.ASGITransport(app=main.app)
    payload = {
        "expected_head_sha": "abc123",
        "decision": "approve",
        "body": "Ready to ship.",
    }
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple"},
        )
        headers = {
            "Origin": "https://test",
            "X-CSRF-Token": client.cookies["stackchain_csrf"],
        }
        missing = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json=payload,
            headers=headers,
        )
        wrong_grant = await fresh_grant(
            client,
            "submit_pull_review",
            "stackchain/api#7@different:approve",
        )
        mismatched = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json=payload,
            headers={**headers, "X-Step-Up-Grant": wrong_grant},
        )
        grant = await fresh_grant(
            client,
            "submit_pull_review",
            "stackchain/api#7@abc123:approve",
        )
        approved = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json=payload,
            headers={**headers, "X-Step-Up-Grant": grant},
        )
        replayed = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json=payload,
            headers={**headers, "X-Step-Up-Grant": grant},
        )
        comment = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json={**payload, "decision": "comment"},
            headers=headers,
        )

    assert [missing.status_code, mismatched.status_code] == [428, 428]
    assert missing.json()["detail"] == {
        "detail": "Fresh authorization required",
        "code": "step_up_required",
        "action": "submit_pull_review",
        "target": "stackchain/api#7@abc123:approve",
    }
    assert approved.status_code == 201
    assert replayed.status_code == 428
    assert comment.status_code == 201
    assert calls == [
        ("requested", "stackchain/api", 7),
        ("submit", "stackchain/api", 7, "abc123", "approve", "Ready to ship."),
        ("requested", "stackchain/api", 7),
        ("submit", "stackchain/api", 7, "abc123", "comment", "Ready to ship."),
    ]


@pytest.mark.anyio
async def test_operator_can_review_and_revoke_one_remote_device(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Pixel <script>",
            },
        )
        await laptop.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "Work laptop",
            },
        )

        listed = await laptop.get("/api/v1/sessions")
        devices = listed.json()["devices"]
        phone_device = next(device for device in devices if device["device_label"] == "Pixel <script>")
        current_device = next(device for device in devices if device["current"])

        missing_csrf = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}"
        )
        grant = await fresh_grant(
            laptop, "revoke_device", phone_device["management_id"]
        )
        revoked = await laptop.delete(
            f"/api/v1/sessions/{phone_device['management_id']}",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_status = await phone.get("/api/v1/session")
        phone_page = await phone.get("/", follow_redirects=False)
        laptop_status = await laptop.get("/api/v1/session")

    assert listed.status_code == 200
    assert listed.headers["cache-control"] == "no-store"
    assert current_device["device_label"] == "Work laptop"
    assert missing_csrf.status_code == 403
    assert revoked.json() == {"revoked": True, "current_session": False}
    assert phone_status.status_code == 401
    assert phone_status.json() == {
        "detail": "Authentication required",
        "code": "session_revoked",
    }
    assert phone_page.status_code == 303
    assert phone_page.headers["location"] == "login?reason=session-revoked"
    assert laptop_status.status_code == 200
    assert "session_hash" not in listed.text
    assert "csrf" not in listed.text


@pytest.mark.anyio
async def test_active_device_labels_are_bounded_at_sign_in(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session",
            json={
                "access_token": "correct horse battery staple",
                "device_label": "x" * 65,
            },
        )

    assert response.status_code == 422


@pytest.mark.anyio
async def test_mutation_requires_same_origin_and_session_csrf(access_control, monkeypatch):
    calls = 0

    async def mark_read(notification_id):
        nonlocal calls
        calls += 1
        return {"id": notification_id, "read": True}

    monkeypatch.setattr(main, "mark_notification_read", mark_read)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        missing = await client.patch("/api/v1/notifications/7/read")
        foreign = await client.patch(
            "/api/v1/notifications/7/read",
            headers={
                "Origin": "https://evil.example",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )
        valid = await client.patch(
            "/api/v1/notifications/7/read",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )

    assert missing.status_code == 403
    assert foreign.status_code == 403
    assert valid.status_code == 200
    assert calls == 1


@pytest.mark.anyio
async def test_operator_mode_rejects_unconfigured_or_foreign_public_origin_before_auth(
    monkeypatch,
):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
    monkeypatch.setenv(
        "STACKCHAIN_DASHBOARD_SESSION_SECRET",
        "a-separate-session-signing-secret-with-enough-entropy",
    )
    monkeypatch.delenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", raising=False)
    transport = httpx.ASGITransport(app=main.app)

    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        missing = await client.get("/login")
        monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "http://test")
        insecure = await client.get("/login")
        monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://dashboard.example")
        foreign_host = await client.get("/login")

    assert missing.status_code == 503
    assert insecure.status_code == 503
    assert foreign_host.status_code == 421


@pytest.mark.anyio
async def test_canonical_origin_drives_csrf_and_passkey_defaults(access_control, monkeypatch):
    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://dashboard.example")
    monkeypatch.delenv("STACKCHAIN_PASSKEY_RP_ID", raising=False)
    monkeypatch.delenv("STACKCHAIN_PASSKEY_ORIGIN", raising=False)
    calls = 0

    async def mark_read(notification_id):
        nonlocal calls
        calls += 1
        return {"id": notification_id, "read": True}

    monkeypatch.setattr(main, "mark_notification_read", mark_read)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(
        transport=transport, base_url="https://dashboard.example"
    ) as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]
        valid = await client.patch(
            "/api/v1/notifications/7/read",
            headers={
                "Origin": "https://dashboard.example",
                "X-CSRF-Token": csrf,
            },
        )

    request = Request({
        "type": "http",
        "method": "GET",
        "scheme": "https",
        "path": "/",
        "query_string": b"",
        "headers": [(b"host", b"attacker.invalid")],
        "server": ("attacker.invalid", 443),
        "client": ("127.0.0.1", 1),
    })
    assert main._passkey_relying_party(request) == (
        "dashboard.example",
        "https://dashboard.example",
    )
    assert valid.status_code == 200
    assert calls == 1


@pytest.mark.anyio
async def test_logout_clears_session_and_blocks_private_routes(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]
        response = await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )
        private = await client.get("/api/v1/background-identity")

    assert response.status_code == 200
    assert response.json() == {"authenticated": False, "clear_private_device_data": True}
    assert private.status_code == 401
    assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))


@pytest.mark.anyio
async def test_sign_out_all_devices_revokes_every_existing_session(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
    ):
        await phone.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        await laptop.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        grant = await fresh_grant(phone, "revoke_all_sessions", "all")
        response = await phone.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": phone.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        phone_private = await phone.get("/api/v1/background-identity")
        laptop_private = await laptop.get("/api/v1/background-identity")

    assert response.status_code == 200
    assert response.json() == {
        "authenticated": False,
        "all_sessions_revoked": True,
        "clear_private_device_data": True,
    }
    assert phone_private.status_code == 401
    assert laptop_private.status_code == 401
    assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))


@pytest.mark.anyio
async def test_sign_out_all_devices_rejects_cross_site_requests_without_revoking(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        response = await client.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://evil.example",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
            },
        )
        session = await client.get("/api/v1/session")

    assert response.status_code == 403
    assert response.headers.get_list("set-cookie") == []
    assert session.status_code == 200
    assert session.json()["authenticated"] is True


@pytest.mark.anyio
async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]
        grant = await fresh_grant(client, "revoke_all_sessions", "all")

        class BrokenStore:
            def revoke_all_access(self):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main, "_passkey_store", lambda: BrokenStore())
        response = await client.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": csrf,
                "X-Step-Up-Grant": grant,
            },
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_sign_out_all_devices_failure_preserves_passkeys_and_sessions(access_control):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
        )
        device = (await client.get("/api/v1/sessions")).json()["devices"][0]
        store = main._passkey_store()
        await asyncio.to_thread(
            store.register,
            credential_id=b"phone-credential",
            public_key=b"phone-public-key",
            sign_count=0,
            device_label="Phone",
            management_id=device["management_id"],
        )
        grant = await fresh_grant(client, "revoke_all_sessions", "all")
        with sqlite3.connect(store.path) as connection:
            connection.execute(
                "CREATE TRIGGER block_session_delete BEFORE DELETE ON active_sessions "
                "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
            )

        response = await client.delete(
            "/api/v1/sessions",
            headers={
                "Origin": "https://test",
                "X-CSRF-Token": client.cookies["stackchain_csrf"],
                "X-Step-Up-Grant": grant,
            },
        )
        session = await client.get("/api/v1/session")
        passkeys = await client.get("/api/v1/passkeys")

    assert response.status_code == 503
    assert session.status_code == 200
    assert session.json()["authenticated"] is True
    assert passkeys.status_code == 200
    assert [item["device_label"] for item in passkeys.json()["passkeys"]] == ["Phone"]


@pytest.mark.anyio
async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
    calls = 0

    async def user():
        nonlocal calls
        calls += 1
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        captured = client.cookies["stackchain_session"]
        csrf = client.cookies["stackchain_csrf"]

        async with httpx.AsyncClient(
            transport=transport,
            base_url="https://test",
            headers={"Cookie": f"stackchain_session={captured}"},
        ) as replay:
            before = await replay.get("/api/v1/background-identity")

        await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )

        async with httpx.AsyncClient(
            transport=transport,
            base_url="https://test",
            headers={"Cookie": f"stackchain_session={captured}"},
        ) as replay:
            after = await replay.get("/api/v1/background-identity")

    assert before.status_code == 200
    assert after.status_code == 401
    assert after.headers["cache-control"] == "no-store"
    assert calls == 1


@pytest.mark.anyio
async def test_session_registry_latency_does_not_block_the_event_loop(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class SlowStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                time.sleep(0.15)
                return "active"

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
        private_request = asyncio.create_task(client.get("/api/v1/session"))
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.01)
        heartbeat_elapsed = time.perf_counter() - started
        response = await private_request

    assert response.status_code == 200
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_sign_in_throttle_lookup_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.15)
            return 17

        def record_blocked(self, *, method):
            assert method == "token"

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.10", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post("/api/v1/session", json={"access_token": "wrong"})
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.01)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 429
    assert response.headers["retry-after"] == "17"
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_failed_sign_in_recording_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    recording_started = threading.Event()
    recording_finished = threading.Event()

    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def record_failure(self, source):
            recording_started.set()
            time.sleep(0.15)
            recording_finished.set()

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.11", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post("/api/v1/session", json={"access_token": "wrong"})
        )
        assert await asyncio.to_thread(recording_started.wait, 1)
        await asyncio.sleep(0)
        assert not recording_finished.is_set()
        response = await sign_in

    assert response.status_code == 401


@pytest.mark.anyio
async def test_successful_sign_in_throttle_clear_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    clear_started = threading.Event()
    clear_finished = threading.Event()

    class SlowAttempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def clear(self, source):
            clear_started.set()
            time.sleep(0.15)
            clear_finished.set()

    monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.12", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post(
                "/api/v1/session",
                json={"access_token": "correct horse battery staple"},
            )
        )
        assert await asyncio.to_thread(clear_started.wait, 1)
        await asyncio.sleep(0)
        assert not clear_finished.is_set()
        response = await sign_in

    assert response.status_code == 200


@pytest.mark.anyio
async def test_session_activation_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    class Attempts:
        def retry_after(self, source):
            time.sleep(0.02)
            return 0

        def clear(self, source):
            return None

    class SlowStore:
        def activate(self, session_id, expires_at, **kwargs):
            time.sleep(0.15)

    monkeypatch.setattr(main, "_login_attempt_store", lambda: Attempts())
    monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
    transport = httpx.ASGITransport(app=main.app, client=("203.0.113.13", 1234))
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        sign_in = asyncio.create_task(
            client.post(
                "/api/v1/session",
                json={"access_token": "correct horse battery staple"},
            )
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_in

    assert response.status_code == 200
    assert any(
        "stackchain_session=" in value
        for value in response.headers.get_list("set-cookie")
    )
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_single_session_revocation_does_not_block_the_event_loop(
    access_control, monkeypatch
):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session",
            json={"access_token": "correct horse battery staple"},
        )
        csrf = client.cookies["stackchain_csrf"]

        class SlowStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                time.sleep(0.02)
                return "active"

            def revoke(self, session_id):
                time.sleep(0.15)

        monkeypatch.setattr(
            main.dashboard_auth, "_session_store", lambda now=None: SlowStore()
        )
        sign_out = asyncio.create_task(
            client.delete(
                "/api/v1/session",
                headers={"Origin": "https://test", "X-CSRF-Token": csrf},
            )
        )
        await asyncio.sleep(0)
        started = time.perf_counter()
        await asyncio.sleep(0.04)
        heartbeat_elapsed = time.perf_counter() - started
        response = await sign_out

    assert response.status_code == 200
    assert response.json()["authenticated"] is False
    assert heartbeat_elapsed < 0.08


@pytest.mark.anyio
async def test_session_registry_read_failure_fails_closed_before_gitea(
    access_control, monkeypatch
):
    called = False

    async def user():
        nonlocal called
        called = True
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class BrokenStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        response = await client.get("/api/v1/background-identity")
        health = await client.get("/healthz")

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert "database path" not in response.text
    assert called is False
    assert health.status_code == 200


@pytest.mark.anyio
async def test_sign_in_registry_write_failure_issues_no_cookie(access_control, monkeypatch):
    class BrokenStore:
        def activate(self, session_id, expires_at, **kwargs):
            raise SessionStoreError("database path and secret details")

    monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        response = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_logout_registry_failure_does_not_claim_revocation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )
        csrf = client.cookies["stackchain_csrf"]

        class BrokenStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                return "active"

            def revoke(self, session_id):
                raise SessionStoreError("database path and secret details")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        response = await client.delete(
            "/api/v1/session",
            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
        )

    assert response.status_code == 503
    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
    assert response.headers["cache-control"] == "no-store"
    assert response.headers.get_list("set-cookie") == []
    assert "database path" not in response.text


@pytest.mark.anyio
async def test_public_routes_skip_session_registry_validation(access_control, monkeypatch):
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        signed_in = await client.post(
            "/api/v1/session", json={"access_token": "correct horse battery staple"}
        )

        class BrokenStore:
            def status(self, session_id, expires_at, *, idle_timeout_seconds):
                raise SessionStoreError("public routes must not read the registry")

        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
        login = await client.get("/login")
        manifest = await client.get("/manifest.webmanifest")
        static = await client.get("/static/session.js")
        runtime = await client.get("/" + FRONTEND_BUILD.runtime_name)

    assert signed_in.status_code == 200
    assert [login.status_code, manifest.status_code, static.status_code, runtime.status_code] == [
        200,
        200,
        200,
        200,
    ]
    assert runtime.headers["cache-control"] == "public, max-age=31536000, immutable"


@pytest.mark.anyio
async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    monkeypatch.setattr(main, "current_user", user)
    await main._check_readiness()
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
        health = await client.get("/healthz")
        login = await client.get("/login")
        ready = await client.get("/readyz")

    assert health.status_code == 200
    assert login.status_code == 200
    assert 'name="access_token"' in login.text
    assert ready.status_code == 200
    assert ready.json() == {"status": "ready", "service": "stackchain-dashboard"}


@pytest.mark.anyio
async def test_subpath_deployment_scopes_routes_and_session_cookies(access_control):
    transport = httpx.ASGITransport(app=main.app, root_path="/dashboard")
    async with httpx.AsyncClient(
        transport=transport, base_url="https://test/dashboard/"
    ) as client:
        login = await client.get("login")
        signed_in = await client.post(
            "api/v1/session",
            json={"access_token": "correct horse battery staple"},
        )
        status = await client.get("api/v1/session")

    assert login.status_code == 200
    assert signed_in.status_code == 200
    assert status.status_code == 200
    assert any(
        "stackchain_session=" in value and "Path=/dashboard" in value
        for value in signed_in.headers.get_list("set-cookie")
    )