core/app.py

"""Streamlit 主应用入口（``core/app.py``）— 需求 15 / 17.1 / 17.2。

底座主框架：**仅负责加载、导航与分发**，不承载任何科学计算（计算在 Skill 的
``compute`` 阶段，纯 Python）。运行方式（与设计一致）::

    streamlit run core/app.py --server.port=7860 --server.address=0.0.0.0

此时 ``core/`` 目录位于 ``sys.path[0]``，故 ``kernel`` / ``services`` 以**顶层包**
方式导入。``skills`` 为仓库根下的顶层包，由 ``SkillRegistry.discover("skills")``
自动发现（需求 1.1）。

职责装配（对应 design.md「Architecture / app.py」与需求 15）：

1. **构建 Services 容器**（:func:`build_services`）：实例化全部底座共享服务；
   ``LLMService`` 从环境变量读取密钥；任一可选重依赖缺失时优雅降级为 ``None``，
   不导致应用崩溃（需求 6.5）。
2. **发现 Skill 并合并 i18n**（:func:`discover_skills` / :func:`merge_skill_i18n`）：
   扫描 ``skills/`` 注册 Skill，并把每个 Skill 的 ``i18n`` 资源合并进 ``I18nService``
   （需求 17.6）。
3. **统一布局**（:func:`main` 及各 ``render_*``）：侧边栏参数区（导航 + 各 Skill 的
   ``render_inputs``）、中央主画布区（结果 / 报告）、可折叠审计追踪区（需求 15.1）。
4. **角色可见性**（:func:`visible_skills_for_role` / :func:`can_view_audit`）：按登录
   用户角色显示可见 Skill 与审计区（接 ``AuthService``，需求 15.5）。管理员初始密码
   取自环境变量 ``ADMIN_INIT_PASSWORD``（需求 5.1，明文不入代码 / 日志）。
5. **顶部语言切换**（:func:`set_session_language` / :func:`render_language_switcher`）：
   写入 ``Services.lang`` 并触发重渲染；界面文案一律经 ``I18nService.t`` 取用
   （需求 17.1 / 17.2 / 17.4）。
6. **分发**（:func:`dispatch`）：把选定 Skill + 收集到的 ``RawInput`` 交给固定
   ``Pipeline`` 执行（extract→compute→explain→report），渲染返回的 ``ReportSections``。

可测试性取向：所有**装配逻辑**（构建服务、发现 + 合并 i18n、角色过滤、语言状态、
认证、分发）都放在**纯函数 / 可导入的辅助函数**中，与 ``st.*`` 渲染调用分离，
便于在无 Streamlit 的环境下用 pytest 直接测试（见 ``tests/test_app.py``）。

安全说明：本应用自身**不暴露任何免认证的网络端点**——它是 Streamlit 渲染的 UI；
角色隔离在应用层完成：审计区仅 ``admin`` / ``auditor`` 可见，分析 Skill 的可见性
由 ``AuthService.visible_skills`` 决定。管理员密码来自环境变量、加盐存储，
日志中不出现任何明文密码或密钥。
"""

from __future__ import annotations

import logging
import os
import re
import sys
from typing import Any, Callable, Optional

