stackchain-dashboard/src/main.py

import asyncio
import base64
import binascii
import hashlib
import hmac
import math
import os
import secrets
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, ValidationError, field_validator, model_validator
from starlette.datastructures import UploadFile

from src import dashboard_auth, gitea_proxy, passkeys
from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
from src.compression import NegotiatedGZipMiddleware
from src.gitea_proxy import (
    activity_events,
    current_user,
    is_requested_review,
    issues,
    mark_notification_read,
    notification_detail,
    notifications,
    pull_requests,
    pull_review_detail,
    repos,
)
from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy
from src.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, client_source
from src.live_snapshot_store import (
    LiveSnapshotMetadata,
    LiveSnapshotState,
    LiveSnapshotStore,
    RefreshLeaseLost,
)
from src.models import Issue, Milestone, PullRequest, Repo, User
from src.passkey_store import PasskeyStore
from src.push_notifications import PushConfiguration, dispatch_unread_updates
from src.push_subscription_store import PushSubscriptionStore
from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
from src.security_event_store import SecurityEventStore, SecurityEventStoreError
from src.suggestion_engine import compute
from src.later_store import LaterStore
from src.today_store import TodayPlanFull, TodayStore
from src.views import FRONTEND_BUILD, router as frontend_router


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


async def _push_poll_loop() -> None:
    interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")))
    send_timeout = max(
        1.0, float(os.getenv("STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS", "10"))
    )
    lease_seconds = max(
        send_timeout + 5.0,
        float(os.getenv("STACKCHAIN_PUSH_LEASE_SECONDS", "60")),
    )
    while True:
        await asyncio.sleep(interval)
        try:
            await dispatch_unread_updates(
                _push_subscription_store,
                _push_configuration(),
                notifications,
                session_active=dashboard_auth.managed_session_active,
                lease_seconds=lease_seconds,
                send_timeout_seconds=send_timeout,
            )
        except asyncio.CancelledError:
            raise
        except Exception:
            # Gitea and push endpoints are external; one failed poll must not
            # stop later delivery attempts.
            continue


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


