# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to the bacloud tool.

.. warning::

  This is an internal api and subject to change at any time. Do not use
  it in mod code.
"""

# This module is the canonical bacloud wire contract -- every repo
# imports its types by name -- and roughly a seventh of it is the
# version history that has to live beside the versions it explains.
# Splitting it would churn public import paths across five repos to
# move a comment block, so this is the sanctioned large primary
# module rather than something to spill into a sibling.
# pylint: disable=too-many-lines

from enum import Enum
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Annotated, assert_never, override

from efro.dataclassio import (
    ioprepped,
    will_ioprep,
    ioprep,
    IOAttrs,
    IOMultiType,
)
from bacommon import securedata

if TYPE_CHECKING:
    pass

# Version is sent to the master-server with all commands. Can be incremented
# if we need to change behavior server-side to go along with client changes.
#
# 15 (2026-04): Workspace upload protocol switched from inline gzipped
#               bytes (``uploads_inline``) to direct-to-GCS streaming
#               via signed URLs (``uploads_signed``). Old clients will
#               be rejected by the bumped ``MIN_VERSION`` server-side.
# 16 (2026-04): Dropped ``md5_b64`` from ``DirectoryManifestFile``.
#               The streaming upload pipeline now relies solely on
#               SHA-256 verification at finalize time; md5 was
#               redundant with the content-addressable cloud_file_id
#               check and just doubled hashing cost client-side.
# 17 (2026-04): Workspace download / asset-package file-download
#               switched from inline gzipped bytes (``downloads_inline``)
#               to direct-from-GCS streaming via signed GET URLs
#               (``downloads_signed``). Removes the Cloud Run 32 MB
#               response-size cap for CloudFile-backed downloads.
#               ``downloads_inline`` remains for small ad-hoc blobs
#               (asset manifests, admin test payloads).
# 18 (2026-04): Added ``upload_plan`` response field for a simpler,
#               framework-managed upload protocol used by new admin
#               archive commands (and eventually the workspace put
#               path). Old clients don't understand the new field
#               and would silently skip uploads.
# 19 (2026-05): StreamCall protocol introduced. Adds a ``stream``
#               flag to ``StandardRequestData``, bumps ``end_command`` from a
#               2-tuple to a 3-tuple (the new bool indicates the next
#               request should be made in stream mode), adds a
#               ``stream_frames`` field to ``StandardResponseData`` carrying
#               ``StreamFrame`` instances, and introduces the
#               ``StreamFrame`` / ``StreamOutput`` / ``StreamFinal``
#               IOMultiTypes themselves. ``MIN_VERSION`` is bumped to
#               match so older clients are rejected with a clear
#               upgrade message rather than silently mishandling the
#               new tuple shape.
# 20 (2026-05): StreamCall Phase 2 wire-protocol additions for the
#               basn-WebSocket fan-out path. Adds ``stream_ws`` to
#               ``StandardResponseData`` (carrying a basn URL + signed
#               capability token + expiry) so kickoff responses can
#               redirect a client at a basn node's WebSocket
#               endpoint instead of the existing ``_streamcall_poll``
#               loop. Backwards-compatible with v19 in the wire
#               sense (old clients ignore the unknown field and use
#               ``end_command`` as before), but bumped so logs make
#               version-skew obvious during the rollout.
# 21 (2026-05): WS capability-token wire shape switched from a v1
#               ad-hoc dotted base64 string (HMAC over the basn
#               node's comm-token) to a
#               :class:`bacommon.securedata.Archive`. Embedding
#               directly skips the base64-of-base64 nesting and
#               unifies signing on the same Reader/Writer pipeline
#               every other server-signed artifact uses, which
#               also lets any basn (not just the issuing one)
#               verify the token. ``StreamWS.ws_token`` field type
#               changes accordingly. ``MIN_VERSION`` is bumped to
#               match so older clients fail loudly rather than
#               silently mishandling the new field type.
# 22 (2026-05): Auth unified on ``Authorization: Bearer <value>``
#               for both API-key and session-login flows. Server
#               distinguishes by prefix (``bsac-`` = API key;
#               otherwise session token). Brings bacloud auth in
#               line with universally-recognized standards: every
#               intermediary now sees a Bearer header it can
#               recognize for redaction, masking, etc., instead
#               of a custom body field that looks like opaque
#               payload. ``StandardRequestData.token`` is deprecated —
#               kept on the wire as a soft-default'd ``null`` so
#               v21 basn nodes parse v22 requests OK during the
#               rollout window — and will be removed in v23
#               after rollout completes. Server-side response
#               still ships ``StandardResponseData.login`` for
#               sign-in/sign-out — only the *outbound* direction
#               changes. ``MIN_VERSION`` is bumped to match (v21
#               clients send body-token but no Authorization
#               header; the unified server only reads Bearer, so
#               v21 clients would silently authenticate as
#               anonymous).
# 23 (2026-05): Streamcall WS routing decoupled from kickoff
#               node, plus loose-end cleanup. Three changes:
#               (1) ``StreamWS.basn_url`` becomes ``str | None``;
#               ``None`` means the bacloud client opens the WS to
#               its own kickoff hostname (whatever it sent the
#               original request to). Server only fills in a
#               specific URL when the stream genuinely lives on
#               one basn (Phase 3 game-server-logs case;
#               unused today). All current streams are
#               node-agnostic, so server now sets ``None`` and
#               WS handshakes can land on any basn the LB
#               routes them to (any basn can verify the token
#               post-securedata). (2) ``StreamWS.call_id`` added
#               as an explicit field — needed for the client to
#               construct its own URL when ``basn_url`` is
#               ``None``, and gets call_ids out of URLs (which
#               leak into proxy/CDN logs) along the way.
#               (3) ``StandardRequestData.token`` removed entirely;
#               unified Bearer auth from v22 means it's been
#               dead code for a wire-version cycle. Drops a
#               soft-default field that no client populates and
#               no server reads. ``MIN_VERSION`` matched.
# 24 (2026-06): Workspace get/put optimistic-concurrency (mid-air-
#               collision guard). ``StandardResponseData.workspace_snapshotid``
#               carries the workspace's current snapshot id back on a
#               get/put; the client stashes it in a ``.bacloudstate.json``
#               (a dotfile, so the sync auto-ignores it) and sends it as
#               ``WorkspacePutProcessCommand.expected_snapshotid`` on the
#               next put, which the server rejects if the workspace has
#               moved since. Both fields are optional/soft-default, so old
#               clients/servers just skip the check -- ``MIN_VERSION``
#               unchanged.
# 25 (2026-06): One-shot small-file uploads via the basn relay.
#               ``StandardResponseData.uploads_oneshot`` tells the
#               client to POST
#               small file bodies to the basn node (which relays them to
#               the master one-shot cloud-file endpoint) instead of the
#               signed-URL-to-GCS path; the client sends the resulting
#               cloud_file_ids back under ``uploads_oneshot``. The server
#               only emits this to v25+ clients and falls back to
#               ``uploads_signed`` otherwise, so the field is optional --
#               ``MIN_VERSION`` unchanged.
# 26 (2026-08): Removed the legacy ``downloads`` mechanism (one master
#               round trip per file, via a per-entry ``_assemble_file_
#               download`` command) along with its ``StandardResponseData.
#               Downloads`` type. It survived only as the
#               non-CAS-delivery fallback for asset-package blobs, and
#               an escape hatch reachable solely by a manual flag is one
#               nobody exercises -- so it would have rotted before it
#               was ever needed, while costing a mechanism to carry onto
#               the upcoming SmartSocket transport. CAS delivery is now
#               the only asset-blob path. ``MIN_VERSION`` tracks
#               ``BACLOUD_VERSION``, so older clients get the standard
#               "please update" rejection.
#
#               Also in 26: bacloud's conversation moves onto a single
#               SmartSocket session to the node it already talks to.
#               The channel is created by the WS handshake itself
#               (``/bacloudsession``, no mint round trip), credentials
#               ride the upgrade's ``Authorization`` header on every
#               attach, and the node hands back a resume token in-band
#               as a :class:`SessionHandleResponse` -- the response
#               hierarchy's first non-answer member. Folded into 26
#               rather than given its own number because 26 has not
#               shipped: prod and push-public land together, so there
#               is no released version in between for anything to be
#               compatible with.
# 27 (2026-08-15): A streamed command's output now rides the session
#               itself as StreamOutputResponse messages, instead of
#               the client opening a second WebSocket per stream and
#               following it there. A v26 client cannot decode the new
#               response type, and rather than keep its shape alive as
#               a second path, ``MIN_VERSION`` moves with this and old
#               clients get the standard "please update" rejection --
#               so prod and push-public land together, as with 26.
#               The separate per-stream WebSocket (``streamws.py``)
#               remains only for clients with no session at all, e.g.
#               against a node too old to host one. (It was then
#               deleted outright, along with the HTTP
#               request-per-connection fallback, later the same day.)
# 28 (2026-08-15): The stream-kickoff signal becomes explicit and the
#               client-side stream-polling machinery is deleted. A
#               streamed command's kickoff response now carries
#               ``stream_call_id`` -- consumed by the basn session
#               adapter, which follows the stream server-side and
#               relays output down the session -- replacing the old
#               shape: an ``end_command`` chain pointing at the
#               ``_streamcall_poll`` command, detected by basn via
#               response-sniffing. Deleted with it: the
#               ``_streamcall_poll`` handler, the ``stream_frames``
#               response field, the ``stream`` request flag, the
#               third (stream) element of ``end_command`` tuples, and
#               the long-dead ``stream_ws``/``StreamWS``
#               WebSocket-pickup field.
#               ``StreamFrame``/``StreamOutput``/``StreamFinal``
#               remain: they ride the internal bamaster->basn poll
#               hop and basn's streamcall WS fan-out. ``MIN_VERSION``
#               tracks ``BACLOUD_VERSION`` as usual; prod and
#               push-public land together.
# 29 (2026-08): Responses larger than one SmartSocket message are
#               split across several. Adds ``ResponseTypeID.CHUNK``
#               and ``ChunkedResponse``: the session sender slices a
#               too-large serialized response into ordered pieces and
#               the client rejoins them before decoding. The transport
#               caps a single message deliberately (they are
#               load-bearing for the relay's resend-buffer and linger
#               math), so a big response has to be split *above* it --
#               previously nothing did, and a ~1.5 MB vendored asset
#               package was simply retained and retried by the relay
#               forever, presenting as a half-hour hang. ``MIN_VERSION``
#               tracks ``BACLOUD_VERSION`` as usual, so older clients
#               get the standard "please update" rejection rather than
#               an undecodable type id.
BACLOUD_VERSION = 29


def asset_file_cache_path(filehash: str) -> str:
    """Given a sha256 hex file hash, return a storage path."""

    # We expect a 64 byte hex str with only lowercase letters and
    # numbers. Note to self: I considered base64 hashes to save space
    # but then remembered that lots of filesystems out there ignore case
    # so that would not end well.
    assert len(filehash) == 64
    assert filehash.islower()
    assert filehash.isalnum()

    # Single level of 256 dirs. Modern filesystems handle huge flat
    # dirs fine via b-tree/hashed indexing, but a single shard keeps us
    # under any per-dir entry limits on older/unusual filesystems and
    # avoids the inode overhead of deeper sharding.
    return f'{filehash[:2]}/{filehash[2:]}'


class RequestTypeID(Enum):
    """Type IDs for things a bacloud client can send."""

    STANDARD = 's'


class RequestData(IOMultiType[RequestTypeID]):
    """Root of what a bacloud client sends.

    One member today (:class:`StandardRequestData`, the command
    envelope every request has always used). It exists as a hierarchy
    so the session transport has a declared send type, and so a future
    message that is not a command is an addition here rather than
    another optional field bolted onto the command envelope.
    """

    @override
    @classmethod
    def get_type_id(cls) -> RequestTypeID:
        raise NotImplementedError()

    @override
    @classmethod
    def get_default_type_id(cls) -> RequestTypeID:
        """Assume a command when no type-id is present.

        This is what makes promoting a plain dataclass to a multitype
        wire-compatible in both directions: the default serializes
        WITHOUT its type-id, so today's traffic is byte-identical to
        before, and keyless data from either side of a rollout still
        loads. Per the dataclassio wire invariants, once keyless data
        exists this designation may never change -- which is fine
        here, since 'a command' is what the envelope has always been.
        """
        return RequestTypeID.STANDARD

    @override
    @classmethod
    def get_type(cls, type_id: RequestTypeID) -> type[RequestData]:
        out: type[RequestData]
        if type_id is RequestTypeID.STANDARD:
            out = StandardRequestData
            return out
        raise ValueError(f'Unrecognized type-id {type_id}.')


class ResponseTypeID(Enum):
    """Type IDs for things a bacloud server can send back."""

    STANDARD = 's'
    SESSION_HANDLE = 'sh'
    STREAM_OUTPUT = 'so'
    CHUNK = 'ch'


class ResponseData(IOMultiType[ResponseTypeID]):
    """Root of what a bacloud server sends back.

    Counterpart to :class:`RequestData`; see its note. This is the
    side more likely to grow -- unsolicited notifications and live
    stream output are responses that are not answers to a command --
    and a discriminated union is what lets a reader tell them apart
    without correlation ids.
    """

    @override
    @classmethod
    def get_type_id(cls) -> ResponseTypeID:
        raise NotImplementedError()

    @override
    @classmethod
    def get_default_type_id(cls) -> ResponseTypeID:
        """Assume a standard response when no type-id is present.

        See the note on :meth:`RequestData.get_default_type_id`.
        """
        return ResponseTypeID.STANDARD

    @override
    @classmethod
    def get_type(cls, type_id: ResponseTypeID) -> type[ResponseData]:
        out: type[ResponseData]
        if type_id is ResponseTypeID.STANDARD:
            out = StandardResponseData
            return out
        if type_id is ResponseTypeID.SESSION_HANDLE:
            out = SessionHandleResponse
            return out
        if type_id is ResponseTypeID.STREAM_OUTPUT:
            out = StreamOutputResponse
            return out
        if type_id is ResponseTypeID.CHUNK:
            out = ChunkedResponse
            return out
        raise ValueError(f'Unrecognized type-id {type_id}.')


@ioprepped
@dataclass
class StandardRequestData(RequestData):
    """Request sent to bacloud server.

    Auth: bacloud always passes its current credential (API key or
    login_token) as ``Authorization: Bearer <value>``. There is no
    in-body token field — both flows look identical on the wire.
    """

    command: Annotated[str, IOAttrs('c')]
    payload: Annotated[dict, IOAttrs('p')]
    tzoffset: Annotated[float, IOAttrs('z')]
    isatty: Annotated[bool, IOAttrs('y')]

    #: Whether the originating CLI command is safe to retry even when a
    #: failure is *post-send* (the request may have reached the server).
    #: The client sets this for read-only / content-idempotent commands
    #: (assemble, version listings, etc.). basn's bacloud proxy reads it
    #: to decide whether a post-send upstream timeout should surface as
    #: a retryable 503 (idempotent) or a terminal error (mutating).
    #: Soft-defaults to ``False`` so older clients/payloads without the
    #: field deserialize as non-idempotent (fail-closed); has a Python
    #: default so non-CLI constructors needn't pass it.
    idempotent: Annotated[bool, IOAttrs('i', soft_default=False)] = False

    #: Engine build number of the caller. Master gates asset-package
    #: resolves on it (see ``MIN_SUPPORTED_ASSET_BUILD``): a build too old
    #: to address current source-named manifests gets a clean
    #: update-required error. Soft-defaults to ``0`` (not None) so
    #: requests/payloads lacking it read as build 0 -- always below the
    #: floor -- and gating stays a simple ``build_number < X``.
    build_number: Annotated[int, IOAttrs('b', soft_default=0)] = 0

    @override
    @classmethod
    def get_type_id(cls) -> RequestTypeID:
        return RequestTypeID.STANDARD


# Types used by the UploadPlan protocol. See StandardResponseData.UploadPlan.


@ioprepped
@dataclass
class UploadPlanFileInfo:
    """A single file entry the client is reporting to the server."""

    #: Relative path (from the upload plan's source_dir).
    name: Annotated[str, IOAttrs('n')]

    #: SHA-256 hex digest of the file body.
    sha256: Annotated[str, IOAttrs('h')]

    #: Size in bytes.
    size: Annotated[int, IOAttrs('s')]


@ioprepped
@dataclass
class UploadPlanPrepareRequest:
    """Client → server: request signed upload sessions for files."""

    files: Annotated[list[UploadPlanFileInfo], IOAttrs('f')]

    #: CloudFileCategory value (e.g. 'archive_misc'). Tells the
    #: server which bucket new uploads should land in.
    cloud_file_category: Annotated[str, IOAttrs('ct')]


@ioprepped
@dataclass
class UploadPlanPrepareItem:
    """Server → client: one file's prepare result."""

    #: Relative path matching the request entry.
    name: Annotated[str, IOAttrs('n')]

    #: The cloud_file_id this file maps to. Present when the file
    #: already exists server-side (dedup hit) — no upload needed.
    cloud_file_id: Annotated[str | None, IOAttrs('c', store_default=False)] = (
        None
    )

    #: Signed URL the client should PUT bytes to. Present when the
    #: file does not yet exist and needs to be uploaded.
    upload_url: Annotated[str | None, IOAttrs('u', store_default=False)] = None

    #: Headers the client must include on the PUT. Present when
    #: upload_url is set.
    upload_headers: Annotated[
        dict[str, str], IOAttrs('h', store_default=False)
    ] = field(default_factory=dict)

    #: Server-side upload session id. Present when upload_url is set.
    #: Passed back to the server in the finalize request.
    session_id: Annotated[str | None, IOAttrs('s', store_default=False)] = None


