Compare commits

...

6 Commits

Author SHA1 Message Date
Soulter
319b2570be fix: keep system tools with persona tool lists 2026-06-20 00:57:43 +08:00
Soulter
a7533aacda fix: fall back to stale WebUI when repair fails 2026-06-20 00:01:29 +08:00
Weilong Liao
46a846b88b chore: bump version to 4.26.0-beta.10 (#8905) 2026-06-19 23:59:39 +08:00
lxfight
2d98d38078 fix: inject knowledge base context as temporary user content (#8904) 2026-06-19 22:48:23 +08:00
Weilong Liao
1b0f5cb0d3 fix: keep WebUI assets in sync with core version (#8901)
* fix: keep WebUI assets in sync with core version

* fix: import dashboard version before bundled fallback

* fix: remove stale WebUI dist robustly
2026-06-19 22:46:38 +08:00
Weilong Liao
cdfb0bdf91 fix: restore webui 401 login redirect (#8903) 2026-06-19 22:43:21 +08:00
13 changed files with 627 additions and 120 deletions

View File

@@ -16,8 +16,11 @@ venv*/
ENV/
.conda/
dashboard/
!astrbot/dashboard/
!astrbot/dashboard/dist/
!astrbot/dashboard/dist/**
data/
tests/
.ruff_cache/
.astrbot
astrbot.lock
astrbot.lock

View File

@@ -46,14 +46,21 @@ jobs:
- name: Build Dashboard
run: |
dashboard_version=$(python3 - <<'PY'
import tomllib
with open("pyproject.toml", "rb") as f:
print("v" + tomllib.load(f)["project"]["version"])
PY
)
cd dashboard
npm install
npm run build
mkdir -p dist/assets
echo $(git rev-parse HEAD) > dist/assets/version
echo "$dashboard_version" > dist/assets/version
cd ..
mkdir -p data
cp -r dashboard/dist data/
mkdir -p astrbot/dashboard
rm -rf astrbot/dashboard/dist
cp -r dashboard/dist astrbot/dashboard/dist
- name: Determine test image tags
id: test-meta
@@ -157,10 +164,11 @@ jobs:
npm install
npm run build
mkdir -p dist/assets
echo $(git rev-parse HEAD) > dist/assets/version
echo "${{ steps.release-meta.outputs.version }}" > dist/assets/version
cd ..
mkdir -p data
cp -r dashboard/dist data/
mkdir -p astrbot/dashboard
rm -rf astrbot/dashboard/dist
cp -r dashboard/dist astrbot/dashboard/dist
- name: Set QEMU
uses: docker/setup-qemu-action@v4.1.0

View File

@@ -278,10 +278,11 @@ async def _apply_kb(
)
if not kb_result:
return
if req.system_prompt is not None:
req.system_prompt += (
f"\n\n[Related Knowledge Base Results]:\n{kb_result}"
)
req.extra_user_content_parts.append(
TextPart(
text=f"[Related Knowledge Base Results]:\n{kb_result}",
).mark_as_temp()
)
except Exception as exc: # noqa: BLE001
logger.error("Error occurred while retrieving knowledge base: %s", exc)
else:
@@ -456,10 +457,10 @@ async def _ensure_persona_and_skills(
cfg: dict,
plugin_context: Context,
event: AstrMessageEvent,
) -> set[str] | None:
) -> None:
"""Ensure persona and skills are applied to the request's system prompt or user prompt."""
if not req.conversation:
return None
return
(
persona_id,
@@ -526,13 +527,11 @@ async def _ensure_persona_and_skills(
# inject toolset in the persona
if (persona and persona.get("tools") is None) or not persona:
persona_allowed_tools = None
persona_toolset = tmgr.get_full_tool_set()
for tool in list(persona_toolset):
if not tool.active:
persona_toolset.remove_tool(tool.name)
else:
persona_allowed_tools = {str(tool_name) for tool_name in persona["tools"]}
persona_toolset = ToolSet()
if persona["tools"]:
for tool_name in persona["tools"]:
@@ -613,7 +612,6 @@ async def _ensure_persona_and_skills(
)
except Exception:
pass
return persona_allowed_tools
async def _request_img_caption(
@@ -946,13 +944,12 @@ async def _decorate_llm_request(
plugin_context: Context,
config: MainAgentBuildConfig,
provider: Provider | None = None,
) -> set[str] | None:
) -> None:
cfg = config.provider_settings or plugin_context.get_config(
umo=event.unified_msg_origin
).get("provider_settings", {})
_apply_prompt_prefix(req, cfg)
persona_allowed_tools = None
main_provider_supports_image = provider is not None and _provider_supports_modality(
provider, "image"
@@ -961,9 +958,7 @@ async def _decorate_llm_request(
quote_images_already_captioned = False
if req.conversation:
persona_allowed_tools = await _ensure_persona_and_skills(
req, cfg, plugin_context, event
)
await _ensure_persona_and_skills(req, cfg, plugin_context, event)
if img_cap_prov_id and req.image_urls and not main_provider_supports_image:
await _ensure_img_caption(
@@ -992,7 +987,6 @@ async def _decorate_llm_request(
tz = plugin_context.get_config().get("timezone")
_append_system_reminders(event, req, cfg, tz)
_apply_workspace_extra_prompt(event, req)
return persona_allowed_tools
def _plugin_tool_fix(event: AstrMessageEvent, req: ProviderRequest) -> None:
@@ -1521,9 +1515,7 @@ async def build_main_agent(
else:
return None
persona_allowed_tools = await _decorate_llm_request(
event, req, plugin_context, config, provider=provider
)
await _decorate_llm_request(event, req, plugin_context, config, provider=provider)
await _apply_kb(event, req, plugin_context, config)
@@ -1559,11 +1551,6 @@ async def build_main_agent(
)
)
if persona_allowed_tools is not None and req.func_tool:
req.func_tool.tools = [
tool for tool in req.func_tool.tools if tool.name in persona_allowed_tools
]
fallback_providers = _get_fallback_chat_providers(
provider, plugin_context, config.provider_settings
)

View File

@@ -183,8 +183,22 @@ async def download_file(
path: str,
show_progress: bool = False,
progress_callback=None,
allow_insecure_ssl_fallback: bool = True,
) -> None:
"""从指定 url 下载文件到指定路径 path"""
"""Download a remote file to a local path.
Args:
url: Remote URL to download.
path: Local destination path.
show_progress: Whether to print progress to stdout.
progress_callback: Optional callback for progress payloads.
allow_insecure_ssl_fallback: Whether certificate failures may retry with
TLS certificate verification disabled.
Returns:
None.
"""
try:
ssl_context = ssl.create_default_context(
cafile=certifi.where(),
@@ -259,6 +273,8 @@ async def download_file(
},
)
except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError):
if not allow_insecure_ssl_fallback:
raise
# 关闭SSL验证仅在证书验证失败时作为fallback
logger.warning(
f"SSL certificate verification failed for {_safe_url_for_log(url)}. "
@@ -355,10 +371,22 @@ def get_local_ip_addresses():
return network_ips
def _read_dashboard_dist_version(dist_dir: str | Path) -> str | None:
def get_dashboard_dist_version(dist_dir: str | Path) -> str | None:
"""Read the WebUI version from a dashboard dist directory.
Args:
dist_dir: Dashboard dist directory path.
Returns:
The version string from assets/version, or None when unavailable.
"""
version_file = Path(dist_dir) / "assets" / "version"
if version_file.exists():
return version_file.read_text(encoding="utf-8").strip()
try:
if version_file.exists():
return version_file.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError) as exc:
logger.warning("Failed to read WebUI version from %s: %s", version_file, exc)
return None
@@ -380,42 +408,106 @@ def _normalize_dashboard_version(version: str) -> str:
return version
def should_use_bundled_dashboard_dist(
user_dist: str | Path, current_version: str
def is_dashboard_version_compatible(
dashboard_version: str | None, current_version: str
) -> bool:
user_version = _read_dashboard_dist_version(user_dist)
bundled_dist = get_bundled_dashboard_dist_path()
if user_version is None or not bundled_dist.exists():
"""Check whether a WebUI version matches the current core version.
Args:
dashboard_version: Version read from the WebUI assets/version file.
current_version: Current AstrBot core version.
Returns:
True when both versions are valid SemVer values and compare equal.
"""
if dashboard_version is None:
return False
try:
return (
VersionComparator.compare_version(
_normalize_dashboard_version(dashboard_version),
_normalize_dashboard_version(current_version),
_normalize_dashboard_version(user_version),
)
> 0
== 0
)
except (TypeError, ValueError):
return False
def is_dashboard_dist_compatible(dist_dir: str | Path, current_version: str) -> bool:
"""Check whether a WebUI dist is complete and matches the core version.
Args:
dist_dir: Dashboard dist directory path.
current_version: Current AstrBot core version.
Returns:
True when the dist has an index file and a compatible assets/version.
"""
dist_path = Path(dist_dir)
return (dist_path / "index.html").is_file() and is_dashboard_version_compatible(
get_dashboard_dist_version(dist_path),
current_version,
)
def should_use_bundled_dashboard_dist(
user_dist: str | Path, current_version: str
) -> bool:
"""Decide whether bundled WebUI should replace a user data dist.
Args:
user_dist: Runtime dashboard dist directory under data/.
current_version: Current AstrBot core version.
Returns:
True when user_dist exists but is missing or mismatched against the
current core version, and bundled WebUI matches the current core version.
"""
user_dist = Path(user_dist)
user_version = get_dashboard_dist_version(user_dist)
bundled_dist = get_bundled_dashboard_dist_path()
if not user_dist.exists() or not is_dashboard_dist_compatible(
bundled_dist,
current_version,
):
return False
if user_version is None or not (user_dist / "index.html").is_file():
return True
try:
return not is_dashboard_version_compatible(user_version, current_version)
except (TypeError, ValueError):
return False
async def get_dashboard_version():
"""Return the effective WebUI version for the current runtime.
Returns:
The matching data/dist version, matching bundled version, or the raw
data/dist version when no compatible bundled WebUI is available.
"""
from astrbot.core.config.default import VERSION
# First check user data directory (manually updated / downloaded dashboard).
dist_dir = os.path.join(get_astrbot_data_path(), "dist")
if os.path.exists(dist_dir):
from astrbot.core.config.default import VERSION
user_version = get_dashboard_dist_version(dist_dir)
if is_dashboard_dist_compatible(dist_dir, VERSION):
return user_version
if should_use_bundled_dashboard_dist(dist_dir, VERSION):
bundled_version = _read_dashboard_dist_version(
get_bundled_dashboard_dist_path()
)
if bundled_version is not None:
return bundled_version
return _read_dashboard_dist_version(dist_dir)
bundled = get_bundled_dashboard_dist_path()
if is_dashboard_dist_compatible(bundled, VERSION):
return get_dashboard_dist_version(bundled)
return user_version
bundled = get_bundled_dashboard_dist_path()
if bundled.exists():
return _read_dashboard_dist_version(bundled)
if is_dashboard_dist_compatible(bundled, VERSION):
return get_dashboard_dist_version(bundled)
return None
@@ -427,6 +519,7 @@ async def download_dashboard(
proxy: str | None = None,
progress_callback=None,
extract: bool = True,
allow_insecure_ssl_fallback: bool = True,
) -> None:
"""Download dashboard assets and optionally extract them.
@@ -438,6 +531,8 @@ async def download_dashboard(
proxy: Optional download proxy prefix.
progress_callback: Optional callback for download progress payloads.
extract: Whether to extract the archive after download.
allow_insecure_ssl_fallback: Whether certificate failures may retry with
TLS certificate verification disabled.
Returns:
None.
@@ -460,6 +555,7 @@ async def download_dashboard(
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError(
@@ -491,6 +587,7 @@ async def download_dashboard(
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError(
@@ -506,6 +603,7 @@ async def download_dashboard(
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError("Downloaded dashboard package is not a valid ZIP file")

View File

@@ -22,7 +22,9 @@ from astrbot.core.db import BaseDatabase
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
from astrbot.core.utils.io import (
get_bundled_dashboard_dist_path,
get_dashboard_dist_version,
get_local_ip_addresses,
is_dashboard_dist_compatible,
should_use_bundled_dashboard_dist,
)
from astrbot.dashboard.asgi_runtime import (
@@ -182,21 +184,41 @@ class AstrBotDashboard:
# Path priority:
# 1. Explicit webui_dir argument
# 2. data/dist/ (user-installed / manually updated dashboard)
# 3. astrbot/dashboard/dist/ (bundled with the wheel)
# 2. data/dist/ when it matches the core version
# 3. astrbot/dashboard/dist/ when it matches the core version
if webui_dir and os.path.exists(webui_dir):
self.data_path = os.path.abspath(webui_dir)
else:
user_dist = os.path.join(get_astrbot_data_path(), "dist")
bundled_dist = get_bundled_dashboard_dist_path()
if os.path.exists(user_dist) and not should_use_bundled_dashboard_dist(
user_version = get_dashboard_dist_version(user_dist)
if os.path.exists(user_dist) and is_dashboard_dist_compatible(
user_dist,
VERSION,
):
self.data_path = os.path.abspath(user_dist)
elif bundled_dist.exists():
elif should_use_bundled_dashboard_dist(
user_dist,
VERSION,
) or is_dashboard_dist_compatible(bundled_dist, VERSION):
self.data_path = str(bundled_dist)
logger.info("Using bundled dashboard dist: %s", self.data_path)
elif (
os.path.exists(user_dist) and (Path(user_dist) / "index.html").is_file()
):
logger.warning(
"Using existing data/dist as a fallback even though WebUI version mismatches core: %s, expected v%s. "
"Some dashboard features may not work until the matching WebUI is available.",
user_version,
VERSION,
)
self.data_path = os.path.abspath(user_dist)
elif os.path.exists(user_dist):
logger.warning(
"Ignoring data/dist because WebUI files are incomplete for core v%s.",
VERSION,
)
self.data_path = None
else:
# Fall back to expected user path (will fail gracefully later)
self.data_path = os.path.abspath(user_dist)
@@ -545,7 +567,7 @@ class AstrBotDashboard:
raise Exception(f"端口 {port} 已被占用")
if (Path(self.data_path) / "index.html").is_file():
if self.data_path and (Path(self.data_path) / "index.html").is_file():
webui_status = "WebUI is ready"
else:
webui_status = (

View File

@@ -0,0 +1,22 @@
- [更新日志(简体中文)](#chinese)
- [Changelog(English)](#english)
<a id="chinese"></a>
## What's Changed
### 修复
- 恢复 WebUI 在接口返回 401 时跳转登录页,避免会话失效后停留在异常状态。([#8903](https://github.com/AstrBotDevs/AstrBot/pull/8903))
- 保持 Core 版本与 WebUI 静态资源版本同步,修复打包或升级后可能加载旧 dist、资源版本错配的问题。([#8901](https://github.com/AstrBotDevs/AstrBot/pull/8901))
- 将知识库上下文作为临时 user 内容注入,修复模型请求中知识库上下文角色不准确的问题。([#8904](https://github.com/AstrBotDevs/AstrBot/pull/8904))
<a id="english"></a>
## What's Changed (EN)
### Bug Fixes
- Restored the WebUI login redirect when API requests return 401, preventing expired sessions from staying in a broken state. ([#8903](https://github.com/AstrBotDevs/AstrBot/pull/8903))
- Kept Core and WebUI static asset versions in sync, fixing stale dist loading and asset version mismatches after packaging or upgrades. ([#8901](https://github.com/AstrBotDevs/AstrBot/pull/8901))
- Injected knowledge base context as temporary user content, fixing the role used for knowledge context in model requests. ([#8904](https://github.com/AstrBotDevs/AstrBot/pull/8904))

View File

@@ -48,6 +48,55 @@ function attachAxiosHeaders(config: InternalAxiosRequestConfig) {
}
function normalizeAxiosError(error: AxiosError) {
if (error.response?.status === 401) {
let requestPath = '';
try {
const url = error.config?.url || '';
const baseURL = error.config?.baseURL;
const resolvedUrl =
url && baseURL && !/^([a-z][a-z\d+\-.]*:)?\/\//i.test(url)
? `${baseURL.replace(/\/+$/, '')}/${url.replace(/^\/+/, '')}`
: url;
const requestUrl = new URL(resolvedUrl || '/', window.location.origin);
if (requestUrl.origin === window.location.origin) {
requestPath = requestUrl.pathname;
}
} catch {
requestPath = '';
}
const isAuthChallenge =
[
'/api/auth/login',
'/api/auth/setup',
'/api/auth/setup-status',
'/api/v1/auth/login',
'/api/v1/auth/setup',
'/api/v1/auth/setup-status',
].includes(requestPath) ||
Boolean(
(
error.response.data as
| { data?: { totp_required?: boolean } }
| undefined
)?.data?.totp_required,
);
if (requestPath.startsWith('/api/') && !isAuthChallenge) {
[
'user',
'token',
'change_pwd_hint',
'md5_pwd_hint',
'password_upgrade_required',
].forEach((key) => localStorage.removeItem(key));
if (!window.location.hash.startsWith('#/auth/login')) {
window.location.hash = '/auth/login';
}
}
}
if (error.response?.status === 429) {
const data = error.response.data as { message?: string } | undefined;
if (data?.message) {

97
main.py
View File

@@ -2,6 +2,7 @@ import argparse
import asyncio
import mimetypes
import os
import shutil
import sys
from pathlib import Path
@@ -46,7 +47,10 @@ from astrbot.core.utils.astrbot_path import ( # noqa: E402
from astrbot.core.utils.io import ( # noqa: E402
download_dashboard,
get_bundled_dashboard_dist_path,
get_dashboard_version,
get_dashboard_dist_version,
is_dashboard_dist_compatible,
is_dashboard_version_compatible,
remove_dir,
should_use_bundled_dashboard_dist,
)
from astrbot.core.utils.runtime_env import is_packaged_desktop_runtime # noqa: E402
@@ -91,7 +95,15 @@ def check_env() -> None:
async def check_dashboard_files(webui_dir: str | None = None):
"""下载管理面板文件"""
"""Resolve and repair dashboard static files for startup.
Args:
webui_dir: Optional explicit WebUI directory path from CLI.
Returns:
The directory path to serve, or None when no usable WebUI can be prepared.
"""
# 指定webui目录
if webui_dir:
if os.path.exists(webui_dir):
@@ -99,40 +111,89 @@ async def check_dashboard_files(webui_dir: str | None = None):
return webui_dir
logger.warning("WebUI directory not found: %s. Using default.", webui_dir)
data_dist_path = os.path.join(get_astrbot_data_path(), "dist")
if os.path.exists(data_dist_path):
v = await get_dashboard_version()
data_dist_path = Path(get_astrbot_data_path()) / "dist"
bundled_dist = get_bundled_dashboard_dist_path()
if data_dist_path.exists():
v = get_dashboard_dist_version(data_dist_path)
if is_dashboard_dist_compatible(data_dist_path, VERSION):
logger.info("WebUI is up to date.")
return str(data_dist_path)
if should_use_bundled_dashboard_dist(data_dist_path, VERSION):
bundled_dist = get_bundled_dashboard_dist_path()
logger.info(
"Using bundled WebUI because data/dist is older than core version v%s.",
"Replacing data/dist with bundled WebUI because its version does not match core version v%s.",
VERSION,
)
return str(bundled_dist)
if v is not None:
# 存在文件
if v == f"v{VERSION}":
logger.info("WebUI is up to date.")
else:
try:
remove_dir(str(data_dist_path))
shutil.copytree(bundled_dist, data_dist_path)
return str(data_dist_path)
except Exception as e:
logger.warning(
"WebUI version mismatch: %s, expected v%s.",
v,
"Failed to replace data/dist with bundled WebUI: %s. Using bundled WebUI directly.",
e,
)
return str(bundled_dist)
if is_dashboard_version_compatible(v, VERSION):
logger.warning(
"WebUI files are incomplete for v%s. Re-downloading WebUI.",
VERSION,
)
elif v is not None:
logger.warning(
"WebUI version mismatch: %s, expected v%s. Re-downloading WebUI.",
v,
VERSION,
)
else:
logger.warning(
"WebUI version file is missing. Re-downloading WebUI v%s.",
VERSION,
)
try:
await download_dashboard(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
except Exception as e:
logger.critical(f"下载管理面板文件失败: {e}")
if (data_dist_path / "index.html").is_file():
logger.warning(
"Falling back to existing data/dist WebUI %s even though core expects v%s. "
"Some dashboard features may not work until the matching WebUI is available.",
v or "unknown",
VERSION,
)
return data_dist_path
return str(data_dist_path)
return None
logger.info("管理面板下载完成。")
return str(data_dist_path)
if is_dashboard_dist_compatible(bundled_dist, VERSION):
logger.info(
"Using bundled WebUI v%s.", get_dashboard_dist_version(bundled_dist)
)
return str(bundled_dist)
logger.info(
"Downloading WebUI. If it fails, download dist.zip from https://github.com/AstrBotDevs/AstrBot/releases/latest and extract dist to data/.",
)
try:
await download_dashboard(version=f"v{VERSION}", latest=False)
await download_dashboard(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
except Exception as e:
logger.critical(f"下载管理面板文件失败: {e}")
return None
logger.info("管理面板下载完成。")
return data_dist_path
return str(data_dist_path)
async def main_async(webui_dir_arg: str | None) -> None:

View File

@@ -1,6 +1,6 @@
[project]
name = "AstrBot"
version = "4.26.0-beta.9"
version = "4.26.0-beta.10"
description = "Easy-to-use multi-platform LLM chatbot and development framework"
readme = "README.md"
license = { text = "AGPL-3.0-or-later" }

View File

@@ -273,6 +273,7 @@ def test_dashboard_uses_bundled_dist_when_data_dist_is_stale(
bundled_dist = tmp_path / "bundled-dist"
user_dist.mkdir(parents=True)
bundled_dist.mkdir()
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
monkeypatch.setattr(
"astrbot.dashboard.server.get_astrbot_data_path",
@@ -293,6 +294,59 @@ def test_dashboard_uses_bundled_dist_when_data_dist_is_stale(
assert server.data_path == str(bundled_dist)
def test_dashboard_falls_back_to_mismatched_data_dist_without_bundled(
core_lifecycle_td: AstrBotCoreLifecycle,
monkeypatch,
tmp_path,
):
data_dir = tmp_path / "data"
user_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
(user_dist / "assets").mkdir(parents=True)
(user_dist / "assets" / "version").write_text("v0.0.1", encoding="utf-8")
(user_dist / "index.html").write_text("stale", encoding="utf-8")
monkeypatch.setattr(
"astrbot.dashboard.server.get_astrbot_data_path",
lambda: str(data_dir),
)
monkeypatch.setattr(
"astrbot.dashboard.server.get_bundled_dashboard_dist_path",
lambda: bundled_dist,
)
shutdown_event = asyncio.Event()
server = AstrBotDashboard(core_lifecycle_td, core_lifecycle_td.db, shutdown_event)
assert server.data_path == str(user_dist)
def test_dashboard_ignores_incomplete_mismatched_data_dist_without_bundled(
core_lifecycle_td: AstrBotCoreLifecycle,
monkeypatch,
tmp_path,
):
data_dir = tmp_path / "data"
user_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
(user_dist / "assets").mkdir(parents=True)
(user_dist / "assets" / "version").write_text("v0.0.1", encoding="utf-8")
monkeypatch.setattr(
"astrbot.dashboard.server.get_astrbot_data_path",
lambda: str(data_dir),
)
monkeypatch.setattr(
"astrbot.dashboard.server.get_bundled_dashboard_dist_path",
lambda: bundled_dist,
)
shutdown_event = asyncio.Event()
server = AstrBotDashboard(core_lifecycle_td, core_lifecycle_td.db, shutdown_event)
assert server.data_path is None
async def _set_dashboard_password_change_required(
core_lifecycle_td: AstrBotCoreLifecycle,
required: bool,

View File

@@ -9,7 +9,7 @@ from unittest import mock
import pytest
from astrbot.core.utils.io import should_use_bundled_dashboard_dist
from astrbot.core.utils.io import get_dashboard_version, should_use_bundled_dashboard_dist
from main import (
DASHBOARD_RESET_PASSWORD_ENV,
_apply_startup_env_flags,
@@ -173,49 +173,146 @@ def test_version_info_comparisons():
@pytest.mark.asyncio
async def test_check_dashboard_files_not_exists(monkeypatch):
async def test_check_dashboard_files_not_exists(tmp_path):
"""Tests dashboard download when files do not exist."""
monkeypatch.setattr(os.path, "exists", lambda x: False)
data_dir = tmp_path / "data"
bundled_dist = tmp_path / "bundled-dist"
with mock.patch("main.download_dashboard") as mock_download:
await check_dashboard_files()
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch(
"main.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
with mock.patch("main.download_dashboard") as mock_download:
result = await check_dashboard_files()
from main import VERSION
assert result == str(data_dir / "dist")
mock_download.assert_called_once()
mock_download.assert_called_once_with(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
@pytest.mark.asyncio
async def test_check_dashboard_files_exists_and_version_match(monkeypatch):
async def test_check_dashboard_files_exists_and_version_match(tmp_path):
"""Tests that dashboard is not downloaded when it exists and version matches."""
# Mock os.path.exists to return True
monkeypatch.setattr(os.path, "exists", lambda x: True)
from main import VERSION
# Mock get_dashboard_version to return the current version
with mock.patch("main.get_dashboard_version") as mock_get_version:
# We need to import VERSION from main's context
from main import VERSION
mock_get_version.return_value = f"v{VERSION}"
data_dir = tmp_path / "data"
data_dist = data_dir / "dist"
(data_dist / "assets").mkdir(parents=True)
(data_dist / "assets" / "version").write_text(f"v{VERSION}", encoding="utf-8")
(data_dist / "index.html").write_text("user", encoding="utf-8")
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch("main.download_dashboard") as mock_download:
await check_dashboard_files()
# Assert that download_dashboard was NOT called
result = await check_dashboard_files()
assert result == str(data_dist)
mock_download.assert_not_called()
@pytest.mark.asyncio
async def test_check_dashboard_files_exists_but_version_mismatch(monkeypatch):
"""Tests that a warning is logged when dashboard version mismatches."""
monkeypatch.setattr(os.path, "exists", lambda x: True)
async def test_check_dashboard_files_exists_but_version_mismatch_downloads(tmp_path):
"""Tests that a mismatched dashboard is downloaded on startup."""
from main import VERSION
with mock.patch(
"main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
):
with mock.patch("main.logger.warning") as mock_logger_warning:
await check_dashboard_files()
data_dir = tmp_path / "data"
data_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
(data_dist / "assets").mkdir(parents=True)
(data_dist / "assets" / "version").write_text("v0.0.1", encoding="utf-8")
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch(
"main.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
with mock.patch("main.download_dashboard") as mock_download:
with mock.patch("main.logger.warning") as mock_logger_warning:
result = await check_dashboard_files()
assert result == str(data_dist)
mock_download.assert_called_once_with(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
mock_logger_warning.assert_called_once()
call_args, _ = mock_logger_warning.call_args
assert "WebUI version mismatch" in call_args[0]
@pytest.mark.asyncio
async def test_check_dashboard_files_falls_back_to_stale_dist_when_download_fails(
tmp_path,
):
"""Tests stale dashboard fallback when the matching WebUI cannot be downloaded."""
from main import VERSION
data_dir = tmp_path / "data"
data_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
(data_dist / "assets").mkdir(parents=True)
(data_dist / "assets" / "version").write_text("v0.0.1", encoding="utf-8")
(data_dist / "index.html").write_text("stale", encoding="utf-8")
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch(
"main.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
with mock.patch(
"main.download_dashboard",
side_effect=RuntimeError("missing dashboard asset"),
) as mock_download:
with mock.patch("main.logger.warning") as mock_logger_warning:
result = await check_dashboard_files()
assert result == str(data_dist)
mock_download.assert_called_once_with(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
assert any(
"Falling back to existing data/dist WebUI" in call.args[0]
for call in mock_logger_warning.call_args_list
)
@pytest.mark.asyncio
async def test_check_dashboard_files_downloads_when_matching_dist_is_incomplete(
tmp_path,
):
"""Tests that a version match alone is not enough to serve WebUI."""
from main import VERSION
data_dir = tmp_path / "data"
data_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
(data_dist / "assets").mkdir(parents=True)
(data_dist / "assets" / "version").write_text(f"v{VERSION}", encoding="utf-8")
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch(
"main.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
with mock.patch("main.download_dashboard") as mock_download:
result = await check_dashboard_files()
assert result == str(data_dist)
mock_download.assert_called_once_with(
version=f"v{VERSION}",
latest=False,
allow_insecure_ssl_fallback=False,
)
def test_should_use_bundled_dashboard_dist_when_data_dist_is_stale(tmp_path):
user_dist = tmp_path / "user-dist"
bundled_dist = tmp_path / "bundled-dist"
@@ -223,6 +320,7 @@ def test_should_use_bundled_dashboard_dist_when_data_dist_is_stale(tmp_path):
(bundled_dist / "assets").mkdir(parents=True)
(user_dist / "assets" / "version").write_text("v4.24.2", encoding="utf-8")
(bundled_dist / "assets" / "version").write_text("v4.24.4", encoding="utf-8")
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
with mock.patch(
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
@@ -231,46 +329,94 @@ def test_should_use_bundled_dashboard_dist_when_data_dist_is_stale(tmp_path):
assert should_use_bundled_dashboard_dist(user_dist, "v4.24.4") is True
def test_should_keep_data_dist_when_version_file_is_malformed(tmp_path):
def test_should_use_bundled_dashboard_dist_when_version_file_is_malformed(tmp_path):
user_dist = tmp_path / "user-dist"
bundled_dist = tmp_path / "bundled-dist"
(user_dist / "assets").mkdir(parents=True)
(bundled_dist / "assets").mkdir(parents=True)
(user_dist / "assets" / "version").write_text("not-a-version", encoding="utf-8")
(bundled_dist / "assets" / "version").write_text("v4.24.4", encoding="utf-8")
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
with mock.patch(
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
assert should_use_bundled_dashboard_dist(user_dist, "4.24.4") is False
assert should_use_bundled_dashboard_dist(user_dist, "4.24.4") is True
def test_should_use_bundled_dashboard_dist_when_data_version_file_is_missing(tmp_path):
user_dist = tmp_path / "user-dist"
bundled_dist = tmp_path / "bundled-dist"
(user_dist / "assets").mkdir(parents=True)
(bundled_dist / "assets").mkdir(parents=True)
(bundled_dist / "assets" / "version").write_text("v4.24.4", encoding="utf-8")
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
with mock.patch(
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
assert should_use_bundled_dashboard_dist(user_dist, "4.24.4") is True
@pytest.mark.asyncio
async def test_check_dashboard_files_uses_bundled_dist_when_data_dist_is_stale(
async def test_get_dashboard_version_uses_bundled_dist_when_data_dist_is_missing(
tmp_path,
):
"""Tests that a stale data/dist does not override bundled dashboard assets."""
"""Tests bundled WebUI version lookup when data/dist is absent."""
from main import VERSION
data_dir = tmp_path / "data"
bundled_dist = tmp_path / "bundled-dist"
(bundled_dist / "assets").mkdir(parents=True)
(bundled_dist / "assets" / "version").write_text(f"v{VERSION}", encoding="utf-8")
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
with mock.patch(
"astrbot.core.utils.io.get_astrbot_data_path",
return_value=str(data_dir),
):
with mock.patch(
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
return_value=bundled_dist,
):
assert await get_dashboard_version() == f"v{VERSION}"
@pytest.mark.asyncio
async def test_check_dashboard_files_replaces_stale_data_dist_with_bundled_dist(
tmp_path,
):
"""Tests that a stale data/dist is repaired from bundled dashboard assets."""
from main import VERSION
data_dir = tmp_path / "data"
data_dist = data_dir / "dist"
bundled_dist = tmp_path / "bundled-dist"
data_dist.mkdir(parents=True)
bundled_dist.mkdir()
(data_dist / "assets").mkdir(parents=True)
(bundled_dist / "assets").mkdir(parents=True)
(data_dist / "assets" / "version").write_text("v0.0.1", encoding="utf-8")
(data_dist / "old.txt").write_text("old", encoding="utf-8")
(bundled_dist / "assets" / "version").write_text(f"v{VERSION}", encoding="utf-8")
(bundled_dist / "index.html").write_text("bundled", encoding="utf-8")
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
with mock.patch(
"main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
"main.get_bundled_dashboard_dist_path",
return_value=Path(bundled_dist),
):
with mock.patch(
"main.should_use_bundled_dashboard_dist", return_value=True
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
return_value=Path(bundled_dist),
):
with mock.patch(
"main.get_bundled_dashboard_dist_path",
return_value=Path(bundled_dist),
):
with mock.patch("main.download_dashboard") as mock_download:
result = await check_dashboard_files()
with mock.patch("main.download_dashboard") as mock_download:
result = await check_dashboard_files()
assert result == str(bundled_dist)
assert result == str(data_dist)
assert (data_dist / "assets" / "version").read_text(encoding="utf-8") == f"v{VERSION}"
assert (data_dist / "index.html").read_text(encoding="utf-8") == "bundled"
assert not (data_dist / "old.txt").exists()
mock_download.assert_not_called()
@@ -281,7 +427,7 @@ async def test_check_dashboard_files_with_webui_dir_arg(monkeypatch):
monkeypatch.setattr(os.path, "exists", lambda path: path == valid_dir)
with mock.patch("main.download_dashboard") as mock_download:
with mock.patch("main.get_dashboard_version") as mock_get_version:
with mock.patch("main.get_dashboard_dist_version") as mock_get_version:
result = await check_dashboard_files(webui_dir=valid_dir)
assert result == valid_dir
mock_download.assert_not_called()

View File

@@ -440,6 +440,7 @@ async def test_download_dashboard_falls_back_when_hosted_package_is_not_zip(
path: str,
show_progress: bool = False, # noqa: ARG001
progress_callback=None, # noqa: ARG001
allow_insecure_ssl_fallback: bool = True, # noqa: ARG001
) -> None:
calls.append(url)
parsed = urlparse(url)

View File

@@ -8,6 +8,7 @@ import pytest
from astrbot.core import astr_main_agent as ama
from astrbot.core.agent.mcp_client import MCPTool
from astrbot.core.agent.message import Message, dump_messages_with_checkpoints
from astrbot.core.agent.tool import FunctionTool, ToolSet
from astrbot.core.conversation_mgr import Conversation
from astrbot.core.message.components import File, Image, Plain, Reply, Video
@@ -377,8 +378,18 @@ class TestApplyKb:
):
await module._apply_kb(mock_event, req, mock_context, config)
assert "[Related Knowledge Base Results]:" in req.system_prompt
assert "KB result" in req.system_prompt
assert req.system_prompt == "System prompt"
assert len(req.extra_user_content_parts) == 1
kb_part = req.extra_user_content_parts[0]
assert kb_part.text == "[Related Knowledge Base Results]:\nKB result"
message = Message.model_validate(await req.assemble_context())
assert isinstance(message.content, list)
assert message.content[0].text == "test question"
assert message.content[1].text == "[Related Knowledge Base Results]:\nKB result"
assert dump_messages_with_checkpoints([message]) == [
{"role": "user", "content": [{"type": "text", "text": "test question"}]}
]
@pytest.mark.asyncio
async def test_apply_kb_with_agentic_mode(self, mock_event, mock_context):
@@ -998,7 +1009,7 @@ class TestEnsurePersonaAndSkills:
assert req.func_tool is not None
@pytest.mark.asyncio
async def test_persona_empty_tools_filters_late_builtin_tools(
async def test_persona_empty_tools_keeps_late_builtin_tools(
self, mock_event, mock_context, mock_provider
):
module = ama
@@ -1006,6 +1017,7 @@ class TestEnsurePersonaAndSkills:
mock_context.persona_manager.resolve_selected_persona = AsyncMock(
return_value=("locked", persona, None, False)
)
mock_event.platform_meta.support_proactive_message = False
mock_context.get_config.return_value = {
"provider_settings": {
"web_search": True,
@@ -1019,6 +1031,7 @@ class TestEnsurePersonaAndSkills:
"websearch_provider": "baidu_ai_search",
},
computer_use_runtime="none",
add_cron_tools=False,
)
req = ProviderRequest(prompt="hello")
req.conversation = MagicMock(persona_id="locked", history="[]")
@@ -1041,9 +1054,52 @@ class TestEnsurePersonaAndSkills:
)
assert result is not None
try:
assert result.provider_request.func_tool is None or (
result.provider_request.func_tool.empty()
assert result.provider_request.func_tool is not None
assert result.provider_request.func_tool.names() == ["web_search_baidu"]
finally:
if result.reset_coro:
result.reset_coro.close()
@pytest.mark.asyncio
async def test_persona_empty_tools_keeps_local_runtime_builtin_tools(
self, mock_event, mock_context, mock_provider
):
module = ama
persona = {"name": "locked", "prompt": "No tools.", "tools": []}
mock_context.persona_manager.resolve_selected_persona = AsyncMock(
return_value=("locked", persona, None, False)
)
mock_event.platform_meta.support_proactive_message = False
config = module.MainAgentBuildConfig(
tool_call_timeout=60,
computer_use_runtime="local",
add_cron_tools=False,
)
req = ProviderRequest(prompt="hello")
req.conversation = MagicMock(persona_id="locked", history="[]")
with (
patch("astrbot.core.astr_main_agent.AgentRunner") as mock_runner_cls,
patch("astrbot.core.astr_main_agent.AstrAgentContext"),
):
mock_runner = MagicMock()
mock_runner.reset = AsyncMock()
mock_runner_cls.return_value = mock_runner
result = await module.build_main_agent(
event=mock_event,
plugin_context=mock_context,
config=config,
provider=mock_provider,
req=req,
apply_reset=False,
)
assert result is not None
try:
assert result.provider_request.func_tool is not None
tool_names = result.provider_request.func_tool.names()
assert "astrbot_execute_shell" in tool_names
assert "astrbot_execute_python" in tool_names
finally:
if result.reset_coro:
result.reset_coro.close()