# ---------------------------------------------------------------------------
# 导入路径自举（关键）：
# 运行 ``streamlit run core/app.py`` 时，Streamlit 仅把 ``core/`` 目录放入
# ``sys.path[0]``，使 ``kernel`` / ``services`` 能以顶层包导入；但顶层包
# ``skills`` / ``utils`` / ``layers`` 位于**仓库根目录**，默认不可导入，导致
# Skill 发现为空（导航「暂无数据」）与注册 / 登录报 ``No module named 'utils'``。
# 因此在模块导入期把「``core/`` 目录」与「仓库根目录」都补进 ``sys.path``，
# 使两类顶层包均可被发现（与 tests/conftest.py 的路径设置保持一致）。
_CORE_DIR = os.path.dirname(os.path.abspath(__file__))
_REPO_ROOT = os.path.dirname(_CORE_DIR)
for _p in (_CORE_DIR, _REPO_ROOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

# 底座以**顶层包** kernel / services 导入（core/ 位于 sys.path[0]）。
from kernel.pipeline import Pipeline
from kernel.registry import SkillRegistry
from kernel.services import Services
from kernel.skill_base import PharmaSkill, RawInput, ReportSections

# Streamlit 为前端运行期依赖；导入失败（如纯逻辑单测环境）时降级为 None，
# 使本模块的装配逻辑仍可被导入与测试。
try:  # pragma: no cover - 取决于运行环境是否安装 streamlit
    import streamlit as st
except Exception:  # noqa: BLE001 - 无 streamlit 时仅禁用渲染，不影响逻辑导入
    st = None  # type: ignore[assignment]

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# 常量
# ---------------------------------------------------------------------------

DEFAULT_LANG = "zh"

#: 默认会话角色（未登录）：研发人员——可使用分析 Skill，但不可查看审计区（需求 15.5）。
DEFAULT_ROLE = "researcher"

#: Streamlit 会话状态键。
SESSION_CTX = "pk_app_context"      # (svc, registry) 缓存，跨重渲染复用
SESSION_LANG = "pk_lang"            # 当前会话语言
SESSION_USER = "pk_user"            # 当前登录用户名
SESSION_ROLE = "pk_role"            # 当前会话角色
SESSION_SECTIONS = "pk_sections"    # 最近一次分析返回的 ReportSections
SESSION_SKILL = "pk_skill_id"       # 当前选中的 Skill id
SESSION_USER_MODEL = "pk_user_model"  # 用户自配模型 (UserModelConfig)
SESSION_TASK_SHEET = "pk_task_sheet"  # 最近一次理解层产出的任务单 (dict)
SESSION_SHEET_STATE = "pk_sheet_state"  # "none" | "pending_confirm" | "confirmed"


# ===========================================================================
# 一、服务构建（需求 7 / 6.5：缺库优雅降级）
# ===========================================================================

def build_services(
    env: Optional[dict] = None,
    *,
    skill_ids: Optional[list] = None,
    audit_db_path: str = ":memory:",
) -> Services:
    """构建并返回底座 :class:`Services` 容器。

    实例化全部底座共享服务；``LLMService`` 从环境变量读取密钥（无密钥则自然无可用
    提供商，``complete`` 返回结构化错误而非静默）。**任一服务构建失败都被捕获**并把
    对应字段降级为 ``None``（需求 6.5：可选重依赖缺失不得崩溃），失败信息收集到
    ``svc._build_warnings`` 供 UI 提示。

    参数：
    - ``env``：环境映射，默认 ``os.environ``。
    - ``skill_ids``：已注册 Skill 的 id 列表，用于 ``AuthService.visible_skills``
      的角色可见性判定；省略时回退到 AuthService 的默认集合。
    - ``audit_db_path``：审计库路径，默认 ``":memory:"``（按会话保存）。
    """
    env = os.environ if env is None else env
    warnings: list[tuple[str, str]] = []

    def _try(name: str, factory: Callable[[], Any]) -> Any:
        try:
            return factory()
        except Exception as exc:  # noqa: BLE001 - 单个服务失败即降级，不阻断整体
            logger.warning("构建服务 %s 失败，已降级为不可用：%s", name, exc)
            warnings.append((name, str(exc)))
            return None

    i18n = _try("i18n", _build_i18n)
    usage = _try("usage", _build_usage)
    llm = _try("llm", lambda: _build_llm(env, usage))
    audit = _try("audit", lambda: _build_audit(audit_db_path))
    report = _try("report", _build_report)
    chart = _try("chart", _build_chart)
    file = _try("file", _build_file)
    prompt_guard = _try("prompt_guard", _build_prompt_guard)
    auth = _try("auth", lambda: _build_auth(skill_ids))

    svc = Services(
        llm=llm,
        report=report,
        chart=chart,
        file=file,
        auth=auth,
        audit=audit,
        usage=usage,
        prompt_guard=prompt_guard,
        i18n=i18n,
        lang=(getattr(i18n, "lang", DEFAULT_LANG) if i18n is not None else DEFAULT_LANG),
    )
    # 把构建告警挂在实例上供 UI 友好提示（Services 为普通 dataclass，可附加属性）。
    setattr(svc, "_build_warnings", warnings)
    return svc


# 各服务的惰性构建器：在函数内部 import，使任一服务的重依赖问题被 build_services
# 的 try/except 隔离，而不在模块导入期就连带失败。

def _build_i18n():
    from services.i18n_service import I18nService

    return I18nService()


def _build_usage():
    from services.usage_service import UsageService

    return UsageService()


def _build_llm(env: dict, usage: Any):
    from services.llm_service import LLMService, SlidingWindowRateLimiter

    # 全局滑动窗口限流：默认 60 秒内最多 30 次 LLM 调用（需求 12.1）。
    limiter = SlidingWindowRateLimiter(max_calls=30, window=60.0)
    return LLMService.from_env(env=env, usage=usage, rate_limiter=limiter)


def _build_audit(db_path: str):
    from services.audit_service import AuditService

    return AuditService(db_path=db_path)


def _build_report():
    from services.report_service import ReportService

    return ReportService()


def _build_chart():
    from services.chart_service import ChartService

    return ChartService()


def _build_file():
    from services.file_service import FileService

    return FileService()


def _build_prompt_guard():
    from services.prompt_guard import PromptGuard

    return PromptGuard()


def _build_auth(skill_ids: Optional[list]):
    from services.auth_service import AuthService

    if skill_ids:
        return AuthService(all_skill_ids=skill_ids)
    return AuthService()


# ===========================================================================
# 二、技能发现与 i18n 合并（需求 1.1 / 17.6）
# ===========================================================================

def discover_skills(skills_source: str = "skills", svc: Optional[Services] = None) -> SkillRegistry:
    """扫描 ``skills_source`` 注册全部 Skill，返回 :class:`SkillRegistry`。

    单个 Skill 导入失败由注册器内部容错跳过（需求 1.3）；``svc`` 注入后失败会落审计。
    """
    registry = SkillRegistry(svc)
    registry.discover(skills_source)
    return registry


def merge_skill_i18n(svc: Services, registry: SkillRegistry) -> int:
    """把每个已注册 Skill 的 ``i18n`` 资源合并进底座 ``I18nService``（需求 17.6）。

    兼容两种 Skill i18n 约定，避免命名空间前缀重复：
    - 若 Skill 的 CATALOG 键**已**全部以 ``"<id>."`` 前缀（如 ``general_qa.title``），
      直接合并、不再加前缀。
    - 否则（如稳定性 / 相容性的 ``name`` / ``section.results``）以 ``namespace=<id>``
      合并，键自动加 ``"<id>."`` 前缀。

    返回成功合并资源的 Skill 数量。``i18n`` 不可用时安全返回 0。
    """
    i18n = getattr(svc, "i18n", None)
    merge = getattr(i18n, "merge", None)
    if not callable(merge):
        return 0

    merged = 0
    for skill in registry.all():
        catalog = getattr(skill, "i18n", None)
        if not catalog:
            continue
        skill_id = getattr(skill.meta, "id", "") or ""
        prefix = f"{skill_id}."
        try:
            keys = list(catalog.keys())
            already_prefixed = bool(keys) and all(str(k).startswith(prefix) for k in keys)
            if already_prefixed or not skill_id:
                merge(catalog)
            else:
                merge(catalog, namespace=skill_id)
            merged += 1
        except Exception as exc:  # noqa: BLE001 - 单个 Skill 文案合并失败不阻断其它
            logger.warning("合并 Skill %s 的 i18n 资源失败：%s", skill_id, exc)
    return merged


def bootstrap(
    env: Optional[dict] = None,
    skills_source: str = "skills",
    *,
    audit_db_path: str = ":memory:",
) -> tuple[Services, SkillRegistry]:
    """一次性装配：发现 Skill → 构建 Services（含角色可见性所需的 skill_ids）→ 合并 i18n。

    返回 ``(svc, registry)``。本函数为纯装配、无 ``st.*`` 依赖，便于单测。
    """
    env = os.environ if env is None else env
    registry = discover_skills(skills_source)
    skill_ids = [s.meta.id for s in registry.all()]
    svc = build_services(env, skill_ids=skill_ids, audit_db_path=audit_db_path)
    registry.svc = svc  # 后续失败可落审计
    merge_skill_i18n(svc, registry)
    return svc, registry


# ===========================================================================
# 三、角色可见性与导航（需求 15.5）
# ===========================================================================

def visible_skills_for_role(
    registry: SkillRegistry, auth: Any, role: Optional[str]
) -> list:
    """返回该角色在导航中可见的 Skill 列表（按导航顺序）。

    可见 id 集合由 ``AuthService.visible_skills(role)`` 决定（需求 15.5）：
    admin / researcher 可见全部分析 Skill；auditor 只读、不展示可执行 Skill。
    ``auth`` 不可用时回退为展示全部已注册 Skill（避免空界面）。
    """
    all_skills = registry.all()
    visible = getattr(auth, "visible_skills", None)
    if not callable(visible):
        return all_skills
    try:
        visible_ids = set(visible(role))
    except Exception as exc:  # noqa: BLE001 - 判定异常时回退全部
        logger.warning("计算角色可见 Skill 失败，回退展示全部：%s", exc)
        return all_skills
    return [s for s in all_skills if s.meta.id in visible_ids]


def can_view_audit(auth: Any, role: Optional[str]) -> bool:
    """该角色是否可查看审计追踪区（admin / auditor 可见，需求 15.5）。"""
    checker = getattr(auth, "can_view_audit", None)
    if not callable(checker):
        return False
    try:
        return bool(checker(role))
    except Exception:  # noqa: BLE001
        return False


def skill_nav_label(skill: PharmaSkill, i18n: Any = None) -> str:
    """构造 Skill 的导航标签（图标 + 本地化名称），文案经 i18n 取用（需求 17.4）。

    优先取底座导航键 ``nav.<id>``，其次取 Skill 自身的 ``<id>.name``，再退回
    ``meta.display_name``。``i18n`` 不可用时直接用 display_name。
    """
    meta = skill.meta
    icon = getattr(meta, "icon", "") or ""
    display = getattr(meta, "display_name", meta.id)
    has = getattr(i18n, "has", None)
    t = getattr(i18n, "t", None)
    if callable(has) and callable(t):
        for key in (f"nav.{meta.id}", f"{meta.id}.name"):
            try:
                if has(key):
                    display = t(key)
                    break
            except Exception:  # noqa: BLE001 - 文案取用失败回退 display_name
                pass
    return f"{icon} {display}".strip()


def skill_description(skill: PharmaSkill, i18n: Any = None, lang: Optional[str] = None) -> str:
    """返回 Skill 的本地化说明文字（随语言切换，需求 17.3 / 1 修复）。

    优先取底座 ``desc.<id>``，其次取 Skill 自身的 ``<id>.description``，
    再退回 ``meta.description``（硬编码中文兜底）。
    """
    meta = skill.meta
    fallback = getattr(meta, "description", "") or ""
    has = getattr(i18n, "has", None)
    t = getattr(i18n, "t", None)
    if callable(t):
        for key in (f"desc.{meta.id}", f"{meta.id}.description"):
            try:
                if not callable(has) or has(key):
                    return t(key, lang) if lang else t(key)
            except Exception:  # noqa: BLE001 - 取用失败回退
                pass
    return fallback


def skill_action_label(skill: PharmaSkill, svc: Services) -> str:
    """返回 Skill 主操作按钮文案：问答类用「开始提问」，其余用「开始分析」（需求 5 修复）。"""
    action_key = "btn.ask" if getattr(skill.meta, "id", "") == "general_qa" else "btn.analyze"
    return _t(svc, action_key)


# ===========================================================================
# 四、语言状态（需求 17.1 / 17.2）
# ===========================================================================

def set_session_language(svc: Services, requested: Optional[str]) -> str:
    """设置当前会话语言：写入 ``Services.lang`` 与 ``I18nService.lang``，返回归一语言码。

    未知 / 空输入回退默认语言（不报错，需求 17.5）。这是「写入 Services.lang 并触发
    重渲染」的纯逻辑部分；触发重渲染由 :func:`render_language_switcher` 在 UI 侧完成。
    """
    i18n = getattr(svc, "i18n", None)
    set_lang = getattr(i18n, "set_language", None)
    if callable(set_lang):
        lang = set_lang(requested)
    else:
        lang = (requested or DEFAULT_LANG)
    svc.lang = lang
    return lang


# ===========================================================================
# 五、认证（需求 5.1 / 5.4 / 15.5）
# ===========================================================================

def authenticate(
    username: str,
    password: str,
    *,
    user_lookup: Callable[[str], Optional[dict]],
    auth: Any = None,
) -> Optional[dict]:
    """校验用户名 / 密码，成功返回 ``{"user", "role"}``，失败返回 ``None``。

    - ``user_lookup(email) -> record``：返回含 ``password_hash``（或 ``password``）与
      ``role`` 的记录；返回 ``None`` 表示用户不存在。注入此参数使认证逻辑与具体
      存储后端解耦，便于无数据库的单测。
    - ``auth``：:class:`AuthService`，用于加盐哈希校验（``verify_password``，需求 5.4）
      与角色归一（``normalize_role``）。省略时按需构造。

    本函数不记录任何明文密码（需求 5.2）。
    """
    if not username or not password:
        return None
    if auth is None:
        from services.auth_service import AuthService

        auth = AuthService()

    try:
        record = user_lookup(username)
    except Exception as exc:  # noqa: BLE001 - 存储查询异常视为登录失败（不回显细节）
        logger.warning("用户查询失败，登录被拒绝：%s", exc)
        return None
    if not record:
        return None

    stored = record.get("password_hash") or record.get("password") or ""
    if not auth.verify_password(password, stored):
        return None

    role = auth.normalize_role(record.get("role"))
    return {"user": username, "role": role}


def ensure_admin_from_env(env: Optional[dict] = None) -> bool:
    """尽力按环境变量 ``ADMIN_INIT_PASSWORD`` 初始化管理员账户（需求 5.1）。

    复用 ``utils.database``：``get_db_connection`` 在建表时会按环境变量创建管理员
    （加盐哈希、明文不入库 / 日志）。数据库不可用时安全返回 ``False``，不抛出。
    """
    try:
        from utils import database as db
    except Exception as exc:  # noqa: BLE001 - 无数据库模块时跳过
        logger.info("用户数据库不可用，跳过管理员初始化：%s", exc)
        return False
    try:
        conn = db.get_db_connection()
        conn.close()
        return True
    except Exception as exc:  # noqa: BLE001 - 初始化失败不阻断应用
        logger.warning("管理员初始化失败（已忽略）：%s", exc)
        return False


def _db_user_lookup(username: str) -> Optional[dict]:
    """基于 ``utils.database`` 的用户查询（供 UI 登录使用）；不可用时返回 ``None``。"""
    try:
        from utils import database as db

        return db.get_user_by_email(username)
    except Exception as exc:  # noqa: BLE001
        logger.warning("数据库用户查询不可用：%s", exc)
        return None


# ===========================================================================
# 六、模型配置解析（需求 5：用户 > 管理员 > 环境变量）
# ===========================================================================

def get_admin_model_config() -> Optional[dict]:
    """读取管理员级模型配置（来自 ``utils.database``）；不可用时返回 ``None``。"""
    try:
        from utils import database as db

        return db.get_default_llm_config()
    except Exception as exc:  # noqa: BLE001 - 数据库不可用时忽略管理员档
        logger.info("读取管理员模型配置失败（已忽略）：%s", exc)
        return None


def resolve_session_llm(svc: Services, user_model: Any = None) -> str:
    """按 用户 > 管理员 > 环境变量 解析并重建 ``svc.llm``，返回首选档来源。

    - ``user_model``：本会话用户自配（:class:`UserModelConfig`）或 ``None``。
    - 解析出有序 Fallback 链后用 :func:`build_llm_service` 重建 ``svc.llm``，
      保留底座的用量计量与限流（沿用原 ``svc.llm`` 的 usage / rate_limiter）。
    返回来源标识：``"user"`` / ``"admin"`` / ``"env"`` / ``"none"``。
    """
    try:
        from services.model_config import build_llm_service, resolve_chain

        admin_cfg = get_admin_model_config()
        chain = resolve_chain(user_cfg=user_model, admin_cfg=admin_cfg, env=os.environ)
        # 复用既有 usage / rate_limiter，保持计量与限流连续。
        old = getattr(svc, "llm", None)
        usage = getattr(old, "_usage", None) or getattr(svc, "usage", None)
        rate_limiter = getattr(old, "_rate_limiter", None)
        svc.llm = build_llm_service(chain, usage=usage, rate_limiter=rate_limiter)
        return chain.source
    except Exception as exc:  # noqa: BLE001 - 解析失败不应阻断应用
        logger.warning("解析会话模型配置失败（保留原 LLM）：%s", exc)
        return "env"


# ===========================================================================
# 七、分发（需求 2.2：固定流水线）
# ===========================================================================

def dispatch(
    skill: PharmaSkill,
    raw: RawInput,
    svc: Services,
    *,
    pipeline: Optional[Pipeline] = None,
    user: str = "anonymous",
) -> ReportSections:
    """把选定 Skill + ``RawInput`` 交给固定 ``Pipeline`` 执行并返回 ``ReportSections``。

    底座流水线顺序固定（extract→compute→explain→report），``compute`` 不接收 ``svc``
    （需求 2.1 由契约保证）。本函数另对「分析执行」事件做尽力而为的审计记录
    （需求 7.4）。
    """
    pipeline = pipeline or Pipeline()
    sections = pipeline.run(skill, raw, svc)
    _audit_analysis(svc, user, getattr(skill.meta, "id", ""))
    return sections


def _audit_analysis(svc: Services, user: str, skill_id: str) -> None:
    audit = getattr(svc, "audit", None)
    recorder = getattr(audit, "record_analysis", None)
    if callable(recorder):
        try:
            recorder(user or "anonymous", skill_id, step="execute")
        except Exception:  # noqa: BLE001 - 审计失败不得影响主流程
            logger.debug("记录分析执行审计失败（已忽略）。", exc_info=True)


# ===========================================================================
# 七·b、理解层两阶段编排（intent-understanding-layer）
# ===========================================================================

def build_orchestrator(registry: SkillRegistry, svc: Services):
    """构造 ``AnalysisOrchestrator``（理解→确认→分析两阶段编排）。

    放在独立纯函数中便于单测；导入失败（理解层未就绪）时返回 ``None``，调用方据此
    回退到旧的直接分发路径，保证向后兼容。
    """
    try:
        from kernel.orchestrator import AnalysisOrchestrator

        return AnalysisOrchestrator(registry, svc)
    except Exception as exc:  # noqa: BLE001 - 理解层不可用时回退
        logger.info("理解层编排器不可用，回退直接分发：%s", exc)
        return None


def stale_session_keys_on_switch(prev_skill_id: Optional[str], new_skill_id: str) -> list[str]:
    """切换技能时应清理的会话键（纯函数，供 UI 与测试）。

    首次进入（``prev_skill_id`` 为空）或技能未变化时返回空列表；技能切换时返回需清理的
    残留键（上一技能的报告结果与任务单状态），避免旧报告串显到新技能。
    """
    if not prev_skill_id or prev_skill_id == new_skill_id:
        return []
    return [SESSION_SECTIONS, SESSION_TASK_SHEET, SESSION_SHEET_STATE]


def sheet_is_confirmable(sheet_dict: Optional[dict]) -> bool:
    """任务单是否可被确认执行（存在、且非澄清态）。

    纯函数，供 UI 与测试判定「确认并开始分析」按钮是否可用（需求 5.4 / 5.7）。
    """
    if not sheet_dict:
        return False
    clar = sheet_dict.get("clarification")
    if clar and clar.get("needed"):
        return False
    return True


def items_to_rows(items: list[dict]) -> list[dict]:
    """把任务单提取项转为可编辑表格行（纯函数，供 UI 与测试，需求 5.3）。"""
    rows: list[dict] = []
    for it in items or []:
        group = it.get("group") or {}
        rows.append({
            "use": True,
            "field": it.get("field", ""),
            "value": it.get("value", ""),
            "strength": group.get("strength", ""),
            "batch": group.get("batch", ""),
            "attribute": group.get("attribute", ""),
            "source_ref": it.get("source_ref", ""),
            "origin": it.get("source", "llm"),
        })
    return rows


def rows_to_items(rows: list[dict], original_items: Optional[list[dict]] = None) -> list[dict]:
    """把编辑后的表格行转回任务单提取项（纯函数，需求 5.5）。

    判定每行是否被用户改动：与对应的原始项逐字段比对；改动或新增的行标记
    ``source="user_edited"``（回填校验据此豁免）。未勾选 ``use`` 的行被丢弃（删除）。
    """
    original_items = original_items or []
    # 以 (field,value,attribute) 粗略对齐原始项，判断是否被编辑。
    def _orig_signature(it: dict) -> tuple:
        g = it.get("group") or {}
        return (str(it.get("field", "")), str(it.get("value", "")),
                str(g.get("strength", "")), str(g.get("batch", "")),
                str(g.get("attribute", "")), str(it.get("source_ref", "")))

    orig_sigs = {_orig_signature(it): it for it in original_items}

    items: list[dict] = []
    for row in rows or []:
        if not row.get("use", True):
            continue  # 删除：未勾选纳入
        field = str(row.get("field", "") or "").strip()
        value = str(row.get("value", "") or "").strip()
        if not field and not value:
            continue  # 空行跳过
        sig = (field, value, str(row.get("strength", "") or ""),
               str(row.get("batch", "") or ""), str(row.get("attribute", "") or ""),
               str(row.get("source_ref", "") or ""))
        if sig in orig_sigs:
            # 未改动：保留原项（含其 source / located）。
            items.append(orig_sigs[sig])
        else:
            # 改动 / 新增：标记 user_edited，豁免回填校验。
            items.append({
                "field": field,
                "value": value,
                "source_ref": str(row.get("source_ref", "") or ""),
                "located": True,
                "group": {
                    "strength": str(row.get("strength", "") or ""),
                    "batch": str(row.get("batch", "") or ""),
                    "attribute": str(row.get("attribute", "") or ""),
                },
                "source": "user_edited",
            })
    return items


def apply_sheet_edits(
    sheet_dict: dict,
    edited_rows: list[dict],
    proposed_skill_id: Optional[str] = None,
) -> dict:
    """把用户编辑（行 + 功能切换）应用回任务单 dict（纯函数，需求 5.5）。"""
    new_sheet = dict(sheet_dict or {})
    new_sheet["extracted_items"] = rows_to_items(edited_rows, sheet_dict.get("extracted_items"))
    if proposed_skill_id is not None:
        new_sheet["proposed_skill_id"] = proposed_skill_id
    return new_sheet


# —— 意图槽位确认（意图保真，阶段四）——

#: 槽位名 → 本地化展示标签（中文兜底）。
_SLOT_LABELS = {
    "primary_cqa": "主要质量指标",
    "target_timepoints": "目标外推时间点",
    "spec_limit": "规格限度",
    "target_attributes": "目标质量属性",
}
#: 来源态 → 展示文案。
_SLOT_STATE_LABELS = {
    "stated": "用户明示",
    "inferred": "系统推断",
    "missing": "缺失（待确认）",
}


def slots_to_rows(sheet_dict: dict) -> list[dict]:
    """把任务单意图槽位转为可编辑确认表格行（纯函数，供 UI 与测试）。

    每行暴露：槽位名 / 标签 / 当前值 / 来源态 / 依据 / 是否锁定。``missing`` 槽位
    的值为空字符串供用户填写。仅返回影响结论的槽位之外的全部槽位（含规格）。
    """
    rows: list[dict] = []
    for s in sheet_dict.get("intent_slots") or []:
        name = str(s.get("name", ""))
        val = s.get("value")
        if isinstance(val, list):
            val_txt = ", ".join(str(v) for v in val)
        else:
            val_txt = "" if val is None else str(val)
        rows.append({
            "name": name,
            "label": _SLOT_LABELS.get(name, name),
            "value": val_txt,
            "state": str(s.get("state", "missing")),
            "state_label": _SLOT_STATE_LABELS.get(str(s.get("state", "missing")), ""),
            "evidence": str(s.get("evidence", "")),
            "locked": bool(s.get("locked", False)),
            "affects_result": bool(s.get("affects_result", True)),
        })
    return rows


def _parse_slot_value(name: str, raw_value: str):
    """把编辑框文本解析回槽位值类型（时间点 → int 列表；其余 → 去空白字符串）。"""
    txt = str(raw_value or "").strip()
    if not txt:
        return None
    if name == "target_timepoints":
        nums: list[int] = []
        for tok in re.split(r"[，,、\s]+", txt):
            tok = tok.strip()
            if not tok:
                continue
            try:
                nums.append(int(float(tok)))
            except ValueError:
                continue
        return nums or None
    return txt


def apply_slot_edits(sheet_dict: dict, edited_slot_rows: list[dict]) -> dict:
    """把用户对意图槽位的确认/编辑写回任务单（纯函数，采纳真实意图的硬保证，需求 5.5）。

    规则：用户填入/修改了某槽位的值，或显式勾选锁定 → 该槽位变为 ``stated`` + ``locked``
    （置信 1.0），下游计算据此覆盖派生值，且后续重推断不得推翻。未改动且未锁定的槽位
    保持原状。这是「用户说了算」的工程落地。
    """
    new_sheet = dict(sheet_dict or {})
    by_name = {str(r.get("name", "")): r for r in (edited_slot_rows or [])}
    new_slots: list[dict] = []
    for s in sheet_dict.get("intent_slots") or []:
        slot = dict(s)
        name = str(slot.get("name", ""))
        row = by_name.get(name)
        if row is not None:
            new_val = _parse_slot_value(name, row.get("value"))
            user_locked = bool(row.get("locked", False))
            old_val = slot.get("value")
            changed = new_val is not None and new_val != old_val
            if changed or user_locked:
                if new_val is not None:
                    slot["value"] = new_val
                # 有有效值才视为已确认明示；仅勾锁却无值则保持缺失（避免锁定空值）。
                if slot.get("value") not in (None, "", []):
                    slot["state"] = "stated"
                    slot["locked"] = True
                    slot["confidence"] = 1.0
        new_slots.append(slot)
    new_sheet["intent_slots"] = new_slots
    return new_sheet


# ===========================================================================
# 八、Streamlit 渲染层（统一布局，需求 15.1）
# ===========================================================================
#
# 以下函数依赖 ``st``（Streamlit 模块）。装配逻辑均在上方纯函数中完成，这里只做
# 渲染编排：顶部语言切换 → 侧边栏（登录 + 导航 + 当前 Skill 输入面板）→ 中央主画布
# （结果 / 报告）→ 可折叠审计追踪区。无 streamlit 时这些函数不会被调用（``main``
# 在入口处守护），故纯逻辑单测不受影响。


def _t(svc: Services, key: str, **params) -> str:
    """经底座 I18nService 取用文案的便捷封装（缺失安全回退，需求 17.4 / 17.5）。"""
    i18n = getattr(svc, "i18n", None)
    t = getattr(i18n, "t", None)
    if callable(t):
        try:
            return t(key, **params) if params else t(key)
        except Exception:  # noqa: BLE001 - 取用失败回退 key 本身
            return key
    return key


def get_or_create_context() -> tuple[Services, SkillRegistry]:
    """在 Streamlit 会话中构建并缓存 ``(svc, registry)``，跨重渲染复用。"""
    if SESSION_CTX not in st.session_state:
        ensure_admin_from_env()
        svc, registry = bootstrap(
            os.environ, "skills", audit_db_path=_default_audit_db_path()
        )
        st.session_state[SESSION_CTX] = (svc, registry)
    svc, registry = st.session_state[SESSION_CTX]

    # 把会话语言同步进 Services（语言切换后重渲染时生效，需求 17.2）。
    lang = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG)
    set_session_language(svc, lang)
    return svc, registry