@ioprepped
@dataclass
class UploadPlanPrepareResponse:
    """Server → client: prepare results for each file."""

    items: Annotated[list[UploadPlanPrepareItem], IOAttrs('i')]


@ioprepped
@dataclass
class UploadPlanFinalizeRequest:
    """Client → server: finalize uploaded files by session id."""

    #: Map of file name → session_id for files that were uploaded.
    sessions: Annotated[dict[str, str], IOAttrs('s')]


@ioprepped
@dataclass
class UploadPlanFinalizeResponse:
    """Server → client: cloud-file-ids for each finalized upload."""

    #: Map of file name → cloud_file_id.
    cloud_file_ids: Annotated[dict[str, str], IOAttrs('c')]


@ioprepped
@dataclass
class UploadPlanCommit:
    """Client → command's finalize handler: full upload results.

    The client builds this after completing the prepare/upload/
    finalize cycle and invokes the upload-plan's
    ``finalize_command`` with it as the payload.
    """

    #: Complete mapping of {relative_path: cloud_file_id} for every
    #: file in the plan's source_dir.
    files: Annotated[dict[str, str], IOAttrs('f')]

    #: The opaque state dict from the original UploadPlan.
    state: Annotated[dict, IOAttrs('s')]


# ---------------------------------------------------------------- #
# StreamCall protocol types.
# ---------------------------------------------------------------- #


