JOJO
aa9bf377a3
- session.* 传输暴露：新增 gateway_api 蓝图（list/create/history），RuntimeService 补 create_session - host Bearer 通道：gateway_auth 双通道认证（host 模式+回环限定，token 存 DATA_DIR/host_api_token），tasks/approval 路由接入 - api_v1 会话路由转调公共入口消双轨，会话索引/列表补齐 run_mode/model_key/custom_prompt_name/personalization_name - flask 包依赖拆解：_flask_bridge 延迟桥接 + context 子包 PEP 562 懒加载，import server.tasks/runtime 不再拉起 flask - 审批链路：mark_expired 终态回写（超时/软停止/取消三路径）+ 终态 TTL 惰性清理（3600s） - 测试 patch 点随迁；全量 75 测试失败恰为 4 项存量，独立启动验收 4/4 全绿 Co-authored-by: Astrion powered by Kimi-K3 <astrion-agent@users.noreply.github.com>
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
"""显式身份模型与角色解析（契约 docs/runtime_contract.md §4.1）。"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from typing import Optional
|
||
|
||
# 注意：本模块属于任务核心层依赖链（server.tasks → server.context），
|
||
# 禁止顶层 import Web 适配层（server.auth_helpers 依赖 flask）。
|
||
# 兼容模式下的角色解析在使用点函数内延迟导入（见 _resolve_user_role）。
|
||
|
||
|
||
class NoWorkspaceError(RuntimeError):
|
||
"""宿主机模式下尚未创建任何工作区。
|
||
|
||
与一般的 resource_busy 区分：前端可据 code=no_workspace 进入
|
||
「引导创建工作区」流程，而不是视为系统繁忙。
|
||
"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RuntimeIdentity:
|
||
"""资源装配的显式身份与偏好快照（契约 docs/runtime_contract.md §4.1）。
|
||
|
||
传入 get_user_resources 后，资源装配完全不读写 Flask session；
|
||
为 None 时保持既有行为（HTTP 适配层在请求上下文内读取 session）。
|
||
"""
|
||
|
||
host_mode: bool = False
|
||
host_workspace_id: Optional[str] = None
|
||
is_api_user: bool = False
|
||
role: Optional[str] = None
|
||
preferred_model_key: Optional[str] = None
|
||
preferred_run_mode: Optional[str] = None
|
||
preferred_thinking_mode: Optional[bool] = None
|
||
|
||
|
||
def _resolve_user_role(identity: Optional[RuntimeIdentity], record, default: str = "user") -> str:
|
||
"""统一角色解析：显式身份模式不触碰 Flask session（无请求上下文时安全）。"""
|
||
if identity is not None:
|
||
if identity.role:
|
||
return identity.role
|
||
return (record.role if record and getattr(record, "role", None) else default)
|
||
# 兼容模式（HTTP 请求上下文）：延迟导入 Web 认证辅助
|
||
from server.auth_helpers import get_current_user_role
|
||
|
||
return get_current_user_role(record)
|