def _default_audit_db_path() -> str:
    """审计库落盘路径（``<repo>/data/audit.db``）；目录不可用则回退内存库。"""
    try:
        from pathlib import Path

        data_dir = Path(__file__).resolve().parents[1] / "data"
        data_dir.mkdir(parents=True, exist_ok=True)
        return str(data_dir / "audit.db")
    except Exception:  # noqa: BLE001 - 落盘不可用回退内存库
        return ":memory:"


def render_language_switcher(svc: Services) -> None:
    """顶部语言切换入口：写入会话语言与 ``Services.lang`` 并触发重渲染（需求 17.1/17.2）。"""
    i18n = getattr(svc, "i18n", None)
    langs = list(getattr(i18n, "available_languages", lambda: ("zh", "en"))())
    current = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG)

    cols = st.columns([6, 1])
    with cols[1]:
        choice = st.selectbox(
            _t(svc, "label.language"),
            options=langs,
            index=langs.index(current) if current in langs else 0,
            format_func=lambda code: _t(svc, f"lang.{code}"),
            key="pk_lang_select",
        )
    if choice != current:
        st.session_state[SESSION_LANG] = choice
        set_session_language(svc, choice)
        _rerun()


def render_login(svc: Services) -> None:
    """侧边栏账户区：登录 / 注册 / 退出；明文密码不记录（需求 5.1 / 5.2）。"""
    user = st.session_state.get(SESSION_USER)
    if user:
        role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE)
        st.sidebar.caption(f"👤 {user} · {_t(svc, f'role.{role}')}")
        if st.sidebar.button(_t(svc, "btn.logout"), key="pk_logout", use_container_width=True):
            for k in (SESSION_USER, SESSION_ROLE, SESSION_USER_MODEL, "pk_user_model_loaded"):
                st.session_state.pop(k, None)
            _rerun()
        return

    with st.sidebar.expander(_t(svc, "auth.account"), expanded=False):
        login_tab, register_tab = st.tabs(
            [_t(svc, "auth.login_tab"), _t(svc, "auth.register_tab")]
        )

        # —— 登录 ——
        with login_tab:
            username = st.text_input(_t(svc, "label.username"), key="pk_login_user")
            password = st.text_input(
                _t(svc, "label.password"), type="password", key="pk_login_pass"
            )
            if st.button(_t(svc, "btn.login"), key="pk_login_submit", use_container_width=True):
                result = authenticate(
                    username, password,
                    user_lookup=_db_user_lookup,
                    auth=getattr(svc, "auth", None),
                )
                if result:
                    st.session_state[SESSION_USER] = result["user"]
                    st.session_state[SESSION_ROLE] = result["role"]
                    _audit_login(svc, result["user"], True)
                    _rerun()
                else:
                    _audit_login(svc, username or "(unknown)", False)
                    st.error(_t(svc, "auth.login_fail"))

        # —— 注册 ——
        with register_tab:
            reg_email = st.text_input(_t(svc, "auth.email"), key="pk_reg_email")
            reg_pass = st.text_input(
                _t(svc, "label.password"), type="password", key="pk_reg_pass"
            )
            if st.button(_t(svc, "btn.register"), key="pk_reg_submit", use_container_width=True):
                ok, msg = _register_user(reg_email, reg_pass)
                if ok:
                    st.success(_t(svc, "auth.register_ok"))
                else:
                    st.error(f"{_t(svc, 'auth.register_fail')}: {msg}")


