import asyncio
import hmac
import math
import os
import sqlite3
import time
from collections.abc import Awaitable, Coroutine
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Any, Literal
from urllib.parse import urlencode

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, field_validator, model_validator

from src import dashboard_auth, gitea_proxy
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.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, client_source
from src.models import Issue, Milestone, PullRequest, Repo, User
from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
from src.suggestion_engine import compute
from src.later_store import LaterStore
from src.today_store import TodayPlanFull, TodayStore
from src.views import router as frontend_router

@asynccontextmanager
async def lifespan(_app: FastAPI):
    global _live_snapshot_task, _available_issue_snapshot_task
    gitea_proxy.start_client()
    try:
        yield
    finally:
        live_task = _live_snapshot_task
        available_task = _available_issue_snapshot_task
        for task in (live_task, available_task):
            if task is not None and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
        try:
            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


app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
app.add_middleware(RequestBodyLimitMiddleware, limit_for=request_body_limit)
CONTEXT_TIMEOUT_SECONDS = 5.0
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0
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")
)
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
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
_live_snapshot_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
}
_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"))
_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_value: list[dict] | None = None
_available_issue_snapshot_created_at: float | None = None
_available_issue_snapshot_retry_at: float | None = None


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 FreshAuthorization(BaseModel):
    access_token: str = Field(min_length=1, max_length=1_024)
    action: Literal[
        "merge_pull", "close_issue", "revoke_device", "revoke_all_sessions"
    ]
    target: str = Field(min_length=1, max_length=255)


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"]
    item_id: str = Field(min_length=1, max_length=500)
    direction: Literal["up", "down"] | None = None

    @model_validator(mode="after")
    def require_move_direction(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")
        return self


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)


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


class IssueCreation(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    body: str = Field(default="", max_length=10_000)
    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()

    @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 IssueHandoff(BaseModel):
    recipient: str = Field(
        min_length=1, max_length=255, pattern=r"^[A-Za-z0-9_.-]+$"
    )


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)],
            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()
    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 == "issue":
        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)],
                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:
    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)
    ]
    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)
    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"}
        or path.startswith("/static/")
        or (path == "/api/v1/session" 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 == "session_revoked":
                payload["code"] = session_reason
            return JSONResponse(
                payload,
                status_code=401,
                headers={"Cache-Control": "no-store"},
            )
        login_redirect = (
            "login?reason=session-revoked"
            if session_reason == "session_revoked"
            else _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/later"} 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=(), geolocation=(), payment=(), usb=()"
    )
    response.headers["X-Frame-Options"] = "DENY"
    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:
        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"},
        )
    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.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.get("/api/v1/session")
async def session_status(request: Request):
    session = request.state.dashboard_session
    return {
        "authenticated": session is not None,
        "csrf_token": session.csrf if session is not None else "",
    }


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"))
    )


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/today")
async def get_today_plan():
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(_today_store().get, login)
    except (OSError, sqlite3.Error):
        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):
    login = await _confirmed_login()
    try:
        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):
        raise HTTPException(
            status_code=503,
            detail="Today synchronization is unavailable",
            headers={"Retry-After": "1"},
        )