app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
app.add_middleware(RequestBodyLimitMiddleware, limit_for=request_body_limit)
app.add_middleware(NegotiatedGZipMiddleware, minimum_size=1_024)
CONTEXT_TIMEOUT_SECONDS = 5.0
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES = 256
AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS = 600.0
AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES = 256
AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS = float(
    os.getenv("STACKCHAIN_IDEMPOTENCY_LOCK_TIMEOUT_SECONDS", "0.1")
)
AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS = float(
    os.getenv("STACKCHAIN_AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", "5.0")
)
NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
WORK_PAGE_TIMEOUT_SECONDS = 5.0
GLOBAL_SEARCH_TIMEOUT_SECONDS = 3.0
NOTIFICATION_DETAIL_TIMEOUT_SECONDS = 5.0
BULK_NOTIFICATION_CONCURRENCY = 5
BULK_NOTIFICATION_DEADLINE_SECONDS = 6.0
LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
LIVE_SNAPSHOT_RETRY_BASE_SECONDS = 5.0
LIVE_SNAPSHOT_RETRY_MAX_SECONDS = 60.0
AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS = 15.0
AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS = 5.0
AVAILABLE_ISSUE_SNAPSHOT_LEASE_SECONDS = WORK_PAGE_TIMEOUT_SECONDS + 1.0
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
_live_snapshot_task: asyncio.Task | None = None
_push_poll_task: asyncio.Task | None = None
_live_snapshot_value: dict | None = None
_live_snapshot_created_at: float | None = None
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
_live_section_created_at: dict[str, float | None] = {
    section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_failure_count: dict[str, int] = {
    section: 0 for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_retry_at: dict[str, float | None] = {
    section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_snapshot_refreshing_sections: set[str] = set()
_live_section_revisions: dict[str, int] = {
    section: 0 for section in LIVE_SNAPSHOT_SECTIONS
}
_live_revision_generation = secrets.token_hex(8)
_read_notification_ids: set[int] = set()
_authored_action_operations: dict[
    str, tuple[tuple[Any, ...], asyncio.Task, float]
] = {}
_state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"))
_live_snapshot_clock = time.time
_live_snapshot_store = LiveSnapshotStore(
    os.getenv("STACKCHAIN_LIVE_SNAPSHOT_DB", str(_state_dir / "live-snapshot.sqlite3")),
    clock=lambda: _live_snapshot_clock(),
)
_idempotency_ledger = IdempotencyLedger(
    os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")),
    ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS,
    max_entries=(
        AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES
        + ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES
    ),
    lock_timeout_seconds=AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS,
)
_available_issue_snapshot_task: asyncio.Task | None = None
_available_issue_snapshot_lock = asyncio.Lock()
_available_issue_snapshot_value: list[dict] | None = None
_available_issue_snapshot_created_at: float | None = None
_available_issue_snapshot_retry_at: float | None = None
_available_issue_snapshot_store = AvailableIssueSnapshotStore(
    os.getenv(
        "STACKCHAIN_AVAILABLE_ISSUE_SNAPSHOT_DB",
        str(_state_dir / "available-issue-snapshot.sqlite3"),
    )
)
_push_subscription_store = PushSubscriptionStore(
    os.getenv("STACKCHAIN_PUSH_DB", str(_state_dir / "push-subscriptions.sqlite3"))
)


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


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


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


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

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


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

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


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


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


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


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


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


class PasskeyAuthorization(PasskeyCeremony, PasskeyAuthorizationTarget):
    pass


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


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


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


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


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


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


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

    @model_validator(mode="after")
    def validate_action_fields(self):
        if self.action == "move" and self.direction is None:
            raise ValueError("move requires a direction")
        if self.action != "move" and self.direction is not None:
            raise ValueError("direction is only valid for move")
        if self.action != "configure" and (self.capacity_minutes is not None or self.estimates):
            raise ValueError("capacity and estimates are only valid for configure")
        if any(not item_id or len(item_id) > 500 or minutes < 5 or minutes > 1440
               for item_id, minutes in self.estimates.items()):
            raise ValueError("estimates must use bounded item IDs and minutes")
        return self


class 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)
    base_revision: int | None = Field(default=None, ge=0)


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


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


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

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


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

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


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


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


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

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

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


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


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


class IssueBlockerDesiredState(IssueBlockerUpdate):
    present: bool


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

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

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


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

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


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


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

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

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

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

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

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


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


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

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


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


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


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


@app.middleware("http")
async def enforce_browser_security_boundary(request: Request, call_next):
    """Apply one browser trust boundary, including to auth short-circuits."""
    response = await call_next(request)
    response.headers["Content-Security-Policy"] = CONTENT_SECURITY_POLICY
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Permissions-Policy"] = (
        "camera=(), microphone=(), 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:
        try:
            await asyncio.to_thread(attempts.record_blocked, method="token")
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        return JSONResponse(
            {"detail": "Too many sign-in attempts"},
            status_code=429,
            headers={
                "Cache-Control": "no-store",
                "Retry-After": str(retry_after),
            },
        )
    configured_token = dashboard_auth.access_token()
    if not configured_token or not hmac.compare_digest(
        payload.access_token, configured_token
    ):
        try:
            await asyncio.to_thread(attempts.record_failure, source)
        except LoginAttemptStoreError:
            return JSONResponse(
                {"detail": "Sign-in throttling is temporarily unavailable"},
                status_code=503,
                headers={"Cache-Control": "no-store"},
            )
        raise HTTPException(status_code=401, detail="Invalid access token")
    try:
        await asyncio.to_thread(attempts.clear, source)
    except LoginAttemptStoreError:
        return JSONResponse(
            {"detail": "Sign-in throttling is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        signed, session = await asyncio.to_thread(
            dashboard_auth.issue_session, device_label=payload.device_label
        )
    except dashboard_auth.SessionStoreError:
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(
            _security_event_store().record,
            "sign_in",
            method="token",
            device_label=payload.device_label,
            target="dashboard",
        )
    except SecurityEventStoreError:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    path = dashboard_auth.cookie_path(request)
    max_age = max(1, session.expires_at - int(time.time()))
    response.set_cookie(
        dashboard_auth.SESSION_COOKIE,
        signed,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.set_cookie(
        dashboard_auth.CSRF_COOKIE,
        session.csrf,
        max_age=max_age,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {"authenticated": True}


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


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


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


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


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


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


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


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


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


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


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


@app.get("/api/v1/push-subscription")
async def push_status(request: Request):
    configuration = _push_configuration()
    device_id = await dashboard_auth.session_management_id(
        request.state.dashboard_session
    )
    subscribed = await asyncio.to_thread(
        _push_subscription_store.is_subscribed,
        device_id,
    )
    return {
        "available": configuration.enabled,
        "subscribed": subscribed,
        "public_key": configuration.public_key if configuration.enabled else "",
    }


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


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


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


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


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


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 | TodayOperationBatch):
    login = await _confirmed_login()
    try:
        if isinstance(payload, TodayOperationBatch):
            return await asyncio.to_thread(
                _today_store().apply_batch,
                login,
                [operation.model_dump() for operation in payload.operations],
            )
        if payload.action == "configure":
            return await asyncio.to_thread(
                _today_store().apply_batch,
                login,
                [payload.model_dump()],
            )
        return await asyncio.to_thread(
            _today_store().apply,
            login,
            payload.operation_id,
            payload.action,
            payload.item_id,
            direction=payload.direction,
        )
    except TodayPlanFull:
        raise HTTPException(status_code=409, detail="Today is limited to 5 items")
    except (OSError, sqlite3.Error):
        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 | LaterOperationBatch):
    login = await _confirmed_login()
    try:
        if isinstance(payload, LaterOperationBatch):
            return await asyncio.to_thread(
                _later_store().apply_batch,
                login,
                [operation.model_dump() for operation in payload.operations],
            )
        return await asyncio.to_thread(
            _later_store().apply,
            login,
            payload.operation_id,
            payload.action,
            payload.item_id,
            wake_at=payload.wake_at,
            base_revision=payload.base_revision,
        )
    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:
        push_device_id = await dashboard_auth.session_management_id(session)
    except (dashboard_auth.SessionStoreError, AttributeError):
        push_device_id = None
    journal = _security_event_store()
    try:
        operation_id = await asyncio.to_thread(
            journal.reserve,
            "sign_out",
            target="current_device",
        )
    except SecurityEventStoreError:
        return JSONResponse(
            {"detail": "Security activity is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(dashboard_auth.revoke_session, session)
    except dashboard_auth.SessionStoreError:
        try:
            await asyncio.to_thread(journal.discard, operation_id)
        except SecurityEventStoreError:
            pass
        return JSONResponse(
            {"detail": "Session registry is temporarily unavailable"},
            status_code=503,
            headers={"Cache-Control": "no-store"},
        )
    try:
        await asyncio.to_thread(journal.finalize, operation_id)
    except SecurityEventStoreError:
        # The pending reservation is durable evidence; do not report a
        # successful revocation as failed merely because completion could not
        # be marked yet.
        pass
    if push_device_id is not None:
        await asyncio.to_thread(
            _push_subscription_store.delete_session, push_device_id
        )
    path = dashboard_auth.cookie_path(request)
    response.delete_cookie(
        dashboard_auth.SESSION_COOKIE,
        path=path,
        secure=True,
        httponly=True,
        samesite="strict",
    )
    response.delete_cookie(
        dashboard_auth.CSRF_COOKIE,
        path=path,
        secure=True,
        httponly=False,
        samesite="strict",
    )
    response.headers["Cache-Control"] = "no-store"
    return {"authenticated": False, "clear_private_device_data": True}


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


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

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


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


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


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


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


@app.get("/api/v1/search")
async def global_search(
    q: str = Query(min_length=2, max_length=100),
    limit: int = Query(default=10, ge=1, le=25),
) -> 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(lease_owner: str) -> list[dict]:
    global _available_issue_snapshot_value, _available_issue_snapshot_created_at
    global _available_issue_snapshot_retry_at
    try:
        result = await gitea_proxy.available_issue_snapshot()
        shared = await asyncio.to_thread(
            _available_issue_snapshot_store.publish, lease_owner, items=result
        )
        _available_issue_snapshot_value = shared.items
        _available_issue_snapshot_created_at = time.monotonic()
        _available_issue_snapshot_retry_at = None
        return shared.items or []
    except asyncio.CancelledError:
        await asyncio.to_thread(
            _available_issue_snapshot_store.release_refresh, lease_owner
        )
        raise
    except Exception:
        try:
            await asyncio.to_thread(
                _available_issue_snapshot_store.fail_refresh,
                lease_owner,
                retry_at=time.time() + AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS,
            )
        except Exception:
            pass
        raise


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


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


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


def _available_issue_page(items: list[dict], page: int, limit: int = 50) -> 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] = _live_snapshot_clock() + delay


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


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


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


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


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


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


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


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


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


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


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


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

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


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



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


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


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

    semaphore = asyncio.Semaphore(BULK_NOTIFICATION_CONCURRENCY)

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

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

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


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


@app.get("/api/v1/notifications/{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,
        )
    await _remove_notifications_from_live_snapshot([thread_id])
    return JSONResponse({"id": thread_id, "status": "read"})


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


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


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

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

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


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


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


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


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


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


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


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


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


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


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


@app.patch("/api/v1/repos/{owner}/{repo}/issues/{number}/content")
async def update_assigned_issue_content(
    update: IssueContentUpdate,
    owner: str,
    repo: str,
    number: int = PathParam(gt=0),
):
    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():
        if await gitea_proxy.repository_access(repository) is None:
            raise HTTPException(status_code=404, detail="Repository not found")
        return await gitea_proxy.repo_milestones(repository)

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


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

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

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


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

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

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


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

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

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


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

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

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

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


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

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

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


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

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

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


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

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


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

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

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


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

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

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


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

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

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


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

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

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


@app.get("/api/v1/repos/{owner}/{repo}/pulls/{number}/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:
        result = await asyncio.wait_for(
            merge_pull(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
        )
        await asyncio.to_thread(
            _security_event_store().record,
            "pull_merged",
            target=f"{repository}#{number}",
        )
        return result
    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,
    request: Request,
    owner: str,
    repo: str,
    number: int,
    idempotency_key: str | None = Header(default=None, max_length=128),
    step_up_grant: str | None = Header(
        default=None, alias="X-Step-Up-Grant", max_length=128
    ),
):
    repository = f"{owner}/{repo}"
    if submission.decision != "comment":
        await _require_step_up(
            request,
            step_up_grant,
            action="submit_pull_review",
            target=(
                f"{repository}#{number}@{submission.expected_head_sha}:"
                f"{submission.decision}"
            ),
        )

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

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