def _register_user(email: str, password: str):
    """注册新用户（复用 ``utils.auth``）；返回 ``(ok, message)``，不可用时友好失败。"""
    try:
        from utils.auth import register_user

        return register_user(email, password)
    except Exception as exc:  # noqa: BLE001 - 注册后端不可用时不崩溃
        logger.warning("注册功能不可用：%s", exc)
        return False, str(exc)


def _render_login_legacy_unused(svc: Services) -> None:
    """（已废弃，保留空壳避免历史引用报错）"""
    return None


def _audit_login(svc: Services, user: str, success: bool) -> None:
    audit = getattr(svc, "audit", None)
    recorder = getattr(audit, "record_login", None)
    if callable(recorder):
        try:
            recorder(user, success=success)
        except Exception:  # noqa: BLE001
            logger.debug("记录登录审计失败（已忽略）。", exc_info=True)


def _load_persisted_user_model(svc: Services) -> None:
    """登录用户首次进入会话时，把持久化的自配模型载入会话（若有且加密可用）。

    仅在会话尚未载入过、且尚无会话级用户配置时执行，避免覆盖用户本次的临时修改。
    """
    user = st.session_state.get(SESSION_USER)
    if not user:
        return
    if st.session_state.get("pk_user_model_loaded") == user:
        return
    st.session_state["pk_user_model_loaded"] = user
    if st.session_state.get(SESSION_USER_MODEL) is not None:
        return
    try:
        from services.model_config import UserModelConfig
        from utils import database as db

        rec = db.get_user_llm_config(user)
        if rec and rec.get("enabled"):
            cfg = UserModelConfig(
                provider=rec.get("provider", ""),
                model=rec.get("model", ""),
                api_key=rec.get("api_key", ""),
                base_url=rec.get("base_url", ""),
            )
            if cfg.is_usable():
                st.session_state[SESSION_USER_MODEL] = cfg
    except Exception as exc:  # noqa: BLE001 - 持久化不可用时静默退回仅会话
        logger.info("载入持久化用户模型配置不可用：%s", exc)


