diff --git a/lightllm/common/attention/__init__.py b/lightllm/common/attention/__init__.py
new file mode 100644
index 0000000000..c3b355bdbb
--- /dev/null
+++ b/lightllm/common/attention/__init__.py
@@ -0,0 +1,11 @@
+from .api import (
+    AttentionBackend,
+    AttentionBatch,
+    AttentionFamily,
+    AttentionPhase,
+    CacheLayout,
+    ContextCapacity,
+    LayerAttentionBinding,
+    LayerSpec,
+    UnsupportedAttention,
+)
diff --git a/lightllm/common/attention/_environment.py b/lightllm/common/attention/_environment.py
new file mode 100644
index 0000000000..85fdec4981
--- /dev/null
+++ b/lightllm/common/attention/_environment.py
@@ -0,0 +1,17 @@
+import importlib
+
+from .api.backend import UnsupportedAttention
+
+
+def check_attention_environment(caps, *, provider, supported_platforms, min_device_capability, dependency):
+    if caps.platform not in supported_platforms:
+        raise UnsupportedAttention(f"{provider} does not support platform {caps.platform.value}")
+    if not caps.available:
+        raise UnsupportedAttention(f"device {caps.device} is unavailable")
+    if min_device_capability is not None and (caps.capability is None or caps.capability < min_device_capability):
+        raise UnsupportedAttention(f"requires device capability {min_device_capability}")
+    try:
+        if dependency:
+            importlib.import_module(dependency)
+    except (ImportError, OSError) as exc:
+        raise UnsupportedAttention(str(exc)) from exc
diff --git a/lightllm/common/attention/api/__init__.py b/lightllm/common/attention/api/__init__.py
new file mode 100644
index 0000000000..709bf2c792
--- /dev/null
+++ b/lightllm/common/attention/api/__init__.py
@@ -0,0 +1,19 @@
+"""模型层与 attention 实现之间的契约。
+
+初始化时，LayerSpec / CacheLayout 用于选择后端和划分可共享 context 的层组，
+LayerAttentionBinding 保存本层参数和共享 backend，通过 forward(context, inputs) 执行。
+运行时，AttentionBatch 描述整个批次，Inputs 提供当前层的 tensor 输入；
+backend 专用 context 保存派生 metadata 和 provider wrapper，由 InferStateInfo 持有。
+KV / Conv / SSM 的分配、回收和迁移仍由 AttentionStateManager 管理。
+无持久 KV 的视觉/音频编码器使用 VarlenAttentionOp，直接传入 Q/K/V 和序列边界。
+
+初始化集成见 common.basemodel.attention，运行时入口见 InferStateInfo.prepare_attention。
+"""
+
+from .backend import AttentionBackend, LayerAttentionBinding, UnsupportedAttention
+from .batch import AttentionBatch, ContextCapacity
+from .execution import BackendResources, ExecutionRequirements
+from .graph import GraphBinding
+from .inputs import VarlenAttentionInputs
+from .spec import AttentionFamily, AttentionPhase, CacheLayout, LayerSpec
+from .varlen import VarlenAttentionOp
diff --git a/lightllm/common/attention/api/backend.py b/lightllm/common/attention/api/backend.py
new file mode 100644
index 0000000000..22a82d8d4e
--- /dev/null
+++ b/lightllm/common/attention/api/backend.py
@@ -0,0 +1,102 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Callable, Generic, Protocol, TypeVar
+
+import torch
+
+from .batch import AttentionBatch, ContextCapacity
+from .execution import BackendResources
+from .spec import LayerSpec
+
+Context = TypeVar("Context")
+Inputs = TypeVar("Inputs")
+Output = TypeVar("Output")
+Params = TypeVar("Params")
+
+
+class UnsupportedAttention(ValueError):
+    """当前环境或输入不满足此 attention 实现的要求。"""
+
+
+class AttentionExecution(Protocol):
+    def attention_context(self, binding: "LayerAttentionBinding[Params, Context, Inputs, Output]") -> Context:
+        ...
+
+
+@dataclass(frozen=True)
+class LayerAttentionBinding(Generic[Params, Context, Inputs, Output]):
+    """一个逻辑层在 prefill 或 decode 阶段的 attention 执行入口。
+
+    权重和存储准备完成后构造；backend 是同组层共享的实现，params 借用本层的
+    权重和存储视图，存储重分配后须重新构造。group_id 用于取得当前批次的 context；
+    logical_layer_id 是模型层号，不一定等于物理存储下标。
+    """
+
+    group_id: int
+    logical_layer_id: int
+    spec: LayerSpec
+    params: Params
+    backend: "AttentionBackend[Context, Inputs, Output]"
+    alloc_tensor: Callable
+
+    def context(self, execution: AttentionExecution) -> Context:
+        return execution.attention_context(self)
+
+    def forward(self, context: Context, inputs: Inputs, *, out: torch.Tensor | None = None) -> Output:
+        return self.backend.forward(self, context, inputs, out=out)
+
+
+class AttentionBackend(ABC, Generic[Context, Inputs, Output]):
+    """同组层共用的执行实现；每批次变化的 metadata 和 wrapper 放在 context 中。
+
+    Context 的具体类型由 backend 定义，实例由 InferStateInfo 保存，不能跨 backend
+    或并发 microbatch 共用。调用顺序为 allocate_context -> prepare -> update，
+    随后同组各层用各自的 LayerAttentionBinding / Inputs 调用 forward。
+    """
+
+    @abstractmethod
+    def allocate_context(self, capacity: ContextCapacity, *, capture: bool, workspace) -> Context:
+        """创建归属此 backend 的 context，workspace 由模型按 group / microbatch slot 提供。
+
+        capture=True 的 context 随 decode CUDA Graph 保存，在后续 replay 中复用。
+        """
+        pass
+
+    @abstractmethod
+    def prepare(self, context: Context, batch: AttentionBatch) -> None:
+        """在图外绑定批次输入并完成 provider planning，这里允许 CPU 侧准备工作。
+
+        decode replay 时仍会调用；图中引用的 buffer 必须保持地址，只更新内容。
+        """
+        pass
+
+    @abstractmethod
+    def update(self, context: Context) -> None:
+        """在各层 forward 前生成 GPU 侧派生 metadata。
+
+        decode Graph 会捕获这段 GPU 工作；不能依赖 replay 时重新执行 Python，
+        也不能把当前批次的 GPU 值读回 CPU 来决定执行路径。
+        """
+        pass
+
+    @abstractmethod
+    def forward(
+        self, layer: LayerAttentionBinding, context: Context, inputs: Inputs, *, out: torch.Tensor | None = None
+    ) -> Output:
+        """使用已准备的 context 执行一层，允许写入 params 借用的 KV / Conv / SSM 存储。
+
+        out 是调用方拥有的目标缓冲区，dtype、device 必须与输出一致。
+        provider 直接写入时可返回其中有效部分的 view（例如去掉 head padding）；
+        返回独立 tensor 时元素数必须一致，由基类按 out 的形状复制。
+        provider 可直接写入，或返回临时结果由基类复制；不承诺零分配。返回 tuple 时只对应首项，
+        其余项保留模型所需的附加输出，例如 GDN 的 gate 输入。
+        """
+        pass
+
+    def resource_requirements(self, capacity: ContextCapacity, *, capture: bool):
+        """返回分配前可计算的资源上界，供共享容量规划和捕获前预留使用。"""
+        return BackendResources()
+
+    def bind_graph(self, context: Context):
+        """为此捕获实例创建可选的 GraphBinding；普通原生 Graph 无需额外绑定。"""
+        return None
diff --git a/lightllm/common/attention/api/batch.py b/lightllm/common/attention/api/batch.py
new file mode 100644
index 0000000000..41a24d464a
--- /dev/null
+++ b/lightllm/common/attention/api/batch.py
@@ -0,0 +1,113 @@
+from dataclasses import dataclass, fields
+from typing import Optional
+
+import torch
+
+from .spec import AttentionPhase
+
+
+@dataclass(frozen=True)
+class AttentionBatch:
+    """一次执行的批次描述，tensor 借用自 InferStateInfo / ReqManager。
+
+    b_* 的首维为 batch_size，b1_cu_* 是带起始 0 的 batch_size + 1 个累积偏移。
+    token_num 是当前 query token 数；total_token_num 包含历史 KV 的逻辑 token 数，
+    捕获 decode Graph 时后者可替换为容量上界。frozen 只固定字段引用，tensor 内容仍可更新。
+    """
+
+    phase: AttentionPhase
+    batch_size: int
+    token_num: int
+    total_token_num: int
+    max_q_seq_len: int
+    max_kv_seq_len: int
+    max_cache_len: int
+    b_req_idx: torch.Tensor
+    # KV 长度包含当前输入；query 长度只包含本次实际计算的 token。
+    b_seq_len: torch.Tensor
+    b_q_seq_len: torch.Tensor
+    b1_cu_q_seq_len: torch.Tensor
+    b1_cu_kv_seq_len: torch.Tensor
+    req_to_token_indexs: torch.Tensor
+    hold_req_id: int
+    b_ready_cache_len: Optional[torch.Tensor] = None
+    b_mtp_index: Optional[torch.Tensor] = None
+    b_shared_seq_len: Optional[torch.Tensor] = None
+    b_shared_radix_node_id: Optional[torch.Tensor] = None
+    mem_index: Optional[torch.Tensor] = None
+    # Absolute positions of the query rows in the request KV sequence.  Sparse
+    # MLA indexers use this to expand per-request metadata without reaching
+    # back into InferStateInfo.
+    position_ids: Optional[torch.Tensor] = None
+    b_image_token_end: Optional[torch.Tensor] = None
+
+    @property
+    def device(self):
+        return self.b_req_idx.device
+
+    @property
+    def is_prefill(self):
+        return self.phase is AttentionPhase.PREFILL
+
+    @property
+    def b_q_start_loc(self):
+        return self.b1_cu_q_seq_len[:-1]
+
+    @property
+    def b_kv_start_loc(self):
+        return self.b1_cu_kv_seq_len[:-1]
+
+    def validate(self):
+        if self.batch_size < 0 or self.token_num < 0:
+            raise ValueError("attention batch sizes must be nonnegative")
+        for name in ("b_req_idx", "b_seq_len", "b_q_seq_len"):
+            tensor = getattr(self, name)
+            if tensor.ndim != 1 or tensor.numel() != self.batch_size:
+                raise ValueError(f"{name} must have batch_size elements")
+        if self.b1_cu_q_seq_len.numel() != self.batch_size + 1:
+            raise ValueError("query offsets must have batch_size + 1 elements")
+
+    def copy_inputs(self, source: "AttentionBatch") -> None:
+        """向捕获时的 tensor 原址拷入新批次，标量字段保留捕获配置。
+
+        先校验所有 tensor 再写入，避免校验失败留下半更新的输入。
+        request table 是共享存储，必须保持原地址，不能通过整表拷贝替换。
+        """
+        if (self.phase, self.batch_size, self.token_num) != (source.phase, source.batch_size, source.token_num):
+            raise ValueError("attention batch does not match the captured shape")
+        copies = []
+        for field in fields(self):
+            dst, src = getattr(self, field.name), getattr(source, field.name)
+            if dst is src:
+                continue
+            if isinstance(dst, torch.Tensor) or isinstance(src, torch.Tensor):
+                if not isinstance(dst, torch.Tensor) or not isinstance(src, torch.Tensor):
+                    raise ValueError(f"captured input {field.name} changed presence")
+                if (dst.shape, dst.dtype, dst.device) != (src.shape, src.dtype, src.device):
+                    raise ValueError(f"captured input {field.name} changed shape, dtype or device")
+                if field.name == "req_to_token_indexs" and dst.data_ptr() != src.data_ptr():
+                    raise ValueError("request table storage changed; recapture attention")
+                if dst.data_ptr() != src.data_ptr():
+                    copies.append((dst, src))
+        for dst, src in copies:
+            dst.copy_(src, non_blocking=True)
+
+
+@dataclass(frozen=True)
+class ContextCapacity:
+    """context 的形状契约：batch_size / token_num 要与输入批次逐项一致，序列长度只检查上界。
+
+    runner 负责选择 Graph bucket 和 padding；backend 不自行改变批次形状。
+    """
+
+    phase: AttentionPhase
+    batch_size: int
+    token_num: int
+    max_kv_seq_len: int
+    max_q_seq_len: int
+
+    def check(self, batch: AttentionBatch):
+        if self.phase is not batch.phase or self.batch_size != batch.batch_size or self.token_num != batch.token_num:
+            raise ValueError("attention context capacity does not match the batch shape")
+        if batch.max_kv_seq_len > self.max_kv_seq_len or batch.max_q_seq_len > self.max_q_seq_len:
+            raise ValueError("attention batch exceeds context sequence capacity")
diff --git a/lightllm/common/attention/api/execution.py b/lightllm/common/attention/api/execution.py
new file mode 100644
index 0000000000..d1f89f3774
--- /dev/null
+++ b/lightllm/common/attention/api/execution.py
@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+from typing import NamedTuple
+
+import torch
+
+from .spec import AttentionPhase
+
+
+@dataclass(frozen=True)
+class ExecutionRequirements:
+    """启动时确定的执行约束，与 LayerSpec 的计算语义分开保存。"""
+
+    graph_phases: frozenset[AttentionPhase] = frozenset()
+    microbatch_count: int = 1
+    max_batch_size: int = 1
+    max_seq_len: int = 1
+    max_verify_tokens: int = 1
+    tp_world_size: int = 1
+    expert_parallel: bool = False
+    transfer_paths: frozenset[str] = frozenset()
+
+    @classmethod
+    def from_model(cls, model):
+        args = model.args
+        phases = set()
+        if not args.disable_cudagraph:
+            phases.add(AttentionPhase.DECODE)
+        if args.enable_prefill_cudagraph:
+            phases.add(AttentionPhase.PREFILL)
+        transfers = set()
+        if args.enable_cpu_cache:
+            transfers.add("cpu")
+        if args.enable_dp_prompt_cache_fetch:
+            transfers.add("dp")
+        if args.run_mode in ("prefill", "decode"):
+            transfers.add("pd")
+        return cls(
+            frozenset(phases),
+            2 if args.enable_decode_microbatch_overlap or args.enable_prefill_microbatch_overlap else 1,
+            getattr(model, "max_req_num", 1),
+            max(getattr(model, "max_seq_length", 1), getattr(model, "graph_max_len_in_batch", 1)),
+            model.mtp_manager.get_decode_draft_step(model.is_mtp_draft_model) + 1,
+            model.tp_world_size_,
+            args.enable_ep_moe,
+            frozenset(transfers),
+        )
+
+
+class WorkspaceRequirement(NamedTuple):
+    name: str
+    elements: int
+    dtype: torch.dtype
+
+
+@dataclass(frozen=True)
+class BackendResources:
+    """捕获前的资源上界；workspace 项为 (名称, 元素数, dtype)。
+
+    同一执行槽位按名称取最大容量，并发槽位分别分配。主机长度快照只在需要
+    它的实现中创建；持久存储的初始化由 State 执行，不能在 forward 中反复清零。
+    """
+
+    workspace: tuple[WorkspaceRequirement, ...] = ()
+    needs_host_lengths: bool = False
+    zero_initialize_cache: bool = False
diff --git a/lightllm/common/attention/api/graph.py b/lightllm/common/attention/api/graph.py
new file mode 100644
index 0000000000..16b257f02f
--- /dev/null
+++ b/lightllm/common/attention/api/graph.py
@@ -0,0 +1,24 @@
+from typing import Protocol
+
+
+class GraphBinding(Protocol):
+    """一个捕获实例的 provider 资源，不能挂在跨批次共享的 Backend 上。
+
+    after_submit 在原生 Graph 提交后立即调用，不能先等待 Graph 完成；某些设备
+    的 Graph 正在等待此方法提交的任务更新。返回的事件须覆盖所有辅助流工作。
+    wait_for_update 只等待上次对主机 metadata 的使用，保证 prepare 可以安全改写。
+    close 在相关设备工作完成后、原生 Graph 销毁前调用，并须可重复执行。
+    """
+
+    # Native Graph-only providers inherit these no-ops; auxiliary work overrides them.
+    def wait_for_update(self) -> None:
+        pass
+
+    def before_replay(self) -> None:
+        pass
+
+    def after_submit(self) -> tuple:
+        return ()
+
+    def close(self) -> None:
+        pass
diff --git a/lightllm/common/attention/api/inputs.py b/lightllm/common/attention/api/inputs.py
new file mode 100644
index 0000000000..ae5a3e1fd5
--- /dev/null
+++ b/lightllm/common/attention/api/inputs.py
@@ -0,0 +1,144 @@
+"""区分初始化时绑定的层资源与每次 forward 的 tensor 输入。
+
+*Params 随 LayerAttentionBinding 保存，借用权重以及 State 管理的 KV / Conv / SSM 视图；
+*Inputs 由模型层为当前调用构造，批次级请求索引和 provider 派生 metadata 则由 context 提供。
+这些数据结构只固定字段绑定，不禁止 backend 原地更新所借用的状态 tensor。
+"""
+
+from dataclasses import dataclass
+from typing import NamedTuple, Optional, Protocol
+
+import torch
+
+
+@dataclass(frozen=True)
+class DenseParams:
+    key: torch.Tensor
+    value: torch.Tensor
+    alibi: Optional[torch.Tensor] = None
+    sliding_window: Optional[tuple[int, int]] = None
+    sink: Optional[torch.Tensor] = None
+    query_scale: Optional[torch.Tensor] = None
+    key_scale: torch.Tensor | float | None = None
+    value_scale: torch.Tensor | float | None = None
+
+
+@dataclass(frozen=True)
+class MLAParams:
+    kv: torch.Tensor
+    softmax_scale: float
+    kv_lora_rank: int
+    qk_rope_head_dim: int
+
+
+class ReplaySSMState(Protocol):
+    """Borrowed replay storage; State retains allocation, checkpoint and release ownership."""
+
+    ssm_state: torch.Tensor
+    keys: torch.Tensor
+    deltas: torch.Tensor
+    gates: torch.Tensor
+    raw_keys: torch.Tensor
+    raw_values: torch.Tensor
+    betas: torch.Tensor
+    cursors: torch.Tensor
+    tile_sync_mark: torch.Tensor
+    num_key_heads: int
+    capacity: int
+    verify_width: int
+    compute_mode: str
+    gate_lower_bound: float
+    hold_req_id: int
+    kda: bool
+
+    def storage_layer(self, logical_layer_id: int) -> int:
+        ...
+
+    def finish_decode_token(self, logical_layer_id: int, reqs: torch.Tensor, cu_seqlens: torch.Tensor) -> None:
+        ...
+
+
+@dataclass(frozen=True)
+class GDNParams:
+    conv: torch.Tensor
+    ssm: torch.Tensor
+    conv_weight: torch.Tensor
+    conv_bias: Optional[torch.Tensor]
+    a_log: torch.Tensor
+    dt_bias: torch.Tensor
+    mtp_state_index: torch.Tensor | None = None
+    replay_state: ReplaySSMState | None = None
+
+
+@dataclass(frozen=True)
+class KDAParams:
+    conv: torch.Tensor
+    ssm: torch.Tensor
+    conv_weight: torch.Tensor
+    a_log: torch.Tensor
+    dt_bias: torch.Tensor
+    mtp_state_index: torch.Tensor | None = None
+    replay_state: ReplaySSMState | None = None
+
+
+@dataclass(frozen=True)
+class DenseInputs:
+    """Dense 路径在调用前已写入 KV cache，因此这里仅传 query。"""
+
+    query: torch.Tensor
+
+
+class VarlenAttentionInputs(NamedTuple):
+    """非因果 self-attention 的 Q/K/V，布局均为 [total_tokens, heads, head_dim]。
+
+    cu_seqlens 是同设备上的 int32 前缀和，始于 0、终于 total_tokens；相邻边界之间
+    独立做 attention，允许空序列。max_seqlen 是最长序列长度，由调用方提供以避免 GPU 同步。
+    每层只打包 tensor 引用，使用 NamedTuple 保持字段不可变并减少构造开销。
+    """
+
+    query: torch.Tensor
+    key: torch.Tensor
+    value: torch.Tensor
+    cu_seqlens: torch.Tensor
+    max_seqlen: int
+
+
+@dataclass(frozen=True)
+class MLAPrefillInputs:
+    query: torch.Tensor
+    key_nope: torch.Tensor
+    key_rope: torch.Tensor
+    value: torch.Tensor
+
+
+@dataclass(frozen=True)
+class MLADecodeInputs:
+    query_nope: torch.Tensor
+    query_rope: torch.Tensor
+
+
+@dataclass(frozen=True)
+class SparseMLAPrefillInputs:
+    query: torch.Tensor
+    topk_mem_indices: torch.Tensor
+    topk_indices: Optional[torch.Tensor] = None
+    prefill_cache_kv: Optional[torch.Tensor] = None
+
+
+@dataclass(frozen=True)
+class SparseMLADecodeInputs:
+    query_nope: torch.Tensor
+    query_rope: torch.Tensor
+    topk_mem_indices: torch.Tensor
+
+
+@dataclass(frozen=True)
+class GDNInputs:
+    mixed_qkvzba: torch.Tensor
+
+
+@dataclass(frozen=True)
+class KDAInputs:
+    mixed_qkv: torch.Tensor
+    raw_gate: torch.Tensor
+    raw_beta: torch.Tensor
diff --git a/lightllm/common/attention/api/spec.py b/lightllm/common/attention/api/spec.py
new file mode 100644
index 0000000000..c6b13c8bf1
--- /dev/null
+++ b/lightllm/common/attention/api/spec.py
@@ -0,0 +1,74 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+import torch
+
+
+class AttentionFamily(str, Enum):
+    """计算类型；Triton、FA3 等实现库由 selector 另行选择。"""
+
+    DENSE = "dense"
+    MLA = "mla"
+    SPARSE_MLA = "sparse_mla"
+    GDN = "gdn"
+    KDA = "kda"
+
+
+class AttentionPhase(str, Enum):
+    PREFILL = "prefill"
+    DECODE = "decode"
+
+
+@dataclass(frozen=True)
+class CacheLayout:
+    """存储布局的静态描述，不持有或分配存储 tensor。
+
+    prefill / decode 必须接受同一布局；dtype、量化粒度和 scale 布局也属于兼容性条件。
+    storage_schema 记录 State 显式声明的布局版本及配置，排除逻辑层号，使同布局的层可以归为一组。
+    """
+
+    format: str
+    page_size: int
+    dtype: torch.dtype
+    key_heads: int
+    key_dim: int
+    value_dim: int
+    state_dtype: Optional[torch.dtype] = None
+    storage_schema: tuple = ()
+    quant_group_size: int = 0
+    scale_mode: str = "none"
+
+    def __post_init__(self):
+        if self.page_size <= 0:
+            raise ValueError("cache page_size must be positive")
+
+
+@dataclass(frozen=True)
+class LayerSpec:
+    """一个层在一个 phase 下的静态计算需求，用于能力检查和分组。
+
+    初始化按完整的 (prefill_spec, decode_spec) 分组，只有相同需求的层才共享 backend / context。
+    层权重和存储视图放在 LayerAttentionBinding.params，批次长度与请求索引放在 AttentionBatch。
+    """
+
+    family: AttentionFamily
+    phase: AttentionPhase
+    dtype: torch.dtype
+    cache: CacheLayout
+    num_q_heads: int
+    head_size: int
+    value_head_size: int
+    sliding_window: Optional[tuple[int, int]] = None
+    use_alibi: bool = False
+    use_sink: bool = False
+    causal: bool = True
+    use_mtp: bool = False
+    dynamic_verify: bool = False
+    variant: str = ""
+
+    def __post_init__(self):
+        if min(self.num_q_heads, self.head_size, self.value_head_size) <= 0:
+            raise ValueError("attention head counts and dimensions must be positive")
+        if self.family in (AttentionFamily.GDN, AttentionFamily.KDA) and self.sliding_window is not None:
+            raise ValueError("recurrent attention cannot use a paged sliding window")
diff --git a/lightllm/common/attention/api/varlen.py b/lightllm/common/attention/api/varlen.py
new file mode 100644
index 0000000000..953e812896
--- /dev/null
+++ b/lightllm/common/attention/api/varlen.py
@@ -0,0 +1,32 @@
+from abc import ABC, abstractmethod
+
+import torch
+
+from .inputs import VarlenAttentionInputs
+
+
+class VarlenAttentionOp(ABC):
+    """无持久 KV 的双向 dense attention，供视觉和音频编码器使用。
+
+    Q/K/V 与变长边界由调用方直接传入，不需要文本推理的请求表或 cache context。
+    实现不保留批次 tensor；调用方持有 out，返回值必须引用同一块输出存储。
+    SDPA / xFormers 需要在 CPU 读取序列边界；只有声明 supports_graph 的实现可整段捕获。
+    """
+
+    supports_graph = False
+
+    @classmethod
+    @abstractmethod
+    def check_environment(cls, caps) -> None:
+        """检查运行环境；不支持时抛出 UnsupportedAttention。"""
+        pass
+
+    @classmethod
+    @abstractmethod
+    def check_support(cls, inputs: VarlenAttentionInputs, caps=None) -> None:
+        """初始化或探针校验输入；不支持时抛出 UnsupportedAttention。"""
+        pass
+
+    @abstractmethod
+    def forward(self, inputs: VarlenAttentionInputs, *, out: torch.Tensor) -> torch.Tensor:
+        pass
diff --git a/lightllm/common/attention/backends/__init__.py b/lightllm/common/attention/backends/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/_shared/__init__.py b/lightllm/common/attention/backends/_shared/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/_shared/linear/__init__.py b/lightllm/common/attention/backends/_shared/linear/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/_shared/prefill/__init__.py b/lightllm/common/attention/backends/_shared/prefill/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/_workspace.py b/lightllm/common/attention/backends/_workspace.py
new file mode 100644
index 0000000000..4600879f8e
--- /dev/null
+++ b/lightllm/common/attention/backends/_workspace.py
@@ -0,0 +1,89 @@
+from typing import Callable, TypeVar, cast
+
+import torch
+
+from lightllm.platform import get_device_runtime
+
+T = TypeVar("T")
+
+
+class AttentionWorkspace:
+    def __init__(self, device=None, *, runtime=None):
+        self.runtime = runtime if runtime is not None else get_device_runtime(device)
+        self.device = self.runtime.device
+        self._buffers = {}
+        self._objects = {}
+        self._completion = None
+        self._last_stream = None
+        self._frozen = False
+        self._pool = None
+
+    def fork(self):
+        """捕获实例借用当前池的地址；池后续扩容时，旧图仍持有原 tensor。
+
+        同一槽位的借用者共享完成事件，因此共享 scratch 不允许并发执行；不同
+        microbatch 槽位由模型创建不同的池。只有实际扩容才需要额外存储。
+        """
+        workspace = AttentionWorkspace(runtime=self.runtime)
+        workspace._pool = self._pool or self
+        workspace._objects = self._objects
+        return workspace
+
+    def get_object(self, name: str, factory: Callable[[], T]) -> T:
+        """Share provider resources across mutually exclusive contexts in this execution slot."""
+        if name not in self._objects:
+            if self._frozen:
+                raise RuntimeError(f"captured workspace object {name} must be created before capture")
+            self._objects[name] = factory()
+        return cast(T, self._objects[name])
+
+    def begin(self):
+        if self._pool is not None:
+            return self._pool.begin()
+        if self._completion is not None:
+            stream = self.runtime.current_stream()
+            if stream != self._last_stream:
+                stream.wait_event(self._completion)
+
+    def finish(self, dependencies=()):
+        if self._pool is not None:
+            return self._pool.finish(dependencies)
+        stream = self.runtime.current_stream()
+        if stream is not None:
+            for event in dependencies:
+                stream.wait_event(event)
+            if self._completion is None:
+                self._completion = self.runtime.create_event()
+            self._last_stream = stream
+            self._completion.record(self._last_stream)
+
+    def freeze(self):
+        """捕获后禁止替换任何地址；互斥 bucket 可继续借用已预留的大缓冲区。"""
+        self._frozen = True
+
+    def reserve(self, resources):
+        for name, size, dtype in resources.workspace:
+            self.get(name, size, dtype)
+
+    def synchronize(self):
+        if self._pool is not None:
+            return self._pool.synchronize()
+        if self._completion is not None:
+            self._completion.synchronize()
+
+    def get(self, name: str, size: int, dtype=torch.uint8):
+        if not name or size <= 0:
+            raise ValueError("workspace name and positive size are required")
+        key = name, dtype
+        tensor = self._buffers.get(key)
+        if tensor is None or tensor.numel() < size:
+            if self._frozen:
+                raise RuntimeError(f"captured workspace {name} cannot grow; reserve capacity before capture")
+            tensor = (
+                self._pool.get(name, size, dtype)
+                if self._pool is not None
+                else self.runtime.allocate(size, dtype=dtype)
+            )
+            self._buffers[key] = tensor
+
+        return tensor
diff --git a/lightllm/common/attention/backends/base.py b/lightllm/common/attention/backends/base.py
new file mode 100644
index 0000000000..c76afad8ed
--- /dev/null
+++ b/lightllm/common/attention/backends/base.py
@@ -0,0 +1,211 @@
+from dataclasses import dataclass, replace
+from types import MappingProxyType
+from typing import Generic, TypeVar
+
+import torch
+
+from lightllm.common.attention._environment import check_attention_environment
+from lightllm.common.attention.api.backend import AttentionBackend, LayerAttentionBinding, UnsupportedAttention
+from lightllm.common.attention.api.batch import AttentionBatch, ContextCapacity
+from lightllm.common.attention.api.spec import AttentionFamily, AttentionPhase, LayerSpec
+from lightllm.common.attention.backends._workspace import AttentionWorkspace
+from lightllm.platform import DeviceCaps, Platform
+
+
+@dataclass(slots=True)
+class AttentionExecutionContext:
+    capacity: ContextCapacity
+    capture: bool
+    workspace: AttentionWorkspace
+    batch: AttentionBatch | None = None
+    current_max_kv_seq_len: int = 0
+    backend_identity: AttentionBackend | None = None
+
+
+Context = TypeVar("Context", bound=AttentionExecutionContext)
+Inputs = TypeVar("Inputs")
+Output = TypeVar("Output")
+
+
+class BaseAttentionBackend(AttentionBackend[Context, Inputs, Output], Generic[Context, Inputs, Output]):
+    requires_cpu_cache_sync = False
+    family = AttentionFamily.DENSE
+    provider = "triton"
+    cache_formats = frozenset({"None"})
+    supports_attention_sink = False
+    supports_noncausal_decode = False
+    supported_platforms = frozenset({Platform.CUDA})
+    supported_phases = frozenset(AttentionPhase)
+    supported_graph_phases = frozenset(AttentionPhase)
+    activation_dtypes = frozenset({torch.float16, torch.bfloat16, torch.float32})
+    dependency = "triton"
+    min_device_capability = None
+    supports_alibi = False
+    supports_sliding_window = False
+    supports_speculative_window = True
+    supports_expert_parallel = True
+    max_head_size = None
+
+    def __init__(self, model):
+        self.runtime = model.device_runtime
+        self.execution = model.attention_execution
+        self.page_size = model.args.page_size
+        self.network_config = MappingProxyType(dict(model.config))
+        self.is_draft_model = model.is_mtp_draft_model
+        self.decode_draft_step = model.mtp_manager.get_decode_draft_step(model.is_mtp_draft_model)
+        self.dynamic_verify = not self.is_draft_model and self.decode_draft_step > 0 and model.args.mtp_dynamic_verify
+        self.causal = not (self.is_draft_model and model.args.mtp_mode in ("dspark", "dflash", "dflash2"))
+        self.hold_req_id = model.req_manager.HOLD_REQUEST_ID
+
+    @classmethod
+    def create(cls, model, spec: LayerSpec):
+        backend = cls(model)
+        backend.configure_spec(spec)
+        return backend
+
+    def configure_spec(self, spec: LayerSpec) -> None:
+        """Apply per-group dimensions before allocating contexts; default backends use layer params."""
+        pass
+
+    @classmethod
+    def check_environment(cls, caps: DeviceCaps) -> None:
+        check_attention_environment(
+            caps,
+            provider=cls.provider,
+            supported_platforms=cls.supported_platforms,
+            min_device_capability=cls.min_device_capability,
+            dependency=cls.dependency,
+        )
+
+    @classmethod
+    def check_support(cls, spec: LayerSpec, caps=None, execution=None) -> None:
+        if spec.family is not cls.family:
+            raise UnsupportedAttention(f"{cls.__name__} does not implement {spec.family.value}")
+        if spec.phase not in cls.supported_phases:
+            raise UnsupportedAttention(f"{cls.provider} does not implement {spec.phase.value}")
+        if spec.cache.format not in cls.cache_formats:
+            raise UnsupportedAttention(f"cache encoding {spec.cache.format} is unsupported")
+        if spec.dtype not in cls.activation_dtypes:
+            raise UnsupportedAttention(f"activation dtype {spec.dtype} is unsupported")
+        if spec.use_alibi and not cls.supports_alibi:
+            raise UnsupportedAttention("this backend does not support ALiBi")
+        if spec.use_alibi and spec.sliding_window is not None:
+            raise UnsupportedAttention("ALiBi with sliding-window attention is unsupported")
+        if spec.use_sink and not cls.supports_attention_sink:
+            raise UnsupportedAttention("this backend does not support attention sinks")
+        if spec.sliding_window is not None and not cls.supports_sliding_window:
+            raise UnsupportedAttention("this backend does not support sliding-window attention")
+        if cls.max_head_size is not None and spec.head_size > cls.max_head_size:
+            raise UnsupportedAttention(f"head size exceeds {cls.max_head_size}")
+        if spec.phase is AttentionPhase.DECODE and not spec.causal and not cls.supports_noncausal_decode:
+            raise UnsupportedAttention("this backend does not support non-causal block decode")
+        if spec.use_mtp and spec.sliding_window is not None and not cls.supports_speculative_window:
+            raise UnsupportedAttention("speculative decode does not implement sliding windows")
+        if execution is not None:
+            if execution.expert_parallel and not cls.supports_expert_parallel:
+                raise UnsupportedAttention("this backend is incompatible with expert-parallel MoE execution")
+            if spec.phase in execution.graph_phases and (
+                spec.phase not in cls.supported_graph_phases or (caps is not None and not caps.supports_graph)
+            ):
+                raise UnsupportedAttention(f"{spec.phase.value} Graph is unsupported")
+
+    def allocate_context(self, capacity: ContextCapacity, *, capture: bool, workspace) -> AttentionExecutionContext:
+        if capacity.phase is AttentionPhase.PREFILL:
+            context = self._allocate_prefill_context(capacity, capture, workspace)
+        else:
+            context = self._allocate_decode_context(capacity, capture, workspace)
+        context.backend_identity = self
+        return context
+
+    def _allocate_prefill_context(self, capacity, capture, workspace):
+        return AttentionExecutionContext(capacity, capture, workspace)
+
+    def _allocate_decode_context(self, capacity, capture, workspace):
+        return AttentionExecutionContext(capacity, capture, workspace)
+
+    def prepare(self, context: AttentionExecutionContext, batch: AttentionBatch) -> None:
+        if context.backend_identity is not self:
+            raise ValueError("context belongs to a different attention backend")
+        batch.validate()
+        context.capacity.check(batch)
+        context.current_max_kv_seq_len = batch.max_kv_seq_len
+        if context.capture and context.batch is not None:
+            context.batch.copy_inputs(batch)
+        elif context.capture:
+            # 图内分配和 launch 使用容量上界；图外 provider planning 仍用 current_max_kv_seq_len。
+            context.batch = replace(
+                batch,
+                max_kv_seq_len=context.capacity.max_kv_seq_len,
+                total_token_num=context.capacity.batch_size * context.capacity.max_kv_seq_len,
+            )
+        else:
+            context.batch = batch
+        if batch.phase is AttentionPhase.PREFILL:
+            self._prepare_prefill(context)
+        else:
+            self._prepare_decode(context)
+
+    def _prepare_prefill(self, context):
+        """Optional host planning; kernels without provider plans need no preparation."""
+        pass
+
+    def _prepare_decode(self, context):
+        """Optional host planning, including updates before Graph replay."""
+        pass
+
+    def update(self, context: AttentionExecutionContext) -> None:
+        if context.backend_identity is not self:
+            raise ValueError("context belongs to a different attention backend")
+        if context.batch is None:
+            raise RuntimeError("prepare attention inputs before update")
+        if context.capacity.phase is AttentionPhase.PREFILL:
+            self._update_prefill(context)
+        else:
+            self._update_decode(context)
+
+    def _update_prefill(self, context):
+        """Optional derived metadata; direct consumers of batch tensors need no update."""
+        pass
+
+    def _update_decode(self, context):
+        """Optional GPU metadata generation, captured alongside layer execution."""
+        pass
+
+    def forward(self, layer: LayerAttentionBinding, context: AttentionExecutionContext, inputs, *, out=None):
+        if context.batch is None:
+            raise RuntimeError("prepare attention inputs before forward")
+        if context.backend_identity is not self:
+            raise ValueError("context belongs to a different attention backend")
+        if layer.spec.phase is not context.capacity.phase:
+            raise ValueError("layer attention and context phases differ")
+        method = self._forward_prefill if context.capacity.phase is AttentionPhase.PREFILL else self._forward_decode
+        result = method(layer, context, inputs, out=out)
+        if out is None:
+            return result
+        attention = result[0] if isinstance(result, tuple) else result
+        if attention.dtype != out.dtype or attention.device != out.device:
+            raise ValueError("out must match the attention result dtype and device")
+        if attention.untyped_storage().data_ptr() == out.untyped_storage().data_ptr():
+            # Providers may return the valid heads of a padded output buffer.
+            begin = attention.storage_offset()
+            end = begin + sum((size - 1) * stride for size, stride in zip(attention.shape, attention.stride())) + 1
+            out_end = (
+                out.storage_offset() + sum((size - 1) * stride for size, stride in zip(out.shape, out.stride())) + 1
+            )
+            if attention.numel() and (begin < out.storage_offset() or end > out_end):
+                raise ValueError("attention result view extends beyond out")
+            return result
+        if attention.numel() != out.numel():
+            raise ValueError("out must match the attention result element count")
+        out.copy_(attention.reshape(out.shape))
+        return (out, *result[1:]) if isinstance(result, tuple) else out
+
+    @property
+    def infer_page_size(self) -> int:
+        return self.page_size
+
+    def uses_dynamic_spec_verify_layout(self):
+        return self.dynamic_verify
+
+    def uses_causal_attention(self):
+        return self.causal
diff --git a/lightllm/common/attention/backends/dense/__init__.py b/lightllm/common/attention/backends/dense/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/fa3/__init__.py b/lightllm/common/attention/backends/dense/fa3/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/flashinfer/__init__.py b/lightllm/common/attention/backends/dense/flashinfer/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/sdpa/__init__.py b/lightllm/common/attention/backends/dense/sdpa/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/__init__.py b/lightllm/common/attention/backends/dense/triton/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/flash/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/flash/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/mtp_diverse/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/gqa/mtp_diverse/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/int4kv/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/int4kv/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/int8kv/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/int8kv/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/int8kv/normal/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/int8kv/normal/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/flash/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/flash/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/stage3_decode_att/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/decode/mha/stage3_decode_att/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/triton/kernels/prefill/__init__.py b/lightllm/common/attention/backends/dense/triton/kernels/prefill/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/dense/xformers/__init__.py b/lightllm/common/attention/backends/dense/xformers/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/gdn/__init__.py b/lightllm/common/attention/backends/gdn/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/gdn/flashqla/__init__.py b/lightllm/common/attention/backends/gdn/flashqla/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/gdn/replay/__init__.py b/lightllm/common/attention/backends/gdn/replay/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/gdn/triton/__init__.py b/lightllm/common/attention/backends/gdn/triton/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/gdn/triton/kernels/__init__.py b/lightllm/common/attention/backends/gdn/triton/kernels/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/kda/__init__.py b/lightllm/common/attention/backends/kda/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/kda/replay/__init__.py b/lightllm/common/attention/backends/kda/replay/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/kda/triton/__init__.py b/lightllm/common/attention/backends/kda/triton/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/kda/triton/kernels/__init__.py b/lightllm/common/attention/backends/kda/triton/kernels/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/__init__.py b/lightllm/common/attention/backends/mla/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/fa3/__init__.py b/lightllm/common/attention/backends/mla/fa3/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/flashinfer/__init__.py b/lightllm/common/attention/backends/mla/flashinfer/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/triton/__init__.py b/lightllm/common/attention/backends/mla/triton/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/triton/kernels/__init__.py b/lightllm/common/attention/backends/mla/triton/kernels/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/triton/kernels/decode/__init__.py b/lightllm/common/attention/backends/mla/triton/kernels/decode/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/mla/triton/kernels/prefill/__init__.py b/lightllm/common/attention/backends/mla/triton/kernels/prefill/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/sparse_mla/__init__.py b/lightllm/common/attention/backends/sparse_mla/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/common/attention/backends/sparse_mla/flashmla/__init__.py b/lightllm/common/attention/backends/sparse_mla/flashmla/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lightllm/platform/__init__.py b/lightllm/platform/__init__.py
new file mode 100644
index 0000000000..7ccfa98bab
--- /dev/null
+++ b/lightllm/platform/__init__.py
@@ -0,0 +1,56 @@
+"""CUDA/CPU device operations used by the attention backend runtime."""
+
+import os
+from functools import lru_cache
+
+import torch
+
+from .api import DeviceCaps, DeviceRuntime, Platform
+
+
+def get_device_runtime(device=None, *, platform=None):
+    platform = Platform(platform) if platform is not None else None
+    if platform not in (None, Platform.CPU, Platform.CUDA):
+        raise ValueError(f"attention backend refactor supports CUDA and CPU only, got {platform.value}")
+    use_worker_default = device is None
+    default_type = "cpu" if platform is Platform.CPU else "cuda"
+    if isinstance(device, int):
+        device = f"{default_type}:{device}"
+    device = device if device is not None else default_type
+    if isinstance(device, str) and device.split(":")[0] not in ("cpu", "cuda", "meta"):
+        raise ValueError(f"attention backend refactor supports CUDA and CPU devices only, got {device}")
+    device = torch.device(device)
+    platform = (
+        Platform(platform)
+        if platform is not None
+        else {
+            "cpu": Platform.CPU,
+            "meta": Platform.CPU,
+            "cuda": Platform.CUDA,
+        }.get(device.type)
+    )
+    if platform is None:
+        raise ValueError(f"no device runtime for {device}")
+    if platform is Platform.CPU and device.type not in ("cpu", "meta"):
+        raise ValueError(f"CPU runtime cannot operate on {device}")
+    if platform is not Platform.CPU and device.type in ("cpu", "meta"):
+        raise ValueError(f"{platform.value} runtime cannot operate on {device}")
+    # 保留调用方指定的 CUDA device；没有显式 ordinal 时使用 worker 绑定的 ordinal。
+    if device.index is None and device.type not in ("cpu", "meta"):
+        index = os.environ.get("LIGHTLLM_CURRENT_DEVICE_ID") if use_worker_default else None
+        if index is None:
+            api = torch.cuda
+            index = api.current_device() if api.is_available() else 0
+        device = torch.device(device.type, int(index))
+    return _runtime(platform, device)
+
+
+@lru_cache(maxsize=None)
+def _runtime(platform, device):
+    if platform is Platform.CPU:
+        return DeviceRuntime(device)
+    if platform is Platform.CUDA:
+        from .cuda import CudaRuntime
+
+        return CudaRuntime(device)
+    raise ValueError(f"no device runtime for platform {platform.value}")
diff --git a/lightllm/platform/api.py b/lightllm/platform/api.py
new file mode 100644
index 0000000000..1b4d5b6a75
--- /dev/null
+++ b/lightllm/platform/api.py
@@ -0,0 +1,96 @@
+from contextlib import nullcontext
+from dataclasses import dataclass
+from functools import cached_property
+
+import torch
+
+from lightllm.utils.device_utils import Platform
+
+
+@dataclass(frozen=True)
+class DeviceCaps:
+    """实际设备的能力快照；torch 的 device.type 不足以区分 CUDA、ROCm 和兼容层。"""
+
+    platform: Platform
+    device: torch.device
+    available: bool = True
+    architecture: str = ""
+    capability: tuple[int, int] | None = None
+    supports_graph: bool = False
+
+
+class DeviceRuntime:
+    """一个 worker/device 的设备操作。实现库和模型语义不属于 runtime。
+
+    构造不切换当前设备；调用方先完成 worker 的设备绑定。caps 延迟读取，
+    注册表和 CPU 契约测试可以在不初始化设备驱动的情况下导入这些接口。
+    """
+
+    platform = Platform.CPU
+    graph_type = ""
+    distributed_backend = "gloo"
+
+    def __init__(self, device, api=None):
+        self.device = torch.device(device)
+        self.api = api
+
+    @cached_property
+    def caps(self):
+        if self.api is None:
+            return DeviceCaps(self.platform, self.device, available=self.device.type != "meta")
+        available = self.api.is_available()
+        capability = None
+        if available and hasattr(self.api, "get_device_capability"):
+            capability = tuple(self.api.get_device_capability(self.device))
+        return DeviceCaps(
+            self.platform,
+            self.device,
+            available,
+            self.api.get_device_name(self.device) if available else "",
+            capability,
+            available and hasattr(self.api, self.graph_type),
+        )
+
+    def allocate(self, shape, dtype=torch.uint8):
+        return torch.empty(shape, dtype=dtype, device=self.device)
+
+    def current_stream(self):
+        return self.api.current_stream(self.device) if self.api is not None else None
+
+    def create_stream(self):
+        return self.api.Stream(device=self.device)
+
+    def use_stream(self, stream):
+        return self.api.stream(stream) if self.api is not None else nullcontext()
+
+    def create_event(self, *, enable_timing=False):
+        return self.api.Event(enable_timing=enable_timing) if self.api is not None else None
+
+    def synchronize(self):
+        if self.api is not None:
+            self.api.synchronize(self.device)
+
+    def is_capturing(self):
+        return self.api is not None and self.api.is_current_stream_capturing()
+
+    def empty_cache(self):
+        if self.api is not None:
+            self.api.empty_cache()
+
+    def memory_info(self):
+        if self.api is None:
+            raise ValueError("host/meta storage requires an explicit capacity or a memory-budget runtime")
+        return self.api.mem_get_info(self.device)
+
+    def graph_pool(self):
+        return self.api.graph_pool_handle() if self.caps.supports_graph else None
+
+    def create_graph(self):
+        if not self.caps.supports_graph:
+            raise RuntimeError(f"{self.platform.value} device {self.device} does not support Graph capture")
+        return getattr(self.api, self.graph_type)()
+
+    def capture_graph(self, graph, *, pool=None, memory_saver=None):
+        if memory_saver is not None and getattr(memory_saver, "enabled", False):
+            raise ValueError(f"memory saver is not supported on {self.platform.value}")
+        return self.api.graph(graph, pool=pool)
diff --git a/lightllm/platform/cuda.py b/lightllm/platform/cuda.py
new file mode 100644
index 0000000000..0db5e4891b
--- /dev/null
+++ b/lightllm/platform/cuda.py
@@ -0,0 +1,17 @@
+import torch
+
+from .api import DeviceRuntime, Platform
+
+
+class CudaRuntime(DeviceRuntime):
+    platform = Platform.CUDA
+    graph_type = "CUDAGraph"
+    distributed_backend = "nccl"
+
+    def __init__(self, device):
+        super().__init__(device, torch.cuda)
+
+    def capture_graph(self, graph, *, pool=None, memory_saver=None):
+        if memory_saver is not None:
+            return memory_saver.cuda_graph(graph, pool=pool)
+        return self.api.graph(graph, pool=pool)
diff --git a/unit_tests/common/attention/helpers.py b/unit_tests/common/attention/helpers.py
new file mode 100644
index 0000000000..f0633f640e
--- /dev/null
+++ b/unit_tests/common/attention/helpers.py
@@ -0,0 +1,285 @@
+from dataclasses import replace
+from types import SimpleNamespace
+
+import torch
+
+from lightllm.common.attention.api import (
+    AttentionBatch,
+    AttentionFamily,
+    AttentionPhase,
+    CacheLayout,
+    ContextCapacity,
+    ExecutionRequirements,
+    LayerAttentionBinding,
+    LayerSpec,
+)
+from lightllm.common.attention.backends._workspace import AttentionWorkspace
+from lightllm.common.attention.backends.base import BaseAttentionBackend
+from lightllm.platform import get_device_runtime
+from lightllm.server.core.objs.start_args_type import StartArgs
+
+
+def make_batch(lengths=(5, 2), *, phase=AttentionPhase.DECODE, device="cpu", **kwargs):
+    total = sum(lengths)
+    maximum = max(lengths) if lengths else 0
+    lengths = torch.tensor(lengths, dtype=torch.int32, device=device)
+    count = lengths.numel()
+    q_lengths = torch.ones_like(lengths)
+    offsets = torch.arange(count + 1, dtype=torch.int32, device=device)
+    values = dict(
+        phase=phase,
+        batch_size=count,
+        token_num=count,
+        total_token_num=total,
+        max_q_seq_len=1,
+        max_kv_seq_len=maximum,
+        max_cache_len=0,
+        b_req_idx=torch.arange(count, dtype=torch.int32, device=device),
+        b_seq_len=lengths,
+        b_q_seq_len=q_lengths,
+        b1_cu_q_seq_len=offsets,
+        b1_cu_kv_seq_len=torch.cat((offsets[:1], lengths.cumsum(0).int())),
+        req_to_token_indexs=torch.zeros((max(count, 8), 256), dtype=torch.int32, device=device),
+        hold_req_id=7,
+        b_mtp_index=torch.zeros_like(lengths),
+        b_ready_cache_len=torch.zeros_like(lengths),
+    )
+    values.update(kwargs)
+    return AttentionBatch(**values)
+
+
+def make_spec(family=AttentionFamily.DENSE, phase=AttentionPhase.DECODE, **kwargs):
+    values = dict(
+        family=family,
+        phase=phase,
+        dtype=torch.bfloat16,
+        cache=CacheLayout(
+            family.value if family in (AttentionFamily.GDN, AttentionFamily.KDA) else "None", 1, torch.bfloat16, 2, 8, 8
+        ),
+        num_q_heads=4,
+        head_size=8,
+        value_head_size=8,
+    )
+    values.update(kwargs)
+    return LayerSpec(**values)
+
+
+def make_backend(cls=BaseAttentionBackend, **kwargs):
+    backend = cls.__new__(cls)
+    values = dict(
+        page_size=1,
+        causal=True,
+        dynamic_verify=False,
+        decode_draft_step=0,
+        is_draft_model=False,
+        hold_req_id=7,
+    )
+    values.update(kwargs)
+    if "infer_page_size" in values:
+        values["page_size"] = values.pop("infer_page_size")
+    vars(backend).update(values)
+    return backend
+
+
+def make_context(backend, batch=None, *, capture=False, max_length=None, prepare=False):
+    batch = make_batch() if batch is None else batch
+    capacity = ContextCapacity(
+        batch.phase, batch.batch_size, batch.token_num, max_length or batch.max_kv_seq_len, batch.max_q_seq_len
+    )
+    context = backend.allocate_context(capacity, capture=capture, workspace=AttentionWorkspace(batch.device))
+    if prepare:
+        backend.prepare(context, batch)
+    else:
+        context.batch = batch
+        context.current_max_kv_seq_len = batch.max_kv_seq_len
+    return context
+
+
+def make_layer_attention(backend, context, params, **spec_kwargs):
+    return LayerAttentionBinding(
+        group_id=0,
+        logical_layer_id=0,
+        spec=make_spec(backend.family, context.capacity.phase, **spec_kwargs),
+        params=params,
+        backend=backend,
+        alloc_tensor=lambda shape, dtype, device="cpu": torch.empty(shape, dtype=dtype, device=context.batch.device),
+    )
+
+
+def model_stub(**args):
+    args.setdefault("disable_cudagraph", not torch.cuda.is_available())
+    model = SimpleNamespace(
+        args=StartArgs(**args),
+        device_runtime=get_device_runtime("cuda" if torch.cuda.is_available() else "cpu"),
+        att_state_role="main",
+        att_state_manager=SimpleNamespace(
+            get_full_attention_states=lambda *args: (), get_bounded_attention_states=lambda *args: ()
+        ),
+        config={},
+        is_mtp_draft_model=False,
+        tp_world_size_=1,
+        req_manager=SimpleNamespace(HOLD_REQUEST_ID=7),
+        mtp_manager=SimpleNamespace(get_decode_draft_step=lambda _: 0),
+    )
+    model.attention_execution = ExecutionRequirements.from_model(model)
+    return model
+
+
+def batch_from_fixture(fixture, phase, token_count=None, device=None):
+    indices = getattr(fixture, "b_req_idx", None)
+    lengths = getattr(fixture, "b_seq_len", None)
+    device = device or (
+        indices.device if indices is not None else getattr(getattr(fixture, "mem_index", None), "device", "cpu")
+    )
+    count = indices.numel() if indices is not None else (token_count or getattr(fixture, "batch_size", None) or 1)
+    if lengths is None:
+        lengths = torch.ones(count, dtype=torch.int32, device=device)
+    values = {}
+    for name in AttentionBatch.__dataclass_fields__:
+        value = getattr(fixture, name, None)
+        if value is not None and name not in ("phase", "token_num", "batch_size"):
+            values[name] = value
+    manager = getattr(fixture, "req_manager", None)
+    if manager is not None:
+        for name in ("req_to_token_indexs",):
+            value = getattr(manager, name, None)
+            if value is not None:
+                values[name] = value
+        values["hold_req_id"] = getattr(manager, "HOLD_REQUEST_ID", 7)
+
+    max_len = getattr(fixture, "max_kv_seq_len", None) or 1
+    cache_key = (phase, count, str(device))
+    cache = getattr(fixture, "_attention_test_batches", None)
+    if cache is None:
+        cache = fixture._attention_test_batches = {}
+    if cache_key not in cache:
+        cache[cache_key] = make_batch((1,) * count, phase=phase, device=device)
+    defaults = cache[cache_key]
+    values.update(
+        phase=phase, batch_size=count, token_num=token_count or count, b_seq_len=lengths, max_kv_seq_len=max_len
+    )
+    return replace(defaults, **values)
+
+
+def backend_context(cls, phase, settings=None, **values):
+    phase = AttentionPhase(phase)
+    if isinstance(settings, BaseAttentionBackend):
+        backend = settings
+    else:
+        config = dict(vars(settings)) if settings is not None else {}
+        model = config.pop("model", None)
+        if model is not None:
+            manager = getattr(model, "mtp_manager", None)
+            if manager is not None:
+                config["decode_draft_step"] = manager.get_decode_draft_step(getattr(model, "is_mtp_draft_model", False))
+            req_manager = getattr(model, "req_manager", None)
+            if req_manager is not None:
+                config["hold_req_id"] = getattr(req_manager, "HOLD_REQUEST_ID", 7)
+        backend = make_backend(cls, **config)
+    defaults = dict(
+        network_config={"index_topk": 2048, "index_kpool": 4},
+        prefill_graph_enabled=False,
+        decode_only=False,
+        prefill_token_capacity=0,
+        mtp_step=0,
+        decode_draft_step=0,
+        dynamic_verify=False,
+        causal=True,
+    )
+    for name, value in defaults.items():
+        if not hasattr(backend, name):
+            setattr(backend, name, value)
+    context = make_context(backend, make_batch((1,), phase=phase))
+    if hasattr(backend, "_test_workspace"):
+        context.workspace = backend._test_workspace
+    for name, value in values.items():
+        setattr(context, name, value)
+    return context
+
+
+def update_context(context, fixture):
+    context.batch = batch_from_fixture(fixture, context.capacity.phase)
+    context.capacity = ContextCapacity(
+        context.batch.phase,
+        context.batch.batch_size,
+        context.batch.token_num,
+        context.batch.max_kv_seq_len,
+        context.batch.max_q_seq_len,
+    )
+    context.backend_identity.prepare(context, context.batch)
+    context.backend_identity.update(context)
+
+
+def run_sparse(context, fixture, query, kv, parameters, *, out=None, alloc_func=None):
+    from lightllm.common.attention.api.inputs import MLAParams, SparseMLADecodeInputs, SparseMLAPrefillInputs
+
+    q = query[0] if isinstance(query, tuple) else query
+    context.capture = bool(getattr(fixture, "is_cuda_graph_capture", context.capture))
+    if getattr(fixture, "prefill_cuda_graph_exe_list", None) is not None:
+        context.capture = True
+    context.batch = batch_from_fixture(fixture, context.capacity.phase, q.shape[0], q.device)
+    backend = context.backend_identity
+    if context.workspace.device != q.device:
+        context.workspace = AttentionWorkspace(q.device)
+        backend._test_workspace = context.workspace
+    if backend.__class__.__module__.startswith("lightllm.models.deepseek_v4"):
+        from lightllm.models.deepseek_v4.attention.sparse_mla import DeepseekV4Inputs, DeepseekV4Params
+
+        params = DeepseekV4Params(
+            kv[0],
+            kv[0],
+            kv[1],
+            parameters["compress_ratio"],
+            parameters["head_dim_v"],
+            parameters["softmax_scale"],
+            parameters["attn_sink"],
+        )
+        inputs = DeepseekV4Inputs(
+            q,
+            parameters["swa_indices"],
+            parameters["swa_lengths"],
+            parameters.get("extra_indices"),
+            parameters.get("extra_lengths"),
+        )
+    else:
+        params = MLAParams(
+            kv, parameters["softmax_scale"], parameters["kv_lora_rank"], parameters.get("qk_rope_head_dim", 0)
+        )
+        if context.capacity.phase is AttentionPhase.PREFILL:
+            inputs = SparseMLAPrefillInputs(
+                q,
+                parameters.get("topk_mem_indices"),
+                parameters.get("topk_indices"),
+                parameters.get("prefill_cache_kv"),
+            )
+        else:
+            inputs = SparseMLADecodeInputs(q, query[1], parameters["topk_mem_indices"])
+    attention = make_layer_attention(backend, context, params)
+    if alloc_func is not None:
+        attention = replace(attention, alloc_tensor=alloc_func)
+
+    if (
+        hasattr(context, "capture_flashmla_sched_meta")
+        and context.capture_flashmla_sched_meta is None
+        and torch.cuda.is_available()
+    ):
+        import flash_mla
+
+        context.capture_flashmla_sched_meta = flash_mla.get_mla_metadata()[0]
+    return attention.forward(context, inputs, out=out)
+
+
+def run_dense(context, fixture, query, key, value, *, out=None, alloc_func=None, params=None):
+    from lightllm.common.attention.api.inputs import DenseInputs, DenseParams
+
+    context.batch = batch_from_fixture(fixture, context.capacity.phase, query.shape[0], query.device)
+    if params is None:
+        ks = vs = None
+        if isinstance(key, tuple):
+            key, ks = key
+            value, vs = value
+        params = DenseParams(key, value, key_scale=ks, value_scale=vs)
+    attention = make_layer_attention(context.backend_identity, context, params)
+    if alloc_func is not None:
+        attention = replace(attention, alloc_tensor=alloc_func)
+    return attention.forward(context, DenseInputs(query), out=out)
diff --git a/unit_tests/common/attention/test_contracts.py b/unit_tests/common/attention/test_contracts.py
new file mode 100644
index 0000000000..e52a18cacc
--- /dev/null
+++ b/unit_tests/common/attention/test_contracts.py
@@ -0,0 +1,94 @@
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from lightllm.common.attention.api import AttentionFamily, AttentionPhase, UnsupportedAttention
+from lightllm.common.attention.backends.base import BaseAttentionBackend
+from unit_tests.common.attention.helpers import make_batch, make_context, make_layer_attention, make_spec, model_stub
+
+
+def test_graph_copy_validates_all_inputs_before_writing():
+    captured = make_batch()
+    incoming = make_batch((9, 7))
+    incoming = replace(incoming, req_to_token_indexs=captured.req_to_token_indexs, b_seq_len=incoming.b_seq_len.long())
+    before = captured.b_req_idx.clone()
+    with pytest.raises(ValueError, match="b_seq_len"):
+        captured.copy_inputs(incoming)
+    torch.testing.assert_close(captured.b_req_idx, before)
+
+
+def test_graph_copy_keeps_tensor_addresses_and_uses_new_lengths():
+    captured = make_batch()
+    incoming = make_batch((9, 7), req_to_token_indexs=captured.req_to_token_indexs)
+    pointer = captured.b_seq_len.data_ptr()
+    captured.copy_inputs(incoming)
+    assert captured.b_seq_len.data_ptr() == pointer
+    assert captured.b_seq_len.tolist() == [9, 7]
+    with pytest.raises(ValueError, match="request table"):
+        captured.copy_inputs(make_batch((9, 7)))
+
+
+@pytest.mark.parametrize("capture", [False, True])
+def test_context_capacity_and_backend_identity(capture):
+    backend = BaseAttentionBackend(model_stub())
+    context = make_context(backend, capture=capture, max_length=16)
+    backend.prepare(context, make_batch(req_to_token_indexs=context.batch.req_to_token_indexs))
+    with pytest.raises(ValueError, match="capacity"):
+        backend.prepare(context, make_batch((17, 2)))
+    with pytest.raises(ValueError, match="different attention backend"):
+        BaseAttentionBackend(model_stub()).prepare(context, make_batch())
+    with pytest.raises(ValueError, match="different attention backend"):
+        BaseAttentionBackend(model_stub()).update(context)
+    attention = make_layer_attention(BaseAttentionBackend(model_stub()), context, params=None)
+    with pytest.raises(ValueError, match="different attention backend"):
+        attention.forward(context, inputs=None)
+
+
+def test_layer_attention_rejects_wrong_phase():
+    backend = BaseAttentionBackend(model_stub())
+    context = make_context(backend, prepare=True)
+    attention = make_layer_attention(backend, context, params=None)
+    attention = replace(attention, spec=replace(attention.spec, phase=AttentionPhase.PREFILL))
+    with pytest.raises(ValueError, match="phases differ"):
+        attention.forward(context, inputs=None)
+
+
+class _OutputBackend(BaseAttentionBackend):
+    def _forward_decode(self, layer, context, inputs, *, out=None):
+        return inputs
+
+
+@pytest.mark.parametrize("strided", [False, True])
+def test_output_copy_and_padded_view_preserve_auxiliary_result(strided):
+    backend = _OutputBackend(model_stub())
+    context = make_context(backend, prepare=True)
+    layer = make_layer_attention(backend, context, params=None)
+    out = torch.empty(2, 8 if strided else 4, 8)[:, :4]
+    gate = torch.ones(2)
+    output, actual_gate = layer.forward(context, (torch.ones(2, 32), gate), out=out)
+    assert output is out and actual_gate is gate
+    assert torch.all(out == 1)
+    valid_heads = out[:, :2]
+    output, actual_gate = layer.forward(context, (valid_heads, gate), out=out)
+    assert output is valid_heads and actual_gate is gate
+
+
+@pytest.mark.parametrize("invalid", ["dtype", "device", "size", "view_bounds"])
+def test_output_rejects_incompatible_buffer(invalid):
+    backend = _OutputBackend(model_stub())
+    context = make_context(backend, prepare=True)
+    layer = make_layer_attention(backend, context, params=None)
+    out = torch.empty(2, 4)
+    result = torch.ones_like(out)
+    if invalid == "dtype":
+        result = result.double()
+    elif invalid == "device":
+        result = result.to("meta")
+    elif invalid == "size":
+        result = result[:1]
+    else:
+        result, out = out, out[:1]
+    with pytest.raises(ValueError, match="out"):
+        layer.forward(context, result, out=out)
diff --git a/unit_tests/common/attention/test_platform_runtime.py b/unit_tests/common/attention/test_platform_runtime.py
new file mode 100644
index 0000000000..a644796df0
--- /dev/null
+++ b/unit_tests/common/attention/test_platform_runtime.py
@@ -0,0 +1,50 @@
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from lightllm.common.attention.api import AttentionPhase, GraphBinding
+from lightllm.common.attention.backends._workspace import AttentionWorkspace
+from lightllm.common.attention_state import AttentionStateManager
+from lightllm.common.attention_state.full_att.gqa.gqa_state import GQAAttentionConfig, GQAFullAttentionState
+from lightllm.platform import DeviceRuntime, get_device_runtime
+
+
+def test_runtime_resolves_host_platform_without_querying_cuda(monkeypatch):
+    monkeypatch.setattr(torch.cuda, "is_available", lambda: pytest.fail("queried CUDA for a CPU runtime"))
+    assert get_device_runtime(platform="cpu").device == torch.device("cpu")
+    with pytest.raises(ValueError, match="cannot operate"):
+        get_device_runtime("cuda:0", platform="cpu")
+    with pytest.raises(ValueError, match="cannot operate"):
+        get_device_runtime("cpu", platform="cuda")
+
+
+def test_runtime_rejects_unimplemented_hardware():
+    with pytest.raises(ValueError, match="CUDA and CPU"):
+        get_device_runtime(platform="musa")
+    with pytest.raises(ValueError, match="CUDA and CPU"):
+        get_device_runtime("npu:0")
+
+
+def test_explicit_current_device_is_not_overridden_by_worker_default(monkeypatch):
+    monkeypatch.setenv("LIGHTLLM_CURRENT_DEVICE_ID", "0")
+    monkeypatch.setattr(torch.cuda, "is_available", lambda: True)
+    monkeypatch.setattr(torch.cuda, "current_device", lambda: 1)
+    assert get_device_runtime("cuda").device == torch.device("cuda:1")
+    assert get_device_runtime("cuda:2").device == torch.device("cuda:2")
+    assert get_device_runtime().device == torch.device("cuda:0")
+
+
+def test_captured_workspace_keeps_address_when_pool_grows():
+    pool = AttentionWorkspace("cpu")
+    first, second = pool.fork(), pool.fork()
+    old = first.get("scratch", 8, torch.int32)
+    assert second.get("scratch", 8, torch.int32).data_ptr() == old.data_ptr()
+    first.freeze()
+    grown = pool.get("scratch", 64, torch.int32)
+    old.fill_(17)
+    grown.fill_(29)
+    assert first.get("scratch", 8, torch.int32).tolist() == [17] * 8
+    with pytest.raises(RuntimeError, match="cannot grow"):
+        first.get("scratch", 64, torch.int32)
+    assert AttentionWorkspace("cpu").get("scratch", 8, torch.int32).data_ptr() != old.data_ptr()
diff --git a/unit_tests/common/attention/test_workspace.py b/unit_tests/common/attention/test_workspace.py
new file mode 100644
index 0000000000..af9e192ada
--- /dev/null
+++ b/unit_tests/common/attention/test_workspace.py
@@ -0,0 +1,28 @@
+import pytest
+import torch
+
+from lightllm.common.attention.backends._workspace import AttentionWorkspace
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+def test_workspace_cross_stream_reuse_waits_for_writer():
+    workspace = AttentionWorkspace("cuda")
+    buffer = workspace.get("shared", 4096, torch.int32)
+    buffer.zero_()
+    producer, consumer = torch.cuda.Stream(), torch.cuda.Stream()
+    producer.wait_stream(torch.cuda.current_stream())
+    consumer.wait_stream(torch.cuda.current_stream())
+    with torch.cuda.stream(producer):
+        torch.cuda._sleep(2_000_000)
+        buffer.fill_(17)
+        workspace.finish()
+    with torch.cuda.stream(consumer):
+        workspace.begin()
+        result = buffer.clone()
+        workspace.finish()
+    consumer.synchronize()
+    assert result.eq(17).all()
+
+    larger = workspace.get("shared", 8192, torch.int32)
+    assert larger.data_ptr() != buffer.data_ptr()
+    assert buffer.eq(17).all()
diff --git a/unit_tests/conftest.py b/unit_tests/conftest.py
index 622c8087ad..6c27ade302 100644
--- a/unit_tests/conftest.py
+++ b/unit_tests/conftest.py
@@ -26,6 +26,7 @@ def single_rank_environment(monkeypatch: pytest.MonkeyPatch):
 def runtime_start_args(monkeypatch: pytest.MonkeyPatch):
     import json
     from dataclasses import asdict
+
     from lightllm.server.core.objs.start_args_type import StartArgs
     from lightllm.utils.envs_utils import get_env_start_args