import asyncio
import base64
import binascii
import hashlib
import hmac
import json
import math
import os
import re
import secrets
import sqlite3
import time
from collections.abc import Awaitable, Coroutine
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Literal
from urllib.parse import urlencode, urlsplit

from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field, PositiveInt, ValidationError, field_validator, model_validator
from starlette.datastructures import UploadFile

from src import dashboard_auth, gitea_proxy, passkeys
from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
from src.completed_filed_review_store import CompletedFiledReviewStore
from src.compression import NegotiatedGZipMiddleware
from src.gitea_proxy import (
    activity_events,
    current_user,
    is_requested_review,
    issues,
    mark_notification_read,
    notification_detail,
    notifications,
    pull_requests,
    pull_review_detail,
    repos,
)
from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy
from src.image_sanitizer import sanitize_image
from src.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, client_source
from src.live_snapshot_store import (
    LiveSnapshotMetadata,
    LiveSnapshotState,
    LiveSnapshotStore,
    RefreshLeaseLost,
)
from src.models import Issue, Milestone, PullRequest, Repo, User
from src.passkey_store import PasskeyStore
from src.push_notifications import (
    PushConfiguration,
    dispatch_deadline_reminders,
    dispatch_start_day_reminders,
    dispatch_unread_updates,
    send_web_push,
)
from src.push_endpoint_policy import UnsafePushEndpoint, validate_public_push_endpoint
from src.push_subscription_store import build_push_subscription_store
from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
from src.saved_search_store import SavedSearchConflict, SavedSearchStore
from src.unfiled_draft_store import (
    UnfiledDraftConflict,
    UnfiledDraftEncryptionError,
    UnfiledDraftStore,
    decode_unfiled_draft_encryption_key,
    decode_unfiled_draft_encryption_keyring,
)
from src.security_event_store import SecurityEventStore, SecurityEventStoreError
from src.suggestion_engine import compute
from src.later_store import LaterStore
from src.today_store import (
    TodayPlanFull, TodayPromotionConflict, TodaySessionConflict, TodayStore,
    TomorrowPlanConflict, TomorrowPlanNotDue, WeekPlanConflict,
)
from src.state_encryption import PrivateStateEncryptionError
from src.views import FRONTEND_BUILD, router as frontend_router


async def _drain_authored_action_operations() -> None:
    for key, operation in list(_authored_action_operations.items()):
        if operation[1].done():
            _authored_action_operations.pop(key, None)
    tasks = {
        operation[1]
        for operation in _authored_action_operations.values()
        if not operation[1].done()
    }
    if not tasks:
        return
    try:
        await asyncio.wait_for(
            asyncio.gather(*tasks, return_exceptions=True),
            timeout=AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS,
        )
    except TimeoutError:
        pass
    finally:
        for key, operation in list(_authored_action_operations.items()):
            if operation[1] in tasks and operation[1].done():
                _authored_action_operations.pop(key, None)