def _persist_user_model(cfg: Any) -> bool:
    """把登录用户的自配模型加密持久化；未登录 / 加密不可用时返回 False（仅会话）。"""
    user = st.session_state.get(SESSION_USER)
    if not user:
        return False
    try:
        from utils import database as db

        return db.set_user_llm_config(
            user, cfg.provider, cfg.api_key, model=cfg.model,
            base_url=cfg.base_url, enabled=True,
        )
    except Exception as exc:  # noqa: BLE001
        logger.info("持久化用户模型配置不可用：%s", exc)
        return False


def _clear_persisted_user_model() -> None:
    """删除登录用户持久化的自配模型（若已登录）。"""
    user = st.session_state.get(SESSION_USER)
    if not user:
        return
    try:
        from utils import database as db

        db.clear_user_llm_config(user)
    except Exception as exc:  # noqa: BLE001
        logger.info("清除持久化用户模型配置不可用：%s", exc)


def render_model_settings(svc: Services) -> None:
    """侧边栏「模型设置」面板：用户自配模型（最高优先级），并显示当前生效来源。

    用户可选择提供商 / 模型并填写自己的密钥；留空则使用管理员 / 环境配置（需求 5）。
    登录用户在配置了 ``PHARMAK_SECRET_KEY`` 时，密钥会**加密后持久化**并随 DB 同步到
    私有 Dataset，跨会话免重填；否则仅存会话状态（``SESSION_USER_MODEL``）。
    """
    try:
        from layers.llm_providers import AVAILABLE_MODELS, LLMProvider, get_available_providers
        from services.model_config import UserModelConfig
    except Exception as exc:  # noqa: BLE001 - 依赖不可用时跳过该面板
        logger.info("模型设置面板不可用：%s", exc)
        return

    # 登录用户：尝试载入持久化配置（仅首次）。
    _load_persisted_user_model(svc)

    providers = get_available_providers()
    with st.sidebar.expander(_t(svc, "model.settings"), expanded=False):
        st.caption(_t(svc, "model.use_admin_hint"))
        # 提示持久化能力：登录 + 加密可用时密钥跨会话保留，否则仅本次会话。
        try:
            from utils import crypto

            if st.session_state.get(SESSION_USER) and crypto.is_available():
                st.caption(_t(svc, "model.persist_on"))
            else:
                st.caption(_t(svc, "model.persist_off"))
        except Exception:  # noqa: BLE001
            st.caption(_t(svc, "model.persist_off"))

        ids = [p["id"] for p in providers]
        names = {p["id"]: p["name"] for p in providers}
        # 默认选中：会话已存用户配置则定位之，否则第一个。
        existing: Optional[UserModelConfig] = st.session_state.get(SESSION_USER_MODEL)
        default_idx = ids.index(existing.provider) if (existing and existing.provider in ids) else 0
        pid = st.selectbox(
            _t(svc, "model.provider"), options=ids,
            index=default_idx, format_func=lambda i: names.get(i, i),
            key="pk_user_provider",
        )

        # 模型：已知提供商给下拉 + "其他(手动)"；custom 直接手填。
        model_value = ""
        try:
            prov_enum = LLMProvider(pid)
        except ValueError:
            prov_enum = None
        known_models = list(AVAILABLE_MODELS.get(prov_enum, [])) if prov_enum else []
        if pid == "custom":
            model_value = st.text_input(_t(svc, "model.model_custom"), key="pk_user_model_custom") or ""
            base_url = st.text_input(
                _t(svc, "model.base_url"), help=_t(svc, "model.base_url_hint"),
                key="pk_user_base_url",
            ) or ""
        else:
            base_url = ""
            if known_models:
                choice = st.selectbox(
                    _t(svc, "model.model"),
                    options=known_models + ["—"],
                    key="pk_user_model_pick",
                )
                model_value = "" if choice == "—" else choice
            custom_m = st.text_input(_t(svc, "model.model_custom"), key="pk_user_model_free") or ""
            if custom_m.strip():
                model_value = custom_m.strip()

        api_key = st.text_input(_t(svc, "model.api_key"), type="password", key="pk_user_api_key") or ""

        cols = st.columns(2)
        with cols[0]:
            if st.button(_t(svc, "model.save"), key="pk_user_model_save"):
                cfg = UserModelConfig(
                    provider=pid, model=model_value, api_key=api_key, base_url=base_url
                )
                if pid == "custom" and not _custom_base_url_ok(svc, base_url):
                    pass  # 错误已在 _custom_base_url_ok 内提示
                elif cfg.is_usable():
                    st.session_state[SESSION_USER_MODEL] = cfg
                    persisted = _persist_user_model(cfg)
                    st.success(_t(svc, "model.saved_persisted" if persisted else "model.saved"))
                    _rerun()
                else:
                    st.warning(_t(svc, "model.test_fail"))
        with cols[1]:
            if st.button(_t(svc, "model.clear"), key="pk_user_model_clear"):
                st.session_state.pop(SESSION_USER_MODEL, None)
                _clear_persisted_user_model()
                st.info(_t(svc, "model.cleared"))
                _rerun()

        if st.button(_t(svc, "model.test"), key="pk_user_model_test"):
            cfg = UserModelConfig(provider=pid, model=model_value, api_key=api_key, base_url=base_url)
            if pid == "custom" and not _custom_base_url_ok(svc, base_url):
                pass  # 错误已提示
            elif _test_model_config(cfg):
                st.success(_t(svc, "model.test_ok"))
            else:
                st.error(_t(svc, "model.test_fail"))

        # 显示当前生效来源（用户/管理员/环境/无）。
        source = resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL))
        st.caption(_t(svc, f"model.source.{source}"))


def _custom_base_url_ok(svc: Any, base_url: str) -> bool:
    """校验用户填写的 custom base_url 是否安全（防 SSRF）；不安全时在 UI 提示并返回 False。"""
    try:
        from utils.url_guard import is_safe_public_url

        ok, reason = is_safe_public_url(base_url)
        if not ok:
            st.error(f"{_t(svc, 'model.base_url_unsafe')}：{reason}")
        return ok
    except Exception as exc:  # noqa: BLE001 - 校验器不可用时保守拒绝自定义端点
        logger.warning("base_url 安全校验不可用：%s", exc)
        st.error(_t(svc, "model.base_url_unsafe"))
        return False


def _test_model_config(cfg: Any) -> bool:
    """对单个用户模型配置做一次轻量真实调用测试（失败返回 False，不抛出）。"""
    try:
        from services.llm_service import LLMService
        from services.model_config import UserModelConfig  # noqa: F401

        spec = cfg.to_spec()
        if spec is None:
            return False
        svc = LLMService([spec])
        res = svc.complete("只回复一个词。", "OK", scope="model-test")
        return bool(res.ok)
    except Exception as exc:  # noqa: BLE001
        logger.info("模型连接测试失败：%s", exc)
        return False