class StreamFrameTypeID(Enum):
    """Type IDs for the :class:`StreamFrame` IOMultiType."""

    OUTPUT = 'o'
    FINAL = 'f'


class StreamFrame(IOMultiType[StreamFrameTypeID]):
    """One frame in a streamed bacloud command's output.

    Producers emit a sequence of :class:`StreamOutput` frames as the
    underlying call generates output, then exactly one
    :class:`StreamFinal` frame carrying the terminal
    :class:`StandardResponseData`. These ride the internal
    bamaster→basn poll hop and basn's streamcall WS fan-out;
    the bacloud client itself never sees them (its stream output
    arrives as :class:`StreamOutputResponse` session messages).
    """

    @override
    @classmethod
    def get_type_id(cls) -> StreamFrameTypeID:
        raise NotImplementedError()

    @override
    @classmethod
    def get_type_id_storage_name(cls) -> str:
        # Pin to the original default for back-compat with stored data.
        return '_dciotype'

    @override
    @classmethod
    def get_type(cls, type_id: StreamFrameTypeID) -> type[StreamFrame]:
        # pylint: disable=cyclic-import
        t = StreamFrameTypeID
        if type_id is t.OUTPUT:
            return StreamOutput
        if type_id is t.FINAL:
            return StreamFinal
        assert_never(type_id)