async def _push_channel_loop(dispatch, *, interval: float) -> None:
    """Run one push channel on fixed ticks without overlapping or catch-up bursts."""
    loop = asyncio.get_running_loop()
    next_tick = loop.time() + interval
    while True:
        await asyncio.sleep(max(0.0, next_tick - loop.time()))
        try:
            await dispatch()
        except asyncio.CancelledError:
            raise
        except Exception:
            pass
        now = loop.time()
        elapsed_intervals = max(1, int((now - next_tick) // interval) + 1)
        next_tick += elapsed_intervals * interval


async def _start_day_plan_snapshot() -> dict:
    user = await current_user()
    login = user.get("login") if isinstance(user, dict) else None
    if not isinstance(login, str) or not login:
        return {}
    return await asyncio.to_thread(_today_store().get_tomorrow, login)


async def _push_poll_loop() -> None:
    interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")))
    deadline_interval = max(
        60.0, float(os.getenv("STACKCHAIN_DEADLINE_POLL_SECONDS", "600"))
    )
    send_timeout = max(
        1.0, float(os.getenv("STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS", "10"))
    )
    lease_seconds = max(
        send_timeout + 5.0,
        float(os.getenv("STACKCHAIN_PUSH_LEASE_SECONDS", "60")),
    )
    max_concurrency = max(
        1, int(os.getenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "8"))
    )
    max_individual_notifications = max(
        0,
        int(os.getenv("STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS", "3")),
    )
    async def dispatch_unread() -> None:
        await dispatch_unread_updates(
                _push_subscription_store,
                _push_configuration(),
                gitea_proxy.unread_notification_snapshot,
                session_statuses=dashboard_auth.managed_session_statuses,
                lease_seconds=lease_seconds,
                send_timeout_seconds=send_timeout,
                max_concurrency=max_concurrency,
                max_individual_notifications=max_individual_notifications,
            )

    async def dispatch_deadlines() -> None:
        await dispatch_deadline_reminders(
                _push_subscription_store,
                _push_configuration(),
                gitea_proxy.assigned_issue_snapshot,
                session_statuses=dashboard_auth.managed_session_statuses,
                send_timeout_seconds=send_timeout,
                lease_seconds=lease_seconds,
                max_concurrency=max_concurrency,
            )

    async def dispatch_start_day() -> None:
        await dispatch_start_day_reminders(
                _push_subscription_store,
                _push_configuration(),
                _start_day_plan_snapshot,
                session_statuses=dashboard_auth.managed_session_statuses,
                send_timeout_seconds=send_timeout,
                lease_seconds=lease_seconds,
                max_concurrency=max_concurrency,
            )

    channel_tasks = (
        asyncio.create_task(_push_channel_loop(dispatch_unread, interval=interval)),
        asyncio.create_task(
            _push_channel_loop(dispatch_deadlines, interval=deadline_interval)
        ),
        asyncio.create_task(
            _push_channel_loop(dispatch_start_day, interval=deadline_interval)
        ),
    )
    try:
        done, _pending = await asyncio.wait(
            channel_tasks, return_when=asyncio.FIRST_COMPLETED
        )
        await next(iter(done))
    finally:
        for task in channel_tasks:
            if not task.done():
                task.cancel()
        await asyncio.gather(*channel_tasks, return_exceptions=True)


async def _readiness_monitor() -> None:
    while True:
        await _check_readiness()
        await asyncio.sleep(READINESS_INTERVAL_SECONDS)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    global _live_snapshot_task, _available_issue_snapshot_task, _push_poll_task
    global _readiness_task
    gitea_proxy.start_client()
    _readiness_task = asyncio.create_task(_readiness_monitor())
    if _push_configuration().enabled:
        _push_poll_task = asyncio.create_task(_push_poll_loop())
    try:
        yield
    finally:
        live_task = _live_snapshot_task
        available_task = _available_issue_snapshot_task
        push_task = _push_poll_task
        readiness_task = _readiness_task
        for task in (live_task, available_task, push_task, readiness_task):
            if task is not None and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
        try:
            await _drain_authored_action_operations()
            await gitea_proxy.stop_client()
        finally:
            if _live_snapshot_task is live_task:
                _live_snapshot_task = None
            if _available_issue_snapshot_task is available_task:
                _available_issue_snapshot_task = None
            if _push_poll_task is push_task:
                _push_poll_task = None
            if _readiness_task is readiness_task:
                _readiness_task = None


app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
app.add_middleware(RequestBodyLimitMiddleware, limit_for=request_body_limit)
app.add_middleware(NegotiatedGZipMiddleware, minimum_size=1_024)
CONTEXT_TIMEOUT_SECONDS = 5.0
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0
READINESS_INTERVAL_SECONDS = max(
    READINESS_TIMEOUT_SECONDS,
    float(os.getenv("STACKCHAIN_READINESS_INTERVAL_SECONDS", "30")),
)
READINESS_MAX_AGE_SECONDS = max(
    READINESS_INTERVAL_SECONDS + READINESS_TIMEOUT_SECONDS + 5.0,
    float(os.getenv("STACKCHAIN_READINESS_MAX_AGE_SECONDS", "65")),
)


@dataclass(frozen=True)
class ReadinessState:
    checked_at: float
    ready: bool
    login: str | None = None
    error: str | None = None
    timed_out: bool = False


_readiness_state: ReadinessState | None = None


async def _check_readiness() -> None:
    global _readiness_state
    checked_at = time.monotonic()
    try:
        user = await asyncio.wait_for(current_user(), timeout=READINESS_TIMEOUT_SECONDS)
        if not isinstance(user, dict) or not user.get("login"):
            raise ReadinessPayloadError(
                "Gitea current-user response did not include a login"
            )
    except Exception as exc:
        timed_out = isinstance(exc, TimeoutError)
        error = (
            f"Gitea readiness check timed out after {READINESS_TIMEOUT_SECONDS:g}s"
            if timed_out
            else (
                str(exc)
                if isinstance(exc, ReadinessPayloadError)
                else "Gitea readiness check is temporarily unavailable"
            )
        )
        _readiness_state = ReadinessState(
            checked_at=checked_at,
            ready=False,
            error=error,
            timed_out=timed_out,
        )
        return
    _readiness_state = ReadinessState(
        checked_at=checked_at,
        ready=True,
        login=user["login"],
    )
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES = 256
AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS = 600.0
AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES = 256
AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS = float(
    os.getenv("STACKCHAIN_IDEMPOTENCY_LOCK_TIMEOUT_SECONDS", "0.1")
)
AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS = float(
    os.getenv("STACKCHAIN_AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", "5.0")
)
NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
WORK_PAGE_TIMEOUT_SECONDS = 5.0
GLOBAL_SEARCH_TIMEOUT_SECONDS = 3.0
NOTIFICATION_DETAIL_TIMEOUT_SECONDS = 5.0
BULK_NOTIFICATION_CONCURRENCY = 5
BULK_NOTIFICATION_DEADLINE_SECONDS = 6.0
LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
LIVE_SNAPSHOT_RETRY_BASE_SECONDS = 5.0
LIVE_SNAPSHOT_RETRY_MAX_SECONDS = 60.0
AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS = 15.0
AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS = 5.0
AVAILABLE_ISSUE_SNAPSHOT_LEASE_SECONDS = WORK_PAGE_TIMEOUT_SECONDS + 1.0
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
_live_snapshot_task: asyncio.Task | None = None
_push_poll_task: asyncio.Task | None = None
_readiness_task: asyncio.Task | None = None
_live_snapshot_value: dict | None = None
_live_snapshot_created_at: float | None = None
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
_live_section_created_at: dict[str, float | None] = {
    section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_failure_count: dict[str, int] = {
    section: 0 for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_retry_at: dict[str, float | None] = {
    section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_snapshot_refreshing_sections: set[str] = set()
_live_section_revisions: dict[str, int] = {
    section: 0 for section in LIVE_SNAPSHOT_SECTIONS
}
_live_revision_generation = secrets.token_hex(8)
_read_notification_ids: set[int] = set()
_authored_action_operations: dict[
    str, tuple[tuple[Any, ...], asyncio.Task, float]
] = {}
_state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"))
_live_snapshot_clock = time.time
_live_snapshot_store = LiveSnapshotStore(
    os.getenv("STACKCHAIN_LIVE_SNAPSHOT_DB", str(_state_dir / "live-snapshot.sqlite3")),
    clock=lambda: _live_snapshot_clock(),
)
_idempotency_ledger = IdempotencyLedger(
    os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")),
    ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS,
    max_entries=(
        AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES
        + ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES
    ),
    lock_timeout_seconds=AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS,
)
_available_issue_snapshot_task: asyncio.Task | None = None
_available_issue_snapshot_lock = asyncio.Lock()
_available_issue_snapshot_value: list[dict] | None = None
_available_issue_snapshot_created_at: float | None = None
_available_issue_snapshot_retry_at: float | None = None
_available_issue_snapshot_store = AvailableIssueSnapshotStore(
    os.getenv(
        "STACKCHAIN_AVAILABLE_ISSUE_SNAPSHOT_DB",
        str(_state_dir / "available-issue-snapshot.sqlite3"),
    )
)
_push_subscription_store = build_push_subscription_store(
    os.getenv("STACKCHAIN_PUSH_DB", str(_state_dir / "push-subscriptions.sqlite3")),
    push_enabled=all(
        os.getenv(name, "").strip()
        for name in (
            "STACKCHAIN_VAPID_PUBLIC_KEY",
            "STACKCHAIN_VAPID_PRIVATE_KEY",
            "STACKCHAIN_VAPID_SUBJECT",
        )
    ),
)


def _push_configuration() -> PushConfiguration:
    return PushConfiguration(
        os.getenv("STACKCHAIN_VAPID_PUBLIC_KEY", "").strip(),
        os.getenv("STACKCHAIN_VAPID_PRIVATE_KEY", "").strip(),
        os.getenv("STACKCHAIN_VAPID_SUBJECT", "").strip(),
    )


class ContextPayloadError(ValueError):
    """Raised when Gitea returns a structurally invalid context payload."""


class ReadinessPayloadError(ValueError):
    """Raised when Gitea returns a structurally invalid readiness payload."""


class DashboardSignIn(BaseModel):
    access_token: str = Field(min_length=1, max_length=1_024)
    device_label: str = Field(default="This device", min_length=1, max_length=64)

    @field_validator("device_label")
    @classmethod
    def normalize_device_label(cls, value: str) -> str:
        normalized = " ".join(value.split())
        if not normalized:
            raise ValueError("device label cannot be blank")
        return normalized


class PushSubscriptionPayload(BaseModel):
    endpoint: str = Field(min_length=12, max_length=2_048, pattern=r"^https://")
    keys: dict[str, str]

    @field_validator("keys")
    @classmethod
    def validate_push_keys(cls, value: dict[str, str]) -> dict[str, str]:
        if set(value) != {"p256dh", "auth"} or any(
            not item or len(item) > 1_024 for item in value.values()
        ):
            raise ValueError("Valid Web Push keys are required")
        return value


class DeadlineReminderPayload(BaseModel):
    enabled: bool
    timezone: str = Field(min_length=1, max_length=64)
    reminder_hour: int = Field(default=9, ge=0, le=23)
    reminder_days: Literal[0, 2, 7] = 2

    @field_validator("timezone")
    @classmethod
    def validate_timezone(cls, value: str) -> str:
        from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

        try:
            ZoneInfo(value)
        except ZoneInfoNotFoundError as error:
            raise ValueError("Valid IANA timezone required") from error
        return value


class StartDayReminderPayload(BaseModel):
    enabled: bool
    timezone: str = Field(min_length=1, max_length=64)
    reminder_hour: int = Field(default=9, ge=0, le=23)

    @field_validator("timezone")
    @classmethod
    def validate_timezone(cls, value: str) -> str:
        from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

        try:
            ZoneInfo(value)
        except ZoneInfoNotFoundError as error:
            raise ValueError("Valid IANA timezone required") from error
        return value


StepUpAction = Literal[
    "merge_pull",
    "submit_pull_review",
    "close_issue",
    "delete_comment",
    "log_recap_time",
    "revoke_device",
    "revoke_all_sessions",
    "enroll_passkey",
    "revoke_passkey",
]


class FreshAuthorization(BaseModel):
    access_token: str = Field(min_length=1, max_length=1_024)
    action: StepUpAction
    target: str = Field(min_length=1, max_length=255)


class PasskeyCeremony(BaseModel):
    challenge: str = Field(min_length=20, max_length=200)
    credential: dict = Field()


class PasskeyAuthentication(PasskeyCeremony):
    device_label: str = Field(default="This device", min_length=1, max_length=64)
    action: Literal[
        "sign_in", "merge_pull", "close_issue", "revoke_device", "revoke_all_sessions"
    ] = "sign_in"
    target: str = Field(default="dashboard", min_length=1, max_length=255)


class PasskeyAuthorizationTarget(BaseModel):
    action: StepUpAction
    target: str = Field(min_length=1, max_length=255)


class PasskeyAuthorization(PasskeyCeremony, PasskeyAuthorizationTarget):
    pass


def _passkey_store() -> PasskeyStore:
    state_dir = os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state")
    database = os.getenv(
        "STACKCHAIN_SESSION_DB", os.path.join(state_dir, "sessions.sqlite3")
    )
    return PasskeyStore(
        database,
        clock=time.time,
        max_challenges=int(os.getenv("STACKCHAIN_PASSKEY_MAX_CHALLENGES", "10000")),
        max_challenges_per_source=int(
            os.getenv("STACKCHAIN_PASSKEY_MAX_CHALLENGES_PER_SOURCE", "10")
        ),
    )


def _security_event_store() -> SecurityEventStore:
    state_dir = os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state")
    database = os.getenv(
        "STACKCHAIN_SECURITY_EVENT_DB",
        os.path.join(state_dir, "security-events.sqlite3"),
    )
    return SecurityEventStore(database, clock=time.time)


def _passkey_relying_party(request: Request) -> tuple[str, str]:
    canonical_origin = dashboard_auth.public_origin()
    canonical_host = urlsplit(canonical_origin).hostname if canonical_origin else None
    rp_id = os.getenv("STACKCHAIN_PASSKEY_RP_ID", canonical_host or request.url.hostname or "")
    origin = os.getenv(
        "STACKCHAIN_PASSKEY_ORIGIN",
        canonical_origin or f"{request.url.scheme}://{request.url.netloc}",
    )
    return rp_id, origin


def _login_attempt_store() -> LoginAttemptStore:
    state_dir = os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state")
    return LoginAttemptStore(
        os.getenv(
            "STACKCHAIN_LOGIN_ATTEMPT_DB",
            os.path.join(state_dir, "login-attempts.sqlite3"),
        ),
        clock=time.time,
        max_failures=int(os.getenv("STACKCHAIN_LOGIN_MAX_FAILURES", "5")),
        window_seconds=int(os.getenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "300")),
        max_entries=int(os.getenv("STACKCHAIN_LOGIN_MAX_ENTRIES", "10000")),
    )


async def _require_step_up(
    request: Request,
    grant: str | None,
    *,
    action: str,
    target: str,
) -> None:
    if dashboard_auth.mode() != dashboard_auth.OPERATOR_MODE:
        return
    try:
        valid = bool(grant) and await dashboard_auth.consume_step_up(
            grant, request.state.dashboard_session, action=action, target=target
        )
    except dashboard_auth.SessionStoreError:
        raise HTTPException(
            status_code=503, detail="Session registry is temporarily unavailable"
        )
    if not valid:
        return_payload = {
            "detail": "Fresh authorization required",
            "code": "step_up_required",
            "action": action,
            "target": target,
        }
        raise HTTPException(status_code=428, detail=return_payload)


class NotificationReadBatch(BaseModel):
    ids: list[PositiveInt] = Field(min_length=1, max_length=50)


class TodayOperation(BaseModel):
    operation_id: str = Field(min_length=1, max_length=100)
    action: Literal["add", "remove", "move", "configure", "rollover", "activate"]
    item_id: str = Field(min_length=1, max_length=500)
    direction: Literal["up", "down"] | None = None
    base_revision: int | None = Field(default=None, ge=0)
    capacity_minutes: int | None = Field(default=None, ge=15, le=1440)
    estimates: dict[str, int] = Field(default_factory=dict, max_length=5)
    plan_date: str | None = Field(default=None, min_length=10, max_length=10)
    timezone: str | None = Field(default=None, min_length=1, max_length=100)
    ids: list[str] = Field(default_factory=list, max_length=5)
    activation_state: Literal["coaching", "complete"] | None = None

    @model_validator(mode="after")
    def validate_action_fields(self):
        if self.action == "move" and self.direction is None:
            raise ValueError("move requires a direction")
        if self.action != "move" and self.direction is not None:
            raise ValueError("direction is only valid for move")
        if self.action not in {"configure", "rollover"} and (
            self.capacity_minutes is not None or self.estimates
        ):
            raise ValueError("capacity and estimates are only valid for configure or rollover")
        if self.action == "rollover":
            if self.item_id != "plan" or self.plan_date is None or self.timezone is None:
                raise ValueError("rollover requires plan date, timezone, and plan item")
            if len(set(self.ids)) != len(self.ids):
                raise ValueError("rollover IDs must be unique")
        elif self.plan_date is not None or self.timezone is not None or self.ids:
            raise ValueError("date, timezone, and IDs are only valid for rollover")
        if self.action == "activate":
            if self.item_id != "first-task" or self.activation_state is None:
                raise ValueError("activate requires a first-task activation state")
        elif self.activation_state is not None:
            raise ValueError("activation state is only valid for activate")
        if any(not item_id or len(item_id) > 500 or minutes < 5 or minutes > 1440
               for item_id, minutes in self.estimates.items()):
            raise ValueError("estimates must use bounded item IDs and minutes")
        return self

    def model_dump(self, *args, **kwargs):
        data = super().model_dump(*args, **kwargs)
        if self.action != "rollover":
            data.pop("plan_date", None)
            data.pop("timezone", None)
            data.pop("ids", None)
        if self.action != "activate":
            data.pop("activation_state", None)
        return data


class TodayRecapItem(BaseModel):
    identity: str = Field(min_length=1, max_length=500)
    estimate_minutes: int | None = Field(default=None, ge=5, le=1440)
    actual_minutes: int = Field(ge=0, le=1440)


class TodayRecap(BaseModel):
    session_id: str = Field(min_length=1, max_length=100)
    items: list[TodayRecapItem] = Field(min_length=1, max_length=20)


class TodaySessionUpdate(BaseModel):
    base_revision: int = Field(ge=0)
    device_id: str = Field(min_length=1, max_length=100)
    identity: str = Field(max_length=500)
    elapsed_ms: int = Field(ge=0, le=7 * 24 * 60 * 60 * 1000)
    running: bool
    break_deadline_at: int | None = Field(default=None, ge=0, le=10_000_000_000_000)

    @model_validator(mode="after")
    def validate_break_is_paused(self):
        if self.running and self.break_deadline_at is not None:
            raise ValueError("break must remain paused")
        return self


class TodayRecapTimeLog(TodayRecap):
    log_identities: list[str] = Field(min_length=1, max_length=20)


def _recap_time_log_target(payload: TodayRecapTimeLog) -> str:
    selected_minutes = {
        item.identity: item.actual_minutes
        for item in payload.items
        if item.identity in payload.log_identities
    }
    material = "\0".join(
        [payload.session_id]
        + [f"{identity}\0{selected_minutes.get(identity, '')}" for identity in sorted(payload.log_identities)]
    )
    return f"today-recap:{hashlib.sha256(material.encode()).hexdigest()[:32]}"


class LaterOperation(BaseModel):
    operation_id: str = Field(min_length=1, max_length=100)
    action: Literal["defer", "restore"]
    item_id: str = Field(min_length=1, max_length=500)
    wake_at: str | None = Field(default=None, max_length=100)
    handoff: Literal["today"] | None = None
    base_revision: int | None = Field(default=None, ge=0)


class TodayOperationBatch(BaseModel):
    operations: list[TodayOperation] = Field(min_length=1, max_length=50)


class TomorrowPlanUpdate(BaseModel):
    base_revision: int = Field(ge=0)
    ids: list[str] = Field(max_length=5)
    capacity_minutes: int | None = Field(default=None, ge=15, le=1440)
    estimates: dict[str, int] = Field(default_factory=dict, max_length=5)
    plan_date: str = Field(min_length=10, max_length=10)
    timezone: str = Field(min_length=1, max_length=100)

    @model_validator(mode="after")
    def validate_plan(self):
        if len(set(self.ids)) != len(self.ids):
            raise ValueError("Tomorrow IDs must be unique")
        if any(not item_id or len(item_id) > 500 for item_id in self.ids):
            raise ValueError("Tomorrow IDs must be bounded")
        if any(not item_id or len(item_id) > 500 or minutes < 5 or minutes > 1440
               for item_id, minutes in self.estimates.items()):
            raise ValueError("estimates must use bounded item IDs and minutes")
        return self


class TomorrowPromotion(BaseModel):
    promotion_id: str = Field(min_length=1, max_length=100)
    tomorrow_revision: int = Field(ge=0)
    today_revision: int = Field(ge=0)


class WeekPlanDay(BaseModel):
    plan_date: str = Field(min_length=10, max_length=10)
    ids: list[str] = Field(max_length=5)
    capacity_minutes: int | None = Field(default=None, ge=15, le=1440)
    estimates: dict[str, int] = Field(default_factory=dict, max_length=5)


class WeekPlanUpdate(BaseModel):
    base_revision: int = Field(ge=0)
    days: list[WeekPlanDay] = Field(max_length=7)
    timezone: str = Field(min_length=1, max_length=100)


class WeekPromotion(BaseModel):
    promotion_id: str = Field(min_length=1, max_length=100)
    week_revision: int = Field(ge=0)
    plan_date: str = Field(min_length=10, max_length=10)
    today_revision: int = Field(ge=0)


class LaterOperationBatch(BaseModel):
    operations: list[LaterOperation] = Field(min_length=1, max_length=50)


class SavedSearchView(BaseModel):
    id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")
    name: str = Field(min_length=1, max_length=60)
    query: str = Field(min_length=2, max_length=200)
    kind: Literal["all", "issue", "pull"] = "all"
    state: Literal["all", "open", "closed"] = "all"
    repository: str = Field(default="", max_length=200)


class SavedSearchCollection(BaseModel):
    revision: int = Field(ge=0)
    views: list[SavedSearchView] = Field(max_length=20)


class CompletedFiledReviewReceipt(BaseModel):
    repository: str = Field(
        min_length=3,
        max_length=200,
        pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$",
    )
    number: PositiveInt
    updated_at: str = Field(min_length=1, max_length=100)


class CompletedFiledReviewBatch(BaseModel):
    receipts: list[CompletedFiledReviewReceipt] = Field(max_length=200)


class UnfiledDraftBlocker(BaseModel):
    repository: str = Field(min_length=3, max_length=200)
    number: int = Field(ge=1)
    title: str = Field(default="", max_length=255)


class UnfiledDraftEvidence(BaseModel):
    filename: str = Field(min_length=1, max_length=255)
    content_type: Literal["image/png", "image/jpeg", "image/webp"]
    note: str = Field(default="", max_length=240)
    data: str = Field(max_length=14_000_000)


class UnfiledDraftFilingPlan(BaseModel):
    repository: str = Field(
        min_length=3,
        max_length=200,
        pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$",
    )
    label_ids: list[PositiveInt] = Field(default_factory=list, max_length=20)
    milestone_id: PositiveInt | None = None
    due_date: str | None = Field(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$", max_length=10)
    template_name: str | None = Field(default=None, min_length=1, max_length=80)
    template_id: str | None = Field(default=None, min_length=1, max_length=80)
    captured_body: str | None = Field(default=None, min_length=1, max_length=10_000)
    assignee: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_.-]+$", max_length=255)
    assignee_name: str | None = Field(default=None, min_length=1, max_length=255)
    unassigned: bool = False
    estimate_minutes: int | None = Field(default=None, ge=5, le=1440)
    completion_intent: Literal["create", "create-and-start"] | None = None

    @field_validator("due_date")
    @classmethod
    def validate_filing_due_date(cls, value: str | None) -> str | None:
        if value is not None:
            datetime.strptime(value, "%Y-%m-%d")
        return value

    @model_validator(mode="after")
    def require_consistent_owner_intent(self):
        if self.unassigned and self.assignee is not None:
            raise ValueError("assignee and unassigned cannot be requested together")
        if self.assignee is None and self.assignee_name is not None:
            raise ValueError("assignee name requires an assignee")
        return self


class UnfiledDraft(BaseModel):
    id: str = Field(min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_-]+$")
    title: str = Field(default="", max_length=255)
    body: str = Field(default="", max_length=10_000)
    saved_at: int = Field(ge=0)
    filing_plan: UnfiledDraftFilingPlan | None = None
    blockers: list[UnfiledDraftBlocker] = Field(default_factory=list, max_length=5)
    evidence: list[UnfiledDraftEvidence] = Field(default_factory=list, max_length=5)

    @model_validator(mode="after")
    def require_title_or_evidence(self):
        self.title = self.title.strip()
        if not self.title and not self.evidence:
            raise ValueError("title or evidence is required")
        return self


class UnfiledDraftCollection(BaseModel):
    revision: int = Field(ge=0)
    drafts: list[UnfiledDraft] = Field(max_length=20)


class NotificationLaterRequest(BaseModel):
    wake_at: str = Field(min_length=1, max_length=100)


class NotificationReply(BaseModel):
    body: str = Field(min_length=1, max_length=10_000)

    @field_validator("body")
    @classmethod
    def strip_body(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("reply must not be blank")
        return value


class IssueComment(BaseModel):
    body: str = Field(min_length=1, max_length=10_000)

    @field_validator("body")
    @classmethod
    def strip_body(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("comment must not be blank")
        return value


def _validate_attachment_metadata(filename: str, content_type: str) -> None:
    expected = {
        "image/png": {"png"},
        "image/jpeg": {"jpg", "jpeg"},
        "image/webp": {"webp"},
    }
    if content_type not in expected:
        raise ValueError("attachment content type must be PNG, JPEG, or WebP")
    if "/" in filename or "\\" in filename or any(ord(character) < 32 for character in filename):
        raise ValueError("attachment filename must be a plain file name")
    extension = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
    if extension not in expected[content_type]:
        raise ValueError("attachment filename must match the selected image type")


def _validate_attachment_content(content_type: str, content: bytes) -> bytes:
    if len(content) > 2 * 1024 * 1024:
        raise ValueError("screenshot must be 2 MB or smaller")
    signatures = {
        "image/png": content.startswith(b"\x89PNG\r\n\x1a\n"),
        "image/jpeg": content.startswith(b"\xff\xd8\xff"),
        "image/webp": content.startswith(b"RIFF") and content[8:12] == b"WEBP",
    }
    if not signatures.get(content_type, False):
        raise ValueError("file contents do not match the selected image type")
    return content


class IssueAttachment(BaseModel):
    filename: str = Field(min_length=1, max_length=255)
    content_type: Literal["image/png", "image/jpeg", "image/webp"]
    data: str = Field(min_length=1, max_length=2_800_000)

    @model_validator(mode="after")
    def validate_filename(self):
        _validate_attachment_metadata(self.filename, self.content_type)
        return self

    def content(self) -> bytes:
        try:
            content = base64.b64decode(self.data, validate=True)
        except (ValueError, binascii.Error) as exc:
            raise ValueError("attachment data must be valid base64") from exc
        return _validate_attachment_content(self.content_type, content)


def _validate_binary_attachment(filename: str, content_type: str, content: bytes) -> bytes:
    _validate_attachment_metadata(filename, content_type)
    return _validate_attachment_content(content_type, content)


async def _sanitize_attachment(content_type: str, content: bytes) -> bytes:
    sanitized = await asyncio.to_thread(sanitize_image, content_type, content)
    if len(sanitized) > 2 * 1024 * 1024:
        raise ValueError("sanitized screenshot must be 2 MB or smaller")
    return sanitized


class IssueCreation(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    body: str = Field(default="", max_length=10_000)
    unassigned: bool = False
    assignee: str | None = Field(
        default=None, pattern=r"^[A-Za-z0-9_.-]+$", max_length=255
    )
    label_ids: list[int] = Field(default_factory=list, max_length=20)
    milestone_id: PositiveInt | None = None
    due_date: str | None = Field(
        default=None,
        pattern=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$",
        max_length=20,
    )

    @field_validator("title")
    @classmethod
    def strip_title(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("title must not be blank")
        return value

    @field_validator("body")
    @classmethod
    def strip_issue_body(cls, value: str) -> str:
        return value.strip()

    @model_validator(mode="after")
    def require_one_owner_intent(self):
        if self.unassigned and self.assignee is not None:
            raise ValueError("assignee and unassigned cannot be requested together")
        return self

    @field_validator("due_date")
    @classmethod
    def validate_due_date(cls, value: str | None) -> str | None:
        if value is not None:
            datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
        return value


class IssueContentUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    body: str = Field(default="", max_length=10_000)
    expected_updated_at: str = Field(min_length=1, max_length=64)

    @field_validator("title")
    @classmethod
    def strip_title(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("title must not be blank")
        return value

    @field_validator("body")
    @classmethod
    def strip_body(cls, value: str) -> str:
        return value.strip()


class IssueLabelUpdate(BaseModel):
    label_ids: list[PositiveInt] = Field(max_length=20)


class IssueDueDateUpdate(BaseModel):
    due_date: str | None = Field(
        default=None,
        pattern=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$",
        max_length=20,
    )


class IssueMilestoneUpdate(BaseModel):
    milestone_id: PositiveInt | None = None


class IssueReleasePlanUpdate(BaseModel):
    milestone_id: PositiveInt
    due_date: str | None = Field(
        default=None,
        pattern=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$",
        max_length=20,
    )


class IssueHandoff(BaseModel):
    recipient: str = Field(
        min_length=1, max_length=255, pattern=r"^[A-Za-z0-9_.-]+$"
    )


class IssueReassignment(IssueHandoff):
    expected_assignees: list[str] = Field(min_length=1, max_length=10)

    @field_validator("expected_assignees")
    @classmethod
    def validate_expected_assignees(cls, values: list[str]) -> list[str]:
        if any(
            not isinstance(value, str)
            or not re.fullmatch(r"[A-Za-z0-9_.-]+", value)
            or len(value) > 255
            for value in values
        ) or len(set(values)) != len(values):
            raise ValueError("expected assignees must be unique valid logins")
        return values


class IssueBlockerUpdate(BaseModel):
    repository: str = Field(
        min_length=3,
        max_length=255,
        pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$",
    )
    number: PositiveInt


class IssueBlockerDesiredState(IssueBlockerUpdate):
    present: bool


class PullReviewComment(BaseModel):
    path: str = Field(min_length=1, max_length=1_000)
    body: str = Field(min_length=1, max_length=10_000)
    new_position: PositiveInt | None = None
    old_position: PositiveInt | None = None

    @field_validator("path", "body")
    @classmethod
    def strip_comment_text(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("inline comment values must not be blank")
        return value

    @model_validator(mode="after")
    def require_one_position(self):
        if (self.new_position is None) == (self.old_position is None):
            raise ValueError("inline comments require exactly one old or new position")
        return self


class PullReviewSubmission(BaseModel):
    decision: str
    body: str = Field(max_length=10_000)
    expected_head_sha: str
    comments: list[PullReviewComment] = Field(default_factory=list, max_length=50)

    @field_validator("decision")
    @classmethod
    def validate_decision(cls, value: str) -> str:
        if value not in {"comment", "approve", "request_changes"}:
            raise ValueError("unsupported review decision")
        return value


class PullMergeSubmission(BaseModel):
    expected_head_sha: str = Field(min_length=1, max_length=128)


async def _run_idempotent_authored_action(
    operation: Coroutine[Any, Any, Any],
    *,
    idempotency_key: str | None,
    fingerprint: tuple[Any, ...],
    timeout: float,
):
    if not idempotency_key:
        return await asyncio.wait_for(operation, timeout=timeout)

    reservation = await asyncio.to_thread(
        _idempotency_ledger.reserve, idempotency_key, fingerprint
    )
    if reservation.state == "conflict":
        operation.close()
        raise HTTPException(status_code=409, detail="Idempotency key already used")
    if reservation.state == "completed":
        operation.close()
        return reservation.response
    if reservation.state == "busy":
        operation.close()
        raise HTTPException(
            status_code=503,
            detail="Authored action queue is busy; please retry",
            headers={"Retry-After": "1"},
        )
    existing = _authored_action_operations.get(idempotency_key)
    if reservation.state == "uncertain" and (
        existing is None or existing[0] != fingerprint
    ):
        operation.close()
        raise HTTPException(
            status_code=422,
            detail={
                "code": "delivery_uncertain",
                "message": (
                    "Delivery could not be confirmed. Verify it was not posted before retrying."
                ),
            },
        )

    if reservation.state in {"pending", "uncertain"}:
        operation.close()
        if existing is None or existing[0] != fingerprint:
            raise HTTPException(
                status_code=503,
                detail=(
                    "This action may still be completing; verify its result before retrying"
                ),
                headers={"Retry-After": "5"},
            )
        task = existing[1]
    else:
        async def persist_result():
            result = await operation
            await asyncio.to_thread(_idempotency_ledger.complete, idempotency_key, result)
            return result

        task = asyncio.create_task(persist_result())
        _authored_action_operations[idempotency_key] = (
            fingerprint,
            task,
            time.monotonic(),
        )

        def discard_completed(completed: asyncio.Task) -> None:
            existing_operation = _authored_action_operations.get(idempotency_key)
            if existing_operation is not None and existing_operation[1] is completed:
                _authored_action_operations.pop(idempotency_key, None)

        task.add_done_callback(discard_completed)
    try:
        return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
    except IdempotencyLedgerBusy:
        raise HTTPException(
            status_code=503,
            detail="This action may have completed; verify its result before retrying",
            headers={"Retry-After": "5"},
        )


def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
    user_model = User(
        id=user_data["id"],
        login=user_data["login"],
        full_name=user_data.get("full_name") or "",
        email=user_data.get("email") or "",
    )
    repo_models = [
        Repo(
            id=r["id"], name=r["name"], full_name=r["full_name"],
            description=r.get("description") or "", url=r["html_url"],
            updated_at=r.get("updated_at", ""),
        )
        for r in (repo_data or [])[:50]
        if isinstance(r, dict)
        and all(field in r for field in ("id", "name", "full_name", "html_url"))
    ]
    issue_models = [
        Issue(
            id=i["id"], number=i["number"], title=i["title"], state=i["state"],
            labels=[label.get("name", "") for label in (i.get("labels") or []) if isinstance(label, dict)],
            assignees=[assignee.get("login", "") for assignee in (i.get("assignees") or []) if isinstance(assignee, dict)],
            work_reasons=[reason for reason in (i.get("work_reasons") or []) if reason == "created_by_me"],
            repository=i["repository"].get("full_name", "") if isinstance(i.get("repository"), dict) else "",
            updated_at=i.get("updated_at") or "",
            due_date=i.get("due_date") if isinstance(i.get("due_date"), str) else None,
            milestone=Milestone(**i["milestone"]) if isinstance(i.get("milestone"), dict)
            and isinstance(i["milestone"].get("id"), int)
            and isinstance(i["milestone"].get("title"), str) else None,
            url=i["html_url"],
        )
        for i in (issues_data or [])[:50]
        if isinstance(i, dict)
        and all(field in i for field in ("id", "number", "title", "state", "html_url"))
    ]
    pr_models = [
        PullRequest(
            id=p["id"], number=p["number"], title=p["title"], state=p["state"],
            user=p["user"].get("login", "") if isinstance(p.get("user"), dict) else "",
            labels=[label.get("name", "") for label in (p.get("labels") or []) if isinstance(label, dict)],
            assignees=[assignee.get("login", "") for assignee in (p.get("assignees") or []) if isinstance(assignee, dict)],
            work_reasons=[reason for reason in (p.get("work_reasons") or []) if reason in ("assigned_to_me", "review_requested")],
            repository=p["repository"].get("full_name", "") if isinstance(p.get("repository"), dict) else "",
            updated_at=p.get("updated_at") or "", url=p["html_url"],
        )
        for p in (prs_data or [])[:50]
        if isinstance(p, dict)
        and all(field in p for field in ("id", "number", "title", "state", "html_url"))
    ]
    payload = compute(user_model, repo_models, issue_models, pr_models).model_dump()
    repository_pagination = getattr(repo_data, "pagination", None)
    if isinstance(repository_pagination, dict):
        payload["repository_pagination"] = repository_pagination
    pagination = {}
    pagination.update(getattr(issues_data, "pagination", {}))
    pagination.update(getattr(prs_data, "pagination", {}))
    if pagination:
        payload["work_pagination"] = pagination
    return payload


def _normalize_work_items(stream: str, items: list[dict]) -> list[dict]:
    if stream in {"issue", "filed"}:
        return [
            Issue(
                id=item["id"], number=item["number"], title=item["title"],
                state=item["state"],
                labels=[label.get("name", "") for label in (item.get("labels") or []) if isinstance(label, dict)],
                assignees=[assignee.get("login", "") for assignee in (item.get("assignees") or []) if isinstance(assignee, dict)],
                work_reasons=[reason for reason in (item.get("work_reasons") or []) if reason == "created_by_me"],
                repository=item["repository"].get("full_name", "") if isinstance(item.get("repository"), dict) else "",
                updated_at=item.get("updated_at") or "",
                due_date=item.get("due_date") if isinstance(item.get("due_date"), str) else None,
                milestone=Milestone(**item["milestone"])
                if isinstance(item.get("milestone"), dict)
                and isinstance(item["milestone"].get("id"), int)
                and isinstance(item["milestone"].get("title"), str) else None,
                url=item["html_url"],
            ).model_dump()
            for item in items
            if all(field in item for field in ("id", "number", "title", "state", "html_url"))
        ]
    reason = "assigned_to_me" if stream == "pull" else "review_requested"
    return [
        PullRequest(
            id=item["id"], number=item["number"], title=item["title"],
            state=item["state"],
            user=item["user"].get("login", "") if isinstance(item.get("user"), dict) else "",
            labels=[label.get("name", "") for label in (item.get("labels") or []) if isinstance(label, dict)],
            assignees=[assignee.get("login", "") for assignee in (item.get("assignees") or []) if isinstance(assignee, dict)],
            work_reasons=[reason],
            repository=item["repository"].get("full_name", "") if isinstance(item.get("repository"), dict) else "",
            updated_at=item.get("updated_at") or "", url=item["html_url"],
        ).model_dump()
        for item in items
        if all(field in item for field in ("id", "number", "title", "state", "html_url"))
    ]

app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
app.include_router(frontend_router)


def _share_target_login_redirect(request: Request) -> str:
    search_values = request.query_params.getlist("search")
    preview_values = request.query_params.getlist("preview")
    if search_values or preview_values:
        allowed = {"search", "preview", "search_kind", "search_state", "search_repository"}
        kind_values = request.query_params.getlist("search_kind")
        state_values = request.query_params.getlist("search_state")
        repository_values = request.query_params.getlist("search_repository")
        if (
            set(request.query_params.keys()) - allowed
            or len(search_values) != 1
            or len(preview_values) != 1
            or len(kind_values) > 1
            or len(state_values) > 1
            or len(repository_values) > 1
            or len(search_values[0]) > 200
            or len(preview_values[0]) > 200
            or not re.fullmatch(
                r"(?:issue|pull):[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+:[1-9]\d*",
                preview_values[0],
            )
        ):
            return "login"
        continuation_values = [
            ("search", search_values[0].strip()), ("preview", preview_values[0]),
        ]
        if kind_values and kind_values[0] in {"all", "issue", "pull"}:
            continuation_values.append(("search_kind", kind_values[0]))
        if state_values and state_values[0] in {"all", "open", "closed"}:
            continuation_values.append(("search_state", state_values[0]))
        if repository_values and re.fullmatch(
            r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository_values[0]
        ):
            continuation_values.append(("search_repository", repository_values[0]))
        continuation = urlencode(continuation_values)
        return f"login?{urlencode({'continue': f'./?{continuation}'})}"
    image_markers = request.query_params.getlist("shared")
    if image_markers:
        if (
            set(request.query_params.keys()) != {"launch", "shared"}
            or request.query_params.getlist("launch") != ["new"]
            or image_markers != ["bundle"]
        ):
            return "login"
        continuation = urlencode([("launch", "new"), ("shared", "bundle")])
        return f"login?{urlencode({'continue': f'./?{continuation}'})}"
    limits = {"title": 200, "text": 8000, "url": 2048}
    if any(
        len(request.query_params.get(name, "")) > limit
        for name, limit in limits.items()
    ):
        return "login"
    shared = [
        (name, request.query_params[name])
        for name in limits
        if request.query_params.get(name)
    ]
    launch = request.query_params.get("launch", "")
    if launch in {"continue", "new", "agenda"}:
        shared.append(("launch", launch))
    if dashboard_auth.application_path(request) != "/" or not shared:
        return "login"
    continuation = f"./?{urlencode(shared)}"
    return f"login?{urlencode({'continue': continuation})}"


@app.middleware("http")
async def require_operator_session(request: Request, call_next):
    path = dashboard_auth.application_path(request)
    canonical_origin = dashboard_auth.public_origin()
    if dashboard_auth.mode() == dashboard_auth.OPERATOR_MODE and canonical_origin:
        expected_authority = urlsplit(canonical_origin).netloc
        if request.url.netloc.lower() != expected_authority:
            return JSONResponse(
                {"detail": "Request host does not match the configured public origin"},
                status_code=421,
                headers={"Cache-Control": "no-store"},
            )
    if path == "/healthz":
        return await call_next(request)

    if dashboard_auth.configuration_error() is not None:
        return JSONResponse(
            {"detail": "Dashboard authentication is not configured"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )

    if dashboard_auth.mode() == dashboard_auth.INSECURE_LOCAL_MODE:
        if not dashboard_auth.is_loopback_request(request):
            return JSONResponse(
                {"detail": "Insecure local mode requires a loopback client"},
                status_code=403,
                headers={"Cache-Control": "no-store"},
            )
        return await call_next(request)

    public = (
        path in {
            "/healthz",
            "/readyz",
            "/login",
            "/manifest.webmanifest",
            "/" + FRONTEND_BUILD.runtime_name,
        }
        or path.startswith("/static/")
        or (path == "/api/v1/session" and request.method == "POST")
        or (
            path in {
                "/api/v1/passkeys/authentication/options",
                "/api/v1/passkeys/authentication/verify",
            }
            and request.method == "POST"
        )
    )
    session = None
    session_reason = None
    if not public:
        try:
            verification = await dashboard_auth.request_session_verification(request)
            session = verification.session
            session_reason = verification.reason
        except dashboard_auth.SessionStoreError:
            return JSONResponse(
                {"detail": "Session registry is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
    if not public and session is None:
        if path.startswith("/api/"):
            payload = {"detail": "Authentication required"}
            if session_reason in {"session_revoked", "session_idle"}:
                payload["code"] = session_reason
            return JSONResponse(
                payload,
                status_code=401,
                headers={"Cache-Control": "no-store"},
            )
        if session_reason == "session_revoked":
            login_redirect = "login?reason=session-revoked"
        elif session_reason == "session_idle":
            login_redirect = "login?reason=session-idle"
        else:
            login_redirect = _share_target_login_redirect(request)
        return RedirectResponse(
            login_redirect,
            status_code=303,
            headers={"Cache-Control": "no-store"},
        )

    if not public and request.method not in {"GET", "HEAD", "OPTIONS"}:
        supplied_csrf = request.headers.get("x-csrf-token", "")
        if (
            session is None
            or not dashboard_auth.same_origin(request)
            or not supplied_csrf
            or not hmac.compare_digest(supplied_csrf, session.csrf)
        ):
            return JSONResponse(
                {"detail": "Valid same-origin CSRF proof required"},
                status_code=403,
                headers={"Cache-Control": "no-store"},
            )
    request.state.dashboard_session = session
    response = await call_next(request)
    if path in {"/login", "/api/v1/session"}:
        response.headers["Cache-Control"] = "no-store"
    return response


@app.middleware("http")
async def prevent_live_api_caching(request, call_next):
    response = await call_next(request)
    path = dashboard_auth.application_path(request)
    if path in {"/api/v1/context", "/api/v1/background-identity", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route", "/api/v1/today", "/api/v1/tomorrow", "/api/v1/tomorrow/promote", "/api/v1/week", "/api/v1/week/promote", "/api/v1/today/session", "/api/v1/later", "/api/v1/saved-searches", "/api/v1/completed-filed-reviews", "/api/v1/security-events", "/api/v1/push-subscription"} or path.startswith("/api/v1/work/") or (
        path.startswith("/api/v1/repos/")
        and path.endswith("/review")
    ) or path.startswith("/api/v1/notifications") or (
        path.startswith("/api/v1/repos/")
        and ("/issues/" in path or "/pulls/" in path)
    ) or (
        path.startswith("/api/v1/repos/")
        and path.endswith("/issues")
    ) or (
        path.startswith("/api/v1/repos/")
        and path.endswith(("/labels", "/milestones"))
    ):
        response.headers["Cache-Control"] = "no-store"
    return response


CONTENT_SECURITY_POLICY = "; ".join(
    (
        "default-src 'self'",
        "script-src 'self'",
        "connect-src 'self'",
        "img-src 'self' data:",
        "style-src 'self' 'unsafe-inline'",
        "worker-src 'self'",
        "manifest-src 'self'",
        "object-src 'none'",
        "base-uri 'self'",
        "form-action 'self'",
        "frame-ancestors 'none'",
    )
)


@app.exception_handler(RequestValidationError)
async def redact_request_validation_error(
    _request: Request, error: RequestValidationError
) -> JSONResponse:
    """Return useful validation locations without reflecting submitted values."""
    details = [
        {key: item[key] for key in ("type", "loc", "msg") if key in item}
        for item in error.errors()
    ]
    return JSONResponse(
        {"detail": details},
        status_code=422,
        headers={"Cache-Control": "no-store"},
    )


@app.middleware("http")
async def enforce_browser_security_boundary(request: Request, call_next):
    """Apply one browser trust boundary, including to auth short-circuits."""
    response = await call_next(request)
    response.headers["Content-Security-Policy"] = CONTENT_SECURITY_POLICY
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Permissions-Policy"] = (
        "camera=(), microphone=(self), geolocation=(), payment=(), usb=()"
    )
    response.headers["X-Frame-Options"] = "DENY"
    if dashboard_auth.mode() == dashboard_auth.OPERATOR_MODE:
        response.headers["Strict-Transport-Security"] = "max-age=31536000"
    return response


@app.get("/healthz")
def health() -> dict[str, str]:
    """Return process liveness without depending on Gitea."""
    return {"status": "ok", "service": "stackchain-dashboard"}


@app.post("/api/v1/session")
async def sign_in(payload: DashboardSignIn, request: Request, response: Response):
    peer_host = request.client.host if request.client is not None else "unknown"
    source = client_source(
        peer_host,
        request.headers.get("x-forwarded-for", ""),
        os.getenv("STACKCHAIN_TRUSTED_PROXY_CIDRS", ""),
    )
    attempts = _login_attempt_store()
    try:
        retry_after = await asyncio.to_thread(attempts.retry_after, source)
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if retry_after:
        try:
            await asyncio.to_thread(attempts.record_blocked, method="token")
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        return JSONResponse(
            {"detail": "Too many sign-in attempts"},
            status_code=429,
            headers={
                "Cache-Control": "no-store",
                "Retry-After": str(retry_after),
            },
        )
    configured_token = dashboard_auth.access_token()
    if not configured_token or not hmac.compare_digest(
        payload.access_token, configured_token
    ):
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=401, detail="Invalid access token")
    try:
        await asyncio.to_thread(attempts.clear, source)
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        signed, session = await asyncio.to_thread(
            dashboard_auth.issue_session, device_label=payload.device_label
        )
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(
            _security_event_store().record,
            "sign_in",
            method="token",
            device_label=payload.device_label,
            target="dashboard",
        )
    except SecurityEventStoreError:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    path = dashboard_auth.cookie_path(request)
    max_age = max(1, session.expires_at - int(time.time()))
    response.set_cookie(
        dashboard_auth.SESSION_COOKIE,
        signed,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.set_cookie(
        dashboard_auth.CSRF_COOKIE,
        session.csrf,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {"authenticated": True}


@app.get("/api/v1/security-events")
async def list_security_events(
    limit: int = Query(default=25, ge=1, le=100),
    cursor: int | None = Query(default=None, ge=1),
):
    try:
        page = await asyncio.to_thread(
            _security_event_store().list, limit=limit, cursor=cursor
        )
        authentication_alerts = await asyncio.to_thread(
            _login_attempt_store().list_alerts, limit=24
        )
    except (SecurityEventStoreError, LoginAttemptStoreError):
        raise HTTPException(
            status_code=503, detail="Security activity is temporarily unavailable"
        )
    return JSONResponse(
        {
            "events": [
                {
                    "id": event.id,
                    "kind": event.kind,
                    "method": event.method,
                    "device_label": event.device_label,
                    "target": event.target,
                    "created_at": event.created_at,
                    "status": event.status,
                }
                for event in page.events
            ],
            "authentication_alerts": authentication_alerts,
            "next_cursor": page.next_cursor,
        },
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/v1/fresh-authorization", status_code=201)
async def fresh_authorization(payload: FreshAuthorization, request: Request):
    peer_host = request.client.host if request.client is not None else "unknown"
    source = client_source(
        peer_host,
        request.headers.get("x-forwarded-for", ""),
        os.getenv("STACKCHAIN_TRUSTED_PROXY_CIDRS", ""),
    )
    attempts = _login_attempt_store()
    try:
        retry_after = await asyncio.to_thread(attempts.retry_after, source)
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if retry_after:
        return JSONResponse(
            {"detail": "Too many sign-in attempts"},
            status_code=429,
            headers={"Cache-Control": "no-store", "Retry-After": str(retry_after)},
        )
    configured_token = dashboard_auth.access_token()
    if not configured_token or not hmac.compare_digest(
        payload.access_token, configured_token
    ):
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=401, detail="Invalid access token")
    try:
        await asyncio.to_thread(attempts.clear, source)
        grant = await dashboard_auth.issue_step_up(
            request.state.dashboard_session,
            action=payload.action,
            target=payload.target,
        )
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    return JSONResponse(
        {"grant": grant, "expires_in": dashboard_auth.STEP_UP_TTL_SECONDS},
        status_code=201,
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/v1/passkeys/registration/options", status_code=201)
async def create_passkey_registration_options(
    request: Request,
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    await _require_step_up(
        request,
        step_up_grant,
        action="enroll_passkey",
        target="current_device",
    )
    rp_id, _origin = _passkey_relying_party(request)
    store = _passkey_store()
    existing = await asyncio.to_thread(store.all)
    options, challenge = passkeys.registration_options(
        rp_id=rp_id,
        excluded=[item.credential_id for item in existing],
    )
    await asyncio.to_thread(
        store.issue_challenge,
        challenge,
        session_id=request.state.dashboard_session.session_id,
        purpose="registration",
        action="enroll_passkey",
        target="current_device",
    )
    return JSONResponse(
        options,
        status_code=201,
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/v1/passkeys/registration/verify", status_code=201)
async def verify_passkey_registration(payload: PasskeyCeremony, request: Request):
    try:
        challenge = passkeys.decode(payload.challenge)
    except (ValueError, TypeError):
        raise HTTPException(status_code=400, detail="Invalid passkey ceremony")
    store = _passkey_store()
    valid = await asyncio.to_thread(
        store.consume_challenge,
        challenge,
        session_id=request.state.dashboard_session.session_id,
        purpose="registration",
        action="enroll_passkey",
        target="current_device",
    )
    if not valid:
        raise HTTPException(status_code=409, detail="Passkey challenge expired or already used")
    rp_id, origin = _passkey_relying_party(request)
    try:
        verified = await asyncio.to_thread(
            passkeys.verify_registration,
            credential=payload.credential,
            challenge=challenge,
            rp_id=rp_id,
            origin=origin,
        )
        devices = await dashboard_auth.active_devices(request.state.dashboard_session)
        current = next(device for device in devices if device.current)
    except dashboard_auth.SessionStoreError:
        raise HTTPException(status_code=503, detail="Passkey registry is temporarily unavailable")
    except Exception as exc:
        raise HTTPException(status_code=400, detail="Passkey verification failed") from exc
    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "passkey_enrolled",
            method="passkey",
            device_label=current.device_label,
            target="passkey",
        )
    except SecurityEventStoreError:
        raise HTTPException(
            status_code=503, detail="Security activity is temporarily unavailable"
        )
    try:
        await asyncio.to_thread(
            store.register,
            credential_id=verified.credential_id,
            public_key=verified.credential_public_key,
            sign_count=verified.sign_count,
            device_label=current.device_label,
            management_id=current.management_id,
        )
    except dashboard_auth.SessionStoreError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        raise HTTPException(
            status_code=503, detail="Passkey registry is temporarily unavailable"
        )
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        pass
    return JSONResponse(
        {"enrolled": True}, status_code=201, headers={"Cache-Control": "no-store"}
    )


@app.get("/api/v1/passkeys")
async def list_enrolled_passkeys(request: Request):
    try:
        credentials = await asyncio.to_thread(_passkey_store().all)
        devices = await dashboard_auth.active_devices(request.state.dashboard_session)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Passkey registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    active = {device.management_id: device for device in devices}
    return JSONResponse(
        {
            "passkeys": [
                {
                    "management_id": credential.management_id,
                    "device_label": credential.device_label,
                    "created_at": credential.created_at,
                    "active": credential.management_id in active,
                    "current": bool(
                        credential.management_id in active
                        and active[credential.management_id].current
                    ),
                }
                for credential in credentials
            ]
        },
        headers={"Cache-Control": "no-store"},
    )


@app.delete("/api/v1/passkeys/{management_id}")
async def revoke_enrolled_passkey(
    request: Request,
    management_id: str = PathParam(
        min_length=16, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"
    ),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    await _require_step_up(
        request,
        step_up_grant,
        action="revoke_passkey",
        target=management_id,
    )
    store = _passkey_store()
    try:
        credential = await asyncio.to_thread(store.get_management_id, management_id)
        devices = await dashboard_auth.active_devices(request.state.dashboard_session)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Passkey registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if credential is None:
        raise HTTPException(status_code=404, detail="Enrolled passkey not found")
    active_device = next(
        (device for device in devices if device.management_id == management_id), None
    )
    current_session = bool(active_device and active_device.current)
    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "passkey_revoked",
            device_label=credential.device_label,
            target="passkey",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        revoked, session_revoked = await asyncio.to_thread(
            store.revoke_access,
            management_id,
            preserve_session=current_session,
        )
    except dashboard_auth.SessionStoreError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"detail": "Passkey registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if not revoked:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        raise HTTPException(status_code=404, detail="Enrolled passkey not found")
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        pass
    return JSONResponse(
        {
            "revoked": True,
            "current_session": current_session,
            "session_revoked": session_revoked,
        },
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/v1/passkeys/authentication/options")
async def create_passkey_authentication_options(request: Request):
    store = _passkey_store()
    credentials = await asyncio.to_thread(store.all)
    if not credentials:
        raise HTTPException(status_code=404, detail="No passkeys enrolled")
    peer_host = request.client.host if request.client is not None else "unknown"
    source = client_source(
        peer_host,
        request.headers.get("x-forwarded-for", ""),
        os.getenv("STACKCHAIN_TRUSTED_PROXY_CIDRS", ""),
    )
    attempts = _login_attempt_store()
    try:
        retry_after = await asyncio.to_thread(attempts.retry_after, source)
        if not retry_after:
            retry_after = await asyncio.to_thread(
                attempts.admit,
                "passkey_options",
                source,
                limit=int(os.getenv("STACKCHAIN_PASSKEY_OPTIONS_MAX_ATTEMPTS", "10")),
            )
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if retry_after:
        return JSONResponse(
            {"detail": "Too many passkey sign-in attempts"},
            status_code=429,
            headers={"Cache-Control": "no-store", "Retry-After": str(retry_after)},
        )
    rp_id, _origin = _passkey_relying_party(request)
    options, challenge = passkeys.authentication_options(
        rp_id=rp_id,
        credentials=[item.credential_id for item in credentials],
    )
    await asyncio.to_thread(
        store.issue_challenge,
        challenge,
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
        source=source,
    )
    return JSONResponse(options, headers={"Cache-Control": "no-store"})


@app.post("/api/v1/passkeys/authorization/options")
async def create_passkey_authorization_options(
    payload: PasskeyAuthorizationTarget, request: Request
):
    store = _passkey_store()
    credentials = await asyncio.to_thread(store.all)
    if not credentials:
        raise HTTPException(status_code=404, detail="No passkeys enrolled")
    rp_id, _origin = _passkey_relying_party(request)
    options, challenge = passkeys.authentication_options(
        rp_id=rp_id,
        credentials=[item.credential_id for item in credentials],
    )
    await asyncio.to_thread(
        store.issue_challenge,
        challenge,
        session_id=request.state.dashboard_session.session_id,
        purpose="authorization",
        action=payload.action,
        target=payload.target,
    )
    return JSONResponse(options, headers={"Cache-Control": "no-store"})


@app.post("/api/v1/passkeys/authorization/verify", status_code=201)
async def verify_passkey_authorization(
    payload: PasskeyAuthorization, request: Request
):
    try:
        challenge = passkeys.decode(payload.challenge)
        credential_id = passkeys.decode(str(payload.credential.get("id", "")))
    except (ValueError, TypeError):
        raise HTTPException(status_code=400, detail="Invalid passkey ceremony")
    store = _passkey_store()
    valid = await asyncio.to_thread(
        store.consume_challenge,
        challenge,
        session_id=request.state.dashboard_session.session_id,
        purpose="authorization",
        action=payload.action,
        target=payload.target,
    )
    stored = await asyncio.to_thread(store.get, credential_id)
    if not valid or stored is None:
        raise HTTPException(status_code=409, detail="Passkey challenge expired or already used")
    rp_id, origin = _passkey_relying_party(request)
    try:
        verified = await asyncio.to_thread(
            passkeys.verify_authentication,
            credential=payload.credential,
            challenge=challenge,
            rp_id=rp_id,
            origin=origin,
            stored=stored,
        )
        updated = await asyncio.to_thread(
            store.advance_counter,
            stored.credential_id,
            expected=stored.sign_count,
            new=verified.new_sign_count,
        )
        if not updated:
            await asyncio.to_thread(
                _security_event_store().record,
                "passkey_counter_anomaly",
                method="passkey",
                device_label=stored.device_label,
                target=f"{payload.action}:{payload.target}",
            )
            raise ValueError("stale passkey counter")
        grant = await dashboard_auth.issue_step_up(
            request.state.dashboard_session,
            action=payload.action,
            target=payload.target,
        )
    except dashboard_auth.SessionStoreError:
        raise HTTPException(status_code=503, detail="Passkey registry is temporarily unavailable")
    except Exception as exc:
        raise HTTPException(status_code=401, detail="Passkey authorization failed") from exc
    return JSONResponse(
        {"grant": grant, "expires_in": dashboard_auth.STEP_UP_TTL_SECONDS},
        status_code=201,
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/v1/passkeys/authentication/verify")
async def verify_passkey_authentication(
    payload: PasskeyAuthentication, request: Request, response: Response
):
    if payload.action != "sign_in" or payload.target != "dashboard":
        raise HTTPException(status_code=400, detail="Invalid passkey sign-in target")
    peer_host = request.client.host if request.client is not None else "unknown"
    source = client_source(
        peer_host,
        request.headers.get("x-forwarded-for", ""),
        os.getenv("STACKCHAIN_TRUSTED_PROXY_CIDRS", ""),
    )
    attempts = _login_attempt_store()
    try:
        retry_after = await asyncio.to_thread(attempts.retry_after, source)
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if retry_after:
        return JSONResponse(
            {"detail": "Too many passkey sign-in attempts"},
            status_code=429,
            headers={"Cache-Control": "no-store", "Retry-After": str(retry_after)},
        )
    try:
        challenge = passkeys.decode(payload.challenge)
        credential_id = passkeys.decode(str(payload.credential.get("id", "")))
    except (ValueError, TypeError):
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=400, detail="Invalid passkey ceremony")
    store = _passkey_store()
    valid = await asyncio.to_thread(
        store.consume_challenge,
        challenge,
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
    )
    stored = await asyncio.to_thread(store.get, credential_id)
    if not valid or stored is None:
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=401, detail="Passkey sign-in failed")
    rp_id, origin = _passkey_relying_party(request)
    try:
        verified = await asyncio.to_thread(
            passkeys.verify_authentication,
            credential=payload.credential,
            challenge=challenge,
            rp_id=rp_id,
            origin=origin,
            stored=stored,
        )
        updated = await asyncio.to_thread(
            store.advance_counter,
            stored.credential_id,
            expected=stored.sign_count,
            new=verified.new_sign_count,
        )
        if not updated:
            await asyncio.to_thread(
                _security_event_store().record,
                "passkey_counter_anomaly",
                method="passkey",
                device_label=stored.device_label,
                target="sign_in:dashboard",
            )
            raise ValueError("stale passkey counter")
        await dashboard_auth.revoke_managed_session(stored.management_id)
        signed, session = await asyncio.to_thread(
            dashboard_auth.issue_session,
            device_label=stored.device_label,
            management_id=stored.management_id,
        )
    except dashboard_auth.SessionStoreError:
        raise HTTPException(status_code=503, detail="Passkey registry is temporarily unavailable")
    except Exception as exc:
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=401, detail="Passkey sign-in failed") from exc
    try:
        await asyncio.to_thread(attempts.clear, source)
    except LoginAttemptStoreError:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(
            _security_event_store().record,
            "sign_in",
            method="passkey",
            device_label=stored.device_label,
            target="dashboard",
        )
    except SecurityEventStoreError:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
        raise HTTPException(
            status_code=503, detail="Security activity is temporarily unavailable"
        )
    path = dashboard_auth.cookie_path(request)
    max_age = max(1, session.expires_at - int(time.time()))
    response.set_cookie(
        dashboard_auth.SESSION_COOKIE,
        signed,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.set_cookie(
        dashboard_auth.CSRF_COOKIE,
        session.csrf,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {"authenticated": True, "method": "passkey"}


@app.get("/api/v1/session")
async def session_status(request: Request):
    session = request.state.dashboard_session
    payload = {
        "authenticated": session is not None,
        "csrf_token": session.csrf if session is not None else "",
    }
    if session is not None:
        payload["expires_at"] = session.expires_at
        payload["idle_expires_at"] = session.idle_expires_at
    return payload


@app.get("/api/v1/push-subscription")
async def push_status(request: Request):
    configuration = _push_configuration()
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    subscribed = await asyncio.to_thread(
        _push_subscription_store.is_subscribed,
        device_id,
    )
    preferences = await asyncio.to_thread(
        _push_subscription_store.deadline_preferences, device_id, now=time.time()
    )
    start_day_preferences = await asyncio.to_thread(
        _push_subscription_store.start_day_preferences, device_id
    )
    delivery_health = await asyncio.to_thread(
        _push_subscription_store.delivery_health, device_id
    )
    return {
        "available": configuration.enabled,
        "subscribed": subscribed,
        "public_key": configuration.public_key if configuration.enabled else "",
        "deadline_enabled": preferences["enabled"],
        "timezone": preferences["timezone"],
        "reminder_hour": preferences["reminder_hour"],
        "reminder_days": preferences["reminder_days"],
        "snoozed_until": preferences["snoozed_until"],
        "start_day_enabled": start_day_preferences["enabled"],
        "start_day_timezone": start_day_preferences["timezone"],
        "start_day_reminder_hour": start_day_preferences["reminder_hour"],
        "delivery_health": delivery_health,
    }


@app.put("/api/v1/push-subscription")
async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
    if not _push_configuration().enabled:
        raise HTTPException(status_code=503, detail="Push notifications are not configured")
    try:
        await validate_public_push_endpoint(payload.endpoint)
    except UnsafePushEndpoint as error:
        raise HTTPException(status_code=422, detail=str(error)) from error
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    await asyncio.to_thread(
        _push_subscription_store.upsert,
        device_id,
        payload.model_dump(),
    )
    try:
        current = await gitea_proxy.unread_notification_snapshot(
            deadline_seconds=NOTIFICATION_PAGE_TIMEOUT_SECONDS
        )
    except Exception as error:
        await asyncio.to_thread(_push_subscription_store.delete_session, device_id)
        raise HTTPException(
            status_code=503,
            detail="Unread updates are temporarily unavailable",
            headers={"Retry-After": "1"},
        ) from error
    existing_revisions = {
        int(item["id"]): str(item.get("updated_at") or "")
        for item in current.get("items", [])
        if isinstance(item, dict) and str(item.get("id", "")).isdigit()
    }
    await asyncio.to_thread(
        _push_subscription_store.mark_delivered, device_id, existing_revisions
    )
    return {"subscribed": True}


@app.delete("/api/v1/push-subscription")
async def unsubscribe_push(request: Request):
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    await asyncio.to_thread(
        _push_subscription_store.delete_session,
        device_id,
    )
    return {"subscribed": False}


@app.post("/api/v1/push-subscription/test")
async def test_push_notification(request: Request):
    configuration = _push_configuration()
    if not configuration.enabled:
        raise HTTPException(status_code=503, detail="Push notifications are not configured")
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    subscription = await asyncio.to_thread(
        _push_subscription_store.subscription_for_session, device_id
    )
    if subscription is None:
        raise HTTPException(status_code=409, detail="Enable device notifications first")
    payload = json.dumps({
        "title": "Stackchain notifications are working",
        "body": "This device can receive private work alerts.",
        "route": "#/device-setup",
        "tag": "stackchain-push-test",
    }, separators=(",", ":"))
    try:
        await send_web_push(subscription, payload, configuration)
    except Exception as error:
        status = getattr(getattr(error, "response", None), "status_code", None)
        if isinstance(error, UnsafePushEndpoint) or status in {404, 410}:
            await asyncio.to_thread(_push_subscription_store.delete_session, device_id)
            raise HTTPException(
                status_code=409, detail="Re-enable notifications for this device"
            ) from error
        reason = "timeout" if isinstance(error, (asyncio.TimeoutError, TimeoutError)) else "provider"
        await asyncio.to_thread(
            _push_subscription_store.mark_delivery_failed,
            device_id,
            "unread",
            reason,
        )
        raise HTTPException(
            status_code=503,
            detail="Test notification delivery failed",
            headers={"Retry-After": "1"},
        ) from error
    await asyncio.to_thread(
        _push_subscription_store.mark_delivery_succeeded, device_id, "unread"
    )
    return {"delivery_state": "healthy", "delivered": True}


@app.put("/api/v1/push-subscription/deadlines")
async def update_deadline_reminders(payload: DeadlineReminderPayload, request: Request):
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    if payload.enabled and not await asyncio.to_thread(
        _push_subscription_store.is_subscribed, device_id
    ):
        raise HTTPException(status_code=409, detail="Enable device notifications first")
    await asyncio.to_thread(
        _push_subscription_store.set_deadline_preferences,
        device_id,
        enabled=payload.enabled,
        timezone=payload.timezone,
        reminder_hour=payload.reminder_hour,
        reminder_days=payload.reminder_days,
    )
    return {
        "deadline_enabled": payload.enabled,
        "timezone": payload.timezone,
        "reminder_hour": payload.reminder_hour,
        "reminder_days": payload.reminder_days,
    }


@app.put("/api/v1/push-subscription/start-day")
async def update_start_day_reminders(payload: StartDayReminderPayload, request: Request):
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    if payload.enabled and not await asyncio.to_thread(
        _push_subscription_store.is_subscribed, device_id
    ):
        raise HTTPException(status_code=409, detail="Enable device notifications first")
    await asyncio.to_thread(
        _push_subscription_store.set_start_day_preferences,
        device_id,
        enabled=payload.enabled,
        timezone=payload.timezone,
        reminder_hour=payload.reminder_hour,
    )
    return {
        "start_day_enabled": payload.enabled,
        "start_day_timezone": payload.timezone,
        "start_day_reminder_hour": payload.reminder_hour,
    }


@app.patch("/api/v1/push-subscription/deadlines/snooze")
async def snooze_deadline_reminder(request: Request):
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    now = int(time.time())
    snoozed = await asyncio.to_thread(
        _push_subscription_store.snooze_deadline_reminder,
        device_id,
        now=now,
        delay_seconds=3_600,
    )
    if not snoozed:
        raise HTTPException(status_code=409, detail="Enable deadline reminders first")
    return {"snoozed": True, "snoozed_until": now + 3_600}


@app.delete("/api/v1/push-subscription/deadlines/snooze")
async def resume_deadline_reminders(request: Request):
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    await asyncio.to_thread(
        _push_subscription_store.clear_deadline_snooze,
        device_id,
    )
    return {"snoozed": False}


@app.post("/api/v1/session/activity")
async def record_session_activity(request: Request):
    session = request.state.dashboard_session
    try:
        active = await dashboard_auth.touch_session(session)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if not active:
        return JSONResponse(
            {"detail": "Authentication required"},
            status_code=401,
            headers={"Cache-Control": "no-store"},
        )
    return {
        "active": True,
        "idle_expires_at": min(
            session.expires_at,
            int(time.time()) + dashboard_auth.idle_timeout_seconds(),
        ),
    }


def _today_store() -> TodayStore:
    return TodayStore(
        os.getenv("STACKCHAIN_TODAY_DB", str(_state_dir / "today.sqlite3")), limit=5
    )


def _later_store() -> LaterStore:
    return LaterStore(
        os.getenv("STACKCHAIN_LATER_DB", str(_state_dir / "later.sqlite3"))
    )


def _saved_search_store() -> SavedSearchStore:
    return SavedSearchStore(
        os.getenv("STACKCHAIN_SAVED_SEARCH_DB", str(_state_dir / "saved-searches.sqlite3"))
    )


def _completed_filed_review_store() -> CompletedFiledReviewStore:
    return CompletedFiledReviewStore(
        os.getenv(
            "STACKCHAIN_COMPLETED_FILED_REVIEW_DB",
            str(_state_dir / "completed-filed-reviews.sqlite3"),
        )
    )


def _unfiled_draft_store() -> UnfiledDraftStore:
    encoded_keyring = os.getenv("STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEYS")
    if encoded_keyring is not None:
        keys, active = decode_unfiled_draft_encryption_keyring(
            encoded_keyring,
            os.getenv("STACKCHAIN_UNFILED_DRAFT_ACTIVE_KEY_ID", ""),
        )
        return UnfiledDraftStore(
            os.getenv("STACKCHAIN_UNFILED_DRAFT_DB", str(_state_dir / "unfiled-drafts.sqlite3")),
            encryption_keys=keys,
            active_key_id=active,
        )
    return UnfiledDraftStore(
        os.getenv("STACKCHAIN_UNFILED_DRAFT_DB", str(_state_dir / "unfiled-drafts.sqlite3")),
        encryption_key=decode_unfiled_draft_encryption_key(
            os.getenv("STACKCHAIN_UNFILED_DRAFT_ENCRYPTION_KEY", "")
        ),
    )


async def _confirmed_login() -> str:
    try:
        user = await asyncio.wait_for(
            current_user(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except Exception as exc:
        raise HTTPException(
            status_code=503, detail="Operator identity is unavailable"
        ) from exc
    login = user.get("login", "") if isinstance(user, dict) else ""
    if not isinstance(login, str) or not login.strip():
        raise HTTPException(status_code=503, detail="Operator identity is unavailable")
    return login.strip().lower()


@app.get("/api/v1/completed-filed-reviews")
async def get_completed_filed_reviews(response: Response):
    login = await _confirmed_login()
    try:
        snapshot = await asyncio.to_thread(_completed_filed_review_store().get, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Completed Filed review synchronization is unavailable",
            headers={"Retry-After": "1"},
        )
    response.headers["Cache-Control"] = "no-store"
    return snapshot


@app.post("/api/v1/completed-filed-reviews")
async def merge_completed_filed_reviews(payload: CompletedFiledReviewBatch):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _completed_filed_review_store().merge,
            login,
            [receipt.model_dump() for receipt in payload.receipts],
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Completed Filed review synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/saved-searches")
async def get_saved_searches(response: Response):
    login = await _confirmed_login()
    try:
        snapshot = await asyncio.to_thread(_saved_search_store().get, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Saved Search synchronization is unavailable",
            headers={"Retry-After": "1"},
        )
    response.headers["Cache-Control"] = "no-store"
    return snapshot


@app.put("/api/v1/saved-searches")
async def replace_saved_searches(payload: SavedSearchCollection):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _saved_search_store().replace,
            login,
            payload.revision,
            [view.model_dump() for view in payload.views],
        )
    except SavedSearchConflict as exc:
        raise HTTPException(
            status_code=409,
            detail={
                "message": "Saved searches changed on another device.",
                "snapshot": exc.snapshot,
            },
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Saved Search synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/unfiled-drafts")
async def get_unfiled_drafts(response: Response):
    login = await _confirmed_login()
    try:
        snapshot = await asyncio.to_thread(_unfiled_draft_store().get, login)
    except (OSError, sqlite3.Error, UnfiledDraftEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Draft synchronization is unavailable",
            headers={"Retry-After": "1"},
        )
    response.headers["Cache-Control"] = "no-store"
    return snapshot


@app.put("/api/v1/unfiled-drafts")
async def replace_unfiled_drafts(payload: UnfiledDraftCollection):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _unfiled_draft_store().replace,
            login,
            payload.revision,
            [draft.model_dump() for draft in payload.drafts],
        )
    except UnfiledDraftConflict as exc:
        raise HTTPException(
            status_code=409,
            detail={
                "message": "Drafts changed on another device.",
                "snapshot": exc.snapshot,
            },
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    except (OSError, sqlite3.Error, UnfiledDraftEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Draft synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/today")
async def get_today_plan():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_today_store().get, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Today synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.patch("/api/v1/today")
async def update_today_plan(payload: TodayOperation | TodayOperationBatch):
    login = await _confirmed_login()
    try:
        if isinstance(payload, TodayOperationBatch):
            return await asyncio.to_thread(
                _today_store().apply_batch,
                login,
                [operation.model_dump() for operation in payload.operations],
            )
        if payload.action == "configure":
            return await asyncio.to_thread(
                _today_store().apply_batch,
                login,
                [payload.model_dump()],
            )
        return await asyncio.to_thread(
            _today_store().apply,
            login,
            payload.operation_id,
            payload.action,
            payload.item_id,
            direction=payload.direction,
        )
    except TodayPlanFull:
        raise HTTPException(status_code=409, detail="Today is limited to 5 items")
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Today synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/tomorrow")
async def get_tomorrow_plan():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_today_store().get_tomorrow, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Tomorrow synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.put("/api/v1/tomorrow")
async def replace_tomorrow_plan(payload: TomorrowPlanUpdate):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _today_store().replace_tomorrow, login, **payload.model_dump()
        )
    except TomorrowPlanConflict as error:
        raise HTTPException(
            status_code=409,
            detail={"code": "tomorrow_changed", "snapshot": error.snapshot},
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Tomorrow synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.post("/api/v1/tomorrow/promote")
async def promote_tomorrow_plan(payload: TomorrowPromotion):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _today_store().promote_tomorrow, login, **payload.model_dump()
        )
    except TomorrowPlanConflict as error:
        raise HTTPException(
            status_code=409,
            detail={"code": "tomorrow_changed", "snapshot": error.snapshot},
        )
    except TodayPromotionConflict as error:
        raise HTTPException(
            status_code=409,
            detail={"code": "today_changed", "today": error.today},
        )
    except TomorrowPlanNotDue as error:
        raise HTTPException(
            status_code=409,
            detail={"code": "tomorrow_not_due", "plan_date": error.plan_date},
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Tomorrow promotion is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/week")
async def get_week_plan():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_today_store().get_week, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Week Ahead synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.put("/api/v1/week")
async def replace_week_plan(payload: WeekPlanUpdate):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _today_store().replace_week, login,
            base_revision=payload.base_revision,
            days=[day.model_dump() for day in payload.days],
            timezone=payload.timezone,
        )
    except WeekPlanConflict as error:
        raise HTTPException(
            status_code=409, detail={"code": "week_changed", "snapshot": error.snapshot}
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Week Ahead synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.post("/api/v1/week/promote")
async def promote_week_plan(payload: WeekPromotion):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _today_store().promote_week, login, **payload.model_dump()
        )
    except WeekPlanConflict as error:
        raise HTTPException(
            status_code=409, detail={"code": "week_changed", "snapshot": error.snapshot}
        )
    except TodayPromotionConflict as error:
        if error.week is not None:
            raise HTTPException(
                status_code=409,
                detail={"code": "week_today_in_progress", "today": error.today, "week": error.week},
            )
        raise HTTPException(
            status_code=409, detail={"code": "today_changed", "today": error.today}
        )
    except TomorrowPlanNotDue as error:
        raise HTTPException(
            status_code=409, detail={"code": "week_not_due", "plan_date": error.plan_date}
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Week Ahead promotion is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/today/session")
async def get_today_session():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_today_store().get_session, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Today session synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.patch("/api/v1/today/session")
async def update_today_session(payload: TodaySessionUpdate):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _today_store().update_session, login, **payload.model_dump()
        )
    except TodaySessionConflict as error:
        raise HTTPException(
            status_code=409,
            detail={"code": "session_changed", "session": error.session},
        )
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503, detail="Today session synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/today/recaps")
async def get_today_recaps(response: Response):
    login = await _confirmed_login()
    try:
        recaps = await asyncio.to_thread(_today_store().list_recaps, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Today recap history is unavailable",
            headers={"Retry-After": "1"},
        )
    response.headers["Cache-Control"] = "no-store"
    return {"recaps": recaps}


@app.post("/api/v1/today/recaps")
async def save_today_recap(payload: TodayRecap, response: Response):
    login = await _confirmed_login()
    try:
        recap = await asyncio.to_thread(
            _today_store().save_recap,
            login,
            payload.session_id,
            [item.model_dump() for item in payload.items],
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Today recap could not be saved",
            headers={"Retry-After": "1"},
        )
    response.headers["Cache-Control"] = "no-store"
    return recap


@app.post("/api/v1/today/recaps/log-time")
async def save_today_recap_and_log_time(
    payload: TodayRecapTimeLog,
    request: Request,
    response: Response,
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    login = await _confirmed_login()
    if len(set(payload.log_identities)) != len(payload.log_identities):
        raise HTTPException(status_code=422, detail="time log targets must be unique")
    requested_items = {item.identity: item for item in payload.items}
    selected = []
    try:
        for identity in payload.log_identities:
            gitea_proxy.issue_time_target(identity)
            requested = requested_items.get(identity)
            if requested is None:
                raise ValueError("time log target must match the recap")
            if requested.actual_minutes <= 0:
                raise ValueError("only non-zero recap time can be logged")
            selected.append(requested)
        await _require_step_up(
            request,
            step_up_grant,
            action="log_recap_time",
            target=_recap_time_log_target(payload),
        )
        recap = await asyncio.to_thread(
            _today_store().save_recap,
            login,
            payload.session_id,
            [item.model_dump() for item in payload.items],
        )
        saved_items = {item["identity"]: item for item in recap["items"]}
        for requested in selected:
            saved = saved_items.get(requested.identity)
            if saved is None or saved["actual_minutes"] != requested.actual_minutes:
                raise ValueError("time log target must match the saved recap")
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(status_code=503, detail="Today recap could not be saved", headers={"Retry-After": "1"})

    semaphore = asyncio.Semaphore(3)

    async def log_selected_time(item):
        async with semaphore:
            result = {"identity": item.identity, "status": "retry"}
            try:
                state = await asyncio.to_thread(
                    _today_store().begin_time_log, login, payload.session_id, item.identity, item.actual_minutes
                )
                if state == "claimed":
                    repository, number = gitea_proxy.issue_time_target(item.identity)
                    journal = _security_event_store()
                    try:
                        operation_id = await asyncio.to_thread(
                            journal.reserve,
                            "gitea_time_logged",
                            target=f"{repository}#{number}",
                        )
                    except SecurityEventStoreError:
                        await asyncio.to_thread(
                            _today_store().finish_time_log,
                            login, payload.session_id, item.identity, succeeded=False,
                        )
                        return result
                    try:
                        await gitea_proxy.log_issue_time(item.identity, item.actual_minutes * 60)
                    except Exception as error:
                        if gitea_proxy.time_log_failure_is_retryable(error):
                            await asyncio.to_thread(
                                _today_store().finish_time_log,
                                login, payload.session_id, item.identity, succeeded=False,
                            )
                            try:
                                await asyncio.to_thread(journal.discard, operation_id)
                            except SecurityEventStoreError:
                                pass
                            status = "retry"
                        else:
                            status = "verify"
                        return {"identity": item.identity, "status": status}
                    await asyncio.to_thread(
                        _today_store().finish_time_log,
                        login, payload.session_id, item.identity, succeeded=True,
                    )
                    try:
                        await asyncio.to_thread(journal.finalize, operation_id)
                    except SecurityEventStoreError:
                        pass
                    state = "logged"
                return {
                    "identity": item.identity,
                    "status": "logged" if state == "logged" else ("verify" if state == "pending" else "retry"),
                }
            except (ValueError, OSError, sqlite3.Error):
                return result

    results = await asyncio.gather(*(log_selected_time(item) for item in selected))
    response.headers["Cache-Control"] = "no-store"
    return {**recap, "time_logs": results}


@app.get("/api/v1/later")
async def get_later_plan():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_later_store().get, login)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Later synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.patch("/api/v1/later")
async def update_later_plan(payload: LaterOperation | LaterOperationBatch):
    login = await _confirmed_login()
    try:
        if isinstance(payload, LaterOperationBatch):
            return await asyncio.to_thread(
                _later_store().apply_batch,
                login,
                [operation.model_dump() for operation in payload.operations],
            )
        return await asyncio.to_thread(
            _later_store().apply,
            login,
            payload.operation_id,
            payload.action,
            payload.item_id,
            wake_at=payload.wake_at,
            handoff=payload.handoff,
            base_revision=payload.base_revision,
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        raise HTTPException(
            status_code=503,
            detail="Later synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@app.delete("/api/v1/session")
async def sign_out(request: Request, response: Response):
    session = request.state.dashboard_session
    try:
        push_device_id = await dashboard_auth.session_management_id(session)
    except (dashboard_auth.SessionStoreError, AttributeError):
        push_device_id = None
    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "sign_out",
            target="current_device",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
    except dashboard_auth.SessionStoreError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        # The pending reservation is durable evidence; do not report a
        # successful revocation as failed merely because completion could not
        # be marked yet.
        pass
    if push_device_id is not None:
        await asyncio.to_thread(
            _push_subscription_store.delete_session, push_device_id
        )
    path = dashboard_auth.cookie_path(request)
    response.delete_cookie(
        dashboard_auth.SESSION_COOKIE,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.delete_cookie(
        dashboard_auth.CSRF_COOKIE,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {"authenticated": False, "clear_private_device_data": True}


@app.get("/api/v1/sessions")
async def list_active_devices(request: Request, response: Response):
    try:
        devices = await dashboard_auth.active_devices(request.state.dashboard_session)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    response.headers["Cache-Control"] = "no-store"
    return {
        "devices": [
            {
                "management_id": device.management_id,
                "device_label": device.device_label,
                "created_at": device.created_at,
                "expires_at": device.expires_at,
                "current": device.current,
            }
            for device in devices
        ]
    }


@app.delete("/api/v1/sessions/{management_id}")
async def revoke_active_device(
    request: Request,
    management_id: str = PathParam(
        min_length=16, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"
    ),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    await _require_step_up(
        request,
        step_up_grant,
        action="revoke_device",
        target=management_id,
    )
    try:
        devices = await dashboard_auth.active_devices(request.state.dashboard_session)
        target = next(
            (device for device in devices if device.management_id == management_id), None
        )
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if target is None:
        raise HTTPException(status_code=404, detail="Active device not found")

    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "device_revoked",
            device_label=target.device_label,
            target="device",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        revoked = await asyncio.to_thread(
            _passkey_store().revoke_device_access, management_id
        )
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if not revoked:
        raise HTTPException(status_code=404, detail="Active device not found")
    await asyncio.to_thread(
        _push_subscription_store.delete_session, management_id
    )
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        pass
    return {"revoked": True, "current_session": target.current}


@app.delete("/api/v1/sessions")
async def sign_out_all_devices(
    request: Request,
    response: Response,
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    await _require_step_up(
        request,
        step_up_grant,
        action="revoke_all_sessions",
        target="all",
    )
    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "all_sessions_revoked",
            target="all_devices",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(_passkey_store().revoke_all_access)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    await asyncio.to_thread(_push_subscription_store.delete_all)
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        pass
    path = dashboard_auth.cookie_path(request)
    response.delete_cookie(
        dashboard_auth.SESSION_COOKIE,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.delete_cookie(
        dashboard_auth.CSRF_COOKIE,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {
        "authenticated": False,
        "all_sessions_revoked": True,
        "clear_private_device_data": True,
    }


@app.get("/readyz")
async def readiness():
    """Return the latest readiness snapshot without contacting Gitea."""
    state = _readiness_state
    if state is None or time.monotonic() - state.checked_at > READINESS_MAX_AGE_SECONDS:
        return JSONResponse(
            {
                "status": "not_ready",
                "service": "stackchain-dashboard",
                "error": "Gitea readiness status is not available",
            },
            status_code=503,
        )
    if not state.ready:
        return JSONResponse(
            {
                "status": "not_ready",
                "service": "stackchain-dashboard",
                "error": state.error,
            },
            status_code=503,
            headers={
                "Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
            } if state.timed_out else None,
        )
    payload = {
        "status": "ready",
        "service": "stackchain-dashboard",
    }
    if not dashboard_auth.enabled():
        payload["gitea_user"] = state.login
    return payload


@app.get("/api/v1/context")
async def context() -> JSONResponse:
    try:
        user_data, repo_data, issues_data, prs_data = await asyncio.wait_for(
            asyncio.gather(current_user(), repos(), issues(), pull_requests()),
            timeout=CONTEXT_TIMEOUT_SECONDS,
        )
        if not isinstance(user_data, dict):
            raise ContextPayloadError("Gitea current-user response was not an object")
        if not all(field in user_data for field in ("id", "login")):
            raise ContextPayloadError(
                "Gitea current-user response did not include id and login"
            )
        if any(
            data is not None and not isinstance(data, list)
            for data in (repo_data, issues_data, prs_data)
        ):
            raise ContextPayloadError("Gitea collection response was not a list")
    except Exception as e:
        if isinstance(e, TimeoutError):
            error_message = (
                f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
            )
        elif isinstance(e, ContextPayloadError):
            error_message = str(e)
        else:
            error_message = "Gitea context is temporarily unavailable"
        return JSONResponse({
            "user": {"id": None, "login": "timmy", "full_name": "Timmy Jr", "email": ""},
            "repos": [],
            "issues": [],
            "pull_requests": [],
            "view": "dashboard",
            "deltas": [
                {"panel": "auth", "action": "connect", "target": "stackchain-dashboard backend", "priority": "high"},
                {"panel": "gitea", "action": "set GITEA_URL + GITEA_TOKEN", "target": "/root/stackchain-dashboard/.env or systemd env", "priority": "high"},
            ],
            "error": error_message,
        }, headers={
            "Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))
        } if isinstance(e, TimeoutError) else None)
    return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data))


@app.get("/api/v1/repositories/search")
async def repository_search(
    q: str = Query(min_length=2, max_length=80),
    limit: int = Query(default=20, ge=1, le=20),
) -> JSONResponse:
    """Return minimal repository identities matching a bounded visible-repo search."""
    query = q.strip()
    if len(query) < 2:
        return JSONResponse({"error": "Enter at least 2 search characters."}, status_code=400)
    try:
        matches = await asyncio.wait_for(
            gitea_proxy.search_repositories(query, limit),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "Repository search is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    items = [
        {"id": item["id"], "name": item["name"], "full_name": item["full_name"]}
        for item in matches
        if isinstance(item, dict)
        and all(field in item for field in ("id", "name", "full_name"))
    ]
    return JSONResponse({"items": items}, headers={"Cache-Control": "no-store"})


@app.get("/api/v1/repositories")
async def repository_page(
    page: int = Query(default=1, ge=1, le=1000),
    limit: int = Query(default=50, ge=1, le=50),
) -> JSONResponse:
    """Return one bounded repository page for lazy issue-capture selection."""
    try:
        result = await asyncio.wait_for(
            gitea_proxy.repo_page(page=page, limit=limit),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
        items = [
            Repo(
                id=item["id"], name=item["name"], full_name=item["full_name"],
                description=item.get("description") or "", url=item["html_url"],
                updated_at=item.get("updated_at", ""),
            ).model_dump()
            for item in result.get("items", [])
            if isinstance(item, dict)
            and all(field in item for field in ("id", "name", "full_name", "html_url"))
        ]
        return JSONResponse({**result, "items": items})
    except Exception:
        return JSONResponse(
            {"error": "Repositories could not be loaded. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/background-identity")
async def background_identity() -> JSONResponse:
    """Return only the account key required to safely drain a browser outbox."""
    try:
        user = await current_user()
        login = user.get("login") if isinstance(user, dict) else None
        if not isinstance(login, str) or not login.strip():
            raise ValueError("Gitea user response did not include a login")
    except Exception:
        return JSONResponse(
            {"error": "Gitea identity is temporarily unavailable"},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse({"login": login})


@app.get("/api/v1/search")
async def global_search(
    q: str = Query(min_length=2, max_length=100),
    limit: int = Query(default=10, ge=1, le=25),
    page: int = Query(default=1, ge=1, le=100),
    kind: Literal["all", "issue", "pull"] = Query(default="all"),
    state: Literal["all", "open", "closed"] = Query(default="all"),
    repository: str | None = Query(
        default=None, pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", max_length=161
    ),
    issues_page: int | None = Query(default=None, ge=1, le=100),
    pulls_page: int | None = Query(default=None, ge=1, le=100),
) -> JSONResponse:
    query = q.strip()
    if len(query) < 2:
        raise HTTPException(status_code=422, detail="Search query must contain at least 2 characters")
    try:
        arguments = (query, limit, page, kind, state)
        continuation = None
        if issues_page is not None or pulls_page is not None:
            continuation = {"issues": issues_page, "pulls": pulls_page}
        if continuation is not None:
            operation = gitea_proxy.global_search(
                *arguments, repository, continuation
            )
        elif repository:
            operation = gitea_proxy.global_search(*arguments, repository)
        else:
            operation = gitea_proxy.global_search(*arguments)
        result = await asyncio.wait_for(
            operation,
            timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "Search is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    scope = {"kind": kind, "state": state}
    if repository:
        scope["repository"] = repository
    return JSONResponse({"query": query, "scope": scope, **result})


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/preview")
async def global_search_preview(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    kind: Literal["issue", "pull"] = Query(),
) -> JSONResponse:
    try:
        preview = await asyncio.wait_for(
            gitea_proxy.work_preview(f"{owner}/{repo}", kind, number),
            timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "This work item is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(preview)


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/preview/conversation")
async def global_search_preview_conversation(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    kind: Literal["issue", "pull"] = Query(),
    page: int | None = Query(default=None, ge=1, le=100),
    limit: int = Query(default=20, ge=1, le=50),
) -> JSONResponse:
    del kind  # Issue and pull-request conversations share Gitea's issue-comments API.
    try:
        conversation = await asyncio.wait_for(
            gitea_proxy.issue_conversation_page(
                f"{owner}/{repo}", number, page, limit
            ),
            timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "This conversation is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(conversation)


@app.post(
    "/api/v1/repos/{owner}/{repo}/issues/{number}/preview/comments",
    status_code=201,
)
async def comment_on_global_search_preview(
    comment: IssueComment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    kind: Literal["issue", "pull"] = Query(),
    idempotency_key: str | None = Header(default=None, max_length=128),
) -> JSONResponse:
    repository = f"{owner}/{repo}"

    async def post_comment():
        preview = await gitea_proxy.work_preview(repository, kind, number)
        if (
            preview.get("repository") != repository
            or preview.get("kind") != kind
            or preview.get("number") != number
        ):
            raise HTTPException(status_code=404, detail="Search result not found")
        return await gitea_proxy.comment_on_issue(repository, number, comment.body)

    try:
        result = await _run_idempotent_authored_action(
            post_comment(),
            idempotency_key=idempotency_key,
            fingerprint=("search-preview-comment", repository, kind, number, comment.body),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The reply could not be posted. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.post(
    "/api/v1/repos/{owner}/{repo}/issues/{number}/preview/attachments",
    status_code=201,
)
async def attach_to_global_search_preview(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    kind: Literal["issue", "pull"] = Query(),
    idempotency_key: str | None = Header(default=None, max_length=128),
) -> JSONResponse:
    repository = f"{owner}/{repo}"
    try:
        form = await request.form()
        uploaded = form.get("file")
        if not isinstance(uploaded, UploadFile):
            raise ValueError("screenshot file is required")
        filename = str(uploaded.filename or "")
        content_type = str(uploaded.content_type or "")
        content = _validate_binary_attachment(filename, content_type, await uploaded.read())
        content = await _sanitize_attachment(content_type, content)
    except (ValueError, ValidationError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    async def upload_attachment():
        preview = await gitea_proxy.work_preview(repository, kind, number)
        if (
            preview.get("repository") != repository
            or preview.get("kind") != kind
            or preview.get("number") != number
            or preview.get("commentable") is not True
        ):
            raise HTTPException(status_code=404, detail="Search result not found")
        result = await gitea_proxy.upload_preview_attachment(
            repository, number, filename, content_type, content
        )
        safe_name = (
            result["name"].replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
            .replace("\r", " ").replace("\n", " ")
        )
        safe_url = result["url"].replace("<", "%3C").replace(">", "%3E")
        result["markdown"] = f"![{safe_name}](<{safe_url}>)"
        return result

    try:
        result = await _run_idempotent_authored_action(
            upload_attachment(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "search-preview-attachment", repository, kind, number, filename,
                content_type, hashlib.sha256(content).hexdigest(),
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The photo could not be uploaded. Your reply is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.get("/api/v1/work-route")
async def resolve_work_route(
    kind: Literal["issue", "filed", "pull", "review", "update"] = Query(),
    repository: str | None = Query(default=None, pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"),
    number: int | None = Query(default=None, gt=0),
    notification_id: int | None = Query(default=None, gt=0),
) -> JSONResponse:
    repository_route = repository is not None and number is not None and notification_id is None
    update_route = kind == "update" and notification_id is not None and repository is None and number is None
    if (kind == "update" and not update_route) or (kind != "update" and not repository_route):
        raise HTTPException(status_code=422, detail="Invalid work route identity")
    try:
        result = await asyncio.wait_for(
            gitea_proxy.resolve_work_route(kind, repository, number, notification_id),
            timeout=WORK_PAGE_TIMEOUT_SECONDS,
        )
    except gitea_proxy.WorkRouteUnavailableError:
        raise HTTPException(status_code=404, detail="Work item is no longer in My Work")
    except Exception:
        return JSONResponse(
            {"error": "The shared work item is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/work/{stream}")
async def paged_work(
    stream: Literal["issue", "filed", "pull", "review"],
    page: int = Query(ge=2, le=100),
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.work_page(stream, page), timeout=WORK_PAGE_TIMEOUT_SECONDS
        )
    except Exception:
        return JSONResponse(
            {"error": "Work page is temporarily unavailable"},
            status_code=503,
            headers={"Retry-After": str(math.ceil(WORK_PAGE_TIMEOUT_SECONDS))},
        )
    return JSONResponse({
        "stream": stream,
        "page": result["page"],
        "total": result["total"],
        "has_more": result["has_more"],
        "items": _normalize_work_items(stream, result["items"]),
    })


async def _refresh_available_issue_snapshot(lease_owner: str) -> list[dict]:
    global _available_issue_snapshot_value, _available_issue_snapshot_created_at
    global _available_issue_snapshot_retry_at
    try:
        result = await gitea_proxy.available_issue_snapshot()
        shared = await asyncio.to_thread(
            _available_issue_snapshot_store.publish, lease_owner, items=result
        )
        _available_issue_snapshot_value = shared.items
        _available_issue_snapshot_created_at = time.monotonic()
        _available_issue_snapshot_retry_at = None
        return shared.items or []
    except asyncio.CancelledError:
        await asyncio.to_thread(
            _available_issue_snapshot_store.release_refresh, lease_owner
        )
        raise
    except Exception:
        try:
            await asyncio.to_thread(
                _available_issue_snapshot_store.fail_refresh,
                lease_owner,
                retry_at=time.time() + AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS,
            )
        except Exception:
            pass
        raise


def _observe_available_issue_refresh(task: asyncio.Task) -> None:
    global _available_issue_snapshot_retry_at
    if not task.cancelled():
        if task.exception() is not None:
            _available_issue_snapshot_retry_at = (
                time.monotonic() + AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS
            )


async def _start_available_issue_refresh() -> bool:
    """Atomically start this worker's refresh when the shared lease is available."""
    global _available_issue_snapshot_task
    async with _available_issue_snapshot_lock:
        if (
            _available_issue_snapshot_task is not None
            and not _available_issue_snapshot_task.done()
        ):
            return True
        lease_owner = await asyncio.to_thread(
            _available_issue_snapshot_store.try_acquire_refresh,
            lease_seconds=AVAILABLE_ISSUE_SNAPSHOT_LEASE_SECONDS,
        )
        if lease_owner is None:
            return False
        _available_issue_snapshot_task = asyncio.create_task(
            _refresh_available_issue_snapshot(lease_owner)
        )
        _available_issue_snapshot_task.add_done_callback(_observe_available_issue_refresh)
        return True


async def _available_issue_snapshot() -> tuple[list[dict], bool, bool, bool]:
    global _available_issue_snapshot_task
    global _available_issue_snapshot_value, _available_issue_snapshot_created_at
    shared = await asyncio.to_thread(_available_issue_snapshot_store.load)
    wall_now = time.time()
    if shared.items is not None:
        _available_issue_snapshot_value = shared.items
        if (
            shared.created_at is not None
            and wall_now - shared.created_at < AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS
        ):
            _available_issue_snapshot_created_at = time.monotonic()
            return shared.items, False, False, False
        if shared.retry_at is not None and wall_now < shared.retry_at:
            return shared.items, True, False, True
    now = time.monotonic()
    if (
        _available_issue_snapshot_value is not None
        and _available_issue_snapshot_created_at is not None
        and now - _available_issue_snapshot_created_at
        < AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS
    ):
        return _available_issue_snapshot_value, False, False, False
    if (
        _available_issue_snapshot_value is not None
        and _available_issue_snapshot_retry_at is not None
        and now < _available_issue_snapshot_retry_at
    ):
        return _available_issue_snapshot_value, True, False, True
    local_refresh = await _start_available_issue_refresh()
    if _available_issue_snapshot_value is not None:
        return _available_issue_snapshot_value, True, True, False
    if not local_refresh:
        wait_deadline = time.monotonic() + WORK_PAGE_TIMEOUT_SECONDS
        while time.monotonic() < wait_deadline:
            await asyncio.sleep(0.02)
            shared = await asyncio.to_thread(_available_issue_snapshot_store.load)
            if shared.items is not None:
                return shared.items, False, False, False
            if not shared.refreshing:
                local_refresh = await _start_available_issue_refresh()
                if local_refresh:
                    break
        if not local_refresh:
            raise RuntimeError("available issue catalog refresh is owned by another worker")
    try:
        return await asyncio.shield(_available_issue_snapshot_task), False, False, False
    except Exception:
        if _available_issue_snapshot_value is not None:
            return _available_issue_snapshot_value, True, False, True
        raise


def _available_issue_page(
    items: list[dict], page: int, limit: int = 50, query: str = "",
    repositories: list[str] | None = None, labels: list[str] | None = None,
    include_facets: bool = False,
) -> dict:
    facets = {
        "repositories": sorted({str(item.get("repository")) for item in items if item.get("repository")}, key=str.casefold),
        "labels": sorted({str(label) for item in items for label in (item.get("labels") or [])}, key=str.casefold),
    }
    repository_filter = {value.casefold() for value in (repositories or [])}
    label_filter = {value.casefold() for value in (labels or [])}
    if repository_filter:
        items = [item for item in items if str(item.get("repository") or "").casefold() in repository_filter]
    if label_filter:
        items = [item for item in items if label_filter & {str(label).casefold() for label in (item.get("labels") or [])}]
    normalized_query = query.strip().casefold()
    if normalized_query:
        number_query = normalized_query.removeprefix("#")
        items = [
            item for item in items
            if normalized_query in str(item.get("repository") or "").casefold()
            or normalized_query in str(item.get("title") or "").casefold()
            or (number_query.isdigit() and number_query == str(item.get("number") or ""))
        ]
    start = (page - 1) * limit
    page_items = items[start:start + limit]
    result = {
        "items": page_items,
        "page": page,
        "total": len(items),
        "has_more": start + len(page_items) < len(items),
    }
    if include_facets or repository_filter or label_filter:
        result["facets"] = facets
    return result


@app.get("/api/v1/available-issues")
async def available_issues(
    page: int = Query(default=1, ge=1, le=100),
    q: str = Query(default="", max_length=100),
    repository: list[str] = Query(default=[], max_length=100),
    label: list[str] = Query(default=[], max_length=100),
    facets: bool = Query(default=False),
) -> JSONResponse:
    if len(repository) > 10 or len(label) > 10:
        return JSONResponse({"error": "Select at most 10 repositories or labels."}, status_code=422)
    try:
        items, stale, revalidating, refresh_failed = await asyncio.wait_for(
            _available_issue_snapshot(), timeout=WORK_PAGE_TIMEOUT_SECONDS
        )
    except Exception:
        return JSONResponse(
            {"error": "Available issues are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": str(math.ceil(WORK_PAGE_TIMEOUT_SECONDS))},
        )
    result = _available_issue_page(
        items, page, query=q, repositories=repository, labels=label, include_facets=facets
    )
    if stale:
        result["stale"] = True
    if revalidating:
        result["revalidating"] = True
    if refresh_failed:
        result["refresh_failed"] = True
    return JSONResponse(result)


async def _load_context_for_user(user_data: dict) -> dict:
    repo_data, issues_data, prs_data = await asyncio.gather(
        repos(), issues(), pull_requests()
    )
    if not all(field in user_data for field in ("id", "login")):
        raise ContextPayloadError("Gitea current-user response did not include id and login")
    if any(
        data is not None and not isinstance(data, list)
        for data in (repo_data, issues_data, prs_data)
    ):
        raise ContextPayloadError("Gitea collection response was not a list")
    return _context_payload(user_data, repo_data, issues_data, prs_data)


async def _build_live_snapshot(sections: set[str] | None = None) -> dict:
    requested = set(sections or LIVE_SNAPSHOT_SECTIONS)
    results: dict[str, object] = {}
    user_data: dict | None = None
    if requested & {"context", "events"}:
        try:
            user_data = await current_user()
            if not isinstance(user_data, dict) or not user_data.get("login"):
                raise ContextPayloadError("Gitea current-user response was invalid")
        except Exception as exc:
            for section in requested & {"context", "events"}:
                results[section] = exc

    loads: dict[str, Awaitable[Any]] = {}
    if "context" in requested and "context" not in results:
        assert user_data is not None
        loads["context"] = _load_context_for_user(user_data)
    if "events" in requested and "events" not in results:
        assert user_data is not None
        loads["events"] = activity_events(user_data)
    if "notifications" in requested:
        loads["notifications"] = notifications()
    if loads:
        loaded = await asyncio.gather(*loads.values(), return_exceptions=True)
        results.update(zip(loads, loaded))

    context_result = results.get("context")
    events_result = results.get("events")
    notifications_result = results.get("notifications")
    context_ok = "context" in requested and not isinstance(context_result, BaseException)
    events_ok = "events" in requested and not isinstance(events_result, BaseException)
    notifications_ok = (
        "notifications" in requested
        and not isinstance(notifications_result, BaseException)
    )
    notification_items = notifications_result
    notification_pagination = None
    if notifications_ok and isinstance(notifications_result, dict):
        notification_items = notifications_result.get("items")
        notification_pagination = {
            "page": notifications_result.get("page", 1),
            "total": notifications_result.get("total", 0),
            "has_more": notifications_result.get("has_more") is True,
        }
        notifications_ok = isinstance(notification_items, list)
    return {
        "context": context_result if context_ok else None,
        "events": events_result if events_ok else None,
        "notifications": notification_items if notifications_ok else None,
        "notification_pagination": notification_pagination if notifications_ok else None,
        "sections": {
            section: "fresh" if ok else "temporarily unavailable"
            for section, ok in (
                ("context", context_ok),
                ("events", events_ok),
                ("notifications", notifications_ok),
            )
            if section in requested
        },
    }


async def _build_live_snapshot_before_deadline(sections: set[str]) -> dict:
    async with asyncio.timeout(CONTEXT_TIMEOUT_SECONDS):
        return await _build_live_snapshot(sections)


def _record_live_section_failure(section: str) -> None:
    _live_section_failure_count[section] += 1
    delay = min(
        LIVE_SNAPSHOT_RETRY_MAX_SECONDS,
        LIVE_SNAPSHOT_RETRY_BASE_SECONDS
        * (2 ** (_live_section_failure_count[section] - 1)),
    )
    _live_section_retry_at[section] = _live_snapshot_clock() + delay


def _due_live_sections(now: float) -> set[str]:
    due = set()
    for section in LIVE_SNAPSHOT_SECTIONS:
        retry_at = _live_section_retry_at[section]
        if retry_at is not None and now < retry_at:
            continue
        created_at = _live_section_created_at[section]
        if created_at is None or now - created_at >= LIVE_SNAPSHOT_FRESHNESS_SECONDS:
            due.add(section)
    return due


def _apply_shared_live_state(state: LiveSnapshotState) -> None:
    """Replace this worker's fast local view with one coherent SQLite read."""
    global _live_snapshot_value, _live_snapshot_created_at
    global _live_section_created_at, _live_section_failure_count, _live_section_retry_at
    global _live_section_revisions, _live_revision_generation
    global _live_snapshot_refreshing_sections
    _live_snapshot_value = state.value
    _live_section_created_at = state.created_at
    _live_section_failure_count = state.failure_count
    _live_section_retry_at = state.retry_at
    _live_section_revisions = state.revisions
    _live_revision_generation = state.generation
    _live_snapshot_refreshing_sections = state.refreshing_sections
    successful = [value for value in state.created_at.values() if value is not None]
    _live_snapshot_created_at = max(successful) if successful else None


def _apply_shared_live_metadata(state: LiveSnapshotMetadata) -> None:
    """Update coordination state while retaining the matching local snapshot value."""
    global _live_snapshot_created_at
    global _live_section_created_at, _live_section_failure_count, _live_section_retry_at
    global _live_section_revisions, _live_revision_generation
    global _live_snapshot_refreshing_sections
    _live_section_created_at = state.created_at
    _live_section_failure_count = state.failure_count
    _live_section_retry_at = state.retry_at
    _live_section_revisions = state.revisions
    _live_revision_generation = state.generation
    _live_snapshot_refreshing_sections = state.refreshing_sections
    successful = [value for value in state.created_at.values() if value is not None]
    _live_snapshot_created_at = max(successful) if successful else None


async def _sync_shared_live_state() -> None:
    """Load the large shared value only when this worker's copy is out of date."""
    metadata = await asyncio.to_thread(_live_snapshot_store.load_metadata)
    if (
        _live_snapshot_value is None
        or metadata.generation != _live_revision_generation
        or metadata.revisions != _live_section_revisions
    ):
        _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load))
    else:
        _apply_shared_live_metadata(metadata)


def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
    if previous is None:
        return refreshed
    merged = dict(previous)
    sections = dict(previous.get("sections") or {})
    for section, state in (refreshed.get("sections") or {}).items():
        if state == "fresh":
            merged[section] = refreshed.get(section)
            sections[section] = "fresh"
            if section == "notifications":
                merged["notification_pagination"] = refreshed.get(
                    "notification_pagination"
                )
        elif previous.get(section) is not None:
            sections[section] = "stale"
        else:
            merged[section] = None
            sections[section] = "temporarily unavailable"
    merged["sections"] = sections
    return merged


async def _refresh_live_snapshot(sections: set[str], lease_owner: str) -> dict:
    global _live_snapshot_value, _live_snapshot_created_at
    try:
        refreshed = await _build_live_snapshot_before_deadline(sections)
    except asyncio.CancelledError:
        try:
            await asyncio.shield(
                asyncio.to_thread(_live_snapshot_store.release_refresh, lease_owner)
            )
        finally:
            raise
    except Exception:
        for section in sections:
            _record_live_section_failure(section)
        try:
            state = await asyncio.to_thread(
                _live_snapshot_store.fail_refresh,
                lease_owner,
                failure_count=_live_section_failure_count,
                retry_at=_live_section_retry_at,
            )
            _apply_shared_live_state(state)
        except RefreshLeaseLost:
            _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load))
        raise
    now = _live_snapshot_clock()
    refreshed_states = refreshed.get("sections") or {}
    for section in sections:
        if refreshed_states.get(section) == "fresh":
            _live_section_created_at[section] = now
            _live_section_failure_count[section] = 0
            _live_section_retry_at[section] = None
        else:
            _record_live_section_failure(section)
    previous = _live_snapshot_value
    result = _merge_live_snapshot(previous, refreshed)
    result = _without_read_notifications(result)
    changed_sections = set()
    for section in sections:
        if section not in refreshed_states:
            continue
        changed = previous is None or previous.get(section) != result.get(section)
        if section == "notifications":
            changed = changed or previous is None or previous.get(
                "notification_pagination"
            ) != result.get("notification_pagination")
        if changed:
            changed_sections.add(section)
    try:
        state = await asyncio.to_thread(
            _live_snapshot_store.publish_refresh,
            lease_owner,
            value=result,
            created_at=_live_section_created_at,
            failure_count=_live_section_failure_count,
            retry_at=_live_section_retry_at,
            changed_sections=changed_sections,
        )
    except RefreshLeaseLost:
        state = await asyncio.to_thread(_live_snapshot_store.load)
        if state.value is None:
            raise
    _apply_shared_live_state(state)
    assert state.value is not None
    return state.value


def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
    if task.cancelled():
        return
    task.exception()


def _start_live_snapshot_refresh(
    sections: set[str], lease_owner: str | None = None
) -> asyncio.Task:
    global _live_snapshot_task
    global _live_snapshot_refreshing_sections
    if _live_snapshot_task is None or _live_snapshot_task.done():
        if lease_owner is None:
            lease_owner = _live_snapshot_store.try_acquire_refresh(
                sections, lease_seconds=CONTEXT_TIMEOUT_SECONDS + 1.0
            )
            if lease_owner is None:
                raise RuntimeError("live snapshot refresh lease is already held")
        _live_snapshot_refreshing_sections = set(sections)
        _live_snapshot_task = asyncio.create_task(
            _refresh_live_snapshot(sections, lease_owner)
        )
        _live_snapshot_task.add_done_callback(_consume_live_snapshot_failure)
    return _live_snapshot_task


def _live_snapshot_payload(
    value: dict,
    *,
    stale: bool,
    revalidating: bool,
    known_revisions: dict[str, str | None] | None = None,
) -> dict:
    payload = dict(value)
    now = _live_snapshot_clock()
    section_freshness = {}
    for section in LIVE_SNAPSHOT_SECTIONS:
        created_at = _live_section_created_at[section]
        retry_at = _live_section_retry_at[section]
        age = max(0.0, now - created_at) if created_at is not None else 0.0
        retry_in = max(0.0, retry_at - now) if retry_at is not None else 0.0
        section_freshness[section] = {
            "age_seconds": round(age, 3),
            "stale": (value.get("sections") or {}).get(section) != "fresh"
            or created_at is None
            or age >= LIVE_SNAPSHOT_FRESHNESS_SECONDS,
            "revalidating": revalidating
            and section in _live_snapshot_refreshing_sections,
            "degraded": retry_at is not None,
            "retry_in_seconds": math.ceil(retry_in),
        }
    ages = [item["age_seconds"] for item in section_freshness.values()]
    retries = [item["retry_in_seconds"] for item in section_freshness.values() if item["retry_in_seconds"]]
    degraded = any(item["degraded"] for item in section_freshness.values())
    payload["freshness"] = {
        "age_seconds": max(ages, default=0.0),
        "fresh_for_seconds": LIVE_SNAPSHOT_FRESHNESS_SECONDS,
        "stale": stale or any(item["stale"] for item in section_freshness.values()),
        "revalidating": revalidating,
        "degraded": degraded,
        "last_refresh_failed": degraded,
        "retry_in_seconds": min(retries, default=0),
        "sections": section_freshness,
    }
    revision_tokens = {
        section: f"{_live_revision_generation}.{revision}"
        for section, revision in _live_section_revisions.items()
    }
    payload["revisions"] = revision_tokens
    for section, known_revision in (known_revisions or {}).items():
        if known_revision is None or known_revision != revision_tokens[section]:
            continue
        payload.pop(section, None)
        if section == "notifications":
            payload.pop("notification_pagination", None)
    return payload


async def _remove_notifications_from_live_snapshot(thread_ids: list[int]) -> None:
    global _live_snapshot_value, _read_notification_ids
    read_ids = set(thread_ids)
    if not read_ids:
        return
    _read_notification_ids = _read_notification_ids | read_ids
    if _live_snapshot_value is not None:
        retained_notifications = _live_snapshot_value.get("notifications")
        if isinstance(retained_notifications, list):
            updated = dict(_live_snapshot_value)
            updated["notifications"] = [
                notification
                for notification in retained_notifications
                if not isinstance(notification, dict)
                or notification.get("id") not in read_ids
            ]
            if updated["notifications"] != retained_notifications:
                _live_section_revisions["notifications"] += 1
            _live_snapshot_value = updated
    try:
        await asyncio.to_thread(_live_snapshot_store.remove_notifications, read_ids)
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        # The upstream mutation already succeeded; keep process-local filtering.
        pass


def _without_read_notifications(snapshot: dict) -> dict:
    global _read_notification_ids
    snapshot_notifications = snapshot.get("notifications")
    if not isinstance(snapshot_notifications, list):
        return snapshot
    returned_ids = {
        notification.get("id")
        for notification in snapshot_notifications
        if isinstance(notification, dict)
    }
    updated = dict(snapshot)
    updated["notifications"] = [
        notification
        for notification in snapshot_notifications
        if not isinstance(notification, dict)
        or notification.get("id") not in _read_notification_ids
    ]
    _read_notification_ids = _read_notification_ids.intersection(returned_ids)
    return updated


async def _live_snapshot_response(
    context_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
    events_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
    notifications_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
) -> JSONResponse:
    """Return a freshness-bounded snapshot and share identical upstream loads."""
    global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at
    known_revisions = {
        "context": context_revision,
        "events": events_revision,
        "notifications": notifications_revision,
    }
    await _sync_shared_live_state()
    now = _live_snapshot_clock()
    due_sections = _due_live_sections(now)
    if _live_snapshot_value is not None and not due_sections:
        return JSONResponse(
            _live_snapshot_payload(
                _live_snapshot_value,
                stale=any(
                    state != "fresh"
                    for state in _live_snapshot_value.get("sections", {}).values()
                ),
                revalidating=False,
                known_revisions=known_revisions,
            )
        )
    if not due_sections:
        retries = [
            retry_at - now
            for retry_at in _live_section_retry_at.values()
            if retry_at is not None and retry_at > now
        ]
        return JSONResponse(
            {"error": "Gitea live snapshot is temporarily unavailable"},
            status_code=503,
            headers={"Retry-After": str(max(1, math.ceil(min(retries, default=1.0))))},
        )

    task = _live_snapshot_task
    if task is None or task.done():
        lease_owner = await asyncio.to_thread(
            _live_snapshot_store.try_acquire_refresh,
            due_sections,
            lease_seconds=CONTEXT_TIMEOUT_SECONDS + 1.0,
        )
        if lease_owner is not None:
            latest = await asyncio.to_thread(_live_snapshot_store.load)
            _apply_shared_live_state(latest)
            due_sections = _due_live_sections(_live_snapshot_clock())
            if not due_sections:
                await asyncio.to_thread(
                    _live_snapshot_store.release_refresh, lease_owner
                )
                return JSONResponse(
                    _live_snapshot_payload(
                        _live_snapshot_value,
                        stale=any(
                            state != "fresh"
                            for state in (_live_snapshot_value or {}).get("sections", {}).values()
                        ),
                        revalidating=False,
                        known_revisions=known_revisions,
                    )
                )
            task = _start_live_snapshot_refresh(due_sections, lease_owner)
        else:
            shared = await asyncio.to_thread(_live_snapshot_store.load)
            _apply_shared_live_state(shared)
            if _live_snapshot_value is not None:
                return JSONResponse(
                    _live_snapshot_payload(
                        _live_snapshot_value,
                        stale=bool(_due_live_sections(_live_snapshot_clock())),
                        revalidating=bool(shared.refreshing_sections),
                        known_revisions=known_revisions,
                    )
                )
            deadline = time.perf_counter() + CONTEXT_TIMEOUT_SECONDS
            while time.perf_counter() < deadline:
                await asyncio.sleep(0.01)
                shared = await asyncio.to_thread(_live_snapshot_store.load)
                if shared.value is not None:
                    _apply_shared_live_state(shared)
                    return JSONResponse(
                        _live_snapshot_payload(
                            shared.value,
                            stale=bool(_due_live_sections(_live_snapshot_clock())),
                            revalidating=bool(shared.refreshing_sections),
                            known_revisions=known_revisions,
                        )
                    )
            return JSONResponse(
                {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
                status_code=503,
                headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
            )
    assert task is not None
    if _live_snapshot_value is not None:
        return JSONResponse(
            _live_snapshot_payload(
                _live_snapshot_value,
                stale=True,
                revalidating=True,
                known_revisions=known_revisions,
            )
        )
    try:
        result = await asyncio.shield(task)
        return JSONResponse(
            _live_snapshot_payload(
                result,
                stale=False,
                revalidating=False,
                known_revisions=known_revisions,
            )
        )
    except TimeoutError:
        return JSONResponse(
            {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
            status_code=503,
            headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
        )
    except Exception:
        return JSONResponse(
            {"error": "Gitea live snapshot is temporarily unavailable"},
            status_code=503,
        )


@app.get("/api/v1/live")
async def live_snapshot(
    context_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
    events_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
    notifications_revision: str | None = Query(
        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
    ),
) -> JSONResponse:
    try:
        return await _live_snapshot_response(
            context_revision=context_revision,
            events_revision=events_revision,
            notifications_revision=notifications_revision,
        )
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        return JSONResponse(
            {"error": "Gitea live snapshot state is temporarily unavailable"},
            status_code=503,
            headers={"Retry-After": "1"},
        )



@app.get("/api/v1/events")
async def event_stream():
    try:
        return await asyncio.wait_for(
            activity_events(), timeout=EVENT_STREAM_TIMEOUT_SECONDS
        )
    except TimeoutError:
        return JSONResponse(
            {
                "error": (
                    "Gitea event stream request timed out after "
                    f"{EVENT_STREAM_TIMEOUT_SECONDS:g}s"
                )
            },
            status_code=503,
            headers={
                "Retry-After": str(
                    max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
                )
            },
        )
    except Exception:
        return JSONResponse(
            {"error": "Gitea event stream is temporarily unavailable"},
            status_code=503,
            headers={
                "Retry-After": str(
                    max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
                )
            },
        )


async def _mark_notification_read_result(thread_id: int) -> tuple[int, bool]:
    try:
        await asyncio.wait_for(
            mark_notification_read(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except Exception:
        return thread_id, False
    return thread_id, True


@app.patch("/api/v1/notifications/read")
async def read_notifications(batch: NotificationReadBatch) -> JSONResponse:
    thread_ids = list(dict.fromkeys(batch.ids))

    semaphore = asyncio.Semaphore(BULK_NOTIFICATION_CONCURRENCY)

    async def mark_within_limit(thread_id: int) -> tuple[int, bool]:
        async with semaphore:
            return await _mark_notification_read_result(thread_id)

    tasks = [asyncio.create_task(mark_within_limit(thread_id)) for thread_id in thread_ids]
    done, pending = await asyncio.wait(
        tasks, timeout=BULK_NOTIFICATION_DEADLINE_SECONDS
    )
    for task in pending:
        task.cancel()
    if pending:
        await asyncio.gather(*pending, return_exceptions=True)

    succeeded_ids = {
        thread_id
        for task in done
        if not task.cancelled() and task.exception() is None
        for thread_id, succeeded in [task.result()]
        if succeeded
    }
    await _remove_notifications_from_live_snapshot(
        [thread_id for thread_id in thread_ids if thread_id in succeeded_ids]
    )
    failed = [thread_id for thread_id in thread_ids if thread_id not in succeeded_ids]
    return JSONResponse(
        {
            "marked": [thread_id for thread_id in thread_ids if thread_id in succeeded_ids],
            "failed": failed,
        },
        headers={"Retry-After": "1"} if failed else None,
    )


@app.get("/api/v1/notifications")
async def notification_page(page: int = Query(default=1, ge=1)) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.notification_page(page),
            timeout=NOTIFICATION_PAGE_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Unread updates timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Unread updates are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/notifications/snapshot")
async def notification_snapshot() -> JSONResponse:
    try:
        result = await gitea_proxy.unread_notification_snapshot(
            deadline_seconds=NOTIFICATION_PAGE_TIMEOUT_SECONDS
        )
    except Exception:
        return JSONResponse(
            {"error": "Complete unread updates are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/notifications/{thread_id}")
async def notification_thread_detail(
    thread_id: int = PathParam(gt=0),
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            notification_detail(thread_id),
            timeout=NOTIFICATION_DETAIL_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Loading the update timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The update is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/notifications/{thread_id}/conversation")
async def notification_thread_conversation(
    thread_id: int = PathParam(gt=0),
    page: int | None = Query(default=None, ge=1),
    limit: int = Query(default=20, ge=1, le=50),
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.notification_conversation_page(thread_id, page, limit),
            timeout=NOTIFICATION_DETAIL_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Loading older messages timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Older messages are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/notifications/{thread_id}/read")
async def read_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
    try:
        await asyncio.wait_for(
            mark_notification_read(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Marking the update read timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The update could not be marked read. Please retry."},
            status_code=503,
        )
    await _remove_notifications_from_live_snapshot([thread_id])
    return JSONResponse({"id": thread_id, "status": "read"})


@app.patch("/api/v1/notifications/{thread_id}/unread")
async def unread_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
    try:
        await asyncio.wait_for(
            gitea_proxy.mark_notification_unread(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Restoring the update timed out. Retry Undo."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The update could not be restored. Retry Undo."},
            status_code=503,
        )
    return JSONResponse({"id": thread_id, "status": "unread"})


@app.post("/api/v1/notifications/{thread_id}/mute")
async def mute_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
    try:
        await asyncio.wait_for(
            gitea_proxy.mute_notification(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Muting future updates timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Future updates could not be muted. Please retry."},
            status_code=503,
        )
    try:
        await asyncio.wait_for(
            mark_notification_read(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {
                "id": thread_id,
                "muted": True,
                "status": "unread",
                "error": "Future updates are muted; current item is still unread. Retry mark read & next.",
            },
            status_code=409,
        )
    await _remove_notifications_from_live_snapshot([thread_id])
    return JSONResponse({"id": thread_id, "muted": True, "status": "read"})


@app.patch("/api/v1/notifications/{thread_id}/later")
async def defer_notification(
    payload: NotificationLaterRequest,
    thread_id: int = PathParam(gt=0),
) -> JSONResponse:
    login = await _confirmed_login()
    try:
        item = await asyncio.wait_for(
            gitea_proxy.resolve_work_route("update", None, None, thread_id),
            timeout=NOTIFICATION_DETAIL_TIMEOUT_SECONDS,
        )
        repository = item.get("repository")
        if not isinstance(repository, str) or not repository:
            raise ValueError("The update has no repository identity")
        item_id = f"update:{repository}::{thread_id}"
        result = await asyncio.to_thread(
            _later_store().apply,
            login,
            f"push-tomorrow:{thread_id}:{payload.wake_at}",
            "defer",
            item_id,
            wake_at=payload.wake_at,
        )
    except gitea_proxy.WorkRouteUnavailableError:
        return JSONResponse(
            {"error": "This update is no longer available to defer."},
            status_code=409,
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except TimeoutError:
        return JSONResponse(
            {"error": "Deferring the update timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
        return JSONResponse(
            {"error": "Later synchronization is unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse({
        "id": thread_id,
        "status": "deferred",
        "item_id": item_id,
        "wake_at": payload.wake_at,
        "revision": result["revision"],
    })


@app.post("/api/v1/notifications/{thread_id}/acknowledge")
async def acknowledge_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.acknowledge_notification(thread_id),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except TimeoutError:
        return JSONResponse(
            {"error": "Acknowledging the update timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except ValueError:
        return JSONResponse(
            {"error": "This update has no comment that can be acknowledged."},
            status_code=422,
        )
    except Exception:
        return JSONResponse(
            {"error": "The update could not be acknowledged. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    await _remove_notifications_from_live_snapshot([thread_id])
    return JSONResponse(result)


@app.post("/api/v1/notifications/{thread_id}/reply", status_code=201)
async def reply_to_notification(
    reply: NotificationReply,
    thread_id: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
) -> JSONResponse:
    try:
        result = await _run_idempotent_authored_action(
            gitea_proxy.reply_to_notification(thread_id, reply.body),
            idempotency_key=idempotency_key,
            fingerprint=("notification-reply", thread_id, reply.body),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {
                "error": (
                    "The reply could not be posted. Your draft is safe; please retry."
                )
            },
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.post("/api/v1/notifications/{thread_id}/attachments", status_code=201)
async def attach_to_notification(
    request: Request,
    thread_id: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
) -> JSONResponse:
    try:
        if request.headers.get("content-type", "").lower().startswith("multipart/form-data"):
            form = await request.form()
            uploaded = form.get("file")
            if not isinstance(uploaded, UploadFile):
                raise ValueError("screenshot file is required")
            filename = str(uploaded.filename or "")
            content_type = str(uploaded.content_type or "")
            content = _validate_binary_attachment(filename, content_type, await uploaded.read())
        else:
            attachment = IssueAttachment.model_validate(await request.json())
            filename = attachment.filename
            content_type = attachment.content_type
            content = attachment.content()
        content = await _sanitize_attachment(content_type, content)
    except (ValueError, ValidationError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    async def upload_attachment():
        result = await gitea_proxy.upload_notification_attachment(
            thread_id, filename, content_type, content
        )
        safe_name = (
            result["name"].replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
            .replace("\r", " ").replace("\n", " ")
        )
        safe_url = result["url"].replace("<", "%3C").replace(">", "%3E")
        result["markdown"] = f"![{safe_name}](<{safe_url}>)"
        return result

    try:
        result = await _run_idempotent_authored_action(
            upload_attachment(), idempotency_key=idempotency_key,
            fingerprint=("notification-attachment", thread_id, filename, content_type,
                         hashlib.sha256(content).hexdigest()),
            timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except Exception:
        return JSONResponse(
            {"error": "The screenshot could not be uploaded. Your draft is safe; please retry."},
            status_code=503, headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/claim")
async def claim_available_issue(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    global _available_issue_snapshot_value
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.claim_available_issue(repository, number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        return JSONResponse(
            {"error": "This issue was already claimed or is no longer open."},
            status_code=409,
        )
    except Exception:
        return JSONResponse(
            {"error": "The issue could not be assigned. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    retained = _available_issue_snapshot_value
    shared = await asyncio.to_thread(
        _available_issue_snapshot_store.remove_claimed, repository, number
    )
    if shared.items is not None:
        _available_issue_snapshot_value = shared.items
    elif retained is not None:
        _available_issue_snapshot_value = [
            item for item in retained
            if item.get("repository") != repository or item.get("number") != number
        ]
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/reopen")
async def reopen_closed_issue(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.reopen_issue(repository, number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        return JSONResponse(
            {"error": "This issue is no longer closed or cannot be resumed."},
            status_code=409,
        )
    except Exception:
        return JSONResponse(
            {"error": "The issue could not be reopened. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/release")
async def release_assigned_issue(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    global _available_issue_snapshot_value, _available_issue_snapshot_created_at
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.release_assigned_issue(repository, number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except Exception:
        return JSONResponse(
            {"error": "The assignment could not be released. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    if result.get("available"):
        await asyncio.to_thread(_available_issue_snapshot_store.invalidate)
        _available_issue_snapshot_value = None
        _available_issue_snapshot_created_at = None
    return JSONResponse(result)


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/handoff-candidates")
async def issue_handoff_candidates(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    access: Literal["assigned", "filed"] = Query(default="assigned"),
) -> JSONResponse:
    repository = f"{owner}/{repo}"

    async def load_candidates():
        authorized = await (
            gitea_proxy.is_authored_issue(repository, number)
            if access == "filed"
            else gitea_proxy.is_assigned_issue(repository, number)
        )
        if not authorized:
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        return await gitea_proxy.issue_handoff_candidates(repository)

    try:
        result = await asyncio.wait_for(
            load_candidates(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Teammates could not be loaded. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/repos/{owner}/{repo}/mention-candidates")
async def repository_mention_candidates(
    owner: str,
    repo: str,
    q: str = Query(min_length=2, max_length=39, pattern=r"^[A-Za-z0-9_.-]+$"),
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.mention_candidates(
                f"{owner}/{repo}", q.casefold(), limit=8
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "Teammate suggestions are temporarily unavailable."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, headers={"Cache-Control": "no-store"})


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/handoff")
async def handoff_assigned_issue(
    handoff: IssueHandoff,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.handoff_assigned_issue(
                repository, number, handoff.recipient
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        return JSONResponse(
            {"error": "The issue or recipient changed. Reload before handing off."},
            status_code=409,
        )
    except Exception:
        return JSONResponse(
            {"error": "The handoff could not be confirmed. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/reassign")
async def reassign_authored_issue(
    reassignment: IssueReassignment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.reassign_authored_issue(
                repository,
                number,
                reassignment.recipient,
                reassignment.expected_assignees,
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        return JSONResponse(
            {"error": "The delegate changed. Reload before reassigning."},
            status_code=409,
        )
    except Exception:
        return JSONResponse(
            {"error": "The reassignment could not be confirmed. The current delegate is unchanged; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/detail")
async def assigned_issue_detail(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    access: Literal["assigned", "filed"] = Query(default="assigned"),
):
    repository = f"{owner}/{repo}"

    async def load_assigned_issue():
        authorized = await (
            gitea_proxy.is_authored_issue(repository, number)
            if access == "filed"
            else gitea_proxy.is_assigned_issue(repository, number)
        )
        if not authorized:
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        detail = await gitea_proxy.issue_detail(repository, number)
        return {**detail, "read_only": True} if access == "filed" else detail

    try:
        return await asyncio.wait_for(
            load_assigned_issue(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Loading the issue timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The issue is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


async def _mutate_assigned_issue_blocker(
    update: IssueBlockerUpdate,
    owner: str,
    repo: str,
    number: int,
    remove: bool,
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.mutate_assigned_issue_dependency(
                f"{owner}/{repo}", number, update.repository, update.number, remove
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except gitea_proxy.IssueDependencyInvalidError as exc:
        return JSONResponse({"error": str(exc)}, status_code=422)
    except Exception:
        return JSONResponse(
            {"error": "The blocker change could not be confirmed. Reload the issue before retrying."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, headers={"Cache-Control": "no-store"})


@app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/blockers")
async def add_assigned_issue_blocker(
    update: IssueBlockerUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    return await _mutate_assigned_issue_blocker(update, owner, repo, number, False)


@app.delete("/api/v1/repos/{owner}/{repo}/issues/{number}/blockers")
async def remove_assigned_issue_blocker(
    update: IssueBlockerUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    return await _mutate_assigned_issue_blocker(update, owner, repo, number, True)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/blockers")
async def set_assigned_issue_blocker_state(
    update: IssueBlockerDesiredState,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    return await _mutate_assigned_issue_blocker(
        update, owner, repo, number, not update.present
    )


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/content")
async def update_assigned_issue_content(
    update: IssueContentUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    access: Literal["assigned", "filed"] = Query(default="assigned"),
):
    repository = f"{owner}/{repo}"
    try:
        update_issue = (
            gitea_proxy.update_authored_issue
            if access == "filed"
            else gitea_proxy.update_assigned_issue
        )
        result = await asyncio.wait_for(
            update_issue(
                repository,
                number,
                update.title,
                update.body,
                update.expected_updated_at,
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueEditConflictError:
        return JSONResponse(
            {"error": "This issue changed in Gitea. Your draft is safe; reload the latest issue before saving."},
            status_code=409,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The issue could not be updated. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/due-date")
async def update_assigned_issue_due_date(
    update: IssueDueDateUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.update_assigned_issue_due_date(
                repository, number, update.due_date
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except Exception:
        return JSONResponse(
            {"error": "The due date could not be updated. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/milestone")
async def update_assigned_issue_milestone(
    update: IssueMilestoneUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.update_assigned_issue_milestone(
                repository, number, update.milestone_id
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except ValueError as exc:
        if str(exc) == "Unknown open repository milestone":
            raise HTTPException(status_code=422, detail=str(exc))
        return JSONResponse(
            {"error": "The milestone could not be updated. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The milestone could not be updated. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/release-plan")
async def update_assigned_issue_release_plan(
    update: IssueReleasePlanUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.update_assigned_issue_release_plan(
                repository, number, update.milestone_id, update.due_date
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    except ValueError as exc:
        if str(exc) == "Unknown open repository milestone":
            raise HTTPException(status_code=422, detail=str(exc))
        return JSONResponse(
            {"error": "The release plan could not be confirmed. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The release plan could not be updated. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/repos/{owner}/{repo}/issue-filing-metadata")
async def repository_issue_filing_metadata(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_metadata():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        names = ("labels", "milestones", "templates")
        loaders = (
            gitea_proxy.repo_labels(repository),
            gitea_proxy.repo_milestones(repository),
            gitea_proxy.repo_issue_templates(repository),
        )
        results = await asyncio.gather(*loaders, return_exceptions=True)
        messages = {
            "labels": "Labels could not be loaded.",
            "milestones": "Milestones could not be loaded.",
            "templates": "Issue types could not be loaded.",
        }
        payload = {}
        for name, result in zip(names, results):
            if isinstance(result, BaseException):
                payload[name] = {
                    "available": False,
                    "items": [],
                    "error": messages[name],
                }
            else:
                payload[name] = {"available": True, "items": result}
        return JSONResponse(payload, headers={"Cache-Control": "no-store"})

    try:
        return await asyncio.wait_for(
            load_metadata(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Issue filing details could not be loaded. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1", "Cache-Control": "no-store"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/milestones")
async def repository_milestones(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_milestones():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        return await gitea_proxy.repo_milestones(repository)

    try:
        return await asyncio.wait_for(
            load_milestones(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Milestones could not be loaded. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/labels")
async def repository_labels(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_labels():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        return await gitea_proxy.repo_labels(repository)

    try:
        return await asyncio.wait_for(load_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Labels could not be loaded. Issue creation is still available."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/issue-templates")
async def repository_issue_templates(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_templates():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        return await gitea_proxy.repo_issue_templates(repository)

    try:
        return await asyncio.wait_for(load_templates(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Issue types could not be loaded. Blank issue creation is still available."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/issue-assignees")
async def repository_issue_assignees(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_assignees():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        candidates = [
            item
            for item in await gitea_proxy.issue_handoff_candidates(repository)
            if isinstance(item, dict)
            and isinstance(item.get("login"), str)
            and re.fullmatch(r"[A-Za-z0-9_.-]+", item["login"])
        ][:25]
        return JSONResponse(candidates, headers={"Cache-Control": "no-store"})

    try:
        return await asyncio.wait_for(
            load_assignees(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Owners could not be loaded. You can still assign the issue to yourself."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.post("/api/v1/repos/{owner}/{repo}/issues", status_code=201)
async def create_assigned_issue(
    creation: IssueCreation,
    owner: str,
    repo: str,
    idempotency_key: str | None = Header(default=None, max_length=128),
):
    repository = f"{owner}/{repo}"

    async def create_issue():
        user, accessible = await asyncio.gather(
            gitea_proxy.current_user(), gitea_proxy.repository_access(repository)
        )
        login = user.get("login") if isinstance(user, dict) else None
        if not login or accessible is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        assignee = None if creation.unassigned else login
        if not creation.unassigned and creation.assignee and creation.assignee != login:
            eligible = {
                item["login"]
                for item in await gitea_proxy.issue_handoff_candidates(repository)
            }
            if creation.assignee not in eligible:
                raise HTTPException(
                    status_code=422, detail="Selected owner is no longer eligible"
                )
            assignee = creation.assignee
        if creation.label_ids:
            available_labels = await gitea_proxy.repo_labels(repository)
            valid_label_ids = {
                item.get("id")
                for item in available_labels
                if isinstance(item, dict) and isinstance(item.get("id"), int)
            }
            if any(label_id not in valid_label_ids for label_id in creation.label_ids):
                raise HTTPException(status_code=422, detail="Unknown repository label")
        if creation.milestone_id is not None:
            available_milestones = await gitea_proxy.repo_milestones(repository)
            valid_milestone_ids = {
                item.get("id")
                for item in available_milestones
                if isinstance(item, dict) and isinstance(item.get("id"), int)
            }
            if creation.milestone_id not in valid_milestone_ids:
                raise HTTPException(
                    status_code=422, detail="Unknown open repository milestone"
                )
        if creation.milestone_id is not None or creation.due_date is not None:
            return await gitea_proxy.create_issue(
                repository,
                creation.title,
                creation.body,
                assignee,
                creation.label_ids,
                creation.milestone_id,
                creation.due_date,
            )
        return await gitea_proxy.create_issue(
            repository, creation.title, creation.body, assignee, creation.label_ids
        )

    try:
        result = await _run_idempotent_authored_action(
            create_issue(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "issue-create",
                repository,
                creation.title,
                creation.body,
                creation.assignee,
                creation.unassigned,
                tuple(creation.label_ids),
                creation.milestone_id,
                creation.due_date,
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The issue could not be created. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/comments")
async def assigned_issue_conversation(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    page: int | None = Query(default=None, ge=1, le=100),
    limit: int = Query(default=20, ge=1, le=50),
    access: Literal["assigned", "filed"] = Query(default="assigned"),
):
    repository = f"{owner}/{repo}"

    async def load_conversation():
        authorized = await (
            gitea_proxy.is_authored_issue(repository, number)
            if access == "filed"
            else gitea_proxy.is_assigned_issue(repository, number)
        )
        if not authorized:
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        return await gitea_proxy.issue_conversation_page(repository, number, page, limit)

    try:
        return await asyncio.wait_for(
            load_conversation(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The conversation could not be loaded. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/comments", status_code=201)
async def comment_on_assigned_issue(
    comment: IssueComment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
    access: Literal["assigned", "filed"] = Query(default="assigned"),
):
    repository = f"{owner}/{repo}"

    async def post_comment():
        authorized = await (
            gitea_proxy.is_authored_issue(repository, number)
            if access == "filed"
            else gitea_proxy.is_assigned_issue(repository, number)
        )
        if not authorized:
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        return await gitea_proxy.comment_on_issue(repository, number, comment.body)

    try:
        result = await _run_idempotent_authored_action(
            post_comment(),
            idempotency_key=idempotency_key,
            fingerprint=("issue-comment", repository, number, comment.body),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The comment could not be posted. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


async def _edit_conversation_comment(
    repository: str, number: int, comment_id: int, body: str
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.edit_owned_comment(repository, number, comment_id, body),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.CommentMutationForbiddenError as exc:
        raise HTTPException(
            status_code=403, detail="You can only change your own comments"
        ) from exc
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The comment could not be updated. Your edit is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


async def _delete_conversation_comment(
    request: Request,
    step_up_grant: str | None,
    repository: str,
    number: int,
    comment_id: int,
) -> JSONResponse:
    target = f"{repository}#{number}:{comment_id}"
    await _require_step_up(
        request,
        step_up_grant,
        action="delete_comment",
        target=target,
    )
    try:
        journal = _security_event_store()
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "comment_deleted",
            target=target,
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"error": "Security activity is temporarily unavailable. The comment was not deleted."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    try:
        result = await asyncio.wait_for(
            gitea_proxy.delete_owned_comment(repository, number, comment_id),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.CommentMutationForbiddenError as exc:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        raise HTTPException(
            status_code=403, detail="You can only change your own comments"
        ) from exc
    except HTTPException:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        raise
    except Exception:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"error": "The comment deletion could not be confirmed. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        # Upstream deletion is authoritative; pending evidence remains truthful.
        pass
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/comments/{comment_id}")
async def edit_assigned_issue_comment(
    comment: IssueComment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"
    if not await gitea_proxy.is_assigned_issue(repository, number):
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    return await _edit_conversation_comment(repository, number, comment_id, comment.body)


@app.delete("/api/v1/repos/{owner}/{repo}/issues/{number}/comments/{comment_id}")
async def delete_assigned_issue_comment(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    if not await gitea_proxy.is_assigned_issue(repository, number):
        raise HTTPException(status_code=404, detail="Assigned issue not found")
    return await _delete_conversation_comment(
        request, step_up_grant, repository, number, comment_id
    )


@app.patch("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments/{comment_id}")
async def edit_assigned_pull_comment(
    comment: IssueComment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"
    if not await gitea_proxy.is_assigned_pull(repository, number):
        raise HTTPException(status_code=404, detail="Assigned pull request not found")
    return await _edit_conversation_comment(repository, number, comment_id, comment.body)


@app.delete("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments/{comment_id}")
async def delete_assigned_pull_comment(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    if not await gitea_proxy.is_assigned_pull(repository, number):
        raise HTTPException(status_code=404, detail="Assigned pull request not found")
    return await _delete_conversation_comment(
        request, step_up_grant, repository, number, comment_id
    )


@app.patch("/api/v1/notifications/{thread_id}/comments/{comment_id}")
async def edit_notification_comment(
    comment: IssueComment,
    thread_id: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
):
    try:
        repository, number = await gitea_proxy.notification_conversation_target(thread_id)
    except Exception as exc:
        raise HTTPException(status_code=404, detail="Notification conversation not found") from exc
    return await _edit_conversation_comment(repository, number, comment_id, comment.body)


@app.delete("/api/v1/notifications/{thread_id}/comments/{comment_id}")
async def delete_notification_comment(
    request: Request,
    thread_id: int = PathParam(gt=0),
    comment_id: int = PathParam(gt=0),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    try:
        repository, number = await gitea_proxy.notification_conversation_target(thread_id)
    except Exception as exc:
        raise HTTPException(status_code=404, detail="Notification conversation not found") from exc
    return await _delete_conversation_comment(
        request, step_up_grant, repository, number, comment_id
    )


@app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/attachments", status_code=201)
async def attach_to_assigned_issue(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
):
    repository = f"{owner}/{repo}"
    try:
        if request.headers.get("content-type", "").lower().startswith("multipart/form-data"):
            form = await request.form()
            uploaded = form.get("file")
            if not isinstance(uploaded, UploadFile):
                raise ValueError("screenshot file is required")
            filename = str(uploaded.filename or "")
            content_type = str(uploaded.content_type or "")
            content = _validate_binary_attachment(filename, content_type, await uploaded.read())
        else:
            attachment = IssueAttachment.model_validate(await request.json())
            filename = attachment.filename
            content_type = attachment.content_type
            content = attachment.content()
        content = await _sanitize_attachment(content_type, content)
    except (ValueError, ValidationError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    async def upload_attachment():
        result = await gitea_proxy.upload_assigned_issue_attachment(
            repository,
            number,
            filename,
            content_type,
            content,
        )
        safe_name = (
            result["name"].replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
            .replace("\r", " ").replace("\n", " ")
        )
        safe_url = result["url"].replace("<", "%3C").replace(">", "%3E")
        result["markdown"] = f"![{safe_name}](<{safe_url}>)"
        return result

    try:
        result = await _run_idempotent_authored_action(
            upload_attachment(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "issue-attachment",
                repository,
                number,
                filename,
                content_type,
                hashlib.sha256(content).hexdigest(),
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError as exc:
        raise HTTPException(status_code=404, detail="Assigned issue not found") from exc
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The screenshot could not be uploaded. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/attachments", status_code=201)
async def attach_to_assigned_pull(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
):
    repository = f"{owner}/{repo}"
    try:
        if request.headers.get("content-type", "").lower().startswith("multipart/form-data"):
            form = await request.form()
            uploaded = form.get("file")
            if not isinstance(uploaded, UploadFile):
                raise ValueError("screenshot file is required")
            filename = str(uploaded.filename or "")
            content_type = str(uploaded.content_type or "")
            content = _validate_binary_attachment(filename, content_type, await uploaded.read())
        else:
            attachment = IssueAttachment.model_validate(await request.json())
            filename = attachment.filename
            content_type = attachment.content_type
            content = attachment.content()
        content = await _sanitize_attachment(content_type, content)
    except (ValueError, ValidationError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    async def upload_attachment():
        result = await gitea_proxy.upload_assigned_pull_attachment(
            repository, number, filename, content_type, content
        )
        safe_name = (
            result["name"].replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
            .replace("\r", " ").replace("\n", " ")
        )
        safe_url = result["url"].replace("<", "%3C").replace(">", "%3E")
        result["markdown"] = f"![{safe_name}](<{safe_url}>)"
        return result

    try:
        result = await _run_idempotent_authored_action(
            upload_attachment(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "pull-attachment", repository, number, filename, content_type,
                hashlib.sha256(content).hexdigest(),
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError as exc:
        raise HTTPException(status_code=404, detail="Assigned pull request not found") from exc
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The screenshot could not be uploaded. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/close")
async def close_assigned_issue(
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    access: Literal["assigned", "filed"] = "assigned",
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    await _require_step_up(
        request,
        step_up_grant,
        action="close_issue",
        target=f"{repository}#{number}",
    )

    target = f"{repository}#{number}"
    try:
        authorized = await asyncio.wait_for(
            (
                gitea_proxy.is_open_authored_issue(repository, number)
                if access == "filed"
                else gitea_proxy.is_assigned_issue(repository, number)
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
        if not authorized:
            detail = (
                "Open authored issue not found"
                if access == "filed"
                else "Assigned issue not found"
            )
            raise HTTPException(status_code=404, detail=detail)
        journal = _security_event_store()
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "issue_closed",
            target=target,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The issue could not be closed. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )

    try:
        result = await asyncio.wait_for(
            gitea_proxy.close_issue(repository, number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except Exception:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"error": "The issue could not be closed. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )

    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        # The close result is authoritative; the pending reservation preserves
        # evidence without telling the operator the issue remains open.
        pass
    return result


@app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/labels")
async def assigned_issue_label_options(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"

    async def load_labels():
        if not await gitea_proxy.is_assigned_issue(repository, number):
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        return await gitea_proxy.repo_labels(repository)

    try:
        return await asyncio.wait_for(load_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Labels could not be loaded. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/labels")
async def update_assigned_issue_labels(
    update: IssueLabelUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    repository = f"{owner}/{repo}"

    async def update_labels():
        if not await gitea_proxy.is_assigned_issue(repository, number):
            raise HTTPException(status_code=404, detail="Assigned issue not found")
        available = await gitea_proxy.repo_labels(repository)
        valid_ids = {
            item.get("id") for item in available
            if isinstance(item, dict) and isinstance(item.get("id"), int)
        }
        if any(label_id not in valid_ids for label_id in update.label_ids):
            raise HTTPException(status_code=422, detail="Unknown repository label")
        return await gitea_proxy.update_issue_labels(repository, number, update.label_ids)

    try:
        return await asyncio.wait_for(update_labels(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Labels could not be updated. Your selection is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/review")
async def review_detail(owner: str, repo: str, number: int):
    async def load_requested_review():
        repository = f"{owner}/{repo}"
        if not await is_requested_review(repository, number):
            raise HTTPException(status_code=404, detail="Review request not found")
        return await pull_review_detail(repository, number)

    try:
        return await asyncio.wait_for(
            load_requested_review(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Pull request review details timed out. Please retry."},
            status_code=503,
            headers={
                "Retry-After": str(
                    max(1, math.ceil(REVIEW_DETAIL_TIMEOUT_SECONDS))
                )
            },
        )
    except Exception:
        return JSONResponse(
            {"error": "Pull request review details are temporarily unavailable"},
            status_code=503,
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/review/checks")
async def requested_review_checks(
    owner: str, repo: str, number: int = PathParam(gt=0)
):
    repository = f"{owner}/{repo}"

    async def load_checks():
        if not await is_requested_review(repository, number):
            raise HTTPException(status_code=404, detail="Review request not found")
        return await gitea_proxy.pull_check_status(repository, number)

    try:
        return await asyncio.wait_for(
            load_checks(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Refreshing review checks timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Review checks are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/detail")
async def assigned_pull_detail(owner: str, repo: str, number: int = PathParam(gt=0)):
    repository = f"{owner}/{repo}"

    async def load_assigned_pull():
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        return await gitea_proxy.pull_completion_detail(repository, number)

    try:
        return await asyncio.wait_for(
            load_assigned_pull(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Loading the pull request timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "The pull request is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/review-data")
async def assigned_pull_review_data(
    owner: str, repo: str, number: int = PathParam(gt=0)
):
    repository = f"{owner}/{repo}"

    async def load_assigned_pull_review():
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        return await gitea_proxy.pull_completion_review(repository, number)

    try:
        return await asyncio.wait_for(
            load_assigned_pull_review(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Loading review and merge data timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Review and merge data is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/checks")
async def assigned_pull_checks(
    owner: str, repo: str, number: int = PathParam(gt=0)
):
    repository = f"{owner}/{repo}"

    async def load_checks():
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        return await gitea_proxy.pull_check_status(repository, number)

    try:
        return await asyncio.wait_for(
            load_checks(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except TimeoutError:
        return JSONResponse(
            {"error": "Refreshing pull request checks timed out. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    except Exception:
        return JSONResponse(
            {"error": "Pull request checks are temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/handoff-candidates")
async def pull_handoff_candidates(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    repository = f"{owner}/{repo}"
    try:
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        result = await asyncio.wait_for(
            gitea_proxy.pull_handoff_candidates(repository),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "Teammates could not be loaded. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, headers={"Cache-Control": "no-store"})


@app.patch("/api/v1/repos/{owner}/{repo}/pulls/{number}/handoff")
async def handoff_assigned_pull(
    handoff: IssueHandoff,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.handoff_assigned_pull(
                f"{owner}/{repo}", number, handoff.recipient
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        return JSONResponse(
            {"error": "The pull request or recipient changed. Reload before handing off."},
            status_code=409,
        )
    except Exception:
        return JSONResponse(
            {"error": "The handoff could not be confirmed. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.patch("/api/v1/repos/{owner}/{repo}/pulls/{number}/release")
async def release_assigned_pull(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    try:
        result = await asyncio.wait_for(
            gitea_proxy.release_assigned_pull(f"{owner}/{repo}", number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except gitea_proxy.IssueNotAvailableError:
        raise HTTPException(status_code=404, detail="Assigned pull request not found")
    except Exception:
        return JSONResponse(
            {"error": "The assignment could not be released. It remains in My Work; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result)


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments")
async def assigned_pull_conversation(
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    page: int | None = Query(default=None, ge=1, le=100),
    limit: int = Query(default=20, ge=1, le=50),
):
    repository = f"{owner}/{repo}"

    async def load_conversation():
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        return await gitea_proxy.issue_conversation_page(repository, number, page, limit)

    try:
        return await asyncio.wait_for(
            load_conversation(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The conversation could not be loaded. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )


@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments", status_code=201)
async def comment_on_assigned_pull(
    comment: IssueComment,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    idempotency_key: str | None = Header(default=None, max_length=128),
):
    repository = f"{owner}/{repo}"

    async def post_comment():
        if not await gitea_proxy.is_assigned_pull(repository, number):
            raise HTTPException(status_code=404, detail="Assigned pull request not found")
        return await gitea_proxy.comment_on_issue(repository, number, comment.body)

    try:
        result = await _run_idempotent_authored_action(
            post_comment(),
            idempotency_key=idempotency_key,
            fingerprint=("pull-comment", repository, number, comment.body),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except Exception:
        return JSONResponse(
            {"error": "The comment could not be posted. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)


@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/merge")
async def merge_assigned_pull(
    submission: PullMergeSubmission,
    request: Request,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    await _require_step_up(
        request,
        step_up_grant,
        action="merge_pull",
        target=f"{repository}#{number}",
    )

    try:
        assigned = await asyncio.wait_for(
            gitea_proxy.is_assigned_pull(repository, number),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "The pull request assignment could not be verified. Nothing was merged."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    if not assigned:
        raise HTTPException(status_code=404, detail="Assigned pull request not found")

    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "pull_merged",
            target=f"{repository}#{number}",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"error": "Security activity is temporarily unavailable. Nothing was merged."},
            status_code=503,
            headers={"Retry-After": "1"},
        )

    try:
        result = await asyncio.wait_for(
            gitea_proxy.merge_assigned_pull(
                repository, number, submission.expected_head_sha
            ),
            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
        )
        try:
            await asyncio.to_thread(journal.finalize, operation_id)
        except SecurityEventStoreError:
            pass
        return result
    except gitea_proxy.StalePullError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"error": "New commits were pushed. Refresh before merging."},
            status_code=409,
        )
    except gitea_proxy.PullNotMergeableError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"error": "This pull request is not currently safe to merge. Refresh its status."},
            status_code=409,
        )
    except Exception:
        try:
            merged = await asyncio.wait_for(
                gitea_proxy.is_pull_merged_at_head(
                    repository, number, submission.expected_head_sha
                ),
                timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
            )
        except Exception:
            merged = False
        if merged:
            try:
                await asyncio.to_thread(journal.finalize, operation_id)
            except SecurityEventStoreError:
                pass
            return {"number": number, "merged": True, "state": "closed"}
        return JSONResponse(
            {
                "number": number,
                "merged": False,
                "state": "unknown",
                "confirmation_pending": True,
                "error": "Merge confirmation is pending. Check its status before retrying.",
            },
            status_code=202,
            headers={"Retry-After": "1"},
        )


@app.get("/api/v1/repos/{owner}/{repo}/release-receipt/{commit_sha}")
async def release_receipt(owner: str, repo: str, commit_sha: str):
    result = await gitea_proxy.release_receipt_status(f"{owner}/{repo}", commit_sha)
    return JSONResponse(result, headers={"Cache-Control": "no-store"})


@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
async def submit_review(
    submission: PullReviewSubmission,
    request: Request,
    owner: str,
    repo: str,
    number: int,
    idempotency_key: str | None = Header(default=None, max_length=128),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    if submission.decision != "comment":
        await _require_step_up(
            request,
            step_up_grant,
            action="submit_pull_review",
            target=(
                f"{repository}#{number}@{submission.expected_head_sha}:"
                f"{submission.decision}"
            ),
        )

    async def submit_requested_review():
        if not await is_requested_review(repository, number):
            raise HTTPException(status_code=404, detail="Review request not found")
        args = (
            repository,
            number,
            submission.expected_head_sha,
            submission.decision,
            submission.body,
        )
        journal = None
        operation_id = None
        if submission.decision != "comment":
            journal = _security_event_store()
            try:
                operation_id = await asyncio.to_thread(
                    journal.reserve,
                    (
                        "pull_review_approved"
                        if submission.decision == "approve"
                        else "pull_review_changes_requested"
                    ),
                    target=f"{repository}#{number}",
                )
            except SecurityEventStoreError:
                raise HTTPException(
                    status_code=503,
                    detail="Security activity is temporarily unavailable",
                )
        try:
            if submission.comments:
                result = await gitea_proxy.submit_pull_review(
                    *args,
                    [comment.model_dump() for comment in submission.comments],
                )
            else:
                result = await gitea_proxy.submit_pull_review(*args)
        except (
            HTTPException,
            gitea_proxy.StaleReviewError,
            gitea_proxy.InvalidReviewCommentError,
        ):
            if journal is not None and operation_id is not None:
                try:
                    await asyncio.to_thread(journal.discard, operation_id)
                except SecurityEventStoreError:
                    pass
            raise
        if journal is not None and operation_id is not None:
            try:
                await asyncio.to_thread(journal.finalize, operation_id)
            except SecurityEventStoreError:
                pass
        return result

    try:
        comment_fingerprint = tuple(
            (comment.path, comment.body, comment.new_position, comment.old_position)
            for comment in submission.comments
        )
        result = await _run_idempotent_authored_action(
            submit_requested_review(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "pull-review", repository, number, submission.expected_head_sha,
                submission.decision, submission.body, comment_fingerprint,
            ),
            timeout=REVIEW_DETAIL_TIMEOUT_SECONDS,
        )
    except HTTPException:
        raise
    except gitea_proxy.StaleReviewError:
        return JSONResponse(
            {"error": "New commits were pushed. Refresh the review before submitting."},
            status_code=409,
        )
    except gitea_proxy.InvalidReviewCommentError:
        return JSONResponse(
            {"error": "An inline comment no longer matches this pull request. Refresh the review."},
            status_code=422,
        )
    except Exception:
        return JSONResponse(
            {"error": "The review could not be submitted. Your draft is safe; please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse(result, status_code=201)