def render_admin_model_settings(svc: Services) -> None:
    """管理员「全局模型配置」面板：仅管理员可见，设置后供未自配的用户使用（需求 5）。"""
    role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE)
    auth = getattr(svc, "auth", None)
    is_admin = getattr(auth, "is_admin", None)
    if not (callable(is_admin) and is_admin(role)):
        return
    try:
        from layers.llm_providers import get_available_providers
        from utils import database as db
    except Exception as exc:  # noqa: BLE001
        logger.info("管理员模型面板不可用：%s", exc)
        return

    providers = get_available_providers()
    ids = [p["id"] for p in providers]
    names = {p["id"]: p["name"] for p in providers}
    current = db.get_default_llm_config()

    with st.sidebar.expander(_t(svc, "admin.model.title"), expanded=False):
        # 持久化状态提示 + Space Secrets 引导（HF 免费档）。
        try:
            from utils import hf_storage

            if hf_storage.is_enabled():
                st.caption(_t(svc, "admin.storage.synced"))
            else:
                st.caption(_t(svc, "admin.storage.ephemeral"))
        except Exception:  # noqa: BLE001
            st.caption(_t(svc, "admin.storage.ephemeral"))
        st.caption(_t(svc, "admin.model.secrets_hint"))

        enabled = st.checkbox(
            _t(svc, "admin.model.enable"),
            value=bool(current.get("enabled")), key="pk_admin_enable",
        )
        idx = ids.index(current.get("provider")) if current.get("provider") in ids else 0
        pid = st.selectbox(
            _t(svc, "model.provider"), options=ids, index=idx,
            format_func=lambda i: names.get(i, i), key="pk_admin_provider",
        )
        model = st.text_input(_t(svc, "model.model_custom"), value=current.get("model", ""), key="pk_admin_model") or ""
        api_key = st.text_input(_t(svc, "model.api_key"), type="password", key="pk_admin_api_key") or ""
        base_url = ""
        if pid == "custom":
            base_url = st.text_input(
                _t(svc, "model.base_url"), value=current.get("base_url", ""),
                help=_t(svc, "model.base_url_hint"), key="pk_admin_base_url",
            ) or ""
        if st.button(_t(svc, "admin.model.save"), key="pk_admin_model_save"):
            # 密钥留空时保留原密钥（避免每次都要重输）。
            key_to_save = api_key or current.get("api_key", "")
            db.set_default_llm_config(
                pid, key_to_save, model=model, base_url=base_url, enabled=enabled
            )
            st.success(_t(svc, "admin.model.saved"))
            _rerun()


def render_sidebar(svc: Services, registry: SkillRegistry) -> Optional[PharmaSkill]:
    """渲染侧边栏参数区：登录 → 导航（按角色过滤）→ 当前 Skill 输入面板（需求 15.1/15.5）。

    返回当前选中的 Skill（无可见 Skill 时返回 ``None``）。
    """
    st.sidebar.title(_t(svc, "app.title"))
    render_login(svc)

    # 模型设置（用户自配，最高优先级）+ 管理员全局配置（仅管理员可见）。
    render_model_settings(svc)
    render_admin_model_settings(svc)

    role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE)
    skills = visible_skills_for_role(registry, getattr(svc, "auth", None), role)
    if not skills:
        st.sidebar.info(_t(svc, "msg.no_data"))
        return None

    i18n = getattr(svc, "i18n", None)
    labels = [skill_nav_label(s, i18n) for s in skills]
    st.sidebar.markdown(f"**{_t(svc, 'nav.section')}**")
    idx = st.sidebar.radio(
        _t(svc, "nav.section"),
        options=list(range(len(skills))),
        format_func=lambda i: labels[i],
        key="pk_nav",
        label_visibility="collapsed",
    )
    selected = skills[idx]
    # 切换功能时清理上一个技能的残留报告 / 任务单状态，避免旧报告串显到新技能（用户反馈）。
    prev_id = st.session_state.get(SESSION_SKILL)
    for k in stale_session_keys_on_switch(prev_id, selected.meta.id):
        st.session_state.pop(k, None)
    st.session_state[SESSION_SKILL] = selected.meta.id

    # 当前 Skill 的输入面板（参数区）。
    st.sidebar.markdown("---")
    st.sidebar.subheader(skill_nav_label(selected, i18n))
    return selected


def render_main(svc: Services, skill: PharmaSkill, registry: Optional[SkillRegistry] = None) -> None:
    """中央主画布区：收集输入 → 理解层产出任务单 → 用户确认 → 分析 → 渲染报告。

    两阶段流程（intent-understanding-layer）：
    1. 用户点击主操作按钮 → 理解层 ``build_task_sheet`` 产出《分析任务单》；
    2. 展示任务单（含画像 / 意图 / 提取项 / 缺失），用户**每次都需确认**后才执行；
    3. 澄清态任务单展示澄清提示与可选功能，不进入计算（需求 6）。
    理解层 / 编排器不可用时回退到旧的直接分发路径（向后兼容）。
    """
    i18n = getattr(svc, "i18n", None)
    lang = getattr(svc, "lang", None)
    st.header(skill_nav_label(skill, i18n))
    st.caption(skill_description(skill, i18n, lang))

    raw = _safe_render_inputs(skill, st.sidebar, svc)
    orchestrator = build_orchestrator(registry, svc) if registry is not None else None

    # 输入签名确定的技能（通用问答 / 相容性）：尊重用户在侧边栏的显式选择，
    # 直接走固定流水线（extract→compute→explain），不经理解层意图澄清——
    # 修复「问答被『请选择质量属性梳理/货架期外推』澄清拦截」与「相容性多余确认」。
    # 理解层 / 编排器不可用时同样回退直接分发（向后兼容）。
    if orchestrator is None or not getattr(skill.meta, "two_phase", True):
        _render_main_legacy(svc, skill, raw)
        return

    action_label = skill_action_label(skill, svc)
    if st.button(action_label, key=f"pk_run_{skill.meta.id}", type="primary", use_container_width=True):
        resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL))
        with st.spinner(_t(svc, "app.loading")):
            try:
                sheet = orchestrator.build_task_sheet(raw)
                st.session_state[SESSION_TASK_SHEET] = sheet.to_dict()
                st.session_state[SESSION_SHEET_STATE] = "pending_confirm"
                st.session_state.pop(SESSION_SECTIONS, None)
            except Exception as exc:  # noqa: BLE001
                logger.warning("理解层处理失败：%s", exc, exc_info=True)
                st.error(_t(svc, "error.generic"))

    state = st.session_state.get(SESSION_SHEET_STATE, "none")
    sheet_dict = st.session_state.get(SESSION_TASK_SHEET)

    if state == "pending_confirm" and sheet_dict:
        if sheet_is_confirmable(sheet_dict):
            # 展示文档画像（只读）+ 可编辑提取项 + 可切换功能。
            _render_task_sheet(svc, sheet_dict)
            edited_rows, chosen_skill = _render_task_sheet_editor(svc, sheet_dict, registry)
            edited_slot_rows = _render_intent_slots_editor(svc, sheet_dict)
            st.caption(_t(svc, "task.confirm_required"))
            cols = st.columns(2)
            with cols[0]:
                if st.button(_t(svc, "task.btn.confirm"), key="pk_sheet_confirm",
                             type="primary", use_container_width=True):
                    final_sheet = apply_sheet_edits(sheet_dict, edited_rows, chosen_skill)
                    final_sheet = apply_slot_edits(final_sheet, edited_slot_rows)
                    _audit_sheet_edit(svc, sheet_dict, final_sheet)
                    _run_confirmed_sheet(svc, orchestrator, final_sheet, raw)
            with cols[1]:
                if st.button(_t(svc, "task.btn.cancel"), key="pk_sheet_cancel",
                             use_container_width=True):
                    st.session_state[SESSION_SHEET_STATE] = "none"
                    st.session_state.pop(SESSION_TASK_SHEET, None)
                    _rerun()
        else:
            _render_task_sheet(svc, sheet_dict)
            chosen_intent = _render_clarification(svc, sheet_dict)
            if chosen_intent:
                # 用户在澄清态选定意图 → 解析为可确认任务单（清除澄清、派生槽位），
                # 下一轮进入完整可编辑确认面板（提取项编辑 + 槽位确认）。
                st.session_state[SESSION_TASK_SHEET] = _resolve_clarified_sheet(
                    sheet_dict, chosen_intent, raw
                )
                st.session_state[SESSION_SHEET_STATE] = "pending_confirm"
                _rerun()
        return

    sections = st.session_state.get(SESSION_SECTIONS)
    if sections is not None:
        render_report(svc, sections)
    else:
        st.info(_t(svc, "main.empty_hint"))


def _run_confirmed_sheet(svc: Services, orchestrator, sheet_dict: dict, raw: RawInput) -> None:
    """把用户确认的任务单交编排器执行并渲染（需求 5.4 / 5.5）。"""
    from kernel.task_sheet import AnalysisTaskSheet

    with st.spinner(_t(svc, "app.loading")):
        try:
            sheet = AnalysisTaskSheet.from_dict(sheet_dict)
            sections = orchestrator.run_confirmed(
                sheet, raw, user=st.session_state.get(SESSION_USER, "anonymous")
            )
            st.session_state[SESSION_SECTIONS] = sections
            st.session_state[SESSION_SHEET_STATE] = "confirmed"
        except Exception as exc:  # noqa: BLE001
            logger.warning("确认执行失败：%s", exc, exc_info=True)
            st.error(_t(svc, "error.generic"))
    _rerun()