@ioprepped
@dataclass
class StreamOutput(StreamFrame):
    """Incremental rendered output to print at the consumer."""

    text: Annotated[str, IOAttrs('t')] = ''

    @override
    @classmethod
    def get_type_id(cls) -> StreamFrameTypeID:
        return StreamFrameTypeID.OUTPUT


# Forward-prepped: ``response`` references :class:`StandardResponseData`,
# which is defined further down, so we explicitly :func:`ioprep` this
# class at the bottom of the module once both are defined.
@will_ioprep
@dataclass
class StreamFinal(StreamFrame):
    """Terminal frame carrying the call's :class:`StandardResponseData`.

    The inner ``response`` is processed by the consumer as if it had
    been the response to the originating non-streamed request — it
    can carry messages, ``end_command`` chains, errors, and so on.
    """

    # Lambda needed because :class:`StandardResponseData` isn't defined yet at
    # this point; resolved at instance-construction time.
    response: Annotated['StandardResponseData', IOAttrs('r')] = field(
        default_factory=(
            lambda: StandardResponseData()  # pylint: disable=unnecessary-lambda
        )
    )

    @override
    @classmethod
    def get_type_id(cls) -> StreamFrameTypeID:
        return StreamFrameTypeID.FINAL


@ioprepped
@dataclass
class StandardResponseData(ResponseData):
    """Response sent from the bacloud server to the client."""

    @ioprepped
    @dataclass
    class SignedUploadEntry:
        """Describes one direct-to-GCS streaming upload to perform."""

        #: Local file the client should read and PUT.
        path: Annotated[str, IOAttrs('p')]

        #: Signed GCS URL to PUT the bytes to.
        upload_url: Annotated[str, IOAttrs('u')]

        #: HTTP headers the client must include on the PUT (notably
        #: ``Content-MD5`` and ``Content-Type`` — these are bound into
        #: the URL signature, so the client cannot modify them).
        upload_headers: Annotated[dict[str, str], IOAttrs('h')]

        #: Server-side upload session id; the client passes this back
        #: in the next ``end_command`` args under ``uploads_signed`` to
        #: tell the server which upload to finalize.
        session_id: Annotated[str, IOAttrs('s')]

    @ioprepped
    @dataclass
    class OneshotUploadEntry:
        """Describes one small file to upload via the basn one-shot relay.

        The client POSTs the file body to the basn node it's already
        talking to (which relays it to the master server's one-shot
        cloud-file endpoint and has the master verify the
        content-addressed id), then sends the resulting
        ``cloud_file_id`` back in the next ``end_command`` args under
        ``uploads_oneshot`` (a ``dict[path, cloud_file_id]`` keyed by
        the same local path). Used for small files: the bytes ride one
        request to the node and the master holds them in memory and
        verifies them inline, avoiding both the signed-URL round-trip
        and the server-side blob read-back of the two-step path.
        """

        #: Local file the client should read and upload.
        path: Annotated[str, IOAttrs('p')]

        #: The content-addressed cloud_file_id the client declares for
        #: this file (from the manifest's sha256+size). The basn relay
        #: passes it through; the master recomputes from the bytes and
        #: rejects a mismatch.
        cloud_file_id: Annotated[str, IOAttrs('f')]

    @ioprepped
    @dataclass
    class SignedDownloadEntry:
        """Describes one direct-from-GCS streaming download to perform.

        The client streams bytes from ``download_url`` straight to
        ``path``, hashing as it goes and verifying against ``sha256``
        before atomic-renaming into place. Since the bytes bypass the
        bacloud response envelope entirely, there is no per-file size
        cap — files of any size can be delivered this way.
        """

        #: Local path the client should write the file to. Will be
        #: created (and its parent dirs) if missing; any existing file
        #: at the path is replaced atomically.
        path: Annotated[str, IOAttrs('p')]

        #: Signed GCS URL to GET the bytes from.
        download_url: Annotated[str, IOAttrs('u')]

        #: Hex-encoded SHA-256 of the expected file body. The client
        #: streams through ``hashlib.sha256`` and refuses to commit the
        #: file if the final digest does not match.
        sha256: Annotated[str, IOAttrs('h')]

        #: Expected file size in bytes, for progress reporting and a
        #: belt-and-suspenders sanity check against the streamed total.
        size: Annotated[int, IOAttrs('s')]

    @ioprepped
    @dataclass
    class UploadPlan:
        """Declarative plan for uploading a set of local files.

        When a command returns an ``UploadPlan`` in its response, the
        bacloud client takes over: it walks ``source_dir``, computes
        sha256+size for each file, checks which ones already exist
        server-side (dedup), streams the missing ones direct-to-GCS
        via signed URLs, and then invokes ``finalize_command`` with
        the resulting ``{relative_path: cloud_file_id}`` mapping.

        The entire upload protocol (manifest, dedup, signed uploads,
        finalize) is handled by shared client/server infrastructure.
        Individual commands only need to describe *what* to upload
        and *what to do* with the cloud-file-ids afterward.
        """

        #: Local directory to upload. All files under this directory
        #: (recursively) will be uploaded.
        source_dir: Annotated[str, IOAttrs('sd')]

        #: Command the client should invoke after all uploads are
        #: complete. The client will pass a ``UploadPlanCommit``
        #: payload to this command.
        finalize_command: Annotated[str, IOAttrs('fc')]

        #: Opaque state dict passed back to ``finalize_command`` so
        #: the originating command can remember its context (e.g.
        #: archive-id, version-number) without storing server-side.
        finalize_state: Annotated[dict, IOAttrs('fs')]

        #: Which CloudFileCategory new uploads should land in. Carried
        #: through to the prepare step, which embeds it into the
        #: signed upload session. Stored as the enum value string.
        cloud_file_category: Annotated[str, IOAttrs('ct')]

        #: Optional description shown to the user during the upload.
        description: Annotated[str, IOAttrs('de', store_default=False)] = ''

    @ioprepped
    @dataclass
    class CasDelivery:
        """CAS-delivery info for an asset-package assemble.

        Returned by an assemble run in CAS-delivery mode instead of
        per-blob signed-URL downloads. Carries everything a basn
        assemble-intercept needs to mint a ``/casblob`` capability token
        and warm the node CAS, and everything a CAS-aware bacloud needs to
        fetch the blobs from ``/casblob`` -- so neither side has to
        re-parse the inline flavor-manifests. ``token`` is minted and
        injected by the intercepting basn node (``None`` from the master).
        """

        #: Fully-qualified asset-package version id this bundle is for.
        apverid: Annotated[str, IOAttrs('a')]

        #: Bucket dimensions, carried for the capability token.
        texture_profile: Annotated[str, IOAttrs('tp')]
        texture_tier: Annotated[str, IOAttrs('tq')]
        language: Annotated[str, IOAttrs('l')]

        #: Every data blob the bundle needs as content-sha256 -> byte
        #: size, in manifest order (= the order to warm/fetch). Fetched
        #: from a basn node's ``/casblob`` warm cache.
        blobs: Annotated[dict[str, int], IOAttrs('b')]

        #: Capability token for ``GET /casblob/{hash}``; minted and
        #: injected by the basn node that warms and serves the blobs
        #: (``None`` as sent from the master).
        token: Annotated[
            securedata.Archive | None, IOAttrs('tk', store_default=False)
        ] = None

        #: The flavor-manifest blobs backing ``blobs``
        #: (content-sha256 -> canonical byte size). Sent by the master
        #: so the intercepting basn node can mint a fleet-portable
        #: capability token whose scope any node can verify (see
        #: ``baserver.assetcap.AssetCapabilityPayload.fm_digests``);
        #: the node consumes it and strips it from the response, so it
        #: never reaches the bacloud client. Empty from pre-scope
        #: masters (and, during rollout, when the basn fleet floor
        #: predates scope support).
        flavor_manifest_blobs: Annotated[
            dict[str, int],
            IOAttrs('fmb', store_default=False, soft_default_factory=dict),
        ] = field(default_factory=dict)

        #: DEPRECATED / UNUSED -- always empty. Compression is no longer
        #: carried here: a blob's transfer encoding is negotiated per
        #: ``/casblob`` request (the node reports it via its
        #: ``X-Cas-Compression`` response header and the client decodes per
        #: that). The field is retained (unpopulated, unread) only so the
        #: producer can stop sending it before any reader drops it -- safe
        #: to delete in a later cleanup once every component is updated.
        blob_compression: Annotated[
            dict[str, str],
            IOAttrs('bc', store_default=False, soft_default_factory=dict),
        ] = field(default_factory=dict)

    #: If present, client should print this message before any other
    #: response processing (including error handling) occurs.
    message: Annotated[str | None, IOAttrs('m', store_default=False)] = None

    #: Value for the 'end' arg of the message print() call.
    message_end: Annotated[str, IOAttrs('m_end', store_default=False)] = '\n'

    #: If present, client should print this message before any other
    #: response processing (including error handling) occurs.
    message_stderr: Annotated[
        str | None, IOAttrs('m2', store_default=False)
    ] = None

    #: Value for the 'end' arg of the message print() call.
    message_stderr_end: Annotated[
        str, IOAttrs('m2_end', store_default=False)
    ] = '\n'

    #: If present, client should abort with this error message and
    #: return-code 2.
    error: Annotated[str | None, IOAttrs('e', store_default=False)] = None

    #: If present for an interactive command, specifies the return code
    #: for the process. Note that this only applies if error is not set.
    #: Standard return codes are 0 for success, 1 for a successful run
    #: but negative result, and 2 for errors.
    return_code: Annotated[int | None, IOAttrs('r', store_default=False)] = None

    #: How long to wait before proceeding with remaining response (can
    #: be useful when waiting for server progress in a loop).
    delay_seconds: Annotated[float, IOAttrs('d', store_default=False)] = 0.0

    #: When > 0, the chained ``end_command`` this response carries is an
    #: idempotent polling step the client may safely RETRY on failure
    #: (transport errors and server-reported errors alike), with
    #: backoff, for up to this many seconds before surfacing the
    #: failure. Lets long polling loops (e.g. asset-package bundle
    #: assembles) ride out transient proxy timeouts / server hiccups
    #: instead of dying mid-flight. Servers must set this only on
    #: chains where a repeat call is always safe; the cost of the
    #: blanket any-failure retry policy is just that a genuine error
    #: on such a chain surfaces after the window elapses rather than
    #: instantly. Older clients ignore this field (no retry — the
    #: behavior before it existed).
    retry_window_seconds: Annotated[
        float, IOAttrs('rw', store_default=False)
    ] = 0.0

    #: If present, a token that should be stored client-side and passed
    #: with subsequent commands.
    login: Annotated[str | None, IOAttrs('l', store_default=False)] = None

    #: If True, any existing client-side token should be discarded.
    logout: Annotated[bool, IOAttrs('lo', store_default=False)] = False

    #: If present, client should generate a manifest of this dir.
    #: It should be added to end_command args as 'manifest'.
    dir_manifest: Annotated[str | None, IOAttrs('man', store_default=False)] = (
        None
    )

    #: If present, files the client should PUT directly to GCS via
    #: the provided signed URLs and then send the resulting session
    #: ids back to the server in the next end_command's args under
    #: 'uploads_signed' (a ``dict[path, session_id]`` keyed by the
    #: same local path that was sent in the entry). Streams without
    #: ever buffering the file body in master-server memory, so it
    #: imposes no per-file size limit.
    uploads_signed: Annotated[
        list[SignedUploadEntry] | None,
        IOAttrs('usgn', store_default=False),
    ] = None

    #: If present, small files the client should upload via the basn
    #: one-shot relay (POST the body to the node it's talking to, which
    #: relays to the master one-shot cloud-file endpoint), then send the
    #: resulting cloud_file_ids back in the next end_command's args under
    #: 'uploads_oneshot' (a ``dict[path, cloud_file_id]`` keyed by the
    #: same local path). The size cutoff vs. ``uploads_signed`` is
    #: decided server-side. Only emitted to v25+ clients (older clients
    #: get ``uploads_signed`` instead).
    uploads_oneshot: Annotated[
        list[OneshotUploadEntry] | None,
        IOAttrs('uos', store_default=False),
    ] = None

    #: If present, an upload plan the client should execute. See
    #: :class:`UploadPlan` for details. The client handles the full
    #: upload protocol (manifest, dedup, signed uploads, finalize)
    #: and then invokes the plan's ``finalize_command`` with the
    #: resulting cloud-file-id mapping.
    upload_plan: Annotated[
        UploadPlan | None, IOAttrs('upl', store_default=False)
    ] = None

    #: Free-form structured payload the server can return for
    #: client-internal consumption (e.g. upload-plan prepare/finalize
    #: results). Not displayed to the user and not written to disk.
    raw_result: Annotated[dict | None, IOAttrs('rr', store_default=False)] = (
        None
    )

    #: If present, file paths that should be deleted on the client.
    deletes: Annotated[
        list[str] | None, IOAttrs('dlt', store_default=False)
    ] = None

    #: If present, pathnames mapped to gzipped data to be written to the
    #: client. This should only be used for relatively small files as
    #: they are all included inline as part of the response — Cloud
    #: Run caps buffered response bodies at 32 MB, which places a hard
    #: ceiling on any payload delivered this way. For larger or
    #: CloudFile-backed downloads, use ``downloads_signed`` instead.
    downloads_inline: Annotated[
        dict[str, bytes] | None, IOAttrs('dinl', store_default=False)
    ] = None

    #: If present, files the client should GET directly from GCS via
    #: the provided signed URLs, verifying the streamed content against
    #: the provided SHA-256. Streams without ever buffering the file
    #: body in master-server memory, so it imposes no per-file size
    #: limit and bypasses the Cloud Run 32 MB response cap that
    #: constrains ``downloads_inline``.
    downloads_signed: Annotated[
        list[SignedDownloadEntry] | None,
        IOAttrs('dsgn', store_default=False),
    ] = None

    #: If present, the assemble ran in CAS-delivery mode: data blobs are
    #: delivered from a basn node's ``/casblob`` warm cache (see
    #: :class:`CasDelivery`) instead of per-blob GCS signed URLs.
    cas_delivery: Annotated[
        CasDelivery | None, IOAttrs('cas', store_default=False)
    ] = None

    #: If present, all empty dirs under this one should be removed.
    dir_prune_empty: Annotated[
        str | None, IOAttrs('dpe', store_default=False)
    ] = None

    #: If present, the workspace's current snapshot id after a completed
    #: workspace ``get``/``put`` (bacloud v24+ optimistic-concurrency).
    #: The client stashes it in ``<dir>/.bacloudstate.json`` and sends it
    #: back as the put's ``expected_snapshotid`` to detect mid-air
    #: collisions (the workspace changing between get and put).
    workspace_snapshotid: Annotated[
        str | None, IOAttrs('wss', store_default=False)
    ] = None

    #: If present, url to display to the user.
    open_url: Annotated[str | None, IOAttrs('url', store_default=False)] = None

    #: If present, a line of input is read and placed into end_command
    #: args as 'input'. The first value is the prompt printed before
    #: reading and the second is whether it should be read as a password
    #: (without echoing to the terminal).
    input_prompt: Annotated[
        tuple[str, bool] | None, IOAttrs('inp', store_default=False)
    ] = None

    #: If present, a message that should be printed after all other
    #: response processing is done.
    end_message: Annotated[str | None, IOAttrs('em', store_default=False)] = (
        None
    )

    #: End arg for end_message print() call.
    end_message_end: Annotated[str, IOAttrs('eme', store_default=False)] = '\n'

    #: If present, a message that should be printed after all other
    #: response processing is done.
    end_message_stderr: Annotated[
        str | None, IOAttrs('em2', store_default=False)
    ] = None

    #: End arg for end_message print() call.
    end_message_stderr_end: Annotated[
        str, IOAttrs('em2e', store_default=False)
    ] = '\n'

    #: If present, this command is run with these args at the end of
    #: response processing. Tuple is ``(command_name, args)``.
    end_command: Annotated[
        tuple[str, dict] | None, IOAttrs('ec', store_default=False)
    ] = None

    #: If present, this response kicked off a streamcall whose output
    #: should be followed before the response is treated as final.
    #: Consumed by the basn bacloud-session adapter: it subscribes to
    #: the call server-side, relays output down the session as
    #: :class:`StreamOutputResponse` messages, and answers with the
    #: stream's terminal response — the client never sees this field.
    #: Authorization is implicit: the id only ever rides a response
    #: to the (already-authenticated) request that created the call.
    stream_call_id: Annotated[
        str | None, IOAttrs('sci', store_default=False)
    ] = None

    @override
    @classmethod
    def get_type_id(cls) -> ResponseTypeID:
        return ResponseTypeID.STANDARD