@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):
        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):
    login = await _confirmed_login()
    try:
        return await asyncio.to_thread(
            _later_store().apply,
            login,
            payload.operation_id,
            payload.action,
            payload.item_id,
            wake_at=payload.wake_at,
        )
    except ValueError as error:
        raise HTTPException(status_code=422, detail=str(error))
    except (OSError, sqlite3.Error):
        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:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    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
        )
        revoked = await dashboard_auth.revoke_managed_session(management_id)
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    if target is None or not revoked:
        raise HTTPException(status_code=404, detail="Active device not found")
    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",
    )
    try:
        await dashboard_auth.revoke_all_sessions()
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    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 readiness after verifying the configured Gitea connection."""
    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_message = (
            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"
            )
        )
        return JSONResponse(
            {
                "status": "not_ready",
                "service": "stackchain-dashboard",
                "error": error_message,
            },
            status_code=503,
            headers={
                "Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
            } if timed_out else None,
        )
    payload = {
        "status": "ready",
        "service": "stackchain-dashboard",
    }
    if not dashboard_auth.enabled():
        payload["gitea_user"] = user["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/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),
) -> JSONResponse:
    query = q.strip()
    if len(query) < 2:
        raise HTTPException(status_code=422, detail="Search query must contain at least 2 characters")
    try:
        result = await asyncio.wait_for(
            gitea_proxy.global_search(query, limit),
            timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
        )
    except Exception:
        return JSONResponse(
            {"error": "Search is temporarily unavailable. Please retry."},
            status_code=503,
            headers={"Retry-After": "1"},
        )
    return JSONResponse({"query": query, **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/work-route")
async def resolve_work_route(
    kind: Literal["issue", "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", "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() -> list[dict]:
    global _available_issue_snapshot_value, _available_issue_snapshot_created_at
    global _available_issue_snapshot_retry_at
    result = await gitea_proxy.available_issue_snapshot()
    _available_issue_snapshot_value = result
    _available_issue_snapshot_created_at = time.monotonic()
    _available_issue_snapshot_retry_at = None
    return result


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 _available_issue_snapshot() -> tuple[list[dict], bool, bool, bool]:
    global _available_issue_snapshot_task
    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
    if _available_issue_snapshot_task is None or _available_issue_snapshot_task.done():
        _available_issue_snapshot_task = asyncio.create_task(
            _refresh_available_issue_snapshot()
        )
        _available_issue_snapshot_task.add_done_callback(_observe_available_issue_refresh)
    if _available_issue_snapshot_value is not None:
        return _available_issue_snapshot_value, True, True, False
    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) -> dict:
    start = (page - 1) * limit
    page_items = items[start:start + limit]
    return {
        "items": page_items,
        "page": page,
        "total": len(items),
        "has_more": start + len(page_items) < len(items),
    }


@app.get("/api/v1/available-issues")
async def available_issues(page: int = Query(default=1, ge=1, le=100)) -> JSONResponse:
    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)
    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] = time.monotonic() + 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 _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]) -> dict:
    global _live_snapshot_value, _live_snapshot_created_at
    try:
        refreshed = await _build_live_snapshot_before_deadline(sections)
    except Exception:
        for section in sections:
            _record_live_section_failure(section)
        raise
    now = time.monotonic()
    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)
    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:
            _live_section_revisions[section] += 1
    _live_snapshot_value = result
    successful_times = [value for value in _live_section_created_at.values() if value is not None]
    _live_snapshot_created_at = max(successful_times) if successful_times else None
    return result


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


def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
    global _live_snapshot_task
    global _live_snapshot_refreshing_sections
    if _live_snapshot_task is None or _live_snapshot_task.done():
        _live_snapshot_refreshing_sections = set(sections)
        _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot(sections))
        _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, int | None] | None = None,
) -> dict:
    payload = dict(value)
    now = time.monotonic()
    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,
    }
    payload["revisions"] = dict(_live_section_revisions)
    for section, known_revision in (known_revisions or {}).items():
        if known_revision is None or known_revision != _live_section_revisions[section]:
            continue
        payload.pop(section, None)
        if section == "notifications":
            payload.pop("notification_pagination", None)
    return payload


def _remove_notification_from_live_snapshot(thread_id: int) -> None:
    global _live_snapshot_value, _read_notification_ids
    _read_notification_ids = _read_notification_ids | {thread_id}
    if _live_snapshot_value is None:
        return
    retained_notifications = _live_snapshot_value.get("notifications")
    if not isinstance(retained_notifications, list):
        return
    updated = dict(_live_snapshot_value)
    updated["notifications"] = [
        notification
        for notification in retained_notifications
        if not isinstance(notification, dict) or notification.get("id") != thread_id
    ]
    if updated["notifications"] != retained_notifications:
        _live_section_revisions["notifications"] += 1
    _live_snapshot_value = updated


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


@app.get("/api/v1/live")
async def live_snapshot(
    context_revision: int | None = Query(default=None, ge=0),
    events_revision: int | None = Query(default=None, ge=0),
    notifications_revision: int | None = Query(default=None, ge=0),
) -> 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,
    }
    now = time.monotonic()
    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,
            )
        )
    task = _start_live_snapshot_refresh(due_sections)
    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/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
    _remove_notification_from_live_snapshot(thread_id)
    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
    }
    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/{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 = Query(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,
        )
    _remove_notification_from_live_snapshot(thread_id)
    return JSONResponse({"id": thread_id, "status": "read"})


@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.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/claim")
async def claim_available_issue(
    owner: str, repo: str, number: int = PathParam(gt=0)
) -> JSONResponse:
    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"},
        )
    if _available_issue_snapshot_value is not None:
        _available_issue_snapshot_value[:] = [
            item for item in _available_issue_snapshot_value
            if not (
                item.get("repository") == repository and item.get("number") == number
            )
        ]
    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"):
        _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)
) -> JSONResponse:
    repository = f"{owner}/{repo}"

    async def load_candidates():
        if not await gitea_proxy.is_assigned_issue(repository, number):
            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.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.get("/api/v1/repos/{owner}/{repo}/issues/{number}/detail")
async def assigned_issue_detail(owner: str, repo: str, number: int = PathParam(gt=0)):
    repository = f"{owner}/{repo}"

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

    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"},
        )


@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),
):
    repository = f"{owner}/{repo}"
    try:
        result = await asyncio.wait_for(
            gitea_proxy.update_assigned_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.get("/api/v1/repos/{owner}/{repo}/milestones")
async def repository_milestones(owner: str, repo: str):
    repository = f"{owner}/{repo}"

    async def load_milestones():
        available = await gitea_proxy.repos()
        accessible = {
            item.get("full_name")
            for item in (available if isinstance(available, list) else [])
            if isinstance(item, dict)
        }
        if repository not in accessible:
            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():
        available = await gitea_proxy.repos()
        accessible = {
            item.get("full_name")
            for item in (available if isinstance(available, list) else [])
            if isinstance(item, dict)
        }
        if repository not in accessible:
            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.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, available = await asyncio.gather(
            gitea_proxy.current_user(), gitea_proxy.repos()
        )
        login = user.get("login") if isinstance(user, dict) else None
        accessible = {
            item.get("full_name")
            for item in (available if isinstance(available, list) else [])
            if isinstance(item, dict)
        }
        if not login or repository not in accessible:
            raise HTTPException(status_code=404, detail="Repository not found")
        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,
                login,
                creation.label_ids,
                creation.milestone_id,
                creation.due_date,
            )
        return await gitea_proxy.create_issue(
            repository, creation.title, creation.body, login, creation.label_ids
        )

    try:
        result = await _run_idempotent_authored_action(
            create_issue(),
            idempotency_key=idempotency_key,
            fingerprint=(
                "issue-create",
                repository,
                creation.title,
                creation.body,
                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),
):
    repository = f"{owner}/{repo}"

    async def load_conversation():
        if not await gitea_proxy.is_assigned_issue(repository, number):
            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),
):
    repository = f"{owner}/{repo}"

    async def post_comment():
        if not await gitea_proxy.is_assigned_issue(repository, number):
            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)


@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),
    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}",
    )

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

    try:
        return await asyncio.wait_for(close_issue(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
    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"},
        )


@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}/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}/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}",
    )

    async def merge_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.merge_assigned_pull(
            repository, number, submission.expected_head_sha
        )

    try:
        return await asyncio.wait_for(
            merge_pull(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
    except HTTPException:
        raise
    except gitea_proxy.StalePullError:
        return JSONResponse(
            {"error": "New commits were pushed. Refresh before merging."},
            status_code=409,
        )
    except gitea_proxy.PullNotMergeableError:
        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:
            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.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
async def submit_review(
    submission: PullReviewSubmission,
    owner: str,
    repo: str,
    number: int,
    idempotency_key: str | None = Header(default=None, max_length=128),
):
    repository = f"{owner}/{repo}"

    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,
        )
        if submission.comments:
            return await gitea_proxy.submit_pull_review(
                *args,
                [comment.model_dump() for comment in submission.comments],
            )
        return await gitea_proxy.submit_pull_review(*args)

    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)