def _render_task_sheet(svc: Services, sheet_dict: dict) -> None:
    """展示《分析任务单》：意图 / 文档画像 / 提取项 / 缺失项（需求 5.2）。"""
    st.subheader(_t(svc, "task.sheet.title"))
    st.caption(_t(svc, "task.sheet.intro"))

    intent = sheet_dict.get("intent", "unknown")
    st.markdown(f"**{_t(svc, 'task.intent')}**: {_t(svc, f'task.intent.{intent}')}")

    profile = sheet_dict.get("document_profile") or {}
    tables = profile.get("tables") or []
    if tables:
        st.markdown(f"**{_t(svc, 'task.profile')}**")
        st.dataframe(
            [
                {
                    _t(svc, "task.profile.table"): t.get("title", ""),
                    _t(svc, "task.profile.type"): t.get("table_type", ""),
                    _t(svc, "task.profile.source"): t.get("source_document", ""),
                    _t(svc, "task.profile.is_stability"): "✓" if t.get("is_stability_time_series") else "—",
                }
                for t in tables
            ],
            use_container_width=True,
        )
    if not profile.get("has_stability_time_series", False):
        st.caption(_t(svc, "task.profile.no_stability"))

    items = sheet_dict.get("extracted_items") or []
    missing = sheet_dict.get("missing_items") or []
    if missing:
        st.markdown(f"**{_t(svc, 'task.missing')}**: " + "、".join(str(m) for m in missing))


def _render_task_sheet_editor(svc: Services, sheet_dict: dict, registry: Optional[SkillRegistry]):
    """可编辑提取项 + 可切换分析功能（需求 5.3 / 5.5）。

    返回 ``(edited_rows, chosen_skill_id)``。无 ``st.data_editor`` 的旧版 Streamlit
    回退为只读展示并返回原始行 / 原功能（保证不崩溃）。
    """
    st.markdown(f"**{_t(svc, 'task.edit.items')}**")
    st.caption(_t(svc, "task.edit.hint"))

    rows = items_to_rows(sheet_dict.get("extracted_items") or [])
    column_config = None
    data_editor = getattr(st, "data_editor", None) or getattr(st, "experimental_data_editor", None)
    if callable(data_editor):
        edited = data_editor(
            rows,
            num_rows="dynamic",   # 允许增 / 删行
            use_container_width=True,
            key="pk_sheet_editor",
            column_config=column_config,
        )
        edited_rows = list(edited) if isinstance(edited, list) else rows
    else:  # 旧版回退：只读
        st.dataframe(rows, use_container_width=True)
        edited_rows = rows

    # 分析功能切换。
    chosen_skill = sheet_dict.get("proposed_skill_id", "")
    if registry is not None:
        ids = [s.meta.id for s in registry.all()]
        if ids:
            default_idx = ids.index(chosen_skill) if chosen_skill in ids else 0
            i18n = getattr(svc, "i18n", None)

            def _skill_label(sid: str) -> str:
                # 优先 nav.<id>，否则取 Skill 自身本地化名 / display_name（避免暴露原始键）。
                has = getattr(i18n, "has", None)
                if callable(has) and has(f"nav.{sid}"):
                    return _t(svc, f"nav.{sid}")
                sk = registry.get(sid)
                return skill_nav_label(sk, i18n) if sk is not None else sid

            chosen_skill = st.selectbox(
                _t(svc, "task.edit.skill"),
                options=ids,
                index=default_idx,
                format_func=_skill_label,
                key="pk_sheet_skill_pick",
            )
    return edited_rows, chosen_skill


def _render_intent_slots_editor(svc: Services, sheet_dict: dict) -> list[dict]:
    """渲染意图槽位确认子面板（意图保真，阶段四）；返回可回写的槽位行。

    展示每个意图维度的当前值、来源态（明示/推断/缺失）与依据，允许用户**修正并锁定**。
    用户填值/锁定后经 :func:`apply_slot_edits` 写回为 ``stated+locked``，下游计算据此
    采纳用户真实意图、不被模型推断推翻。无槽位时不渲染（返回空列表）。
    """
    rows = slots_to_rows(sheet_dict)
    if not rows:
        return []

    st.markdown(f"**{_t(svc, 'task.slots.title')}**")
    st.caption(_t(svc, "task.slots.hint"))

    data_editor = getattr(st, "data_editor", None) or getattr(st, "experimental_data_editor", None)
    # 仅暴露可编辑的「值 / 锁定」列；其余为只读上下文。
    display = [
        {
            "label": r["label"],
            "value": r["value"],
            "state_label": r["state_label"],
            "evidence": r["evidence"],
            "locked": r["locked"],
        }
        for r in rows
    ]
    if callable(data_editor):
        edited = data_editor(
            display,
            use_container_width=True,
            key="pk_slot_editor",
            disabled=["label", "state_label", "evidence"],
        )
        edited_list = list(edited) if isinstance(edited, list) else display
    else:  # 旧版回退：只读
        st.dataframe(display, use_container_width=True)
        edited_list = display

    # 回填 name（展示层用 label，回写需 name）。
    out: list[dict] = []
    for r, e in zip(rows, edited_list):
        out.append({
            "name": r["name"],
            "value": e.get("value", r["value"]),
            "locked": bool(e.get("locked", r["locked"])),
        })
    return out


def _audit_sheet_edit(svc: Services, original: dict, edited: dict) -> None:
    """记录用户对任务单的编辑（需求 8.2 可追溯）。"""
    orig_items = original.get("extracted_items") or []
    new_items = edited.get("extracted_items") or []
    user_edited = sum(1 for it in new_items if it.get("source") == "user_edited")
    audit = getattr(svc, "audit", None)
    recorder = getattr(audit, "record_event", None) or getattr(audit, "record_analysis", None)
    if callable(recorder):
        try:
            recorder("task_sheet_edited",
                     edited.get("proposed_skill_id", ""),
                     f"orig={len(orig_items)} final={len(new_items)} user_edited={user_edited}")
        except Exception:  # noqa: BLE001 - 审计失败不影响主流程
            logger.debug("记录任务单编辑审计失败（已忽略）。", exc_info=True)


def _render_clarification(svc: Services, sheet_dict: dict):
    """澄清态：提示 + **交互式意图选择** + 确认按钮（需求 6）。

    返回用户选定的意图值（如 ``"shelf_life_extrapolation"``）当且仅当用户点击确认按钮；
    否则返回 ``None``。本函数本身不进入计算——选定后由调用方解析为可确认任务单。
    """
    clar = sheet_dict.get("clarification") or {}
    st.warning(_t(svc, "understanding.clarify.title"))
    msg_key = clar.get("message_key") or ""
    if msg_key:
        st.info(_t(svc, msg_key))
    options = [o for o in (clar.get("options") or []) if o]
    if not options:
        return None
    st.caption(_t(svc, "understanding.clarify.choose"))
    labels = {o: _t(svc, f"task.intent.{o}") for o in options}
    radio = getattr(st, "radio", None)
    if callable(radio):
        choice = radio(
            _t(svc, "understanding.clarify.choose"),
            options=options,
            format_func=lambda o: labels.get(o, o),
            key="pk_clarify_choice",
        )
    else:  # 极简回退：选择框
        choice = st.selectbox(
            _t(svc, "understanding.clarify.choose"), options=options,
            format_func=lambda o: labels.get(o, o), key="pk_clarify_choice",
        )
    if st.button(_t(svc, "task.clarify.apply"), key="pk_clarify_apply",
                 type="primary", use_container_width=True):
        return choice
    return None


def _resolve_clarified_sheet(sheet_dict: dict, intent_value: str, raw: RawInput) -> dict:
    """把用户在澄清态选定的意图解析为**可确认任务单**（纯函数，供 UI 与测试）。

    - 设定 ``intent`` 与 ``proposed_skill_id``（按意图映射），并**清除澄清态**；
    - 依选定意图与现有提取项派生意图槽位（``intent_slots``），使后续确认面板完整呈现
      来源标记与待确认事项。意图解析 / 槽位派生失败时安全降级（不抛异常）。
    """
    new_sheet = dict(sheet_dict or {})
    new_sheet["intent"] = intent_value
    new_sheet["proposed_skill_id"] = _INTENT_SKILL_MAP.get(intent_value, "")
    new_sheet["clarification"] = None
    try:
        from kernel.task_sheet import ExtractedItem, Intent
        from kernel.understanding import build_intent_slots

        intent = Intent(intent_value)
        items = [ExtractedItem.from_dict(it) for it in (sheet_dict.get("extracted_items") or [])]
        goal = str(getattr(raw, "goal", "") or "")
        new_sheet["intent_slots"] = [s.to_dict() for s in build_intent_slots(goal, intent, items)]
    except Exception:  # noqa: BLE001 - 槽位派生失败不阻断澄清解析
        new_sheet.setdefault("intent_slots", [])
    return new_sheet


#: 意图值 → Skill id（与 kernel.router.INTENT_TO_SKILL 一致）。
_INTENT_SKILL_MAP = {
    "descriptive_summary": "descriptive_summary",
    "shelf_life_extrapolation": "stability",
    "compatibility": "compatibility",
}


def _render_main_legacy(svc: Services, skill: PharmaSkill, raw: RawInput) -> None:
    """旧的直接分发路径（理解层不可用时回退，保持向后兼容）。"""
    action_label = skill_action_label(skill, svc)
    if st.button(action_label, key=f"pk_run_legacy_{skill.meta.id}", type="primary", use_container_width=True):
        resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL))
        with st.spinner(_t(svc, "app.loading")):
            try:
                sections = dispatch(
                    skill, raw, svc,
                    user=st.session_state.get(SESSION_USER, "anonymous"),
                )
                st.session_state[SESSION_SECTIONS] = sections
            except Exception as exc:  # noqa: BLE001
                logger.warning("分析执行失败：%s", exc, exc_info=True)
                st.error(_t(svc, "error.generic"))
                st.session_state.pop(SESSION_SECTIONS, None)

    sections = st.session_state.get(SESSION_SECTIONS)
    if sections is not None:
        render_report(svc, sections)
    else:
        st.info(_t(svc, "main.empty_hint"))