@ioprepped
@dataclass
class StreamOutputResponse(ResponseData):
    """Live output from a streamed command, as it happens.

    Unsolicited, and the reason streaming needs no machinery of its
    own over a session: the server pushes these until the command's
    real answer arrives, and a reader tells them apart from that
    answer by type. Node-side, the basn session adapter follows the
    kickoff response's ``stream_call_id`` and produces these; the
    client plays no part in stream transport at all.

    Print ``text`` verbatim with no separator; the producer decides
    its own line breaks, exactly as with the frames this replaces.
    """

    text: Annotated[str, IOAttrs('t')]

    @override
    @classmethod
    def get_type_id(cls) -> ResponseTypeID:
        return ResponseTypeID.STREAM_OUTPUT


@ioprepped
@dataclass
class ChunkedResponse(ResponseData):
    """One ordered slice of a response too large for a single message.

    The transport caps a single message on purpose -- that cap is what
    the relay's resend-buffer and linger math are sized against, so it
    does not grow to fit one caller's payload. A response past it is
    therefore split here, above the transport, and rejoined by the
    reader before it decodes anything.

    No correlation id and no total-length field: the session delivers
    gaplessly and in order (that is its whole invariant), and a sender
    finishes one response before starting the next, so "collect until
    ``index == count - 1``" is sufficient and cannot interleave. A
    reader that sees a gap has already lost the session.

    ``data`` is a slice of the *serialized* response, not a serialized
    slice -- rejoining is string concatenation, and the result decodes
    exactly as it would have unsplit. That keeps every existing
    response type carryable with no per-type awareness here.
    """

    #: Position of this slice, from zero.
    index: Annotated[int, IOAttrs('i')]

    #: How many slices the whole response was split into.
    count: Annotated[int, IOAttrs('n')]

    #: This slice of the serialized response.
    data: Annotated[str, IOAttrs('d')]

    @override
    @classmethod
    def get_type_id(cls) -> ResponseTypeID:
        return ResponseTypeID.CHUNK


