lightseekorg--tokenspeed/python/tokenspeed/runtime/engine/io_struct.py

# Copyright (c) 2026 LightSeek Foundation
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""
The definition of objects transferred between different
processes (TokenizerManager, DetokenizerManager, Controller).
"""

import copy
import uuid
from abc import ABC
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Literal

from tokenspeed.runtime.engine.request_types import BaseFinishReason
from tokenspeed.runtime.sampling.sampling_params import SamplingParams


def _require(condition: bool, message: str) -> None:
    if not condition:
        raise ValueError(message)


@dataclass
class BaseReq(ABC):
    rid: str | list[str] | None = field(default=None)
    http_worker_ipc: str | None = field(default=None)

    def regenerate_rid(self):
        """Generate a new request ID and return it."""
        if isinstance(self.rid, list):
            self.rid = [uuid.uuid4().hex for _ in range(len(self.rid))]
        else:
            self.rid = uuid.uuid4().hex
        return self.rid


@dataclass
class SessionParams:
    id: str | None = None
    rid: str | None = None
    offset: int | None = None
    replace: bool | None = None


@dataclass
class GenerateReqInput:
    # The input prompt. It can be a single prompt or a batch of prompts.
    text: list[str] | str | None = None
    # The token ids for text; one can specify either text or input_ids
    input_ids: list[list[int]] | list[int] | None = None
    input_multi_ids: list[list[int]] | list[list[int]] | None = None
    # The embeddings for input_ids; one can specify either text or input_ids or input_embeds.
    input_embeds: list[list[list[float]]] | list[list[float]] | None = None
    # Pre-built MultimodalInputs (already produced by an upstream preprocessor,
    # e.g. SMG's Rust crates/multimodal pipeline). The engine's InputProcessor
    # uses this directly (it does no in-process image preprocessing). input_ids
    # must already contain expanded image placeholder tokens at the right
    # offsets — the gateway is responsible for that. Typed as Any to avoid a
    # circular import on MultimodalInputs.
    precomputed_multimodal_inputs: Any | None = None
    # The sampling_params. See descriptions below.
    sampling_params: list[dict] | dict | None = None
    input_extra_infos: list[dict] | dict | None = None
    # Optional client label for logging; defaults to `rid`. Safe to reuse.
    user_rid: list[str] | str | None = None
    # Routing id; always server-assigned during normalize, never caller-settable.
    rid: list[str] | str | None = field(default=None, init=False)
    # --- Logprob request (two dialects, one compute path) ---
    # vLLM-compatible requests use ``sampling_params["logprobs"]``;
    # SGLang-compatible requests use the legacy fields below. A request uses
    # one dialect; the response is rendered to match (override with
    # ``logprob_format``).
    return_logprob: list[bool] | bool | None = None
    # Start location in the prompt for prompt logprobs. -1 (default) = output
    # tokens only.
    logprob_start_len: list[int] | int | None = None
    # Number of top logprobs per position.
    top_logprobs_num: list[int] | int | None = None
    # Specific token ids to score per position.
    token_ids_logprob: list[list[int]] | list[int] | None = None
    # Detokenize tokens in the returned logprobs.
    return_text_in_logprobs: bool = False
    # Output rendering dialect: "vllm" | "sglang" | "both". None = auto (match
    # the request dialect: vllm if sampling_params.logprobs is set, else sglang).
    logprob_format: list[str | None] | str | None = None
    # Whether to stream output.
    stream: bool = False
    # Whether to log metrics for this request (e.g. health_generate calls do not log metrics)
    log_metrics: bool = True

    # Session info for continual prompting
    session_params: list[dict] | dict | None = None

    # Custom logit processor for advanced sampling control. Must be a serialized instance
    # of `CustomLogitProcessor` in python/tokenspeed/runtime/sampling/custom_logit_processor.py
    # Use the processor's `to_str()` method to generate the serialized string.
    custom_logit_processor: list[str | None] | str | None = None

    # Whether to return hidden states
    return_hidden_states: bool = False

    # For disaggregated inference
    bootstrap_host: list[str] | str | None = None
    bootstrap_port: list[int] | int | None = None
    bootstrap_room: list[int] | int | None = None

    def normalize_batch_and_arguments(self):
        if (
            self.text is None and self.input_ids is None and self.input_embeds is None
        ) or (
            self.text is not None
            and self.input_ids is not None
            and self.input_embeds is not None
        ):
            raise ValueError(
                "Either text, input_ids or input_embeds should be provided."
            )

        # Derive the batch size
        if self.text is not None:
            if isinstance(self.text, str):
                self.is_single = True
                self.batch_size = 1
            else:
                self.is_single = False
                self.batch_size = len(self.text)
            self.input_embeds = None
        elif self.input_ids is not None:
            if isinstance(self.input_ids[0], int):
                self.is_single = True
                self.batch_size = 1
            else:
                self.is_single = False
                self.batch_size = len(self.input_ids)
            self.input_embeds = None
        else:
            _require(
                isinstance(self.input_embeds, list), "input_embeds should be a list."
            )
            if isinstance(self.input_embeds[0][0], float):
                # list[list[float]]
                self.is_single = True
                self.batch_size = 1
            else:
                # list[list[list[float]]]
                _require(
                    isinstance(self.input_embeds[0][0], list),
                    "input_embeds should be a list of float lists.",
                )
                _require(
                    isinstance(self.input_embeds[0][0][0], float),
                    "input_embeds should contain floats.",
                )
                self.is_single = False
                self.batch_size = len(self.input_embeds)

        # Handle parallel sampling. Pop "n" out of sampling_params so the
        # downstream SamplingParams(**dict) construction doesn't see it —
        # "n" is a request-level fan-out knob, not a per-sample field.
        if self.sampling_params is None:
            self.parallel_sample_num = 1
        elif isinstance(self.sampling_params, dict):
            self.parallel_sample_num = self.sampling_params.get("n", 1)
        else:  # isinstance(self.sampling_params, list):
            self.parallel_sample_num = self.sampling_params[0].get("n", 1)
            for sp in self.sampling_params[1:]:
                _require(
                    self.parallel_sample_num == sp.get("n", 1),
                    "The parallel_sample_num should be the same for all samples in sample params.",
                )

        if self.parallel_sample_num > 1 and self.is_single:
            self.is_single = False
            if self.text is not None:
                self.text = [self.text]
            if self.input_ids is not None:
                self.input_ids = [self.input_ids]
            if self.input_multi_ids is not None:
                self.input_multi_ids = [self.input_multi_ids]
            if self.input_embeds is not None:
                self.input_embeds = [self.input_embeds]

        # Fill in default arguments
        if self.is_single:
            if self.sampling_params is None:
                self.sampling_params = {}
            if self.rid is None:
                self.rid = uuid.uuid4().hex
            if self.user_rid is None:
                self.user_rid = self.rid
            else:
                if isinstance(self.user_rid, list):
                    _require(
                        len(self.user_rid) == 1,
                        "user_rid list should have length 1 for single request.",
                    )
                    self.user_rid = self.user_rid[0]
                _require(isinstance(self.user_rid, str), "user_rid should be a str.")
            if self.return_logprob is None:
                self.return_logprob = False
            if self.logprob_start_len is None:
                self.logprob_start_len = -1
            if self.top_logprobs_num is None:
                self.top_logprobs_num = 0
            if not self.token_ids_logprob:  # covers both None and []
                self.token_ids_logprob = None
            if isinstance(self.input_extra_infos, dict):
                self.input_extra_infos = [self.input_extra_infos]
        else:
            if self.parallel_sample_num == 1:
                num = self.batch_size
            else:
                # Expand parallel_sample_num
                num = self.batch_size * self.parallel_sample_num

            if self.sampling_params is None:
                self.sampling_params = [{} for _ in range(num)]
            elif not isinstance(self.sampling_params, list):
                self.sampling_params = [dict(self.sampling_params) for _ in range(num)]

            if self.rid is None:
                self.rid = [uuid.uuid4().hex for _ in range(num)]
            else:
                _require(isinstance(self.rid, list), "The rid should be a list.")
            if self.user_rid is None:
                self.user_rid = list(self.rid)
            elif isinstance(self.user_rid, str):
                self.user_rid = [self.user_rid] * num
            else:
                _require(
                    isinstance(self.user_rid, list) and len(self.user_rid) == num,
                    "user_rid should be a str or a list of matching length.",
                )

            if self.return_logprob is None:
                self.return_logprob = [False] * num
            elif not isinstance(self.return_logprob, list):
                self.return_logprob = [self.return_logprob] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "return_logprob cannot be a list when n > 1.",
                )

            if self.logprob_start_len is None:
                self.logprob_start_len = [-1] * num
            elif not isinstance(self.logprob_start_len, list):
                self.logprob_start_len = [self.logprob_start_len] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "logprob_start_len cannot be a list when n > 1.",
                )

            if self.top_logprobs_num is None:
                self.top_logprobs_num = [0] * num
            elif not isinstance(self.top_logprobs_num, list):
                self.top_logprobs_num = [self.top_logprobs_num] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "top_logprobs_num cannot be a list when n > 1.",
                )

            if not self.token_ids_logprob:  # covers both None and []
                self.token_ids_logprob = [None] * num
            elif not isinstance(self.token_ids_logprob, list):
                self.token_ids_logprob = [[self.token_ids_logprob] for _ in range(num)]
            elif not isinstance(self.token_ids_logprob[0], list):
                self.token_ids_logprob = [
                    copy.deepcopy(self.token_ids_logprob) for _ in range(num)
                ]
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "token_ids_logprob cannot be nested lists when n > 1.",
                )

            if self.logprob_format is None or isinstance(self.logprob_format, str):
                self.logprob_format = [self.logprob_format] * num

            if self.custom_logit_processor is None:
                self.custom_logit_processor = [None] * num
            elif not isinstance(self.custom_logit_processor, list):
                self.custom_logit_processor = [self.custom_logit_processor] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "custom_logit_processor cannot be a list when n > 1.",
                )

            if self.bootstrap_host is None:
                self.bootstrap_host = [None] * num
            elif not isinstance(self.bootstrap_host, list):
                self.bootstrap_host = [self.bootstrap_host] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "bootstrap_host cannot be a list when n > 1.",
                )

            if self.bootstrap_port is None:
                self.bootstrap_port = [None] * num
            elif not isinstance(self.bootstrap_port, list):
                self.bootstrap_port = [self.bootstrap_port] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "bootstrap_port cannot be a list when n > 1.",
                )

            if self.bootstrap_room is None:
                self.bootstrap_room = [None] * num
            elif not isinstance(self.bootstrap_room, list):
                self.bootstrap_room = [self.bootstrap_room] * num
            else:
                _require(
                    self.parallel_sample_num == 1,
                    "bootstrap_room cannot be a list when n > 1.",
                )

        # Other checks
        if self.session_params is not None:
            _require(
                isinstance(self.session_params, dict)
                or isinstance(self.session_params[0], dict),
                "session_params should be a dict or a list of dicts.",
            )

    def regenerate_rid(self):
        self.rid = uuid.uuid4().hex
        return self.rid

    def __getitem__(self, i):
        sub = GenerateReqInput(
            text=self.text[i] if self.text is not None else None,
            input_ids=self.input_ids[i] if self.input_ids is not None else None,
            # precomputed_multimodal_inputs is a single prompt's MM; the SMG
            # path only clears is_single via n>1 (batch_size == 1), so all n
            # parallel samples correctly share it. Without this the image is
            # silently dropped on the n>1 fan-out (placeholders -> text path).
            precomputed_multimodal_inputs=self.precomputed_multimodal_inputs,
            input_multi_ids=(
                self.input_multi_ids[i] if self.input_multi_ids is not None else None
            ),
            input_embeds=(
                self.input_embeds[i] if self.input_embeds is not None else None
            ),
            input_extra_infos=(
                self.input_extra_infos[i]
                if self.input_extra_infos is not None
                else None
            ),
            sampling_params=self.sampling_params[i],
            user_rid=self.user_rid[i],
            return_logprob=self.return_logprob[i],
            logprob_start_len=self.logprob_start_len[i],
            top_logprobs_num=self.top_logprobs_num[i],
            token_ids_logprob=self.token_ids_logprob[i],
            return_text_in_logprobs=self.return_text_in_logprobs,
            logprob_format=self.logprob_format[i],
            stream=self.stream,
            log_metrics=self.log_metrics,
            custom_logit_processor=(
                self.custom_logit_processor[i]
                if self.custom_logit_processor is not None
                else None
            ),
            return_hidden_states=self.return_hidden_states,
            # if `__getitem__` is called, the bootstrap_host, bootstrap_port, bootstrap_room must be a list
            bootstrap_host=(
                self.bootstrap_host[i] if self.bootstrap_host is not None else None
            ),
            bootstrap_port=(
                self.bootstrap_port[i] if self.bootstrap_port is not None else None
            ),
            bootstrap_room=(
                self.bootstrap_room[i] if self.bootstrap_room is not None else None
            ),
        )
        sub.rid = self.rid[i]
        return sub


@dataclass
class TokenizedGenerateReqInput:
    # The request id
    rid: str
    # The input text
    input_text: str
    # The input token ids
    input_ids: list[int]
    # The sampling parameters
    sampling_params: SamplingParams
    # Whether to return the sampled token's logprob for this request.
    return_logprob: bool
    # Internal carry-over fields kept for pipeline/PD compatibility. The vLLM
    # output-logprob API only drives ``return_logprob``; InputProcessor sets
    # these to neutral values (logprob_start_len=-1, top_logprobs_num=0,
    # token_ids_logprob=None) since prompt logprobs, output top-k, and token-id
    # logprobs are not supported.
    logprob_start_len: int
    top_logprobs_num: int
    token_ids_logprob: list[int]
    # Whether to stream output
    stream: bool

    # The input embeds
    input_embeds: list[list[list[float]]] | list[list[float]] | None = None

    # Session info for continual prompting
    session_params: SessionParams | None = None

    # Custom logit processor for advanced sampling control. Must be a serialized instance
    # of `CustomLogitProcessor` in python/tokenspeed/runtime/sampling/custom_logit_processor.py
    # Use the processor's `to_str()` method to generate the serialized string.
    custom_logit_processor: str | None = None

    # Whether to return hidden states
    return_hidden_states: bool = False

    # Time at object instantiated
    created_time: float = 0.0

    # For disaggregated inference
    bootstrap_host: str | None = None
    bootstrap_port: int | None = None
    bootstrap_room: int | None = None

    input_multi_ids: list[list[int]] = None
    input_extra_infos: list[dict] | None = None
    # Original prompt ids before multimodal pad/hash replacement. The scheduler
    # uses input_ids, while detokenization must use these tokenizer-valid ids.
    input_ids_unpadded: list[int] | None = None
    multimodal_inputs: Any | None = None


@dataclass
class EmbeddingReqInput:
    # The input prompt. It can be a single prompt or a batch of prompts.
    text: list[str] | str | None = None
    # The token ids for text; one can either specify text or input_ids.
    input_ids: list[list[int]] | list[int] | None = None
    # Optional client label for logging; defaults to `rid`. Safe to reuse.
    user_rid: list[str] | str | None = None
    # Routing id; always server-assigned during normalize, never caller-settable.
    rid: list[str] | str | None = field(default=None, init=False)
    # Optional placeholder so non-generation callers can still instantiate the
    # shared request shape without real sampling params.
    sampling_params: list[dict] | dict = None
    # Optional placeholder for callers that do not provide input embeddings.
    input_embeds: list[list[list[float]]] | list[list[float]] | None = None
    # Whether to log metrics for this request (e.g. health_generate calls do not log metrics)
    log_metrics: bool = True

    def normalize_batch_and_arguments(self):
        if (self.text is None and self.input_ids is None) or (
            self.text is not None and self.input_ids is not None
        ):
            raise ValueError("Either text or input_ids should be provided.")

        # Derive the batch size
        if self.text is not None:
            if isinstance(self.text, str):
                self.is_single = True
                self.batch_size = 1
            else:
                self.is_single = False
                self.batch_size = len(self.text)
        else:
            if isinstance(self.input_ids[0], int):
                self.is_single = True
                self.batch_size = 1
            else:
                self.is_single = False
                self.batch_size = len(self.input_ids)

        # Fill in default arguments
        if self.is_single:
            if self.rid is None:
                self.rid = uuid.uuid4().hex
            if self.user_rid is None:
                self.user_rid = self.rid
            else:
                if isinstance(self.user_rid, list):
                    _require(
                        len(self.user_rid) == 1,
                        "user_rid list should have length 1 for single request.",
                    )
                    self.user_rid = self.user_rid[0]
                _require(isinstance(self.user_rid, str), "user_rid should be a str.")
            if self.sampling_params is None:
                self.sampling_params = {}
            self.sampling_params["max_new_tokens"] = 0
        else:
            if self.rid is None:
                self.rid = [uuid.uuid4().hex for _ in range(self.batch_size)]
            else:
                _require(isinstance(self.rid, list), "The rid should be a list.")
            if self.user_rid is None:
                self.user_rid = list(self.rid)
            elif isinstance(self.user_rid, str):
                self.user_rid = [self.user_rid] * self.batch_size
            else:
                _require(
                    isinstance(self.user_rid, list)
                    and len(self.user_rid) == self.batch_size,
                    "user_rid should be a str or a list of matching length.",
                )

            if self.sampling_params is None:
                self.sampling_params = [{} for _ in range(self.batch_size)]
            for i in range(self.batch_size):
                self.sampling_params[i]["max_new_tokens"] = 0

    def regenerate_rid(self):
        self.rid = uuid.uuid4().hex
        return self.rid

    def __getitem__(self, i):
        sub = EmbeddingReqInput(
            text=self.text[i] if self.text is not None else None,
            input_ids=self.input_ids[i] if self.input_ids is not None else None,
            sampling_params=self.sampling_params[i],
            user_rid=self.user_rid[i],
        )
        sub.rid = self.rid[i]
        return sub


@dataclass
class TokenizedEmbeddingReqInput:
    # The request id
    rid: str
    # The input text
    input_text: str
    # The input token ids
    input_ids: list[int]
    # Placeholder sampling params field so request metadata can share one shape
    # with generation-oriented code paths.
    sampling_params: SamplingParams
    # Time at object instantiated
    created_time: float


@dataclass
class BatchTokenIDOut:
    # The request id
    rids: list[str]
    # The finish reason
    finished_reasons: list[BaseFinishReason]
    # For incremental decoding
    decoded_texts: list[str]
    decode_ids: list[list[int]]
    read_offsets: list[int]
    # Only used when `--skip-tokenizer-init` is on
    output_ids: list[int] | None
    output_multi_ids: list[int] | None
    # Detokenization configs
    skip_special_tokens: list[bool]
    spaces_between_special_tokens: list[bool]
    no_stop_trim: list[bool]

    # Token counts
    prompt_tokens: list[int]
    completion_tokens: list[int]
    cached_tokens: list[int]
    spec_verify_ct: list[int]

    # Logprobs
    input_token_logprobs_val: list[float]
    input_token_logprobs_idx: list[int]
    output_token_logprobs_val: list[float]
    output_token_logprobs_idx: list[int]
    input_top_logprobs_val: list[list]
    input_top_logprobs_idx: list[list]
    output_top_logprobs_val: list[list]
    output_top_logprobs_idx: list[list]
    input_token_ids_logprobs_val: list[list]
    input_token_ids_logprobs_idx: list[list]
    output_token_ids_logprobs_val: list[list]
    output_token_ids_logprobs_idx: list[list]

    # Hidden states
    output_hidden_states: list[list[float]]
    batch_accept_draft_tokens: list[float]

    # Store some custom information, such as decoding status in multimodal scenarios, etc.
    output_extra_infos: list[dict[str, Any]]

    generated_time: int


@dataclass
class BatchStrOut:
    # The request id
    rids: list[str]
    # The finish reason
    finished_reasons: list[dict]
    # The output decoded strings
    output_strs: list[str]
    # The token ids
    output_ids: list[int] | None

    # Token counts
    prompt_tokens: list[int]
    completion_tokens: list[int]
    cached_tokens: list[int]
    spec_verify_ct: list[int]

    # Logprobs
    input_token_logprobs_val: list[float]
    input_token_logprobs_idx: list[int]
    output_token_logprobs_val: list[float]
    output_token_logprobs_idx: list[int]
    input_top_logprobs_val: list[list]
    input_top_logprobs_idx: list[list]
    output_top_logprobs_val: list[list]
    output_top_logprobs_idx: list[list]
    input_token_ids_logprobs_val: list[list]
    input_token_ids_logprobs_idx: list[list]
    output_token_ids_logprobs_val: list[list]
    output_token_ids_logprobs_idx: list[list]

    # Hidden states
    output_hidden_states: list[list[float]]
    batch_accept_draft_tokens: list[float]

    # Store some custom information, such as decoding status in multimodal scenarios, etc.
    output_extra_infos: list[dict[str, Any]]

    generated_time: int


@dataclass
class BatchEmbeddingOut:
    # The request id
    rids: list[str]
    # The finish reason
    finished_reasons: list[BaseFinishReason]
    # The output embedding
    embeddings: list[list[float]] | list[dict]
    # Token counts
    prompt_tokens: list[int]


@dataclass
class FlushCacheReqInput:
    pass


@dataclass
class FlushCacheReqOutput:
    success: bool


# How a pause should treat in-flight requests.
# - "abort": kill in-flight requests immediately, then stop admitting new ones.
# - "wait":  stop admitting new ones, keep stepping until running requests drain.
# - "keep":  freeze everything in place; resume picks up where it left off.
PauseMode = Literal["abort", "wait", "keep"]


@dataclass
class PauseSchedulerReqInput:
    # See PauseMode for how each mode treats in-flight requests.
    mode: PauseMode = "abort"


@dataclass
class PauseSchedulerReqOutput:
    success: bool
    message: str = ""


@dataclass
class ResumeSchedulerReqInput:
    pass


@dataclass
class ResumeSchedulerReqOutput:
    success: bool
    message: str = ""


@dataclass
class IsSchedulerPausedReqInput:
    pass


@dataclass
class IsSchedulerPausedReqOutput:
    is_paused: bool


@dataclass
class UpdateWeightFromDiskReqInput:
    # The model path with the new weights
    model_path: str
    # The format to load the weights
    load_format: str | None = None


@dataclass
class UpdateWeightFromDiskReqOutput:
    success: bool
    message: str
    # Number of paused requests during weight sync.
    num_paused_requests: int | None = 0


@dataclass
class UpdateWeightsFromDistributedReqInput:
    name: str
    dtype: str
    shape: list[int]


@dataclass
class UpdateWeightsFromDistributedReqOutput:
    success: bool
    message: str


@dataclass
class UpdateWeightsFromTensorReqInput:
    serialized_named_tensors: bytes  # indeed Dict[str, torch.Tensor]
    load_format: str | None
    flush_cache: bool


@dataclass
class UpdateWeightsFromTensorReqOutput:
    success: bool
    message: str


@dataclass
class InitWeightsUpdateGroupReqInput:
    # The master address
    master_address: str
    # The master port
    master_port: int
    # The rank offset
    rank_offset: int
    # The world size
    world_size: int
    # The group name
    group_name: str = "weight_update_group"
    # The backend
    backend: str = "nccl"


@dataclass
class InitWeightsUpdateGroupReqOutput:
    success: bool
    message: str


@dataclass
class GetWeightsByNameReqInput:
    name: str
    truncate_size: int = 100


@dataclass
class GetWeightsByNameReqOutput:
    parameter: list


@dataclass
class ReleaseMemoryOccupationReqInput:
    # Memory regions to release. None ⇒ all ("weights" and "kv_cache").
    tags: list[str] | None = None


@dataclass
class ReleaseMemoryOccupationReqOutput:
    success: bool = True
    message: str = ""


@dataclass
class ResumeMemoryOccupationReqInput:
    # Memory regions to resume. None ⇒ all previously released tags.
    tags: list[str] | None = None


@dataclass
class ResumeMemoryOccupationReqOutput:
    success: bool = True
    message: str = ""


@dataclass
class IsSleepingReqInput:
    pass


@dataclass
class IsSleepingReqOutput:
    is_sleeping: bool


@dataclass
class AbortReq:
    # The request id
    rid: str


@dataclass
class GetInternalStateReq:
    pass


@dataclass
class GetInternalStateReqOutput:
    internal_state: dict[Any, Any]


@dataclass
class SetInternalStateReq:
    server_args: dict[str, Any]


@dataclass
class SetInternalStateReqOutput:
    updated: bool
    server_args: dict[str, Any]


class ExpertDistributionReq(Enum):
    START_RECORD = 1
    STOP_RECORD = 2
    DUMP_RECORD = 3


@dataclass
class ExpertDistributionReqOutput:
    pass


class ProfileReqType(Enum):
    START_PROFILE = 1
    STOP_PROFILE = 2


@dataclass
class ProfileReq:
    type: ProfileReqType
    output_dir: str | None = None
    start_step: int | None = None
    num_steps: int | None = None
    activities: list[str] | None = None
    profile_by_stage: bool = False
    with_stack: bool | None = None
    record_shapes: bool | None = None
    profile_id: str | None = None


@dataclass
class ProfileReqOutput:
    success: bool
    message: str


@dataclass
class ConfigureLoggingReq:
    log_requests: bool | None = None
    log_requests_level: int | None = None
    dump_requests_folder: str | None = None
    dump_requests_threshold: int | None = None


@dataclass
class OpenSessionReqInput:
    capacity_of_str_len: int
    session_id: str | None = None


@dataclass
class CloseSessionReqInput:
    session_id: str


@dataclass
class OpenSessionReqOutput:
    session_id: str | None
    success: bool


@dataclass
class HealthCheckOutput:
    pass


@dataclass
class RpcReqInput:
    method: str
    parameters: dict | None = None


@dataclass
class RpcReqOutput:
    success: bool
    message: str


@dataclass
class GetLoadReqInput(BaseReq):
    pass


@dataclass
class GetLoadReqOutput(BaseReq):
    dp_rank: int = 0
    num_reqs: int = 0
    num_waiting_reqs: int = 0
    num_pages: int = 0


@dataclass
class WatchLoadUpdateReq(BaseReq):
    loads: list[GetLoadReqOutput] = field(default_factory=list)


class BlockReqType(Enum):
    BLOCK = 1
    UNBLOCK = 2


@dataclass
class BlockReqInput(BaseReq):
    type: BlockReqType = field(default_factory=BlockReqType.BLOCK)