def _safe_render_inputs(skill: PharmaSkill, st_ctx, svc: Optional[Services] = None) -> RawInput:
    """调用 Skill 的 ``render_inputs``，异常时降级空输入（不崩溃）。

    渲染前把一个**已绑定底座 i18n 与命名空间**的翻译函数注入到 Skill 实例
    （``skill._i18n_t``），使各 Skill 的输入面板文案经底座 ``I18nService`` 取用、
    随语言切换（需求 3 / 17.4）。注入失败或无 ``svc`` 时，面板按其内置中文兜底，
    不影响无 GUI 的纯逻辑单测。
    """
    if svc is not None:
        skill_id = getattr(skill.meta, "id", "") or ""

        def _translate(key: str) -> str:
            # 面板传入局部键（如 "input.goal"）；底座目录以 "<id>." 前缀合并。
            return _t(svc, f"{skill_id}.{key}")

        try:
            setattr(skill, "_i18n_t", _translate)
        except Exception:  # noqa: BLE001 - 注入失败不阻断渲染
            pass
    try:
        raw = skill.render_inputs(st_ctx)
        return raw if isinstance(raw, RawInput) else RawInput()
    except Exception as exc:  # noqa: BLE001
        logger.warning("渲染 Skill %s 输入面板失败：%s", skill.meta.id, exc)
        return RawInput()


def render_report(svc: Services, sections: ReportSections) -> None:
    """渲染分析返回的 ``ReportSections``：优先完整 HTML，否则逐段展示，并提供下载。"""
    html = getattr(sections, "html", "") or ""
    if html:
        # 用 iframe 承载完整自包含报告（含样式）；跨 Streamlit 版本兼容。
        _render_html_component(html, height=760)
        cols = st.columns(2)
        with cols[0]:
            st.download_button(
                _t(svc, "btn.download_report"),
                data=html,
                file_name="report.html",
                mime="text/html",
                key="pk_download_report",
            )
        # 专业 Word(.docx) 下载：表格化、分级标题、去 AI/Markdown 噪声。
        with cols[1]:
            _render_word_download(svc, html)
        return

    section_map = getattr(sections, "sections", {}) or {}
    if not section_map:
        st.info(_t(svc, "msg.no_data"))
        return
    for key, value in section_map.items():
        st.subheader(_t(svc, f"report.section.{key}") if key else key)
        text = value if isinstance(value, str) else str(value)
        if text.strip().startswith("<"):
            st.markdown(text, unsafe_allow_html=True)
        else:
            st.write(text)


def _render_word_download(svc: Services, html: str) -> None:
    """生成并提供专业 Word(.docx) 下载；python-docx 不可用时给出友好提示。"""
    try:
        report = getattr(svc, "report", None)
        to_docx = getattr(report, "to_docx", None)
        data = to_docx(html) if callable(to_docx) else None
    except Exception:  # noqa: BLE001 - 导出失败不影响主流程
        data = None
    if data:
        st.download_button(
            _t(svc, "btn.download_word"),
            data=data,
            file_name="report.docx",
            mime="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            key="pk_download_word",
        )
    else:
        st.caption(_t(svc, "msg.word_unavailable"))


def _render_html_component(html: str, *, height: int = 760) -> None:
    """跨版本兼容地嵌入 HTML：优先 ``st.components.v1.html``，回退 ``st.html``。"""
    try:
        comp = getattr(getattr(st, "components", None), "v1", None)
        if comp is not None and hasattr(comp, "html"):
            comp.html(html, height=height, scrolling=True)
            return
    except Exception:  # noqa: BLE001 - 回退到 st.html
        pass
    html_fn = getattr(st, "html", None)
    if callable(html_fn):
        html_fn(html)
    else:
        st.markdown(html, unsafe_allow_html=True)


def render_audit(svc: Services) -> None:
    """可折叠审计追踪区：仅 admin / auditor 可见（需求 15.1 / 15.5）。"""
    role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE)
    if not can_view_audit(getattr(svc, "auth", None), role):
        return

    audit = getattr(svc, "audit", None)
    recent = getattr(audit, "recent", None)
    with st.expander(_t(svc, "nav.audit"), expanded=False):
        if not callable(recent):
            st.info(_t(svc, "msg.no_data"))
            return
        try:
            events = recent(50)
        except Exception as exc:  # noqa: BLE001
            logger.warning("查询审计事件失败：%s", exc)
            events = []
        if not events:
            st.info(_t(svc, "msg.no_data"))
            return
        rows = [
            {
                "ts": getattr(e, "ts_iso", ""),
                "user": getattr(e, "user", ""),
                "event": getattr(e, "event_type", ""),
                "skill": getattr(e, "skill_id", "") or "",
                "step": getattr(e, "step", "") or "",
            }
            for e in events
        ]
        st.dataframe(rows, use_container_width=True)


def render_build_warnings(svc: Services) -> None:
    """渲染服务构建期降级告警（如某可选重依赖缺失），帮助用户理解功能降级。"""
    warnings = getattr(svc, "_build_warnings", None) or []
    if not warnings:
        return
    with st.expander("⚠️ " + _t(svc, "warn.pdf_fallback"), expanded=False):
        for name, msg in warnings:
            st.caption(f"{name}: {msg}")


def _rerun() -> None:
    """跨 Streamlit 版本兼容的重渲染触发。"""
    rerun = getattr(st, "rerun", None) or getattr(st, "experimental_rerun", None)
    if callable(rerun):
        rerun()


def load_css() -> None:
    """注入 Nordic 极简主题 CSS（``assets/style.css``）；缺失时静默跳过。"""
    try:
        from pathlib import Path

        css_path = Path(__file__).resolve().parents[1] / "assets" / "style.css"
        if css_path.is_file():
            css = css_path.read_text(encoding="utf-8")
            st.markdown(f"<style>{css}</style>", unsafe_allow_html=True)
    except Exception as exc:  # noqa: BLE001 - 样式缺失不应影响功能
        logger.info("加载样式表失败（已忽略）：%s", exc)


def render_top_bar(svc: Services) -> None:
    """顶部品牌条 + 右侧语言切换：左品牌名/副标题，右语言选择（修复 #0 布局）。"""
    left, right = st.columns([5, 1])
    with left:
        st.markdown(
            f"<div style='display:flex;align-items:baseline;gap:.6rem;'>"
            f"<span style='font-size:1.5rem;font-weight:800;color:#1f6f78;'>"
            f"💊 {_html_escape(_t(svc, 'app.title'))}</span>"
            f"<span style='font-size:.85rem;color:#5d6b76;'>"
            f"{_html_escape(_t(svc, 'app.subtitle'))}</span></div>",
            unsafe_allow_html=True,
        )
    with right:
        i18n = getattr(svc, "i18n", None)
        langs = list(getattr(i18n, "available_languages", lambda: ("zh", "en"))())
        current = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG)
        choice = st.selectbox(
            _t(svc, "label.language"),
            options=langs,
            index=langs.index(current) if current in langs else 0,
            format_func=lambda code: _t(svc, f"lang.{code}"),
            key="pk_lang_select",
            label_visibility="collapsed",
        )
        if choice != current:
            st.session_state[SESSION_LANG] = choice
            set_session_language(svc, choice)
            _rerun()
    st.markdown(
        "<hr style='margin:.4rem 0 1.2rem;border:none;border-top:1px solid #e7ebee;'>",
        unsafe_allow_html=True,
    )


def _html_escape(s: str) -> str:
    import html as _h

    return _h.escape(str(s or ""))


def main() -> None:
    """Streamlit 应用主入口：装配上下文并按统一布局渲染。"""
    if st is None:  # pragma: no cover - 无 streamlit 环境下不渲染
        raise RuntimeError(
            "Streamlit 未安装，无法渲染界面。请通过 "
            "`streamlit run core/app.py` 启动。"
        )

    # set_page_config 必须是第一个 Streamlit 调用。
    st.set_page_config(
        page_title="Pharma K",
        page_icon="💊",
        layout="wide",
        initial_sidebar_state="expanded",
    )
    load_css()

    svc, registry = get_or_create_context()

    # 顶部品牌条 + 语言切换（需求 17.1 / 17.2；修复 #0 布局）。
    render_top_bar(svc)
    render_build_warnings(svc)

    # 侧边栏参数区（账户 + 导航 + 输入面板，需求 15.1 / 15.5）。
    selected = render_sidebar(svc, registry)

    # 中央主画布区（结果 / 报告，需求 15.1）。
    if selected is not None:
        render_main(svc, selected, registry)
    else:
        st.info(_t(svc, "msg.no_data"))

    # 可折叠审计追踪区（按角色，需求 15.1 / 15.5）。
    render_audit(svc)


if __name__ == "__main__":  # pragma: no cover - 由 `streamlit run` 调用
    main()