@ioprepped
@dataclass
class SessionHandleResponse(ResponseData):
    """Unsolicited: the resume token for a session just created.

    A bacloud session's channel is created by the handshake itself --
    the client dials with no token at all -- so the node has to hand
    back the credential for *re*-attaching to that channel. It arrives
    as the session's first message rather than as a field on some
    answer, because it is not an answer to anything: this is exactly
    the case the response hierarchy exists for, and a reader tells it
    apart from a command result by type rather than by correlation.

    The client stores it and presents it as ``X-WS-Token`` on any
    reconnect. It needs no refreshing: it is minted to outlive the
    session's own duration cap, and a channel that has ended is
    tombstoned and refuses every token ever minted for it.
    """

    #: Signed capability token, base64url-encoded. Opaque to the
    #: client, which only ever echoes it back on a reconnect.
    token: Annotated[str, IOAttrs('t')]

    #: Id of the channel this session runs on. Observability only --
    #: the token is the authority -- but load-bearing for support: it
    #: is the single string that ties a client-side failure to the
    #: node-side log lines for the same session, and without it a
    #: bacloud run cannot name the session it was using.
    channel_id: Annotated[str, IOAttrs('c')]

    #: Where to reconnect, naming the node that holds this session.
    #:
    #: Load-bearing, and not the same host the client dialed. In prod
    #: bacloud dials a regional endpoint that routes each connection
    #: to *some* node; session state lives in one node's process, so
    #: re-dialing that endpoint would usually land somewhere with no
    #: session to resume. (Streamcall can be node-agnostic because its
    #: state is bamaster's; this is not.) The node fills in its own
    #: address here and the client uses it for reconnects only.
    ws_url: Annotated[str, IOAttrs('u')]

    #: Unix-seconds expiry, for diagnostics. A client that finds
    #: itself past this has been away longer than the session could
    #: possibly have survived, so it should start a fresh one.
    expiry_unix_seconds: Annotated[int, IOAttrs('e')]

    @override
    @classmethod
    def get_type_id(cls) -> ResponseTypeID:
        return ResponseTypeID.SESSION_HANDLE


# Now that :class:`StandardResponseData` exists in the module namespace,
# finish prep of :class:`StreamFinal` (which references StandardResponseData
# via its ``response`` field).
ioprep(StreamFinal)
