Pinned

class PinnedAPI:
    def __init__(self, ctx: HostableContext) -> None
    async def authorize(self, conn: HTTPConnection) -> Any;  async def lifespan(self) -> AsyncIterator[None];  def stats(self) -> dict[str, Any] | None
    id: str; lease: InstanceLease; domain: Domain; cluster: httpx.AsyncClient; frontier: Frontier; placement_epoch: int   # properties
    def abort(self) -> None;  def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]

class PinnedContext:
    def __init__(self, id: str, *, lease: InstanceLease, abort: Callable[[], None], create_task: Callable[[Coroutine[Any, Any, Any]], "asyncio.Task[Any]"], durability: InstanceDurability | None = None, cluster: httpx.AsyncClient | None = None, domain: Domain) -> None

class AuthRejected(Exception):
    def __init__(self, message: str, *, status_code: Literal[401, 403] = 401, payload: Any = None, code: str | None = None)
    message: str; status_code: int; payload: Any; code: str | None

async def auth_result(conn: HTTPConnection) -> Any

class PinnedWorker(APIRouter):
    def __init__(self, *, hosts: Mapping[str, type[PinnedAPI] | Host] | None = None, ..., **kwargs: Any) -> None   # keyword parameters below; kwargs reach APIRouter
    def host(self, namespace: str, api: type[PinnedAPI] | Host, **options: Any) -> PinnedHost
    pinboard_url; advertise_url; hostname; env; takeover; epoch; state_store; domain: Domain
    dispatch_slots: asyncio.BoundedSemaphore; lease: Lease; registration: RegistrationClient | None; secret: str   # the registration's secret, or a fresh 32-byte hex in local mode
    ready: bool; superseded: bool; draining: bool; drained: bool   # properties: registered (control plane mode) and inside the lease; displaced by a higher-epoch worker
    def abort_all(self) -> None
    async def drain(self, deadline: float | None) -> None
    def install_sigterm_drain(self, deadline: float = 30.0, sig: int = signal.SIGTERM) -> None   # needs a running loop; drains, then hands sig to a handler installed with loop.add_signal_handler

class Host:
    def __init__(self, cls: type[PinnedAPI], **options: Any) -> None
    cls; options

class Lease:
    def __init__(self, *, domain: Domain) -> None
    healthy: bool; expires_at: datetime | None; expires_at_monotonic: float | None; expired: bool; buffer: float   # properties; buffer default 5.0
    ttl: float | None; flaps: int
    def cap(self, deadline: float | None) -> None                          # monotonic ceiling renewals may not pass; None lifts it
    def renewed(self, ttl: float, *, anchor: float | None = None) -> None   # ack: runs until anchor (default now) + ttl
    def configure(self, *, buffer: float | None = None, on_unhealthy: Callable[[], None] | None = None, on_restored: Callable[[], None] | None = None) -> None
    async def close(self) -> None

class InstanceLease:
    def __init__(self, worker: Lease, eviction_deadline_monotonic: Callable[[], float | None], *, winding_down: Callable[[], bool] = lambda: False, wind_down: Callable[[float | None], None] | None = None) -> None
    healthy: bool; expires_at: datetime | None; expired: bool; buffer: float; eviction_deadline: datetime | None; winding_down: bool   # properties
    def wind_down(self, deadline: float | None = None) -> None
    on_wind_down: Callable[[datetime | None], None] | None; on_idle: Callable[[], None] | None
    def configure(self, *, buffer=None, on_unhealthy=None, on_restored=None, on_wind_down: Callable[[datetime | None], None] | None = None, on_idle: Callable[[], None] | None = None) -> None

def attestation(conn: HTTPConnection) -> Attestation.Instance | Attestation.Account | None

@dataclass(frozen=True, match_args=False)
class Attestation.Instance:
    ns: str; id: str; depth: int = 0
    __match_args__ = ("ns", "id")

@dataclass(frozen=True, match_args=False)
class Attestation.Account:
    sub: str; claims: Mapping[str, Any] = field(default_factory=dict); scopes: frozenset[str] = frozenset()
    __match_args__ = ("sub",)

def pinned.pinboard.wire.invalid_id_reason(id: str) -> str | None

@runtime_checkable
class Domain(Protocol):
    def now(self) -> float;  def monotonic(self) -> float;  def token_hex(self, nbytes: int) -> str;  def rng(self, label: str) -> random.Random
    async def sleep(self, seconds: float) -> None;  async def wait(self, event: asyncio.Event, timeout: float) -> bool
    def spawn(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]

class RealDomain:
    def __init__(self, *, seed: int = 0) -> None

class PinnedHost(APIRouter):
    def __init__(self, cls: type[PinnedAPI], *, worker: PinnedWorker, namespace: str, ...) -> None   # keyword parameters below
    cls; worker; namespace; env; domain: Domain; directory: LocalDirectory
    lease: Lease; registration: RegistrationClient | None; ready: bool; draining: bool; drained: bool   # properties; the worker's, drained = draining with no live instances
    def wind_down(self, id: str, deadline: float | None) -> None
    async def lifespan(self, app: Any = None) -> AsyncIterator[None]   # async context manager: starts the sweep; closes instances, the cluster client, then the worker

@runtime_checkable
class HostableContext(Protocol):
    id: str; lease: InstanceLease; durability: InstanceDurability | None; cluster: httpx.AsyncClient | None; domain: Domain   # properties
    def abort(self) -> None;  def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]

@runtime_checkable
class Hostable(Protocol):
    def lifespan(self) -> AsyncIterator[None]

class route:
    get / post / put / delete / patch / websocket = (path: str, *, auth: AuthResolver | None | object = DEFAULT_AUTH) -> Callable[[F], F]

async with instance(cls: type[T], id: str = "test", *, abort: Callable[[], None] = lambda: None, state_store: StateStore | None = None) as api: ...
def pinned_host(cls: type[T], *, namespace: str = "/threads", **kwargs: Any) -> PinnedHost
async with pinned_app(router: PinnedHost) as app: ...            # FastAPI
def client_for(app: FastAPI, headers: dict | None = None) -> httpx.AsyncClient
async with pinned_client(router: PinnedHost, *, attested: bool = True) as client: ...
async def wake_until_ok(client: httpx.AsyncClient, url: str, *, timeout: float = 5.0) -> None

async with ASGIStream(app, path: str, method: str = "GET") as stream:
    stream.status: int | None
    await stream.next_chunk(timeout: float = 5) -> bytes | None   # None ends the body
    await stream.drain(timeout: float = 5) -> None

class BarrierStateStore(MemoryStateStore):
    parked: asyncio.Event
    def hold(self) -> None;  def release(self) -> None