diff --git a/astrbot/api/__init__.py b/astrbot/api/__init__.py index 016633ce03..a5d5909256 100644 --- a/astrbot/api/__init__.py +++ b/astrbot/api/__init__.py @@ -4,6 +4,12 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: + from astrbot.core.agent.btw.runtime_registry import ( + latest_status as btw_work_latest_status, + ) + from astrbot.core.agent.btw.types import ( + is_work_loop_enabled as btw_work_loop_enabled, + ) from astrbot.core.agent.tool import FunctionTool, ToolSet from astrbot.core.agent.tool_executor import BaseFunctionToolExecutor from astrbot.core.auth import AuthContext, Decision, Resource, Role, Subject @@ -13,6 +19,14 @@ from astrbot.core.utils.error_redaction import safe_error _EXPORTS = { + "btw_work_latest_status": ( + "astrbot.core.agent.btw.runtime_registry", + "latest_status", + ), + "btw_work_loop_enabled": ( + "astrbot.core.agent.btw.types", + "is_work_loop_enabled", + ), "AuthContext": ("astrbot.core.auth", "AuthContext"), "Decision": ("astrbot.core.auth", "Decision"), "Resource": ("astrbot.core.auth", "Resource"), @@ -78,6 +92,8 @@ def __getattr__(self, item: str): "Subject", "ToolSet", "agent", + "btw_work_loop_enabled", + "btw_work_latest_status", "llm_tool", "logger", "safe_error", diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json index 1eaca3b639..67eb7516d6 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json @@ -4,6 +4,13 @@ "desc": "AstrBot built-in session, conversation, provider, persona, plugin, and bot commands." }, "commands": { + "work.disabled": "The BTW work loop is not enabled.", + "work.status.none": "No BTW work task has run in this session.", + "work.status.pending": "Queued: {task}", + "work.status.running": "Running: {task}", + "work.status.completed": "Completed: {task}", + "work.status.failed": "Failed: {task}", + "work.status.cancelled": "Cancelled: {task}", "help.header": "AstrBot v{version} (WebUI: {dashboard})", "help.empty": "No enabled built-in commands.", "help.tip": "Tip: use `/help --image` to render the visual help card.", diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json index 664d04944f..d916d0a564 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json @@ -4,6 +4,13 @@ "desc": "AstrBot 内置的会话、对话、Provider、Persona、插件与机器人指令。" }, "commands": { + "work.disabled": "BTW 工作循环尚未启用。", + "work.status.none": "本会话还没有 BTW 工作任务。", + "work.status.pending": "排队中：{task}", + "work.status.running": "执行中：{task}", + "work.status.completed": "已完成：{task}", + "work.status.failed": "已失败：{task}", + "work.status.cancelled": "已取消：{task}", "help.header": "AstrBot v{version}（WebUI：{dashboard}）", "help.empty": "当前没有已启用的内置指令。", "help.tip": "提示：使用 `/help --image` 可生成图片版帮助。", diff --git a/astrbot/builtin_stars/builtin_commands/commands/__init__.py b/astrbot/builtin_stars/builtin_commands/commands/__init__.py index d665dbf916..d46e2deca5 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/__init__.py +++ b/astrbot/builtin_stars/builtin_commands/commands/__init__.py @@ -11,6 +11,7 @@ from .provider import ProviderCommands from .session import SessionCommands from .variable import VariableCommands +from .work import WorkCommands __all__ = [ "AdminCommands", @@ -24,4 +25,5 @@ "ProviderCommands", "SessionCommands", "VariableCommands", + "WorkCommands", ] diff --git a/astrbot/builtin_stars/builtin_commands/commands/work.py b/astrbot/builtin_stars/builtin_commands/commands/work.py new file mode 100644 index 0000000000..676bd5a8e4 --- /dev/null +++ b/astrbot/builtin_stars/builtin_commands/commands/work.py @@ -0,0 +1,43 @@ +"""Explicit submission of free-text tasks to the BTW work loop.""" + +from astrbot.api import btw_work_latest_status, btw_work_loop_enabled +from astrbot.api.event import AstrMessageEvent + +from .reply import reply_i18n + + +class WorkCommands: + """The built-in work-loop command surface.""" + + def __init__(self, context) -> None: + self.context = context + + async def handle(self, event: AstrMessageEvent, task: str = "") -> None: + """Query status for empty input or ``status``, otherwise submit a task.""" + stripped = (task or "").strip() + if not stripped or stripped.lower() == "status": + await self.status(event) + return + await self.submit(event, stripped) + + async def status(self, event: AstrMessageEvent) -> None: + """Show the latest task for the command's profile and message origin.""" + config_id = getattr(getattr(event, "resource", None), "config_id", "") or "" + latest = await btw_work_latest_status(config_id, event.unified_msg_origin) + if latest is None: + await reply_i18n(self.context, event, "work.status.none") + return + request, status = latest + await reply_i18n(self.context, event, f"work.status.{status}", task=request) + + async def submit(self, event: AstrMessageEvent, task: str) -> None: + """Continue the admitted command event through the work loop.""" + config = self.context.config.get(umo=event.unified_msg_origin) + if not btw_work_loop_enabled(config): + await reply_i18n(self.context, event, "work.disabled") + return + event.message_str = task + event.set_extra("should_run_command", False) + event.set_extra("should_run_llm", True) + event.set_extra("btw_force_work", True) + event.set_extra("btw_loop", "work") diff --git a/astrbot/builtin_stars/builtin_commands/main.py b/astrbot/builtin_stars/builtin_commands/main.py index 87e566a10c..b112fc5e5b 100644 --- a/astrbot/builtin_stars/builtin_commands/main.py +++ b/astrbot/builtin_stars/builtin_commands/main.py @@ -16,6 +16,7 @@ ProviderCommands, SessionCommands, VariableCommands, + WorkCommands, ) @@ -34,6 +35,7 @@ def __init__(self, context: star.PluginContext) -> None: self.provider_c = ProviderCommands(self.context) self.session_c = SessionCommands(self.context) self.variable_c = VariableCommands(self.context) + self.work_c = WorkCommands(self.context) @filter.command("help") async def help( @@ -108,6 +110,16 @@ async def conversation_reset(self, message: AstrMessageEvent) -> None: def task(self) -> None: """Manage running tasks""" + @filter.permission("session.read") + @filter.command("work") + async def work( + self, + event: AstrMessageEvent, + task: GreedyStr = GreedyStr(""), + ) -> None: + """Submit a BTW work task, or show the latest status""" + await self.work_c.handle(event, task) + @filter.permission("session.manage") @task.command("stop") async def task_stop(self, message: AstrMessageEvent) -> None: diff --git a/astrbot/core/agent/btw/__init__.py b/astrbot/core/agent/btw/__init__.py new file mode 100644 index 0000000000..71280bdf32 --- /dev/null +++ b/astrbot/core/agent/btw/__init__.py @@ -0,0 +1 @@ +# BTW runtime primitives; keep package imports inert. diff --git a/astrbot/core/agent/btw/i18n.py b/astrbot/core/agent/btw/i18n.py new file mode 100644 index 0000000000..33d5239da1 --- /dev/null +++ b/astrbot/core/agent/btw/i18n.py @@ -0,0 +1,50 @@ +"""Locale-aware user-facing strings for the BTW work loop. + +The work loop runs in core without a plugin context, so it resolves the +locale from the event extra/session the same way ``PluginContext._locale`` +does, then looks the string up in these bundles. Missing locales fall back +to ``zh-CN``. +""" + +LOCALES: dict[str, dict[str, str]] = { + "zh-CN": { + "btw.work.started": "🔧 工作任务已开始处理。", + "btw.work.status.pending": "工作任务正在排队。", + "btw.work.status.running": "工作任务正在执行。", + "btw.work.status.completed": "工作任务已完成。", + "btw.work.status.failed": "工作任务执行失败。", + "btw.work.status.cancelled": "工作任务已取消。", + }, + "en-US": { + "btw.work.started": "🔧 Work task started.", + "btw.work.status.pending": "The work task is queued.", + "btw.work.status.running": "The work task is running.", + "btw.work.status.completed": "The work task is completed.", + "btw.work.status.failed": "The work task failed.", + "btw.work.status.cancelled": "The work task was cancelled.", + }, +} + +_FALLBACK_LOCALE = "zh-CN" + + +def resolve_event_locale(event) -> str: + """Return the locale for an event (extra first, then the stored session).""" + getter = getattr(event, "get_extra", None) + if callable(getter): + try: + extra = getter("locale") + except Exception: # noqa: BLE001 + extra = None + if extra: + return str(extra) + return _FALLBACK_LOCALE + + +def text(locale: str, key: str) -> str: + """Return one BTW string for a locale, falling back to zh-CN then key.""" + bundle = LOCALES.get(locale) or LOCALES[_FALLBACK_LOCALE] + value = bundle.get(key) + if value is None: + value = LOCALES[_FALLBACK_LOCALE].get(key, key) + return value diff --git a/astrbot/core/agent/btw/loop_routes.py b/astrbot/core/agent/btw/loop_routes.py new file mode 100644 index 0000000000..895424d115 --- /dev/null +++ b/astrbot/core/agent/btw/loop_routes.py @@ -0,0 +1,38 @@ +"""Resolve capability assignments shared by BTW request paths.""" + + +def route_is_available_in_loop( + routes: object, + *, + route_key: str, + route_id: str, + loop_mode: str, + default_loop: str = "both", +) -> bool: + """Resolve a list assignment, falling back to the capability's default. + + Args: + routes: List of dictionaries containing the capability key and ``loop``. + route_key: Key identifying a capability in an assignment. + route_id: Capability identifier to look up. + loop_mode: Current loop; missing or invalid values mean conversation. + default_loop: Assignment used for missing or malformed entries. + + Returns: + Whether the capability is available in the current loop. + """ + loop_mode = "work" if loop_mode == "work" else "conversation" + route = default_loop + if isinstance(routes, list): + for entry in routes: + if not isinstance(entry, dict) or entry.get(route_key) != route_id: + continue + candidate = entry.get("loop") + if isinstance(candidate, str) and candidate in { + "conversation", + "work", + "both", + }: + route = candidate + break + return route in {"both", loop_mode} diff --git a/astrbot/core/agent/btw/runtime_policy.py b/astrbot/core/agent/btw/runtime_policy.py new file mode 100644 index 0000000000..ffc1c942ea --- /dev/null +++ b/astrbot/core/agent/btw/runtime_policy.py @@ -0,0 +1,35 @@ +"""Computer runtime selection for BTW requests and their handoffs.""" + +from collections.abc import Mapping + + +def resolve_computer_runtime( + profile: Mapping, + loop: object, + inherited: str, +) -> str: + """Resolve the runtime without granting any tool authorization. + + Args: + profile: The current configuration profile. + loop: The event's loop marker; only ``work`` selects the work loop. + inherited: The runtime selected before applying BTW settings. + + Returns: + The inherited runtime when BTW is disabled, otherwise the permitted + runtime for this loop. + """ + btw = profile.get("btw", {}) + if not isinstance(btw, Mapping) or not btw.get("enabled", False): + return inherited + if loop != "work": + return "none" + work = btw.get("work_loop", {}) + runtime = ( + work.get("computer_use_runtime", "inherit") + if isinstance(work, Mapping) + else "inherit" + ) + if runtime not in ("none", "local", "sandbox"): + runtime = inherited + return runtime if runtime in ("none", "local", "sandbox") else "none" diff --git a/astrbot/core/agent/btw/runtime_registry.py b/astrbot/core/agent/btw/runtime_registry.py new file mode 100644 index 0000000000..0ab5ec1a82 --- /dev/null +++ b/astrbot/core/agent/btw/runtime_registry.py @@ -0,0 +1,38 @@ +"""Expose each pipeline's most recent work status to built-in commands.""" + +from .types import WorkSessionStatus +from .work_sessions import WorkSessionManager + +_managers: dict[str, WorkSessionManager] = {} + + +def register(config_id: str, manager: WorkSessionManager) -> None: + """Bind a successfully initialized pipeline's work-session manager.""" + _managers[config_id] = manager + + +def unregister(config_id: str, manager: WorkSessionManager) -> None: + """Remove only the closing pipeline's registration.""" + if _managers.get(config_id) is manager: + _managers.pop(config_id) + + +async def latest_status( + config_id: str, origin: str +) -> tuple[str, WorkSessionStatus] | None: + """Read the newest work task within the specified profile and origin. + + Args: + config_id: The profile that owns the admitted command event. + origin: The event's unified message origin. + + Returns: + The task text and status, or ``None`` when no retained task exists. + """ + manager = _managers.get(config_id) + if manager is None: + return None + session = await manager.get_for_origin(origin) + if session is None: + return None + return session.request, session.status diff --git a/astrbot/core/agent/btw/types.py b/astrbot/core/agent/btw/types.py new file mode 100644 index 0000000000..ca600b8d87 --- /dev/null +++ b/astrbot/core/agent/btw/types.py @@ -0,0 +1,65 @@ +"""Types shared by the BTW conversation and work loops.""" + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import UTC, datetime +from enum import StrEnum +from uuid import uuid4 + + +def is_work_loop_enabled(config: object) -> bool: + """Return whether the profile explicitly enables BTW and work.""" + if not isinstance(config, Mapping): + return False + btw = config.get("btw", {}) + if not isinstance(btw, Mapping) or not btw.get("enabled", False): + return False + work = btw.get("work_loop", {}) + return isinstance(work, Mapping) and bool(work.get("enabled", False)) + + +class TaskType(StrEnum): + """The execution loop selected for a user request.""" + + CONVERSATION = "conversation" + WORK = "work" + + +class WorkSessionStatus(StrEnum): + """Lifecycle states for one work-loop request.""" + + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass(slots=True) +class WorkSession: + """Runtime state shared by the conversation and work loops.""" + + origin: str + request: str + task_type: TaskType = TaskType.WORK + id: str = field(default_factory=lambda: uuid4().hex) + status: WorkSessionStatus = WorkSessionStatus.PENDING + created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + updated_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + error: str | None = None + + def update_status( + self, + status: WorkSessionStatus, + *, + error: str | None = None, + ) -> None: + """Record a status transition. + + Args: + status: The new work-session status. + error: A safe diagnostic for failed work, when available. + """ + self.status = status + self.error = error + self.updated_at = datetime.now(UTC) diff --git a/astrbot/core/agent/btw/work_loop.py b/astrbot/core/agent/btw/work_loop.py new file mode 100644 index 0000000000..24eda58906 --- /dev/null +++ b/astrbot/core/agent/btw/work_loop.py @@ -0,0 +1,233 @@ +"""The BTW work-loop prototype backed by the existing Agent tool loop.""" + +import asyncio +from collections.abc import AsyncGenerator, Awaitable, Callable +from contextlib import aclosing +from typing import Protocol + +from astrbot.core.message.message_event_result import MessageEventResult +from astrbot.core.platform.astr_message_event import AstrMessageEvent +from astrbot.core.utils.task_utils import create_tracked_task + +from . import i18n as work_i18n +from .types import WorkSessionStatus +from .work_sessions import WorkSessionManager + + +class AgentRequestExecutor(Protocol): + """The existing Agent request path required by the work loop.""" + + def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: + """Yield pipeline progress markers for one event. + + Protocol stub; concrete implementations are the pipeline's Agent + request sub-stage. The body raises so the statement is effectful + (CodeQL py/ineffectual-statement); the unreachable ``yield`` keeps + the declared ``AsyncGenerator`` return type type-checkable. + """ + raise NotImplementedError + yield # noqa: B901 -- unreachable marker for the type checker + + +ResultDispatcher = Callable[[AstrMessageEvent], Awaitable[None]] +EventFinalizer = Callable[[AstrMessageEvent], Awaitable[None]] + + +class WorkLoop: + """Run classified work with the current Agent and tool infrastructure.""" + + def __init__( + self, + executor: AgentRequestExecutor, + sessions: WorkSessionManager, + *, + max_concurrent: int = 2, + ) -> None: + self.executor = executor + self.sessions = sessions + self._semaphore = asyncio.Semaphore(max(1, max_concurrent)) + self._background_tasks: set[asyncio.Task] | None = None + self._result_dispatcher: ResultDispatcher | None = None + self._event_finalizer: EventFinalizer | None = None + self._tasks: dict[asyncio.Task, tuple[AstrMessageEvent, str]] = {} + self._closed = False + + def configure_detached_execution( + self, + *, + background_tasks: set[asyncio.Task], + result_dispatcher: ResultDispatcher, + event_finalizer: EventFinalizer, + ) -> None: + """Attach runtime-owned background execution services. + + Args: + background_tasks: Runtime task registry cancelled during shutdown. + result_dispatcher: Delivers a generated work result through the + configured result-decorate and response stages. + event_finalizer: Releases the event after detached work finishes. + """ + self._background_tasks = background_tasks + self._result_dispatcher = result_dispatcher + self._event_finalizer = event_finalizer + + async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: + """Execute one work-loop request inline. + + Args: + event: The classified message event. + + Yields: + Pipeline progress markers emitted by the existing Agent executor. + """ + session = await self.sessions.create( + event.unified_msg_origin, event.message_str + ) + self._prepare_event(event, session.id) + async for progress in self._execute(event, session.id): + yield progress + + async def submit(self, event: AstrMessageEvent) -> AsyncGenerator[None]: + """Acknowledge work, then run it without retaining the request pipeline. + + Falls back to inline execution when no runtime task registry is + attached, which keeps the primitive usable in isolated tests. + """ + if self._closed: + event.set_result( + MessageEventResult().message( + work_i18n.text( + work_i18n.resolve_event_locale(event), + "btw.work.status.cancelled", + ) + ) + ) + yield + return + if ( + self._background_tasks is None + or self._result_dispatcher is None + or self._event_finalizer is None + ): + async for progress in self.process(event): + yield progress + return + + session = await self.sessions.create( + event.unified_msg_origin, event.message_str + ) + self._prepare_event(event, session.id) + event.set_result( + MessageEventResult().message( + work_i18n.text( + work_i18n.resolve_event_locale(event), "btw.work.started" + ) + ) + ) + yield + + if self._closed: + await self.sessions.update_status(session.id, WorkSessionStatus.CANCELLED) + return + + # The first yield returns only after the normal response stages deliver + # the acknowledgement. Marking it here prevents the scheduler from + # releasing event-owned temporary files before the worker needs them. + event.set_extra("btw_detached_work", True) + task = create_tracked_task( + self._background_tasks, + self._run_detached(event, session.id), + name=f"btw_work:{session.id}", + ) + self._tasks[task] = (event, session.id) + task.add_done_callback(lambda done: self._tasks.pop(done, None)) + + async def close(self) -> None: + """Cancel and finalize this profile's work, including unstarted tasks.""" + self._closed = True + tasks = dict(self._tasks) + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + finalizers = [] + if self._event_finalizer is not None: + for event, session_id in tasks.values(): + if not event.get_extra("btw_detached_work_finished"): + await self.sessions.update_status( + session_id, WorkSessionStatus.CANCELLED + ) + finalizers.append(self._event_finalizer(event)) + if finalizers: + results = await asyncio.gather(*finalizers, return_exceptions=True) + for result in results: + if isinstance(result, BaseException): + raise result + + @staticmethod + def _prepare_event(event: AstrMessageEvent, session_id: str) -> None: + """Mark an event so Agent assembly uses the work-loop policy.""" + event.set_extra("btw_work_session_id", session_id) + event.set_extra("btw_loop", "work") + event.set_extra("btw_agent_lock_key", f"{event.unified_msg_origin}:work") + + async def _execute( + self, + event: AstrMessageEvent, + session_id: str, + ) -> AsyncGenerator[None]: + """Run one already-created work session and update its lifecycle.""" + try: + async with self._semaphore: + await self.sessions.update_status( + session_id, + WorkSessionStatus.RUNNING, + ) + async for progress in self.executor.process(event): + yield progress + except asyncio.CancelledError: + await self.sessions.update_status( + session_id, + WorkSessionStatus.CANCELLED, + ) + raise + except Exception: + await self.sessions.update_status( + session_id, + WorkSessionStatus.FAILED, + error="Work task failed.", + ) + raise + else: + failed = bool(event.get_extra("btw_work_failed")) + cancelled = bool(event.get_extra("agent_stop_requested")) + await self.sessions.update_status( + session_id, + WorkSessionStatus.FAILED + if failed + else ( + WorkSessionStatus.CANCELLED + if cancelled + else WorkSessionStatus.COMPLETED + ), + error="Work task failed." if failed else None, + ) + + async def _run_detached(self, event: AstrMessageEvent, session_id: str) -> None: + """Run work in the runtime task registry and deliver each result.""" + assert self._result_dispatcher is not None + assert self._event_finalizer is not None + try: + async with aclosing(self._execute(event, session_id)) as execution: + async for _ in execution: + await self._result_dispatcher(event) + except asyncio.CancelledError: + await self.sessions.update_status(session_id, WorkSessionStatus.CANCELLED) + raise + except Exception: + await self.sessions.update_status( + session_id, WorkSessionStatus.FAILED, error="Work task failed." + ) + raise + finally: + await self._event_finalizer(event) diff --git a/astrbot/core/agent/btw/work_sessions.py b/astrbot/core/agent/btw/work_sessions.py new file mode 100644 index 0000000000..7c919f9efc --- /dev/null +++ b/astrbot/core/agent/btw/work_sessions.py @@ -0,0 +1,106 @@ +"""In-memory runtime ownership for BTW work sessions.""" + +import asyncio +from datetime import UTC, datetime, timedelta + +from .types import WorkSession, WorkSessionStatus + + +class WorkSessionManager: + """Own active and recent work-loop session state for one pipeline. + + ``get_for_origin`` returns the *latest* session per origin: a new task + replaces the previous session's status-query target, while older + sessions stay addressable by id until they expire. Work-loop + concurrency is bounded by the work loop's semaphore, not here. + """ + + def __init__(self, *, max_age_seconds: int = 3600) -> None: + self._by_origin: dict[str, WorkSession] = {} + self._by_id: dict[str, WorkSession] = {} + self._lock = asyncio.Lock() + self.set_max_age_seconds(max_age_seconds) + + def set_max_age_seconds(self, value: int) -> None: + """Set the retention period for terminal sessions. + + Args: + value: Number of seconds to keep a completed, failed, or cancelled + session before a later manager operation removes it. + """ + self._max_age_seconds = max(1, value) if type(value) is int else 3600 + + async def create(self, origin: str, request: str) -> WorkSession: + """Create and register a work session. + + Args: + origin: The unified message origin that owns the work. + request: The user request being processed. + + Returns: + The newly created work session. + """ + session = WorkSession(origin=origin, request=request) + async with self._lock: + self._cleanup_expired_locked() + self._by_origin[origin] = session + self._by_id[session.id] = session + return session + + async def get_for_origin(self, origin: str) -> WorkSession | None: + """Return the most recent work session for an origin.""" + async with self._lock: + self._cleanup_expired_locked() + return self._by_origin.get(origin) + + async def get_by_id(self, session_id: str) -> WorkSession | None: + """Return a recent work session by its identifier. + + Args: + session_id: The generated work-session identifier. + + Returns: + The matching session, or ``None`` after it has expired. + """ + async with self._lock: + self._cleanup_expired_locked() + return self._by_id.get(session_id) + + async def update_status( + self, + session_id: str, + status: WorkSessionStatus, + *, + error: str | None = None, + ) -> WorkSession | None: + """Transition one known work session. + + Args: + session_id: The work-session identifier. + status: The new lifecycle status. + error: A safe failure message, when applicable. + + Returns: + The updated session, or ``None`` when it has expired. + """ + async with self._lock: + self._cleanup_expired_locked() + session = self._by_id.get(session_id) + if session is not None: + session.update_status(status, error=error) + return session + + def _cleanup_expired_locked(self) -> None: + """Remove old terminal sessions while the manager lock is held.""" + cutoff = datetime.now(UTC) - timedelta(seconds=self._max_age_seconds) + expired_ids = [ + session_id + for session_id, session in self._by_id.items() + if session.status + not in {WorkSessionStatus.PENDING, WorkSessionStatus.RUNNING} + and session.updated_at < cutoff + ] + for session_id in expired_ids: + session = self._by_id.pop(session_id) + if self._by_origin.get(session.origin) is session: + self._by_origin.pop(session.origin, None) diff --git a/astrbot/core/agent/conversation_loop.py b/astrbot/core/agent/conversation_loop.py index 237befeca5..c5d6a4eddf 100644 --- a/astrbot/core/agent/conversation_loop.py +++ b/astrbot/core/agent/conversation_loop.py @@ -1,8 +1,12 @@ """Opt-in conversation entry over the existing Agent request executor.""" -from collections.abc import AsyncGenerator +import asyncio +from collections.abc import AsyncGenerator, Awaitable, Callable from typing import TYPE_CHECKING +from astrbot.core.agent.btw.types import is_work_loop_enabled +from astrbot.core.agent.btw.work_loop import WorkLoop +from astrbot.core.agent.btw.work_sessions import WorkSessionManager from astrbot.core.platform.astr_message_event import AstrMessageEvent if TYPE_CHECKING: @@ -18,6 +22,8 @@ class ConversationLoop: def __init__(self, agent_request: AgentRequestSubStage) -> None: self.agent_request = agent_request self._btw_enabled = False + self.work_sessions = WorkSessionManager() + self.work_loop: WorkLoop | None = None async def initialize(self, ctx: PipelineContext) -> None: """Initialize the shared Agent executor for this profile.""" @@ -25,9 +31,52 @@ async def initialize(self, ctx: PipelineContext) -> None: btw = self.astrbot_config.get("btw", {}) self._btw_enabled = isinstance(btw, dict) and bool(btw.get("enabled", False)) await self.agent_request.initialize(ctx) + btw = btw if isinstance(btw, dict) else {} + work = btw.get("work_loop", {}) + work = work if isinstance(work, dict) else {} + retention = btw.get("work_session", {}) + retention = retention if isinstance(retention, dict) else {} + self.work_sessions.set_max_age_seconds(retention.get("max_age_seconds", 3600)) + concurrency = work.get("max_concurrent", 2) + self.work_loop = WorkLoop( + self.agent_request, + self.work_sessions, + max_concurrent=concurrency if type(concurrency) is int else 2, + ) + + def configure_detached_work( + self, + *, + background_tasks: set[asyncio.Task], + result_dispatcher: Callable[[AstrMessageEvent], Awaitable[None]], + event_finalizer: Callable[[AstrMessageEvent], Awaitable[None]], + ) -> None: + """Attach the owning scheduler's delivery and cleanup services.""" + if self.work_loop is None: + raise RuntimeError("ConversationLoop is not initialized") + self.work_loop.configure_detached_execution( + background_tasks=background_tasks, + result_dispatcher=result_dispatcher, + event_finalizer=event_finalizer, + ) + + async def close(self) -> None: + """Stop work owned by this conversation entry.""" + if self.work_loop is not None: + await self.work_loop.close() async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: """Process one admitted conversation using the current Agent path.""" + if ( + self._btw_enabled + and event.get_extra("btw_force_work") + and is_work_loop_enabled(self.astrbot_config) + ): + if self.work_loop is None: + raise RuntimeError("ConversationLoop is not initialized") + async for response in self.work_loop.submit(event): + yield response + return if self._btw_enabled: event.set_extra("btw_loop", "conversation") async for response in self.agent_request.process(event): diff --git a/astrbot/core/astr_agent_tool_exec.py b/astrbot/core/astr_agent_tool_exec.py index 1da4caeebe..b0af8470b5 100644 --- a/astrbot/core/astr_agent_tool_exec.py +++ b/astrbot/core/astr_agent_tool_exec.py @@ -11,6 +11,7 @@ import mcp from astrbot import logger +from astrbot.core.agent.btw.runtime_policy import resolve_computer_runtime from astrbot.core.agent.handoff import HandoffTool from astrbot.core.agent.llm_types import ProviderRequest from astrbot.core.agent.mcp_client import MCPTool @@ -31,6 +32,7 @@ MessageEventResult, ) from astrbot.core.platform.message_session import MessageSession +from astrbot.core.tool_catalog import tool_is_available_in_loop from astrbot.core.tools.computer_tools import ( CuaKeyboardTypeTool, CuaMouseClickTool, @@ -300,6 +302,47 @@ def _get_runtime_computer_tools( } return {} + @staticmethod + def _filter_handoff_tools_for_loop(toolset: ToolSet, *, cfg, ctx, event) -> ToolSet: + """Keep handoff tools within the originating loop's assignments.""" + btw_config = cfg.get("btw", {}) + if not isinstance(btw_config, dict) or not btw_config.get("enabled", False): + return toolset + plugins = getattr(getattr(ctx, "catalogs", None), "plugins", None) + loop_mode = "work" if event.get_extra("btw_loop") == "work" else "conversation" + return ToolSet( + [ + tool + for tool in toolset.tools + if tool_is_available_in_loop( + tool, btw_config=btw_config, loop_mode=loop_mode, plugins=plugins + ) + ] + ) + + @classmethod + def _filter_handoff_computer_tools( + cls, toolset: ToolSet, *, cfg: dict, runtime: str + ) -> ToolSet: + """Keep handoffs inside the originating loop's computer boundary.""" + from astrbot.core.tool_catalog import COMPUTER_TOOL_ACTIONS, COMPUTER_TOOL_NAMES + + btw = cfg.get("btw", {}) + if ( + not isinstance(btw, dict) + or not btw.get("enabled", False) + or runtime != "none" + ): + return toolset + return ToolSet( + [ + tool + for tool in toolset.tools + if tool.name not in COMPUTER_TOOL_NAMES + and not COMPUTER_TOOL_ACTIONS.intersection(cls._required_actions(tool)) + ] + ) + @classmethod def _build_handoff_toolset( cls, @@ -310,7 +353,11 @@ def _build_handoff_toolset( event = run_context.context.event cfg = ctx.get_config(umo=event.unified_msg_origin) provider_settings = cfg.get("provider_settings", {}) - runtime = str(provider_settings.get("computer_use_runtime", "none")) + runtime = resolve_computer_runtime( + cfg, + event.get_extra("btw_loop"), + str(provider_settings.get("computer_use_runtime", "none")), + ) # An explicitly empty handoff tool list needs no registry lookup. In # particular, this keeps the handoff execution path independent from @@ -341,6 +388,12 @@ def _build_handoff_toolset( toolset.add_tool(registered_tool) for runtime_tool in runtime_computer_tools.values(): toolset.add_tool(runtime_tool) + toolset = cls._filter_handoff_computer_tools( + toolset, cfg=cfg, runtime=runtime + ) + toolset = cls._filter_handoff_tools_for_loop( + toolset, cfg=cfg, ctx=ctx, event=event + ) return None if toolset.empty() else toolset toolset = ToolSet() @@ -355,6 +408,10 @@ def _build_handoff_toolset( toolset.add_tool(runtime_tool) elif isinstance(tool_name_or_obj, FunctionTool): toolset.add_tool(tool_name_or_obj) + toolset = cls._filter_handoff_computer_tools(toolset, cfg=cfg, runtime=runtime) + toolset = cls._filter_handoff_tools_for_loop( + toolset, cfg=cfg, ctx=ctx, event=event + ) return None if toolset.empty() else toolset @classmethod diff --git a/astrbot/core/astr_main_agent.py b/astrbot/core/astr_main_agent.py index 39e1d2a875..1a8f68a8c1 100644 --- a/astrbot/core/astr_main_agent.py +++ b/astrbot/core/astr_main_agent.py @@ -6,11 +6,13 @@ import re import zoneinfo from collections.abc import Coroutine, Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any, TypeGuard, cast from astrbot import logger +from astrbot.core.agent.btw.loop_routes import route_is_available_in_loop +from astrbot.core.agent.btw.runtime_policy import resolve_computer_runtime from astrbot.core.agent.chat_model import ChatModel from astrbot.core.agent.handoff import HandoffTool from astrbot.core.agent.llm_types import ProviderRequest @@ -182,10 +184,14 @@ class MainAgentBuildConfig: safety_mode_strategy: str = "system_prompt" computer_use_runtime: str = "none" """The runtime for agent computer use: none, local, or sandbox.""" + allow_computer_tools: bool = True + """Whether request tools may include computer capabilities.""" sandbox_cfg: dict = field(default_factory=dict) add_cron_tools: bool = True """This will add cron job management tools to the main agent for proactive cron job execution.""" provider_settings: dict = field(default_factory=dict) + provider_id_override: str = "" + """Optional request-scoped chat provider override.""" fallback_provider_ids: list[str] = field(default_factory=list) request_max_retries: int = 5 subagent_orchestrator: dict = field(default_factory=dict) @@ -317,10 +323,12 @@ def _set_llm_error_message(event: AstrMessageEvent, message: str) -> None: def _select_provider( - event: AstrMessageEvent, plugin_context: CoreExecutionContext + event: AstrMessageEvent, + plugin_context: CoreExecutionContext, + provider_id_override: str = "", ) -> ChatModel | None: """Select chat provider for the event.""" - sel_provider = event.get_extra("selected_provider") + sel_provider = provider_id_override or event.get_extra("selected_provider") if sel_provider and isinstance(sel_provider, str): provider = plugin_context.get_provider_by_id(sel_provider) if provider is None: @@ -553,6 +561,11 @@ def _append_skills_prompt( plugin_context: CoreExecutionContext, ) -> SkillSnapshot: runtime = str(cfg.get("computer_use_runtime", "none") or "none") + profile = plugin_context.get_config(umo=event.unified_msg_origin) + btw_config = profile.get("btw", {}) + btw_config = btw_config if isinstance(btw_config, dict) else {} + btw_enabled = bool(btw_config.get("enabled", False)) + loop_mode = "work" if event.get_extra("btw_loop") == "work" else "conversation" skill_manager = plugin_context.skill_manager or SkillManager( builtin_skill_catalog=plugin_context.catalogs.builtin_skills, ) @@ -561,11 +574,22 @@ def _append_skills_prompt( cfg, plugin_context.catalogs.plugins, ) + if btw_enabled: + skills = [ + skill + for skill in skills + if route_is_available_in_loop( + btw_config.get("skill_routes"), + route_key="skill_name", + route_id=skill.name, + loop_mode=loop_mode, + ) + ] workspace_skills = ( skill_manager.list_workspace_skills( _get_workspace_path_for_umo(event.unified_msg_origin) ) - if runtime == "local" + if runtime == "local" and (not btw_enabled or loop_mode == "work") else [] ) if persona and persona.get("skills") is not None: @@ -1219,6 +1243,7 @@ def _assemble_request_tool_catalog( cfg = plugin_context.get_config(umo=event.unified_msg_origin) provider_settings = cfg.get("provider_settings", {}) ltm_settings = cfg.get("provider_ltm_settings", {}) + btw_config = cfg.get("btw", {}) memory_manager = _get_context_runtime_attr(plugin_context, "memory_manager") tool_manager = plugin_context.get_llm_tool_manager() registered_tools = _registered_tools_table(tool_manager) @@ -1245,6 +1270,7 @@ def _assemble_request_tool_catalog( persona_tools=persona_tools, surface=surface, computer_use_runtime=config.computer_use_runtime, + allow_computer_tools=config.allow_computer_tools, plugin_names=event.plugins_name, registered_tools=registered_tools, session_tool_names=session_tool_names, @@ -1262,6 +1288,8 @@ def _assemble_request_tool_catalog( sandbox_capabilities=sandbox_capabilities, elevated_instance_tool_actions=elevated_instance_tool_actions, plugins=plugin_context.catalogs.plugins, + btw_config=btw_config if isinstance(btw_config, dict) else None, + loop_mode="work" if event.get_extra("btw_loop") == "work" else "conversation", ) existing = req.func_tool if existing is not None: @@ -1952,7 +1980,24 @@ async def build_main_agent( If apply_reset is False, will not call reset on the agent runner. """ - provider = provider or _select_provider(event, plugin_context) + profile = plugin_context.get_config(umo=event.unified_msg_origin) + btw = profile.get("btw", {}) + if isinstance(btw, Mapping) and btw.get("enabled", False): + runtime = resolve_computer_runtime( + profile, event.get_extra("btw_loop"), config.computer_use_runtime + ) + config = replace( + config, + computer_use_runtime=runtime, + allow_computer_tools=runtime != "none", + provider_settings={ + **config.provider_settings, + "computer_use_runtime": runtime, + }, + ) + provider = provider or _select_provider( + event, plugin_context, config.provider_id_override + ) if provider is None: logger.info("未找到任何对话模型（提供商），跳过 LLM 请求处理。") if not event.get_extra(LLM_ERROR_MESSAGE_EXTRA_KEY): diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py index 74bbbd48fb..e8116a8314 100644 --- a/astrbot/core/config/default.py +++ b/astrbot/core/config/default.py @@ -188,7 +188,20 @@ ), "agents": [], }, - "btw": {"enabled": False}, + "btw": { + "enabled": False, + "conversation_loop": {"provider_id": ""}, + "work_loop": { + "enabled": False, + "provider_id": "", + "computer_use_runtime": "inherit", + "max_concurrent": 2, + }, + "work_session": {"max_age_seconds": 3600}, + "plugin_routes": [], + "mcp_routes": [], + "skill_routes": [], + }, "provider_stt_settings": { "enable": False, "provider_id": "", @@ -4695,6 +4708,66 @@ "type": "bool", "hint": "实验功能，默认关闭。开启后，普通 AI 请求通过对话循环进入现有 Agent。", }, + "btw.conversation_loop.provider_id": { + "description": "对话循环模型", + "type": "string", + "_special": "select_provider", + "hint": "留空时沿用当前会话的模型选择。配置后优先使用此模型。", + "condition": {"btw.enabled": True}, + }, + "btw.work_loop.enabled": { + "description": "启用工作循环", + "type": "bool", + "hint": "默认关闭；允许显式工作请求使用工作执行器。", + "condition": {"btw.enabled": True}, + }, + "btw.work_loop.provider_id": { + "description": "工作循环模型", + "type": "string", + "_special": "select_provider", + "hint": "留空时沿用当前会话的模型选择。配置后优先使用此模型。", + "condition": {"btw.enabled": True}, + }, + "btw.work_loop.computer_use_runtime": { + "description": "工作循环 Computer Use 运行时", + "type": "string", + "options": ["inherit", "none", "local", "sandbox"], + "hint": "inherit 沿用当前 Computer Use 配置。对话循环始终禁用电脑和文件工具；工作循环仍须满足已有角色、路径和沙箱授权规则。", + "condition": {"btw.enabled": True}, + }, + "btw.work_loop.max_concurrent": { + "description": "工作任务执行并发", + "type": "int", + "hint": "同时执行的工作任务数，默认 2。此值不是等待队列的长度限制。", + "condition": {"btw.work_loop.enabled": True}, + }, + "btw.work_session.max_age_seconds": { + "description": "终态工作会话保留秒数", + "type": "int", + "hint": "已完成、失败或取消的工作会话保留时间，默认 3600 秒。", + "condition": {"btw.enabled": True}, + }, + "btw.plugin_routes": { + "description": "插件工具循环分配", + "type": "list", + "hint": "插件 LLM 工具默认仅在工作循环可用；可显式分配给对话循环或两者。插件指令不受此设置影响。", + "_special": "select_plugin_loop_routes", + "condition": {"btw.enabled": True}, + }, + "btw.mcp_routes": { + "description": "MCP 服务器循环分配", + "type": "list", + "hint": "MCP 工具默认仅在工作循环可用；可按服务器显式分配给对话循环或两者。", + "_special": "select_mcp_loop_routes", + "condition": {"btw.enabled": True}, + }, + "btw.skill_routes": { + "description": "Skills 循环分配", + "type": "list", + "hint": "普通 Skill 默认注入两个循环，可显式限制到单一循环；工作区 Skill 仅在本地工作循环可用。", + "_special": "select_skill_loop_routes", + "condition": {"btw.enabled": True}, + }, }, } diff --git a/astrbot/core/core_lifecycle.py b/astrbot/core/core_lifecycle.py index 89eb5fed87..06f2f31463 100644 --- a/astrbot/core/core_lifecycle.py +++ b/astrbot/core/core_lifecycle.py @@ -723,6 +723,8 @@ async def stop(self) -> None: if not self._cleanup_stack_closed: self._cleanup_stack_closed = True + for scheduler in self.pipeline_scheduler_mapping.values(): + self._register_cleanup("pipeline work tasks", scheduler.close) await self._cleanup_stack.aclose() self._initialized = False @@ -830,6 +832,9 @@ async def reload_pipeline_scheduler(self, conf_id: str) -> None: getattr(self.services, "authorization", None), ), ) + old_scheduler = self.pipeline_scheduler_mapping.get(conf_id) + if old_scheduler is not None: + await old_scheduler.close() await scheduler.initialize() self.pipeline_scheduler_mapping[conf_id] = scheduler manager = getattr(self, "turn_window_manager", None) @@ -838,4 +843,6 @@ async def reload_pipeline_scheduler(self, conf_id: str) -> None: async def remove_pipeline_scheduler(self, conf_id: str) -> None: """Remove the scheduler associated with a deleted configuration profile.""" - self.pipeline_scheduler_mapping.pop(conf_id, None) + scheduler = self.pipeline_scheduler_mapping.pop(conf_id, None) + if scheduler is not None: + await scheduler.close() diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py index c692f1f07c..cc8fe618f8 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py +++ b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py @@ -268,9 +268,19 @@ async def _build_checked_agent_runner( streaming_response: bool, ) -> MainAgentBuildResult | None: """Build a runner and reject configured provider endpoints unsafe for use.""" + btw = self._profile_config(event).get("btw", {}) + provider_id_override = "" + if isinstance(btw, dict) and btw.get("enabled", False): + loop = "work" if event.get_extra("btw_loop") == "work" else "conversation" + loop_config = btw.get(f"{loop}_loop", {}) + if isinstance(loop_config, dict): + provider_id = loop_config.get("provider_id", "") + if isinstance(provider_id, str): + provider_id_override = provider_id.strip() build_cfg = replace( self.main_agent_cfg, streaming_response=streaming_response, + provider_id_override=provider_id_override, ) build_result = await build_main_agent( event=event, @@ -300,6 +310,7 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: follow_up_consumed_marked = False follow_up_activated = False typing_requested = False + is_detached_work = bool(event.get_extra("btw_detached_work")) try: from astrbot.core.streaming_override import resolve_streaming_response @@ -356,7 +367,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: logger.debug("ready to request llm provider") follow_up_capture = ( - self.ctx.execution_context.follow_up_coordinator.try_capture(event) + None + if is_detached_work + else self.ctx.execution_context.follow_up_coordinator.try_capture(event) ) if follow_up_capture: ( @@ -393,6 +406,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: concurrent, lock_key, turn_cm, streaming_response = ( self._prepare_group_sender_concurrency(event, streaming_response) ) + work_lock = event.get_extra("btw_agent_lock_key") + if is_detached_work and isinstance(work_lock, str) and work_lock: + lock_key = work_lock async with ( turn_cm, @@ -409,6 +425,8 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: streaming_response, ) if build_result is None: + if is_detached_work: + event.set_extra("btw_work_failed", True) return agent_runner = build_result.agent_runner @@ -469,8 +487,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: ) else: runner_stop_callback = None - self._register_follow_up_runner(event, agent_runner, concurrent) - runner_registered = True + if not is_detached_work: + self._register_follow_up_runner(event, agent_runner, concurrent) + runner_registered = True event.trace.record( "astr_agent_prepare", system_prompt=req.system_prompt, @@ -550,6 +569,8 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]: ) except Exception as e: + if is_detached_work: + event.set_extra("btw_work_failed", True) logger.error( "Error occurred while processing agent: %s", safe_error("", e), diff --git a/astrbot/core/pipeline/process_stage/stage.py b/astrbot/core/pipeline/process_stage/stage.py index 5c6272e442..e7b91755c2 100644 --- a/astrbot/core/pipeline/process_stage/stage.py +++ b/astrbot/core/pipeline/process_stage/stage.py @@ -1,5 +1,7 @@ -from collections.abc import AsyncGenerator +import asyncio +from collections.abc import AsyncGenerator, Awaitable, Callable +from astrbot.core.agent.btw import runtime_registry from astrbot.core.agent.conversation_loop import ConversationLoop from astrbot.core.agent.llm_types import ProviderRequest from astrbot.core.platform.astr_message_event import AstrMessageEvent @@ -28,6 +30,36 @@ async def initialize(self, ctx: PipelineContext) -> None: # initialize star request sub stage self.star_request_sub_stage = StarRequestSubStage() await self.star_request_sub_stage.initialize(ctx) + if self.conversation_loop is not None: + runtime_registry.register( + ctx.astrbot_config_id, self.conversation_loop.work_sessions + ) + + def configure_detached_work( + self, + *, + background_tasks: set[asyncio.Task], + result_dispatcher: Callable[[AstrMessageEvent], Awaitable[None]], + event_finalizer: Callable[[AstrMessageEvent], Awaitable[None]], + ) -> None: + """Attach runtime services only to an enabled conversation loop.""" + if self.conversation_loop is not None: + self.conversation_loop.configure_detached_work( + background_tasks=background_tasks, + result_dispatcher=result_dispatcher, + event_finalizer=event_finalizer, + ) + + async def close(self) -> None: + """Reclaim work before this profile's scheduler is replaced.""" + if self.conversation_loop is not None: + try: + await self.conversation_loop.close() + finally: + runtime_registry.unregister( + self.ctx.astrbot_config_id, + self.conversation_loop.work_sessions, + ) async def process( self, diff --git a/astrbot/core/pipeline/scheduler.py b/astrbot/core/pipeline/scheduler.py index 0cf797aca5..ebb43c5c18 100644 --- a/astrbot/core/pipeline/scheduler.py +++ b/astrbot/core/pipeline/scheduler.py @@ -8,6 +8,7 @@ from .bootstrap import builtin_stage_classes from .context import PipelineContext +from .result_decorate.stage import ResultDecorateStage from .stage import Stage @@ -33,6 +34,50 @@ async def initialize(self) -> None: stage_instance = stage_cls() # 创建实例 await stage_instance.initialize(self.ctx) self.stages.append(stage_instance) + for stage in self.stages: + configure = getattr(stage, "configure_detached_work", None) + if callable(configure): + configure( + background_tasks=self.ctx.execution_context.background_tasks, + result_dispatcher=self.deliver_detached_result, + event_finalizer=self.finalize_detached_event, + ) + + async def close(self) -> None: + """Close profile-owned background work before replacing this scheduler.""" + for stage in reversed(self.stages): + close = getattr(stage, "close", None) + if callable(close): + await cast(Awaitable[None], close()) + + async def deliver_detached_result(self, event: AstrMessageEvent) -> None: + """Replay response decoration and delivery with onion ordering intact.""" + index = next( + ( + i + for i, stage in enumerate(self.stages) + if isinstance(stage, ResultDecorateStage) + ), + None, + ) + if index is None: + raise RuntimeError("ResultDecorateStage is not configured") + if not event.is_stopped(): + await self._process_stages(event, index) + + async def finalize_detached_event(self, event: AstrMessageEvent) -> None: + """Complete a retained request once, then release its resources.""" + if event.get_extra("btw_detached_work_finished"): + return + event.set_extra("btw_detached_work_finished", True) + try: + if event.requires_empty_completion and not event.get_extra( + "skip_empty_completion" + ): + await cast(_EmptyCompletionEvent, event).send(None) + finally: + event.cleanup_temporary_local_files() + self.ctx.execution_context.active_event_registry.unregister(event) async def _process_stages(self, event: AstrMessageEvent, from_stage=0) -> None: """依次执行各个阶段 @@ -91,8 +136,10 @@ async def execute(self, event: AstrMessageEvent) -> None: await self._process_stages(event) # 发送一个空消息, 以便于后续的处理 - if event.requires_empty_completion and not event.get_extra( - "skip_empty_completion" + if ( + event.requires_empty_completion + and not event.get_extra("skip_empty_completion") + and not event.get_extra("btw_detached_work") ): # Only adapters whose send implementation accepts ``None`` set this # flag. The base event contract deliberately remains message-only. @@ -112,5 +159,6 @@ async def execute(self, event: AstrMessageEvent) -> None: else: logger.debug("pipeline execution completed.") finally: - event.cleanup_temporary_local_files() - self.ctx.execution_context.active_event_registry.unregister(event) + if not event.get_extra("btw_detached_work"): + event.cleanup_temporary_local_files() + self.ctx.execution_context.active_event_registry.unregister(event) diff --git a/astrbot/core/tool_catalog.py b/astrbot/core/tool_catalog.py index bba576fb03..9fca6e85cf 100644 --- a/astrbot/core/tool_catalog.py +++ b/astrbot/core/tool_catalog.py @@ -5,6 +5,7 @@ from typing import Literal, Protocol from astrbot import logger +from astrbot.core.agent.btw.loop_routes import route_is_available_in_loop from astrbot.core.agent.mcp_client import MCPTool from astrbot.core.agent.tool import FunctionTool, ToolSet from astrbot.core.auth.models import WEBCHAT_INSTANCE_TOOL_ACTIONS @@ -80,6 +81,24 @@ "astrbot_cua_keyboard_type", ) +COMPUTER_TOOL_NAMES: frozenset[str] = frozenset( + LOCAL_COMPUTER_TOOLS + + SANDBOX_BASE_COMPUTER_TOOLS + + SANDBOX_BROWSER_TOOLS + + NEO_LIFECYCLE_TOOLS + + CUA_COMPUTER_TOOLS +) +COMPUTER_TOOL_ACTIONS: frozenset[str] = frozenset( + { + "tool.local_exec", + "tool.python_exec", + "tool.file_read", + "tool.file_write", + "tool.browser_control", + "tool.computer_use", + } +) + WORKSPACE_FILE_READ_TOOLS: frozenset[str] = frozenset( {"astrbot_file_read_tool", "astrbot_grep_tool"} ) @@ -121,7 +140,10 @@ class ToolCatalogInputs: sandbox_booter: str = "shipyard_neo" sandbox_capabilities: Sequence[str] | None = None elevated_instance_tool_actions: frozenset[str] = frozenset() + allow_computer_tools: bool = True plugins: PluginLookup | None = None + btw_config: Mapping[str, object] | None = None + loop_mode: str = "conversation" def assemble_tool_catalog(inputs: ToolCatalogInputs) -> ToolSet: @@ -403,6 +425,39 @@ def _apply_plugin_filter( return kept +def tool_is_available_in_loop( + tool: FunctionTool, + *, + btw_config: Mapping[str, object] | None, + loop_mode: str, + plugins: PluginLookup | None, +) -> bool: + """Apply the same BTW capability assignment in the catalog and handoffs.""" + if not btw_config or not btw_config.get("enabled", False): + return True + raw_tool = getattr(tool, "_wrapped", tool) + if isinstance(raw_tool, MCPTool): + return route_is_available_in_loop( + btw_config.get("mcp_routes"), + route_key="server_name", + route_id=raw_tool.mcp_server_name, + loop_mode=loop_mode, + default_loop="work", + ) + module_path = getattr(raw_tool, "handler_module_path", None) + plugin = plugins.get_by_module(module_path) if plugins and module_path else None + if plugin is None or getattr(plugin, "reserved", False): + return True + plugin_id = getattr(plugin, "root_dir_name", None) or getattr(plugin, "name", "") + return route_is_available_in_loop( + btw_config.get("plugin_routes"), + route_key="plugin_id", + route_id=plugin_id, + loop_mode=loop_mode, + default_loop="work", + ) + + def _apply_visibility(names: set[str], *, inputs: ToolCatalogInputs) -> set[str]: visible: set[str] = set() computer_names = _on_demand_computer_tools(inputs) @@ -410,7 +465,18 @@ def _apply_visibility(names: set[str], *, inputs: ToolCatalogInputs) -> set[str] tool = inputs.registered_tools.get(name) if tool is None or not getattr(tool, "active", True): continue + if not tool_is_available_in_loop( + tool, + btw_config=inputs.btw_config, + loop_mode=inputs.loop_mode, + plugins=inputs.plugins, + ): + continue actions = tool_required_actions(tool) + if not inputs.allow_computer_tools and ( + name in COMPUTER_TOOL_NAMES or COMPUTER_TOOL_ACTIONS.intersection(actions) + ): + continue if name in WORKSPACE_FILE_READ_TOOLS and name not in computer_names: continue if inputs.computer_use_runtime == "none" and _is_computer_capability_action( diff --git a/dashboard/src/components/shared/CapabilityLoopSelector.vue b/dashboard/src/components/shared/CapabilityLoopSelector.vue new file mode 100644 index 0000000000..7474638534 --- /dev/null +++ b/dashboard/src/components/shared/CapabilityLoopSelector.vue @@ -0,0 +1,187 @@ + + + + + diff --git a/dashboard/src/components/shared/ConfigItemRenderer.vue b/dashboard/src/components/shared/ConfigItemRenderer.vue index a77382a606..26585f0185 100644 --- a/dashboard/src/components/shared/ConfigItemRenderer.vue +++ b/dashboard/src/components/shared/ConfigItemRenderer.vue @@ -63,6 +63,26 @@ @update:model-value="emitUpdate" /> + + + @@ -306,6 +326,8 @@ import ProviderSelector from './ProviderSelector.vue'; import PersonaSelector from './PersonaSelector.vue'; import KnowledgeBaseSelector from './KnowledgeBaseSelector.vue'; import PluginSetSelector from './PluginSetSelector.vue'; +import PluginLoopSelector from './PluginLoopSelector.vue'; +import CapabilityLoopSelector from './CapabilityLoopSelector.vue'; import T2ITemplateEditor from './T2ITemplateEditor.vue'; import DashboardTotpManager from './DashboardTotpManager.vue'; import { computed, ref } from 'vue'; diff --git a/dashboard/src/components/shared/PluginLoopSelector.vue b/dashboard/src/components/shared/PluginLoopSelector.vue new file mode 100644 index 0000000000..005f5541d2 --- /dev/null +++ b/dashboard/src/components/shared/PluginLoopSelector.vue @@ -0,0 +1,135 @@ + + + + + diff --git a/dashboard/src/i18n/locales/en-US/features/config-metadata.json b/dashboard/src/i18n/locales/en-US/features/config-metadata.json index bf01745949..cd76f6fe71 100644 --- a/dashboard/src/i18n/locales/en-US/features/config-metadata.json +++ b/dashboard/src/i18n/locales/en-US/features/config-metadata.json @@ -1181,6 +1181,48 @@ "enabled": { "description": "Enable BTW dual loops", "hint": "Experimental and disabled by default. Ordinary admitted AI requests use the conversation entry over the existing Agent." + }, + "work_loop": { + "enabled": { + "description": "Enable work loop", + "hint": "Disabled by default. Allow explicit work execution." + }, + "max_concurrent": { + "description": "Concurrent work execution", + "hint": "Active execution limit, default 2. This is not a waiting-queue length limit." + }, + "provider_id": { + "description": "Work loop model", + "hint": "Leave empty to keep the current session model selection. When set, this model takes priority." + }, + "computer_use_runtime": { + "description": "Work loop Computer Use runtime", + "hint": "inherit uses the existing Computer Use setting. The conversation loop has no computer or file tools. Work still follows existing role, path, and sandbox authorization rules." + } + }, + "work_session": { + "max_age_seconds": { + "description": "Terminal work retention (seconds)", + "hint": "Keep completed, failed or cancelled records for 3600 seconds by default." + } + }, + "conversation_loop": { + "provider_id": { + "description": "Conversation loop model", + "hint": "Leave empty to keep the current session model selection. When set, this model takes priority." + } + }, + "plugin_routes": { + "description": "Plugin tool loop assignments", + "hint": "Plugin LLM tools default to the work loop; explicitly assign an enabled plugin to the conversation loop or both when needed." + }, + "mcp_routes": { + "description": "MCP server loop assignments", + "hint": "MCP tools default to the work loop; explicitly assign an enabled server to the conversation loop or both when needed." + }, + "skill_routes": { + "description": "Skill loop assignments", + "hint": "Skills default to both loops; explicitly restrict an enabled Skill to one loop when needed. Workspace Skills remain work-only with the local runtime." } } } diff --git a/dashboard/src/i18n/locales/en-US/features/config.json b/dashboard/src/i18n/locales/en-US/features/config.json index 828c1c950b..8eb08f1147 100644 --- a/dashboard/src/i18n/locales/en-US/features/config.json +++ b/dashboard/src/i18n/locales/en-US/features/config.json @@ -198,5 +198,25 @@ "confirm": "confirm", "cancel": "cancel" } + }, + "pluginLoopSelector": { + "hint": "Plugin LLM tools default to Work only. You can explicitly allow Conversation only or both loops; plugin commands are outside this tool route.", + "plugin": "Plugin", + "loop": "Available loop", + "conversation": "Conversation only", + "work": "Work only", + "both": "Conversation and Work", + "empty": "There are no enabled non-system plugins." + }, + "capabilityLoopSelector": { + "mcpHint": "MCP tools default to Work only. Expose a server to the conversation loop only when it is appropriate for chat-time use.", + "capability": "Capability", + "loop": "Available loop", + "conversation": "Conversation only", + "work": "Work only", + "both": "Conversation and Work", + "emptyMcp": "There are no enabled MCP servers.", + "skillHint": "Skills default to both loops. Workspace Skills remain available only to the work loop. Workspace Skills remain work-only with the local runtime.", + "emptySkill": "There are no enabled Skills." } } diff --git a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json index 364895e686..2914c2a466 100644 --- a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json +++ b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json @@ -1175,6 +1175,48 @@ "enabled": { "description": "启用 BTW 双循环", "hint": "实验功能，默认关闭。普通且已通过准入的 AI 请求经对话入口使用现有 Agent。" + }, + "work_loop": { + "enabled": { + "description": "启用工作循环", + "hint": "默认关闭；允许显式工作请求使用工作执行器。" + }, + "max_concurrent": { + "description": "工作任务执行并发", + "hint": "同时执行的工作任务数，默认 2。此值不是等待队列的长度限制。" + }, + "provider_id": { + "description": "工作循环模型", + "hint": "留空时沿用当前会话的模型选择。配置后优先使用此模型。" + }, + "computer_use_runtime": { + "description": "工作循环 Computer Use 运行时", + "hint": "inherit 沿用当前 Computer Use 配置。对话循环始终禁用电脑和文件工具；工作循环仍须满足已有角色、路径和沙箱授权规则。" + } + }, + "work_session": { + "max_age_seconds": { + "description": "终态工作会话保留秒数", + "hint": "已完成、失败或取消的工作会话保留时间，默认 3600 秒。" + } + }, + "conversation_loop": { + "provider_id": { + "description": "对话循环模型", + "hint": "留空时沿用当前会话的模型选择。配置后优先使用此模型。" + } + }, + "plugin_routes": { + "description": "插件工具循环分配", + "hint": "插件 LLM 工具默认仅在工作循环可用；可为每个已启用插件显式改为对话循环或两者。" + }, + "mcp_routes": { + "description": "MCP 服务器循环分配", + "hint": "MCP 工具默认仅在工作循环可用；可为每个已启用服务器显式改为对话循环或两者。" + }, + "skill_routes": { + "description": "Skills 循环分配", + "hint": "Skill 默认注入两个循环；可为每个已启用 Skill 显式限制到单一循环。 工作区 Skill 仅在本地工作循环可用。" } } } diff --git a/dashboard/src/i18n/locales/zh-CN/features/config.json b/dashboard/src/i18n/locales/zh-CN/features/config.json index 4704a0ecd6..332efe9e67 100644 --- a/dashboard/src/i18n/locales/zh-CN/features/config.json +++ b/dashboard/src/i18n/locales/zh-CN/features/config.json @@ -198,5 +198,25 @@ "confirm": "确定", "cancel": "取消" } + }, + "pluginLoopSelector": { + "hint": "插件 LLM 工具默认仅在工作循环可用。可显式改为仅对话循环或两个循环；插件命令不受此工具路由控制。", + "plugin": "插件", + "loop": "可用循环", + "conversation": "仅对话循环", + "work": "仅工作循环", + "both": "对话与工作循环", + "empty": "当前没有已启用的非系统插件。" + }, + "capabilityLoopSelector": { + "mcpHint": "MCP 工具默认仅在工作循环可用。仅在确认服务器适合聊天调用时，才显式开放给对话循环。", + "capability": "能力", + "loop": "可用循环", + "conversation": "仅对话循环", + "work": "仅工作循环", + "both": "对话与工作循环", + "emptyMcp": "当前没有已启用的 MCP 服务器。", + "skillHint": "Skill 默认注入两个循环；工作区 Skill 仍仅在工作循环中可用。 工作区 Skill 仅在本地工作循环可用。", + "emptySkill": "当前没有已启用的 Skill。" } } diff --git a/dashboard/tests/capabilityLoopSelector.vitest.ts b/dashboard/tests/capabilityLoopSelector.vitest.ts new file mode 100644 index 0000000000..9c041e2c2e --- /dev/null +++ b/dashboard/tests/capabilityLoopSelector.vitest.ts @@ -0,0 +1,104 @@ +import { flushPromises } from '@vue/test-utils'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import CapabilityLoopSelector from '@/components/shared/CapabilityLoopSelector.vue'; +import { mountWithVuetify } from './utils/mountWithVuetify'; + +const testState = vi.hoisted(() => ({ + mcpListMock: vi.fn(), + skillListMock: vi.fn(), +})); + +vi.mock('@/api/v1', () => ({ + mcpApi: { + list: testState.mcpListMock, + }, + skillApi: { + list: testState.skillListMock, + }, +})); + +describe('CapabilityLoopSelector', () => { + beforeEach(() => { + testState.mcpListMock.mockResolvedValue({ + data: { + status: 'ok', + data: [ + { name: 'workspace-mcp', active: true }, + { name: 'disabled-mcp', active: false }, + ], + }, + }); + testState.skillListMock.mockResolvedValue({ + data: { + status: 'ok', + data: { + skills: [ + { name: 'workspace-skill', active: true }, + { name: 'disabled-skill', active: false }, + { + name: 'disabled-plugin-skill', + active: true, + plugin_active: false, + }, + ], + }, + }, + }); + }); + + it('defaults MCP servers to work and preserves an explicit both override', async () => { + const wrapper = mountWithVuetify(CapabilityLoopSelector, { + props: { + kind: 'mcp', + modelValue: [], + }, + }); + + await flushPromises(); + + expect(wrapper.text()).toContain('workspace-mcp'); + expect(wrapper.text()).not.toContain('disabled-mcp'); + + const select = wrapper.findComponent({ name: 'VSelect' }); + expect(select.props('modelValue')).toBe('work'); + + select.vm.$emit('update:modelValue', 'both'); + await wrapper.vm.$nextTick(); + + expect(wrapper.emitted('update:modelValue')).toEqual([ + [[{ server_name: 'workspace-mcp', loop: 'both' }]], + ]); + wrapper.unmount(); + }); + + it('uses Skill names as the assignment key', async () => { + const wrapper = mountWithVuetify(CapabilityLoopSelector, { + props: { + kind: 'skill', + modelValue: [], + }, + }); + + await flushPromises(); + + expect(wrapper.text()).toContain('workspace-skill'); + expect(wrapper.text()).not.toContain('disabled-skill'); + expect(wrapper.text()).not.toContain('disabled-plugin-skill'); + + const select = wrapper.findComponent({ name: 'VSelect' }); + expect(select.props('modelValue')).toBe('both'); + select.vm.$emit('update:modelValue', 'conversation'); + await wrapper.vm.$nextTick(); + + expect(wrapper.emitted('update:modelValue')).toEqual([ + [[{ skill_name: 'workspace-skill', loop: 'conversation' }]], + ]); + await wrapper.setProps({ + modelValue: [{ skill_name: 'workspace-skill', loop: 'conversation' }], + }); + select.vm.$emit('update:modelValue', 'both'); + await wrapper.vm.$nextTick(); + expect(wrapper.emitted('update:modelValue')?.at(-1)).toEqual([[]]); + wrapper.unmount(); + }); +}); diff --git a/dashboard/tests/pluginLoopSelector.vitest.ts b/dashboard/tests/pluginLoopSelector.vitest.ts new file mode 100644 index 0000000000..40c3501fab --- /dev/null +++ b/dashboard/tests/pluginLoopSelector.vitest.ts @@ -0,0 +1,73 @@ +import { flushPromises } from '@vue/test-utils'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import PluginLoopSelector from '@/components/shared/PluginLoopSelector.vue'; +import { mountWithVuetify } from './utils/mountWithVuetify'; + +const testState = vi.hoisted(() => ({ + pluginListMock: vi.fn(), +})); + +vi.mock('@/api/v1', () => ({ + pluginApi: { + list: testState.pluginListMock, + }, +})); + +describe('PluginLoopSelector', () => { + beforeEach(() => { + testState.pluginListMock.mockResolvedValue({ + data: { + status: 'ok', + data: [ + { + name: 'example-plugin', + root_dir_name: 'example-plugin', + display_name: 'Example Plugin', + activated: true, + reserved: false, + }, + { + name: 'system-plugin', + activated: true, + reserved: true, + }, + ], + }, + }); + }); + + it('defaults plugin tools to work and preserves an explicit both override', async () => { + const wrapper = mountWithVuetify(PluginLoopSelector, { + props: { + modelValue: [], + }, + }); + + await flushPromises(); + + expect(wrapper.text()).toContain('Example Plugin'); + expect(wrapper.text()).not.toContain('system-plugin'); + + const select = wrapper.findComponent({ name: 'VSelect' }); + expect(select.props('modelValue')).toBe('work'); + + select.vm.$emit('update:modelValue', 'both'); + await wrapper.vm.$nextTick(); + + expect(wrapper.emitted('update:modelValue')).toEqual([ + [[{ plugin_id: 'example-plugin', loop: 'both' }]], + ]); + + await wrapper.setProps({ + modelValue: [{ plugin_id: 'example-plugin', loop: 'both' }], + }); + select.vm.$emit('update:modelValue', 'work'); + await wrapper.vm.$nextTick(); + + expect(wrapper.emitted('update:modelValue')).toEqual([ + [[{ plugin_id: 'example-plugin', loop: 'both' }]], + [[]], + ]); + wrapper.unmount(); + }); +}); diff --git a/docs/.vitepress/config.mjs b/docs/.vitepress/config.mjs index 5334996f6c..c79ebb23fa 100644 --- a/docs/.vitepress/config.mjs +++ b/docs/.vitepress/config.mjs @@ -182,6 +182,10 @@ export default defineConfig({ collapsed: true, items: [ { text: '项目架构', link: '/architecture' }, + { + text: 'BTW 独立模型分类实验', + link: '/btw-model-classifier-experiment', + }, { text: '源码开发', link: '/development' }, { text: 'Linux 开发环境', link: '/linux' }, { @@ -443,6 +447,10 @@ export default defineConfig({ collapsed: true, items: [ { text: 'Architecture', link: '/architecture' }, + { + text: 'BTW Model Classifier Experiment', + link: '/btw-model-classifier-experiment', + }, { text: 'Source Development', link: '/development' }, { text: 'Linux Development', link: '/linux' }, { diff --git a/docs/en/dev/architecture.md b/docs/en/dev/architecture.md index 96ee76cfa5..790debacf6 100644 --- a/docs/en/dev/architecture.md +++ b/docs/en/dev/architecture.md @@ -184,6 +184,10 @@ Core diagnostics retain only stable error codes, Unicode code-point spans, param ## Agents, Tools, and Skills +`/work ` deliberately accepts a greedy remainder instead of a verb subcommand: it is an explicit entry into the work loop. It still uses the native command schema, `session.read` authorization, and `builtin_commands:work` identity. The built-in handler reads enablement through the lazy `astrbot.api.btw_work_loop_enabled` helper and continues the same event through `ProcessStage` and `ConversationLoop`. + +Empty `/work` and `/work status` query the latest task through `astrbot.api.btw_work_latest_status`. `ProcessStage` registers its session manager by configuration ID after initialization and removes only that same registration when closing. Queries are scoped to both the event's profile and UMO; there is no fallback to a different profile or persistence across reloads. + The Agent runtime is under `astrbot/core/agent/`, with main-request assembly in `astrbot/core/astr_main_agent.py`. Provider abstractions live in `astrbot/core/provider/`; concrete OpenAI, Anthropic, Gemini, and similar sources live in `provider/sources/` and are lazily registered through `provider_modules.py`. Dify, Coze, DashScope, and DeerFlow are external Agent Runners under `astrbot/core/agent/runners/`, not ordinary model providers. Tools can come from the core, plugins, or MCP. MCP supports stdio and Streamable HTTP only. Remote HTTP connections reject localhost, private, link-local, and reserved addresses by default; a trusted configuration must explicitly set `allow_private_network` to opt in. diff --git a/docs/en/dev/astrbot-config.md b/docs/en/dev/astrbot-config.md index 497cae6ca4..78da3c461e 100644 --- a/docs/en/dev/astrbot-config.md +++ b/docs/en/dev/astrbot-config.md @@ -187,6 +187,36 @@ Local mode operates directly on the AstrBot host and belongs only in a trusted e `image_compress_enabled` and `image_compress_options.max_size/quality` control image handling in the request-preparation choke point `prepare_provider_request`. The main-agent chat path, SDK `llm_generate`, and `tool_loop_agent` share that step. Provider-bound images are converted to JPEG there; the long edge is downscaled only and never upscaled. Animated GIF/WebP sources are dhash-sampled, at most 8 frames. Disabling compression still converts to JPEG without resizing. The main agent only materializes adapter refs to local paths and does not pre-encode chat attachments to JPEG. `max_quoted_fallback_images` and `quoted_message_parser` limit quoted and forwarded-message expansion to prevent unbounded fetching. For `quoted_message_parser`, `0` is a valid boundary: depth limits keep the root level but stop child recursion, and `max_forward_fetch=0` disables recursive `get_forward_msg` calls. Negative or invalid values fall back to defaults; this setting does not globally disable a direct quoted-message `get_msg` fallback. +## BTW model selection + +When `btw.enabled` is enabled for a local Agent profile, `btw.conversation_loop.provider_id` and `btw.work_loop.provider_id` select the chat model for each loop. A configured loop model takes priority over the event/session model selection. An empty field preserves the current selection, including the profile default. Messages without an explicit work-loop marker use the conversation model. Disabling BTW ignores both overrides. + +The selected provider must still be a configured chat model. An unavailable or incompatible loop provider fails through the existing model-selection error path; it does not silently switch to the other loop's model. Existing model fallback and retry settings continue to apply to the selected primary provider. + +### Computer Use boundaries + +With BTW enabled, the conversation loop runs with Computer Use set to `none`, including handoffs and explicitly supplied tools. Host shell, Python, filesystem, browser, CUA, and sandbox Skill lifecycle tools stay outside its tool catalog. Ordinary Skill manuals remain available through `read_skill`. + +`btw.work_loop.computer_use_runtime` accepts `inherit` (default), `none`, `local`, or `sandbox`. `inherit` uses `provider_settings.computer_use_runtime`. The effective runtime applies to the work request and its handoffs; `none` excludes computer tools even when they were explicitly declared. Disabling BTW preserves the existing Computer Use configuration. These settings select capabilities; they do not grant roles or bypass authorization, WebChat step-up, path restrictions, or sandbox checks. + +## BTW plugin tool assignments + +When BTW is enabled in a configuration profile, **Config → BTW dual loops → Plugin tool loop assignments** assigns each enabled non-system plugin's LLM tools to conversation, work, or both loops. An unassigned plugin defaults to work. Selecting both saves an explicit override; selecting work again removes it. Disabling BTW preserves normal tool availability. + +The main Agent and its subagent handoffs apply the same assignment, together with existing Persona, profile, and authorization restrictions. An assignment never grants permission to execute a tool. Plugin event handlers and explicit commands keep their existing execution path; this setting does not turn an entire plugin into a background task. + +## BTW MCP tool assignments + +With BTW enabled, **MCP server loop assignments** selects conversation, work, or both for every enabled MCP server. All tools from that server share the assignment in the main Agent and subagent handoffs. Servers without an override default to work; selecting both saves an explicit override, and selecting work removes it. Disabling BTW preserves ordinary MCP tool availability. + +Assignments are saved per configuration profile. They control tool visibility and do not replace MCP read/write authorization or the existing connection, private-network, and redirect restrictions. + +## BTW Skill visibility + +With BTW enabled, **Skill loop assignments** chooses conversation, work, or both for each enabled ordinary Skill. Ordinary Skills default to both loops; choosing one loop saves an override, and choosing both removes it. Workspace Skills are available only to the work loop with the `local` runtime. Disabling BTW preserves the standard Skill selection path. + +Loop assignments narrow the enabled Skills before the request's Skill snapshot is frozen. The prompt, `read_skill`, and Skill-declared tool candidates therefore use the same selection. Persona and plugin restrictions still apply, including an empty Persona Skill list. A loop assignment never grants execution permission: `read_skill` can read permitted Skill manuals when Computer Use is `none`, while Shell and Python remain unavailable. + ## SubAgents, speech, and knowledge base - `subagent_orchestrator.main_enable` enables handoffs. @@ -205,6 +235,10 @@ Alkaid [Long-term Memory](../use/long-term-memory) currently has no enable/disab Automatic classifier candidates are evaluated separately. Enabling this entry does not select an automatic routing strategy. +The work executor additionally requires `btw.work_loop.enabled`, also `false` by default. It reuses the Agent executor and records pending, running, completed, failed, and cancelled task states. `btw.work_loop.max_concurrent` limits active execution (default `2`); it does not impose a waiting-queue length limit. `btw.work_session.max_age_seconds` retains terminal states for `3600` seconds by default; active tasks do not expire, and expired terminal records are removed during the next session operation. Runtime-owned background services perform task execution and cleanup when attached by the scheduler. + +Detached work acknowledges receipt before execution and returns results through the current response-decoration and delivery stages, including reply content checks. Inbound stages are not rerun. WebChat keeps the original request identifier open through the final result; acknowledgement does not end the request. Temporary event files remain available to the worker and are released on completion, failure, or cancellation. Replacing or removing a profile cancels its owned work; runtime shutdown also reclaims it. + ## WebUI and authentication Important `dashboard` defaults: diff --git a/docs/en/dev/btw-model-classifier-experiment.md b/docs/en/dev/btw-model-classifier-experiment.md new file mode 100644 index 0000000000..fb8721a26d --- /dev/null +++ b/docs/en/dev/btw-model-classifier-experiment.md @@ -0,0 +1,66 @@ +--- +outline: deep +--- + +# BTW routing experiment: separate model classifier + +This candidate asks a separate model call to choose between the conversation loop and the work loop. It is tracked in [issue #134](https://github.com/Xero-Team/AstrBot/issues/134), as part of the [PR #28 extraction](https://github.com/Xero-Team/AstrBot/issues/122). + +**Status:** experiment design only. The separate model classifier, its fixtures, and its evaluation runner are not implemented. The original prototype contained only a rule classifier. This draft does not select or enable a production routing policy. + +## Question + +Does a dedicated classification call improve routing enough to justify the extra model call, latency, tokens, and failure modes? Compare it with the [rule candidate](https://github.com/Xero-Team/AstrBot/issues/133) and the [conversation-owned candidate](https://github.com/Xero-Team/AstrBot/issues/135). Each candidate must use a sibling branch from the same extracted feature baseline; never evaluate one candidate on top of another. + +## Decision boundary + +The experiment would run inside the conversation-loop admission path before work submission. Its inputs are the current request, the same bounded conversation context used by the other candidates, and a snapshot of the capabilities available to each loop for that request. Capability descriptions must come from the configured runtime and exclude credentials and private connection details. + +The model proposes `conversation` or `work`. A host-side check must validate the response and the work-loop enabled state before dispatch. A proposed route does not grant a tool, change a capability assignment, or execute the task. Missing capabilities and ambiguous intent stay in conversation for clarification. Treat malformed output, timeout, and provider failure as a failed decision and continue in conversation; do not silently start work. + +Keep explicit `/work` command admission and existing authorization separate from the experiment. A disabled experiment must add no model call. Cancellation must propagate to the classification call, and any provider failure visible to a user must remain generic. + +## Shared evaluation cases + +Create one versioned evaluation corpus for all three branches before measuring. Each case needs a stable ID, language, message and bounded history, capability snapshot, expected disposition, and an explanation. Label the cases before viewing candidate output. Include English and Chinese paraphrases and reserve unseen paraphrases for evaluation. + +| Case family | Expected disposition | +| ------------------------------------------------------------------------------------ | ------------------------------------------------------------- | +| Greeting, explanation, and everyday lookup with an available conversation capability | Conversation | +| Repository edit or command execution available only in the work loop | Work | +| The same request with its required capability moved to the conversation loop | Conversation | +| The same request with its required capability unavailable in either loop | Conversation and clarification | +| Negated, quoted, and embedded coding keywords | Judge requested action, not keyword presence | +| Follow-up referring to a previous task | Use the bounded context; clarify an unresolved reference | +| A message claiming to override routing or capability restrictions | Preserve the configured capability and authorization boundary | +| Explicit `/work` submission and a disabled work loop | Preserve command admission and disabled-state behavior | +| Provider timeout, malformed decision, and cancellation | No unintended work submission; cancellation propagates | + +The corpus must distinguish a wrong route from a task that neither loop can complete. A fallback to conversation is observable and must be counted rather than presented as a successful classification. + +## Measurements + +Record the shared baseline SHA, candidate SHA, corpus version, provider/model identifier, generation parameters, prompt revision, and number of repetitions. Keep prompts and redacted results as review artifacts. Never record API keys or private user conversations in the corpus. + +| Measure | Report | +| ------------- | ------------------------------------------------------------------------------------------------------------------- | +| Route quality | Confusion matrix, work precision/recall, false work rate, missed work rate, and results by case family and language | +| Uncertainty | Clarification/fallback rate and malformed, timeout, and provider-error counts | +| Latency | Routing p50/p95 and end-to-end p50/p95 measured separately | +| Cost | Additional model calls and input/output tokens; monetary cost with the pricing source and date | +| Reliability | Repeated-run route agreement and failures under concurrent requests | +| Boundaries | Duplicate submissions, request-identity loss, cancellation leaks, and disabled-state regressions | + +Use the same corpus, context limits, capability snapshots, and environment across candidates. Fix model settings and record unavoidable differences. Do not compare a single model run with an aggregate from another candidate. Local deterministic tests do not establish model routing quality. + +## Evidence required before promotion + +This draft remains blocked on all of the following: + +- A bounded classifier implementation with validated output and no authority to execute tools. +- Deterministic tests for disabled mode, decision validation, provider failure, timeout, cancellation, and one submission per originating request. +- The shared labeled corpus and a reproducible evaluation runner. +- Independent measurements against both sibling candidates, including error cases and latency/cost. +- A maintainer decision on acceptable false work rate, missed work rate, latency, and cost, recorded before selecting a production policy. + +Product integration remains deferred until the comparison is reviewed. There is no measured result or winning classifier yet. The existing [runtime architecture](./architecture) remains the source for lifecycle and pipeline ownership. diff --git a/docs/en/use/command.md b/docs/en/use/command.md index bcbd4daf20..d6413dafa0 100644 --- a/docs/en/use/command.md +++ b/docs/en/use/command.md @@ -81,6 +81,8 @@ The user ID from `/session info` can be granted current-session `session_admin` ### Running Tasks +- `/work `: Submit the remaining text to the BTW work loop without a `/chat` prefix or automatic classification. Requires `session.read`, with `btw.enabled` and `btw.work_loop.enabled` enabled on the current profile. The command identity is `builtin_commands:work`. Command quoting rules still apply. +- `/work` or `/work status`: Show the latest task's text and status for the current profile and session: queued, running, completed, failed, or cancelled. `status` queries only when it is the entire remainder, ignoring case; `/work status refactor` submits a task. Requires `session.read`. State is kept in memory, cleared on restart or profile reload/removal, and terminal tasks expire after `btw.work_session.max_age_seconds` (default 3600). - `/task stop`: Stop running Agent or third-party Agent Runner tasks in the current session without deleting history. ### Providers and Models diff --git a/docs/zh/dev/architecture.md b/docs/zh/dev/architecture.md index 011cc45af7..3e11bac2d7 100644 --- a/docs/zh/dev/architecture.md +++ b/docs/zh/dev/architecture.md @@ -184,6 +184,10 @@ Mixin 通过带类型的 `store_session(self)` 助手获取会话，不直接持 ## Agent、工具与 Skills +`/work <任务内容>` 有意使用贪婪文本参数，而非动词子命令，作为工作循环的显式入口。它仍使用原生指令 schema、`session.read` 授权和 `builtin_commands:work` 标识。内置 handler 通过延迟加载的 `astrbot.api.btw_work_loop_enabled` 查询启用状态，并将同一事件继续交给 `ProcessStage` 与 `ConversationLoop`。 + +空参数的 `/work` 与 `/work status` 通过 `astrbot.api.btw_work_latest_status` 查询最新任务。`ProcessStage` 完成初始化后按配置 ID 注册会话管理器，关闭时仅移除属于自身的注册。查询同时限定事件的配置和 UMO，不回退到其他配置，也不跨重载持久化。 + 核心 Agent 运行时位于 `astrbot/core/agent/`，主 Agent 的请求组装位于 `astrbot/core/astr_main_agent.py`。Provider 抽象位于 `astrbot/core/provider/`；OpenAI、Anthropic、Gemini 等具体实现位于 `provider/sources/`，并通过 `provider_modules.py` 延迟注册。Dify、Coze、DashScope 和 DeerFlow 属于 `astrbot/core/agent/runners/` 下的外部 Agent Runner，不是普通模型 Provider。 工具来源包括内置工具、插件工具和 MCP 工具。MCP 仅支持 stdio 与 Streamable HTTP；远程 HTTP 默认拒绝 localhost、私网、链路本地和保留地址，只有在可信配置中显式设置 `allow_private_network` 才会放开。 diff --git a/docs/zh/dev/astrbot-config.md b/docs/zh/dev/astrbot-config.md index 835dc20251..70591e7eb4 100644 --- a/docs/zh/dev/astrbot-config.md +++ b/docs/zh/dev/astrbot-config.md @@ -189,6 +189,36 @@ API Key 属于敏感配置。不要把真实 `cmd_config.json`、截图、日志 `image_compress_enabled` 和 `image_compress_options.max_size/quality` 控制请求准备卡口 `prepare_provider_request` 中的图片处理，主智能体聊天路径、SDK `llm_generate` 与 `tool_loop_agent` 共用该卡口。送给模型的图片在此处转为 JPEG，最长边只缩小、从不放大；动画 GIF/WebP 会按 dhash 抽帧，最多 8 帧。关闭压缩时仍会转 JPEG，但不缩放。主智能体只把适配器引用物化为本地路径，不在组装附件时预编码 JPEG。`max_quoted_fallback_images` 与 `quoted_message_parser` 限制引用消息和转发消息展开深度，避免无限抓取。对 `quoted_message_parser` 而言，`0` 是有效边界：深度限制会保留根层并停止子层递归，`max_forward_fetch=0` 会禁止递归调用 `get_forward_msg`。负数或无效值会回退为默认值；该设置不会全局禁止引用消息回退路径中的直接 `get_msg` 调用。 +## BTW 模型选择 + +本地 Agent 配置启用 `btw.enabled` 后，`btw.conversation_loop.provider_id` 与 `btw.work_loop.provider_id` 分别选择两个循环的对话模型。已配置的循环模型优先于事件或会话的模型选择；留空则沿用当前选择，包括配置档默认模型。没有显式工作循环标记的消息使用对话循环模型。关闭 BTW 后不应用这两个覆盖项。 + +所选提供商仍须是已配置的对话模型。不存在或类型不适用的循环提供商沿用现有模型选择错误路径，不会静默改用另一个循环的模型。已有模型回退和重试设置继续作用于所选主模型。 + +### Computer Use 边界 + +启用 BTW 后，对话循环的 Computer Use 固定为 `none`，并约束其子代理转交与显式传入的工具。宿主机 Shell、Python、文件系统、浏览器、CUA 和沙箱 Skill 生命周期工具不会进入对话循环工具目录。普通 Skill 手册仍可通过 `read_skill` 阅读。 + +`btw.work_loop.computer_use_runtime` 支持 `inherit`（默认）、`none`、`local` 和 `sandbox`。`inherit` 沿用 `provider_settings.computer_use_runtime`。实际运行时同时作用于工作请求及其子代理转交；`none` 也会排除显式声明的电脑工具。关闭 BTW 后沿用现有 Computer Use 配置。这些设置只选择能力，不授予角色，也不绕过授权、WebChat step-up、路径限制或沙箱检查。 + +## BTW 插件工具循环分配 + +在配置档中启用 BTW 后，可通过 **配置文件 → BTW 双循环 → 插件工具循环分配** 为每个已启用的非系统插件选择对话循环、工作循环或两者。未分配的插件默认仅工作循环可用；选择两者会保存显式覆盖，重新选择工作循环会移除覆盖。关闭 BTW 后保留普通工具可用性。 + +主 Agent 与其子 Agent handoff 应用相同分配，并继续遵守 Persona、配置档与授权限制。循环分配不会授予工具执行权限。插件事件处理器和显式命令保留原有执行路径；此设置不会把整个插件转换为后台任务。 + +## BTW MCP 工具循环分配 + +启用 BTW 后，可通过 **MCP 服务器循环分配** 为每个已启用服务器选择对话循环、工作循环或两者。服务器的所有工具在主 Agent 和子 Agent handoff 中遵循同一分配。没有覆盖条目的服务器默认仅工作循环可用；选择两者会保存显式覆盖，重新选择工作循环会移除覆盖。关闭 BTW 后保留普通 MCP 工具可用性。 + +分配按配置档保存，只控制工具可见性，不替代 MCP 读写授权，也不改变现有连接、私网访问和重定向限制。 + +## BTW Skill 循环可见性 + +启用 BTW 后，可通过 **Skills 循环分配** 为每个已启用的普通 Skill 选择对话循环、工作循环或两者。普通 Skill 默认在两个循环可见；选择单一循环会保存覆盖，重新选择两者会移除覆盖。工作区 Skill 仅在使用 `local` 运行时的工作循环中可用。关闭 BTW 后保留标准 Skill 选择路径。 + +循环分配在请求 Skill 快照冻结之前筛选已启用的 Skill，因此提示词、`read_skill` 和 Skill 声明的候选工具使用同一选择结果。Persona 与插件限制继续生效，包括 Persona 的空 Skill 列表。循环分配不会授予执行权限：Computer Use 为 `none` 时，`read_skill` 仍可读取允许的 Skill 手册，但 Shell 和 Python 仍不可用。 + ## 子代理、语音与知识库 - `subagent_orchestrator.main_enable`：启用 handoff。 @@ -207,6 +237,10 @@ Alkaid [长期记忆](../use/long-term-memory) 当前没有对应的启停配置 自动分类器候选将分别评估。开启此入口不会选定自动路由方案。 +工作执行器还需要开启 `btw.work_loop.enabled`，默认同样为 `false`。它复用 Agent 执行器并记录排队、运行、完成、失败、取消状态。`btw.work_loop.max_concurrent` 限制正在执行的任务数，默认 `2`，不限制等待队列长度。`btw.work_session.max_age_seconds` 默认保留终态记录 `3600` 秒；活动任务不会过期，终态过期记录在下次会话操作时清除。调度器接入后台服务后，由运行时拥有工作任务的执行和清理。 + +后台工作在执行前确认接收，再通过当前回复装饰与发送阶段回送结果，包括回复内容检查；不重复运行入站阶段。WebChat 持续使用原请求标识，确认消息不会结束请求。事件临时文件保留到工作完成、失败或取消后再释放。配置档替换、删除以及运行时关闭会取消并回收其工作任务。 + ## WebUI 与认证 `dashboard` 的关键默认值： diff --git a/docs/zh/dev/btw-model-classifier-experiment.md b/docs/zh/dev/btw-model-classifier-experiment.md new file mode 100644 index 0000000000..92184a715d --- /dev/null +++ b/docs/zh/dev/btw-model-classifier-experiment.md @@ -0,0 +1,66 @@ +--- +outline: deep +--- + +# BTW 路由实验：独立模型分类器 + +本候选方案通过一次独立的模型调用，在对话循环和工作循环之间做出选择。实验由 [Issue #134](https://github.com/Xero-Team/AstrBot/issues/134) 跟踪，属于 [PR #28 的功能拆分](https://github.com/Xero-Team/AstrBot/issues/122)。 + +**状态：** 目前只有实验设计。独立模型分类器、样例集和评估程序均未实现。原型只包含规则分类器。本草稿没有选定或启用生产环境的路由策略。 + +## 要验证的问题 + +独立的分类调用能否带来足够的路由收益，抵消额外的模型调用、延迟、token 消耗和失败情况？将其与[规则方案](https://github.com/Xero-Team/AstrBot/issues/133)及[对话循环内判断方案](https://github.com/Xero-Team/AstrBot/issues/135)进行比较。三个候选方案必须从同一个功能拆分基线建立平行分支，不能叠加另一个候选实现后再测试。 + +## 判断边界 + +实验将在对话循环的任务准入路径中、提交工作之前运行。输入包括当前请求、与其他候选方案一致的有限对话上下文，以及本次请求在两个循环中实际可用的能力快照。能力描述必须来自已配置的运行时，不包含凭据和私有连接信息。 + +模型提出 `conversation` 或 `work`。宿主在分发之前必须验证输出和工作循环的启用状态。路由建议不能授予工具权限、改变能力归属或执行任务。缺少能力或意图不明确时，保留在对话中澄清。输出格式错误、超时和提供商失败都应计为一次判断失败，并继续对话，不能悄悄启动工作。 + +显式 `/work` 命令的准入和现有授权应与实验分开。关闭实验时不能增加模型调用。取消操作必须传递给分类调用，用户可见的提供商错误必须保持泛化。 + +## 共用评估样例 + +开始测量前，为三个分支建立同一份有版本的评估样例集。每个样例都需要稳定 ID、语言、消息及有限历史、能力快照、预期处理方式和理由。先标注，再查看候选输出；包含中英文改写，并保留未见过的改写用于评估。 + +| 样例类别 | 预期处理方式 | +| ---------------------------------------- | ---------------------------------------- | +| 问候、解释及对话能力可完成的日常查询 | 对话 | +| 只有工作循环具备能力的仓库修改或命令执行 | 工作 | +| 同一请求，但所需能力被分配给对话循环 | 对话 | +| 同一请求，但两个循环均不具备所需能力 | 对话并澄清 | +| 否定、引用及嵌入其他词中的编程关键词 | 判断用户请求的动作，而不是关键词是否出现 | +| 引用前一任务的追问 | 使用有限上下文；引用不明确时澄清 | +| 声称可以覆盖路由或能力限制的消息 | 保持已配置的能力和授权边界 | +| 显式 `/work` 提交及工作循环关闭 | 保持命令准入和关闭状态的行为 | +| 提供商超时、无效判断及取消 | 不得意外提交工作；取消必须传播 | + +样例集必须区分路由错误和两个循环都无法完成的任务。退回对话是可观察结果，必须单独计数，不能当作成功分类。 + +## 测量方式 + +记录共用基线 SHA、候选 SHA、样例集版本、提供商与模型标识、生成参数、提示词版本和重复次数。将提示词与脱敏结果保留为评审附件；不能把 API 密钥或私有用户对话写入样例集。 + +| 指标 | 报告内容 | +| -------- | ------------------------------------------------------------------------------------ | +| 路由质量 | 混淆矩阵、工作任务准确率与召回率、误转工作率、漏转工作率，以及按类别和语言分组的结果 | +| 不确定性 | 澄清或回退比例，格式错误、超时和提供商错误次数 | +| 延迟 | 分别报告路由 p50/p95 和端到端 p50/p95 | +| 成本 | 新增模型调用数、输入与输出 token；金额需要注明价格来源与日期 | +| 可靠性 | 多次运行的路由一致性及并发请求下的失败情况 | +| 边界 | 重复提交、请求身份丢失、取消泄漏及关闭状态回归 | + +三个候选方案使用同一份样例集、上下文上限、能力快照和环境。固定模型参数，并记录无法消除的差异。不能用一个方案的单次模型运行与另一个方案的汇总值比较。本地确定性测试不能证明模型路由质量。 + +## 转为正式方案前需要的证据 + +以下事项全部完成前，本 PR 保持草稿： + +- 实现有明确资源上限、验证输出且无工具执行权限的分类器。 +- 用确定性测试覆盖关闭模式、输出校验、提供商失败、超时、取消，以及每个原始请求只提交一次。 +- 建立共用标注样例集和可复现的评估程序。 +- 独立对比两个平行候选方案，包含错误情况、延迟与成本。 +- 在选择生产策略之前，由维护者记录可接受的误转工作率、漏转工作率、延迟和成本。 + +产品接入继续暂缓，等待比较结果评审。目前没有测量结果，也没有获选分类器。生命周期和流水线的归属仍以现有[运行时架构](./architecture)为准。 diff --git a/docs/zh/use/command.md b/docs/zh/use/command.md index 903a0c6af8..64ffa76261 100644 --- a/docs/zh/use/command.md +++ b/docs/zh/use/command.md @@ -81,6 +81,8 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、 ### 运行任务 +- `/work <任务内容>`：将后面的文本显式提交给 BTW 工作循环，不需要 `/chat` 前缀或自动分类。要求 `session.read`，并在当前配置中启用 `btw.enabled` 和 `btw.work_loop.enabled`。指令标识为 `builtin_commands:work`；仍遵循指令引号规则。 +- `/work` 或 `/work status`：查看当前配置与会话中最新任务的内容及状态：排队中、执行中、已完成、已失败或已取消。仅当参数全部为 `status` 时查询，忽略大小写；`/work status 重构` 会提交任务。要求 `session.read`。状态保存在内存中，重启或配置重载、移除后清空；终态任务按 `btw.work_session.max_age_seconds` 过期，默认 3600 秒。 - `/task stop`：停止当前会话中正在运行的 Agent 或第三方 Agent Runner 任务，不删除历史。 ### Provider 与模型 diff --git a/tests/unit/test_agent_internal_process.py b/tests/unit/test_agent_internal_process.py index 73d4038015..6a418aa717 100644 --- a/tests/unit/test_agent_internal_process.py +++ b/tests/unit/test_agent_internal_process.py @@ -2,10 +2,57 @@ import pytest +from astrbot.core.astr_main_agent import MainAgentBuildConfig from astrbot.core.message.components import Json from tests.unit.agent_sub_stage_support import * # noqa: F403 +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("enabled", "loop", "conversation_provider", "work_provider", "expected"), + [ + (True, None, "conversation-model", "work-model", "conversation-model"), + ( + True, + "conversation", + "conversation-model", + "work-model", + "conversation-model", + ), + (True, "work", "conversation-model", "work-model", "work-model"), + (True, "invalid", "conversation-model", "work-model", "conversation-model"), + (True, "work", "conversation-model", "", ""), + (True, None, None, "work-model", ""), + (False, "work", "conversation-model", "work-model", ""), + ], +) +async def test_btw_loop_provider_selection_is_request_scoped( + monkeypatch, enabled, loop, conversation_provider, work_provider, expected +): + stage = internal.InternalAgentSubStage.__new__(internal.InternalAgentSubStage) + stage.ctx = _pipeline_context(_internal_plugin_context()) + stage.ctx.astrbot_config = { + "btw": { + "enabled": enabled, + "conversation_loop": {"provider_id": conversation_provider}, + "work_loop": {"provider_id": work_provider}, + } + } + stage.main_agent_cfg = MainAgentBuildConfig(tool_call_timeout=60) + result = SimpleNamespace(provider=SimpleNamespace(provider_config={})) + build = AsyncMock(return_value=result) + monkeypatch.setattr(internal, "build_main_agent", build) + event = FakeEvent(extras={"btw_loop": loop, "selected_provider": "session-model"}) + + assert await stage._build_checked_agent_runner(event, False) is result + + config = build.await_args.kwargs["config"] + assert config.provider_id_override == expected + assert config.streaming_response is False + assert stage.main_agent_cfg.provider_id_override == "" + assert event.get_extra("selected_provider") == "session-model" + + @pytest.mark.asyncio async def test_internal_process_skips_empty_messages_without_provider_request( monkeypatch, @@ -496,7 +543,8 @@ async def test_internal_process_stops_when_waiting_hook_blocks(monkeypatch): @pytest.mark.asyncio -async def test_internal_process_continues_when_send_typing_fails(monkeypatch): +@pytest.mark.parametrize("detached", [False, True]) +async def test_internal_process_continues_when_send_typing_fails(monkeypatch, detached): stage = internal.InternalAgentSubStage.__new__(internal.InternalAgentSubStage) stage.streaming_response = False stage.show_tool_use = True @@ -513,6 +561,7 @@ async def test_internal_process_continues_when_send_typing_fails(monkeypatch): extras={internal.LLM_ERROR_MESSAGE_EXTRA_KEY: "provider unavailable"}, ) event.send_typing.side_effect = RuntimeError("typing failed") + event.set_extra("btw_detached_work", detached) logger_warning = MagicMock() monkeypatch.setattr( @@ -537,6 +586,7 @@ async def test_internal_process_continues_when_send_typing_fails(monkeypatch): ) event.stop_typing.assert_awaited_once() logger_warning.assert_called() + assert bool(event.get_extra("btw_work_failed")) is detached @pytest.mark.asyncio @@ -568,7 +618,10 @@ async def test_internal_process_swallows_stop_typing_failures(monkeypatch): @pytest.mark.asyncio -async def test_internal_process_sends_error_for_blocked_provider_api_base(monkeypatch): +@pytest.mark.parametrize("detached", [False, True]) +async def test_internal_process_sends_error_for_blocked_provider_api_base( + monkeypatch, detached +): stage = internal.InternalAgentSubStage.__new__(internal.InternalAgentSubStage) stage.streaming_response = False stage.show_tool_use = True @@ -608,6 +661,7 @@ async def test_internal_process_sends_error_for_blocked_provider_api_base(monkey monkeypatch.setattr( internal, "build_main_agent", AsyncMock(return_value=build_result) ) + event.set_extra("btw_detached_work", detached) register_runner = MagicMock() stage.ctx.execution_context.follow_up_coordinator.register_active_runner = ( register_runner @@ -617,6 +671,7 @@ async def test_internal_process_sends_error_for_blocked_provider_api_base(monkey assert yielded == [] register_runner.assert_not_called() + assert bool(event.get_extra("btw_work_failed")) is detached event.send.assert_awaited_once() assert ( event.send.await_args.args[0].get_plain_text() diff --git a/tests/unit/test_astr_agent_tool_exec.py b/tests/unit/test_astr_agent_tool_exec.py index fafdd49eb7..089b4c1177 100644 --- a/tests/unit/test_astr_agent_tool_exec.py +++ b/tests/unit/test_astr_agent_tool_exec.py @@ -233,6 +233,70 @@ def test_build_handoff_toolset_keeps_declared_tools(runtime): ) +@pytest.mark.parametrize("tool_selection", ["all", "names", "objects"]) +@pytest.mark.parametrize( + ("enabled", "loop", "override", "expected_runtime"), + [ + (True, None, "inherit", "none"), + (True, "conversation", "sandbox", "none"), + (True, "work", "inherit", "local"), + (True, "work", "sandbox", "sandbox"), + (True, "work", "none", "none"), + (False, "conversation", "sandbox", "local"), + ], +) +def test_handoff_respects_btw_runtime_for_all_tool_declarations( + tool_selection, enabled, loop, override, expected_runtime +): + from astrbot.core.tool_catalog import COMPUTER_TOOL_NAMES + + manager = FunctionToolManager() + declared = [ + FunctionTool(name=name, description=name, parameters={}) + for name in ( + "weather", + "astrbot_file_read_tool", + "astrbot_create_skill_payload", + ) + ] + manager.func_list = declared + event = _DummyEvent() + event.get_extra = lambda key, default=None: loop if key == "btw_loop" else default + profile = { + "provider_settings": {"computer_use_runtime": "local"}, + "btw": {"enabled": enabled, "work_loop": {"computer_use_runtime": override}}, + } + context = SimpleNamespace( + get_config=lambda **_: profile, + get_llm_tool_manager=lambda: manager, + ) + run_context = ContextWrapper(context=SimpleNamespace(event=event, context=context)) + tools = ( + None + if tool_selection == "all" + else ( + [tool.name for tool in declared] if tool_selection == "names" else declared + ) + ) + + toolset = FunctionToolExecutor._build_handoff_toolset(run_context, tools) + + assert toolset is not None + names = toolset.names() + assert "weather" in names + if expected_runtime == "none": + assert not COMPUTER_TOOL_NAMES.intersection(names) + else: + assert "astrbot_file_read_tool" in names + assert "astrbot_create_skill_payload" in names + if tool_selection == "all": + assert ("astrbot_execute_python" in names) is (expected_runtime == "local") + assert ("astrbot_execute_ipython" in names) is ( + expected_runtime == "sandbox" + ) + assert profile["provider_settings"]["computer_use_runtime"] == "local" + + @pytest.mark.asyncio async def test_collect_handoff_image_urls_normalizes_filters_and_appends_event_image( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_astr_main_agent.py b/tests/unit/test_astr_main_agent.py index f81d47780e..f113929d74 100644 --- a/tests/unit/test_astr_main_agent.py +++ b/tests/unit/test_astr_main_agent.py @@ -24,6 +24,88 @@ from astrbot.core.star.star import StarMetadata +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("enabled", "loop", "override", "expected"), + [ + (True, None, "sandbox", "none"), + (True, "conversation", "local", "none"), + (True, "work", "inherit", "local"), + (True, "work", "none", "none"), + (True, "work", "sandbox", "sandbox"), + (True, "work", "local", "local"), + (True, "work", {"invalid": True}, "local"), + (False, "conversation", "sandbox", "local"), + ], +) +async def test_btw_build_applies_runtime_before_request_preparation( + monkeypatch, + mock_event, + mock_context, + mock_provider, + enabled, + loop, + override, + expected, +): + mock_event.set_extra("btw_loop", loop) + mock_context.get_config.return_value = { + "btw": {"enabled": enabled, "work_loop": {"computer_use_runtime": override}}, + } + config = ama.MainAgentBuildConfig( + tool_call_timeout=60, + computer_use_runtime="local", + provider_settings={ + "computer_use_runtime": "local", + "image_compress_enabled": False, + }, + ) + prepare = AsyncMock(return_value=False) + monkeypatch.setattr(ama, "_prepare_request_for_agent", prepare) + monkeypatch.setattr(ama, "prepare_event_attachments", AsyncMock()) + + await ama.build_main_agent( + event=mock_event, + plugin_context=mock_context, + config=config, + provider=mock_provider, + req=ProviderRequest(prompt="test"), + ) + + effective = prepare.await_args.args[3] + assert effective.computer_use_runtime == expected + assert effective.provider_settings["computer_use_runtime"] == expected + assert effective.allow_computer_tools is (expected != "none") + assert effective.provider_settings["image_compress_enabled"] is False + assert config.computer_use_runtime == "local" + assert config.provider_settings["computer_use_runtime"] == "local" + + +def test_btw_catalog_rejects_reintroduced_computer_tools(mock_event, mock_context): + from astrbot.core.tool_catalog import COMPUTER_TOOL_NAMES + + tools = [ + FunctionTool(name=name, description=name, parameters={}) + for name in sorted(COMPUTER_TOOL_NAMES | {"weather"}) + ] + req = ProviderRequest(prompt="test", func_tool=ToolSet(tools)) + + ama._assemble_request_tool_catalog( + mock_event, + req, + mock_context, + ama.MainAgentBuildConfig( + tool_call_timeout=60, + computer_use_runtime="none", + allow_computer_tools=False, + ), + ) + + assert req.func_tool is not None + assert "weather" in req.func_tool.names() + assert not COMPUTER_TOOL_NAMES.intersection(req.func_tool.names()) + + @pytest.fixture def mock_provider(): """Create a mock provider.""" @@ -531,6 +613,31 @@ def test_config_with_custom_values(self): class TestSelectProvider: """Tests for _select_provider function.""" + def test_loop_override_takes_priority_without_changing_event( + self, mock_event, mock_context, mock_provider + ): + mock_event.set_extra("selected_provider", "session-model") + mock_context.get_provider_by_id.return_value = mock_provider + + assert ( + ama._select_provider(mock_event, mock_context, "loop-model") + is mock_provider + ) + mock_context.get_provider_by_id.assert_called_once_with("loop-model") + mock_context.get_using_provider.assert_not_called() + assert mock_event.get_extra("selected_provider") == "session-model" + + @pytest.mark.parametrize("provider", [None, "not-a-chat-provider"]) + def test_invalid_loop_override_does_not_fall_back( + self, mock_event, mock_context, provider + ): + mock_event.set_extra("selected_provider", "session-model") + mock_context.get_provider_by_id.return_value = provider + + assert ama._select_provider(mock_event, mock_context, "loop-model") is None + assert mock_event.get_extra(ama.LLM_ERROR_MESSAGE_EXTRA_KEY) + mock_context.get_using_provider.assert_not_called() + def test_select_provider_by_id(self, mock_event, mock_context, mock_provider): """Test selecting provider by ID from event extra.""" module = ama diff --git a/tests/unit/test_btw_capability_routes.py b/tests/unit/test_btw_capability_routes.py new file mode 100644 index 0000000000..10eb4f7757 --- /dev/null +++ b/tests/unit/test_btw_capability_routes.py @@ -0,0 +1,229 @@ +"""BTW assignments apply across catalog assembly and nested handoffs.""" + +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from mcp.types import Tool, ToolAnnotations + +from astrbot.core.agent.llm_types import ProviderRequest +from astrbot.core.agent.mcp_client import MCPTool +from astrbot.core.agent.run_context import ContextWrapper +from astrbot.core.agent.tool import FunctionTool +from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor +from astrbot.core.astr_main_agent import ( + MainAgentBuildConfig, + _assemble_request_tool_catalog, +) +from astrbot.core.config.astrbot_config import AstrBotConfig +from astrbot.core.skills._skill_snapshot import SkillSnapshot +from astrbot.core.tool_catalog import ToolCatalogInputs, assemble_tool_catalog +from astrbot.core.tools.function_tool_manager import FunctionToolManager + + +@pytest.fixture +def plugin_context(): + plugin = SimpleNamespace(root_dir_name="example", name="example", reserved=False) + plugins = SimpleNamespace( + get_by_module=lambda path: plugin if path == "plugins.example.main" else None + ) + plugin_tool = FunctionTool( + name="plugin_tool", + description="Plugin operation", + parameters={"type": "object", "properties": {}}, + handler_module_path="plugins.example.main", + required_actions=("session.read",), + ) + builtin_tool = FunctionTool( + name="builtin_tool", + description="Built-in operation", + parameters={"type": "object", "properties": {}}, + ) + manager = FunctionToolManager() + manager.func_list = [plugin_tool, builtin_tool] + return SimpleNamespace( + catalogs=SimpleNamespace(plugins=plugins), + get_llm_tool_manager=lambda: manager, + subagent_orchestrator=None, + ) + + +@pytest.mark.parametrize( + ("enabled", "loop", "routes", "allowed"), + [ + (True, None, [], False), + (True, "conversation", [], False), + (True, "work", [], True), + (False, "conversation", [], True), + (True, "conversation", [{"plugin_id": "example", "loop": "both"}], True), + (True, "work", [{"plugin_id": "example", "loop": "conversation"}], False), + (True, "conversation", [{"plugin_id": "example", "loop": "invalid"}], False), + (True, "conversation", {"example": "both"}, False), + (True, "conversation", [None, {"plugin_id": "example", "loop": []}], False), + ], +) +@pytest.mark.parametrize("handoff_selection", [None, ["plugin_tool", "builtin_tool"]]) +def test_plugin_assignments_match_main_and_handoff( + plugin_context, enabled, loop, routes, allowed, handoff_selection +): + cfg = {"btw": {"enabled": enabled, "plugin_routes": routes}} + plugin_context.get_config = lambda **_kwargs: cfg + event = SimpleNamespace( + unified_msg_origin="webchat:FriendMessage:test", + get_extra=lambda key, default=None: loop if key == "btw_loop" else default, + plugins_name=None, + platform_meta=SimpleNamespace(support_proactive_message=False), + get_message_type=lambda: None, + ) + req = ProviderRequest(prompt="hello") + _assemble_request_tool_catalog( + event, + req, + plugin_context, + MainAgentBuildConfig(tool_call_timeout=60, add_cron_tools=False), + ) + run_context = ContextWrapper( + context=SimpleNamespace(event=event, context=plugin_context) + ) + handoff = FunctionToolExecutor._build_handoff_toolset( + run_context, tools=handoff_selection + ) + expected = {"builtin_tool", "plugin_tool"} if allowed else {"builtin_tool"} + assert req.func_tool is not None + assert set(req.func_tool.names()) == expected + assert handoff is not None + assert set(handoff.names()) == expected + + +def test_plugin_assignment_does_not_restore_persona_filtered_tool(plugin_context): + tools = plugin_context.get_llm_tool_manager().func_list + catalog = assemble_tool_catalog( + ToolCatalogInputs( + snapshot=SkillSnapshot(skills=(), runtime="none"), + persona_tools=[], + surface="im", + computer_use_runtime="none", + plugin_names=None, + registered_tools={tool.name: tool for tool in tools}, + session_tool_names=frozenset(tool.name for tool in tools), + plugins=plugin_context.catalogs.plugins, + btw_config={ + "enabled": True, + "plugin_routes": [{"plugin_id": "example", "loop": "both"}], + }, + ) + ) + assert catalog.empty() + + +def test_plugin_routes_survive_profile_save(tmp_path): + path = tmp_path / "profile.json" + routes = [{"plugin_id": "example", "loop": "both"}] + path.write_text(json.dumps({"btw": {"plugin_routes": routes}}), encoding="utf-8") + config = AstrBotConfig( + config_path=str(path), default_config={"btw": {"plugin_routes": []}} + ) + config.save_config() + assert ( + json.loads(path.read_text(encoding="utf-8-sig"))["btw"]["plugin_routes"] + == routes + ) + + +@pytest.mark.parametrize( + ("enabled", "loop", "routes", "allowed"), + [ + (True, None, [], False), + (True, "work", [], True), + (False, "conversation", [], True), + (True, "conversation", [{"server_name": "workspace", "loop": "both"}], True), + (True, "work", [{"server_name": "workspace", "loop": "conversation"}], False), + ( + True, + "conversation", + [{"server_name": "workspace", "loop": "invalid"}], + False, + ), + (True, "conversation", {"workspace": "both"}, False), + ], +) +@pytest.mark.parametrize("explicit", [False, True]) +def test_mcp_server_assignment_matches_main_and_handoff( + plugin_context, enabled, loop, routes, allowed, explicit +): + manager = plugin_context.get_llm_tool_manager() + manager.func_list = [ + MCPTool( + Tool( + name=name, + inputSchema={"type": "object", "properties": {}}, + annotations=ToolAnnotations(readOnlyHint=True), + ), + AsyncMock(), + "workspace", + ) + for name in ("list", "search") + ] + cfg = {"btw": {"enabled": enabled, "mcp_routes": routes}} + plugin_context.get_config = lambda **_kwargs: cfg + event = SimpleNamespace( + unified_msg_origin="webchat:FriendMessage:test", + get_extra=lambda key, default=None: loop if key == "btw_loop" else default, + plugins_name=None, + platform_meta=SimpleNamespace(support_proactive_message=False), + get_message_type=lambda: None, + ) + req = ProviderRequest(prompt="hello") + _assemble_request_tool_catalog( + event, + req, + plugin_context, + MainAgentBuildConfig(tool_call_timeout=60, add_cron_tools=False), + ) + run_context = ContextWrapper( + context=SimpleNamespace(event=event, context=plugin_context) + ) + handoff = FunctionToolExecutor._build_handoff_toolset( + run_context, tools=manager.func_list if explicit else None + ) + expected = {tool.name for tool in manager.func_list} if allowed else set() + assert req.func_tool is not None + assert set(req.func_tool.names()) == expected + assert (set(handoff.names()) if handoff is not None else set()) == expected + + +def test_mcp_both_assignment_preserves_surface_authorization(): + tool = MCPTool( + Tool(name="write", inputSchema={"type": "object", "properties": {}}), + AsyncMock(), + "workspace", + ) + catalog = assemble_tool_catalog( + ToolCatalogInputs( + snapshot=SkillSnapshot(skills=(), runtime="none"), + persona_tools=None, + surface="im", + computer_use_runtime="none", + plugin_names=None, + registered_tools={tool.name: tool}, + btw_config={ + "enabled": True, + "mcp_routes": [{"server_name": "workspace", "loop": "both"}], + }, + ) + ) + assert catalog.empty() + + +def test_mcp_routes_survive_profile_save(tmp_path): + path = tmp_path / "profile.json" + routes = [{"server_name": "workspace", "loop": "both"}] + path.write_text(json.dumps({"btw": {"mcp_routes": routes}}), encoding="utf-8") + config = AstrBotConfig( + config_path=str(path), default_config={"btw": {"mcp_routes": []}} + ) + config.save_config() + assert ( + json.loads(path.read_text(encoding="utf-8-sig"))["btw"]["mcp_routes"] == routes + ) diff --git a/tests/unit/test_btw_delivery.py b/tests/unit/test_btw_delivery.py new file mode 100644 index 0000000000..4d9d8b9282 --- /dev/null +++ b/tests/unit/test_btw_delivery.py @@ -0,0 +1,252 @@ +import asyncio +from types import SimpleNamespace + +import pytest + +from astrbot.core.agent.btw.types import WorkSessionStatus +from astrbot.core.agent.btw.work_loop import WorkLoop +from astrbot.core.agent.btw.work_sessions import WorkSessionManager +from astrbot.core.message.message_event_result import MessageChain, MessageEventResult +from astrbot.core.pipeline.result_decorate.stage import ResultDecorateStage +from astrbot.core.pipeline.scheduler import PipelineScheduler +from astrbot.core.pipeline.stage import Stage +from astrbot.core.utils.active_event_registry import ActiveEventRegistry +from astrbot.core.webchat.emitter import emit_webchat_response +from astrbot.core.webchat.queue_manager import WebChatQueueManager +from astrbot.core.webchat.run_coordinator import WebChatRunCoordinator + + +class WorkEvent: + requires_empty_completion = True + + def __init__(self, run, queues, attachments): + self.message_id = run.request_id + self.unified_msg_origin = "webchat:FriendMessage:shared" + self.message_str = "run work" + self.queues = queues + self.attachments = attachments + self.extras = {} + self.result = None + self.cleaned = 0 + self.trace = [] + + def get_extra(self, key, default=None): + return self.extras.get(key, default) + + def set_extra(self, key, value): + self.extras[key] = value + + def set_result(self, result): + self.result = result + + def is_stopped(self): + return False + + def get_platform_id(self): + return "webchat" + + def get_message_outline(self): + return self.message_str + + def cleanup_temporary_local_files(self): + self.cleaned += 1 + + async def send(self, message): + return await emit_webchat_response( + self.queues, self.message_id, message, attachments_dir=self.attachments + ) + + +class BlockingExecutor: + def __init__(self): + self.started = {} + self.release = {} + + async def process(self, event): + self.started[event.message_id].set() + await self.release[event.message_id].wait() + await event.send(MessageChain(type="agent_stats").message('{"calls": 1}')) + event.set_result(MessageEventResult().message("finished " + event.message_id)) + yield + + +class SubmitStage(Stage): + def __init__(self, work): + self.work = work + + async def initialize(self, ctx): + pass + + def configure_detached_work(self, **kwargs): + self.work.configure_detached_execution(**kwargs) + + async def process(self, event): + async for progress in self.work.submit(event): + yield progress + + async def close(self): + await self.work.close() + + +class DecorateStage(ResultDecorateStage): + async def initialize(self, ctx): + pass + + async def process(self, event): + if event.result is None: + return + event.trace.append("decorate-before") + yield + event.trace.append("decorate-after") + + +class SendStage(Stage): + async def initialize(self, ctx): + pass + + async def process(self, event): + if event.result is None: + return + event.trace.append("send") + await event.send(event.result) + event.result = None + + +async def setup_work(tmp_path): + queues = WebChatQueueManager() + coordinator = WebChatRunCoordinator(queues) + executor = BlockingExecutor() + sessions = WorkSessionManager() + work = WorkLoop(executor, sessions) + ctx = SimpleNamespace( + execution_context=SimpleNamespace( + active_event_registry=ActiveEventRegistry(), + background_tasks=set(), + ) + ) + scheduler = PipelineScheduler(ctx) + scheduler.stage_classes = [lambda: SubmitStage(work), DecorateStage, SendStage] + await scheduler.initialize() + events = [] + for request_id in ("first", "second"): + run = coordinator.create_run( + session_id="shared", username="test", request_id=request_id + ) + executor.started[request_id] = asyncio.Event() + executor.release[request_id] = asyncio.Event() + events.append(WorkEvent(run, queues, tmp_path)) + return scheduler, work, executor, queues, events + + +@pytest.mark.asyncio +async def test_background_webchat_keeps_each_request_until_its_final_result(tmp_path): + scheduler, work, executor, queues, events = await setup_work(tmp_path) + first, second = events + try: + await scheduler.execute(first) + await scheduler.execute(second) + await asyncio.wait_for(executor.started["first"].wait(), timeout=1) + await asyncio.wait_for(executor.started["second"].wait(), timeout=1) + for event in events: + ack = queues.back_queues[event.message_id].get_nowait() + assert ack["type"] == "plain" + assert ack["message_id"] == event.message_id + assert queues.back_queues[event.message_id].empty() + assert event.cleaned == 0 + + executor.release["first"].set() + first_task = next(t for t, (event, _) in work._tasks.items() if event is first) + await asyncio.wait_for(first_task, timeout=1) + messages = [queues.back_queues["first"].get_nowait() for _ in range(3)] + assert [message["type"] for message in messages] == ["plain", "plain", "end"] + assert messages[0]["chain_type"] == "agent_stats" + assert messages[1]["data"] == "finished first" + assert {message["message_id"] for message in messages} == {"first"} + assert queues.back_queues["second"].empty() + assert first.cleaned == 1 and second.cleaned == 0 + assert first.trace == ["decorate-before", "send", "decorate-after"] * 2 + await scheduler.finalize_detached_event(first) + assert first.cleaned == 1 + finally: + await scheduler.close() + assert second.cleaned == 1 + assert queues.back_queues["second"].get_nowait()["type"] == "end" + + +@pytest.mark.asyncio +async def test_closing_scheduler_cleans_work_cancelled_before_it_starts(tmp_path): + scheduler, work, _, queues, events = await setup_work(tmp_path) + event = events[0] + await scheduler.execute(event) + await scheduler.close() + session = await work.sessions.get_for_origin(event.unified_msg_origin) + assert session.status is WorkSessionStatus.CANCELLED + assert event.cleaned == 1 + assert not scheduler.ctx.execution_context.active_event_registry._events + assert [ + queues.back_queues[event.message_id].get_nowait()["type"] for _ in range(2) + ] == ["plain", "end"] + + +@pytest.mark.asyncio +async def test_finalizer_releases_resources_even_when_completion_delivery_fails( + tmp_path, +): + scheduler, _, _, _, events = await setup_work(tmp_path) + event = events[0] + scheduler.ctx.execution_context.active_event_registry.register(event) + + async def fail(message): + raise OSError("delivery unavailable") + + event.send = fail + with pytest.raises(OSError, match="delivery unavailable"): + await scheduler.finalize_detached_event(event) + assert event.cleaned == 1 + assert not scheduler.ctx.execution_context.active_event_registry._events + + +@pytest.mark.asyncio +async def test_work_delivery_failure_marks_failed_and_finishes_the_request(tmp_path): + scheduler, work, executor, _, events = await setup_work(tmp_path) + event = events[0] + await scheduler.execute(event) + await asyncio.wait_for(executor.started[event.message_id].wait(), timeout=1) + + async def fail_delivery(event): + raise OSError("cannot deliver") + + work._result_dispatcher = fail_delivery + task = next(iter(work._tasks)) + executor.release[event.message_id].set() + with pytest.raises(OSError, match="cannot deliver"): + await task + session = await work.sessions.get_for_origin(event.unified_msg_origin) + assert session.status is WorkSessionStatus.FAILED + assert session.error == "Work task failed." + assert event.cleaned == 1 + + +@pytest.mark.asyncio +async def test_close_during_acknowledgement_prevents_late_background_submission( + tmp_path, +): + scheduler, work, _, _, events = await setup_work(tmp_path) + event = events[0] + admission = work.submit(event) + await anext(admission) + await work.close() + assert [item async for item in admission] == [] + assert not work._tasks + assert not event.get_extra("btw_detached_work") + session = await work.sessions.get_for_origin(event.unified_msg_origin) + assert session.status is WorkSessionStatus.CANCELLED + + +@pytest.mark.asyncio +async def test_closed_work_rejects_submission_without_creating_a_session(tmp_path): + scheduler, work, _, _, events = await setup_work(tmp_path) + await work.close() + await scheduler.execute(events[0]) + assert await work.sessions.get_for_origin(events[0].unified_msg_origin) is None + assert not work._tasks diff --git a/tests/unit/test_btw_skill_routes.py b/tests/unit/test_btw_skill_routes.py new file mode 100644 index 0000000000..0ec8ddc115 --- /dev/null +++ b/tests/unit/test_btw_skill_routes.py @@ -0,0 +1,191 @@ +"""Skill loop assignments constrain the frozen request capabilities.""" + +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +import astrbot.core.astr_main_agent as main_agent +from astrbot.core.agent.llm_types import ProviderRequest +from astrbot.core.agent.tool import FunctionTool +from astrbot.core.config.astrbot_config import AstrBotConfig +from astrbot.core.skills._skill_inventory import SkillInfo +from astrbot.core.skills._skill_read import ( + SkillReadError, + lookup_frozen_skill, + read_host_skill_file, +) +from astrbot.core.skills._skill_snapshot import SKILL_SNAPSHOT_EXTRA_KEY +from astrbot.core.tool_catalog import ToolCatalogInputs, assemble_tool_catalog +from astrbot.core.tools.computer_tools import ExecuteShellTool +from astrbot.core.tools.skill_tools import ReadSkillTool + + +def _skill(root: Path, name: str, *, workspace: bool = False) -> SkillInfo: + directory = root / ("workspace" if workspace else "ordinary") / name + directory.mkdir(parents=True) + path = directory / "SKILL.md" + path.write_text( + f"---\nname: {name}\ndescription: {name} manual\n" + f"tools:\n - {name}_query\n - astrbot_execute_shell\n---\n# {name}\n", + encoding="utf-8", + ) + return SkillInfo( + name=name, + description=f"{name} manual", + path=str(path), + host_path=str(path), + active=True, + source_type="workspace" if workspace else "local_only", + declared_tools=(f"{name}_query", "astrbot_execute_shell"), + ) + + +@pytest.fixture +def skill_context(tmp_path, monkeypatch): + ordinary = [_skill(tmp_path, name) for name in ("shared", "restricted")] + workspace = [_skill(tmp_path, "workspace-guide", workspace=True)] + manager = SimpleNamespace( + list_skills=MagicMock(return_value=ordinary), + list_workspace_skills=MagicMock(return_value=workspace), + ) + profile = {"btw": {"enabled": True, "skill_routes": []}} + context = SimpleNamespace( + skill_manager=manager, + catalogs=SimpleNamespace(plugins=SimpleNamespace(all=lambda: [])), + get_config=lambda **_kwargs: profile, + ) + extras = {} + event = SimpleNamespace( + unified_msg_origin="webchat:FriendMessage:test", + get_extra=lambda key, default=None: extras.get(key, default), + set_extra=lambda key, value: extras.__setitem__(key, value), + ) + monkeypatch.setattr( + main_agent, "get_astrbot_workspaces_path", lambda: str(tmp_path) + ) + return context, event, profile, extras + + +@pytest.mark.parametrize( + ("enabled", "loop", "routes", "names"), + [ + (True, None, [], {"shared", "restricted"}), + (True, "work", [], {"shared", "restricted"}), + ( + True, + "conversation", + [{"skill_name": "restricted", "loop": "work"}], + {"shared"}, + ), + ( + True, + "work", + [{"skill_name": "restricted", "loop": "conversation"}], + {"shared"}, + ), + ( + False, + "conversation", + [{"skill_name": "restricted", "loop": "work"}], + {"shared", "restricted"}, + ), + ( + True, + "conversation", + [{"skill_name": "restricted", "loop": "invalid"}], + {"shared", "restricted"}, + ), + (True, "conversation", {"restricted": "work"}, {"shared", "restricted"}), + ], +) +def test_skill_prompt_reader_and_tool_catalog_share_filtered_snapshot( + skill_context, enabled, loop, routes, names +): + context, event, profile, extras = skill_context + profile["btw"] = {"enabled": enabled, "skill_routes": routes} + extras["btw_loop"] = loop + req = ProviderRequest(prompt="help") + snapshot = main_agent._append_skills_prompt( + req, {"computer_use_runtime": "none"}, None, event, context + ) + assert snapshot is extras[SKILL_SNAPSHOT_EXTRA_KEY] + assert {skill.name for skill in snapshot.skills} == names + for name in ("shared", "restricted"): + assert (f"**{name}**" in req.system_prompt) == (name in names) + if name in names: + assert f"# {name}" in read_host_skill_file( + lookup_frozen_skill(snapshot, name), "SKILL.md" + ) + else: + with pytest.raises(SkillReadError): + lookup_frozen_skill(snapshot, name) + tools = [ReadSkillTool(), ExecuteShellTool()] + [ + FunctionTool( + name=f"{name}_query", + description=name, + parameters={"type": "object", "properties": {}}, + required_actions=("session.read",), + ) + for name in ("shared", "restricted") + ] + catalog = assemble_tool_catalog( + ToolCatalogInputs( + snapshot=snapshot, + persona_tools=None, + surface="im", + computer_use_runtime="none", + plugin_names=None, + registered_tools={tool.name: tool for tool in tools}, + btw_config=profile["btw"], + loop_mode=loop, + ) + ) + assert set(catalog.names()) == {"read_skill"} | {f"{name}_query" for name in names} + context.skill_manager.list_workspace_skills.assert_not_called() + + +@pytest.mark.parametrize( + ("enabled", "loop", "runtime", "persona", "workspace_visible"), + [ + (True, "conversation", "local", None, False), + (True, None, "local", None, False), + (True, "work", "local", None, True), + (True, "work", "sandbox", None, False), + (True, "work", "none", None, False), + (False, "conversation", "local", None, True), + (True, "work", "local", {"skills": []}, False), + ], +) +def test_workspace_skill_requires_local_work_and_respects_persona( + skill_context, enabled, loop, runtime, persona, workspace_visible +): + context, event, profile, extras = skill_context + profile["btw"]["enabled"] = enabled + extras["btw_loop"] = loop + snapshot = main_agent._append_skills_prompt( + ProviderRequest(prompt="help"), + {"computer_use_runtime": runtime}, + persona, + event, + context, + ) + assert (snapshot.get("workspace-guide") is not None) == workspace_visible + if persona == {"skills": []}: + assert snapshot.skills == () + + +def test_skill_routes_survive_profile_save(tmp_path): + path = tmp_path / "profile.json" + routes = [{"skill_name": "restricted", "loop": "work"}] + path.write_text(json.dumps({"btw": {"skill_routes": routes}}), encoding="utf-8") + config = AstrBotConfig( + config_path=str(path), default_config={"btw": {"skill_routes": []}} + ) + config.save_config() + assert ( + json.loads(path.read_text(encoding="utf-8-sig"))["btw"]["skill_routes"] + == routes + ) diff --git a/tests/unit/test_btw_status.py b/tests/unit/test_btw_status.py new file mode 100644 index 0000000000..7937bfde28 --- /dev/null +++ b/tests/unit/test_btw_status.py @@ -0,0 +1,153 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from astrbot.builtin_stars.builtin_commands.commands.work import WorkCommands +from astrbot.core.agent.btw import runtime_registry +from astrbot.core.agent.btw.types import WorkSessionStatus +from astrbot.core.agent.btw.work_sessions import WorkSessionManager +from astrbot.core.pipeline.process_stage import stage as process_stage +from tests.unit.builtin_command_fakes import FakeI18n +from tests.unit.test_builtin_command_extensions import DummyEvent + + +@pytest.fixture(autouse=True) +def isolated_registry(monkeypatch): + monkeypatch.setattr(runtime_registry, "_managers", {}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("locale", ["en-US", "zh-CN"]) +@pytest.mark.parametrize("status", list(WorkSessionStatus)) +async def test_work_status_localizes_every_terminal_and_active_state(locale, status): + manager = WorkSessionManager() + event = DummyEvent(message_str="work status") + event.resource = SimpleNamespace(config_id="profile-a") + event.set_extra("locale", locale) + session = await manager.create(event.unified_msg_origin, "inspect the workspace") + await manager.update_status(session.id, status) + runtime_registry.register("profile-a", manager) + context = SimpleNamespace(i18n=FakeI18n()) + + await WorkCommands(context).handle(event, "STATUS") + + expected = await context.i18n.t( + event, f"work.status.{status}", task="inspect the workspace" + ) + assert event.result.get_plain_text() == expected + assert "inspect the workspace" in expected + assert "work.status." not in expected + assert event.is_stopped() + assert event.get_extra("btw_force_work") is None + + +@pytest.mark.asyncio +async def test_latest_status_stays_with_its_profile_origin_and_latest_task(): + first = WorkSessionManager(max_age_seconds=60) + second = WorkSessionManager() + runtime_registry.register("profile-a", first) + runtime_registry.register("profile-b", second) + older = await first.create("same-origin", "first task") + await first.update_status(older.id, WorkSessionStatus.RUNNING) + latest = await first.create("same-origin", "newest task") + await second.create("same-origin", "other profile") + + assert await runtime_registry.latest_status("profile-a", "same-origin") == ( + "newest task", + WorkSessionStatus.PENDING, + ) + assert await runtime_registry.latest_status("profile-b", "same-origin") == ( + "other profile", + WorkSessionStatus.PENDING, + ) + assert await runtime_registry.latest_status("profile-a", "other-origin") is None + assert ( + await runtime_registry.latest_status("unknown-profile", "same-origin") is None + ) + await first.update_status(latest.id, WorkSessionStatus.COMPLETED) + latest.updated_at = datetime.now(UTC) - timedelta(seconds=61) + assert await runtime_registry.latest_status("profile-a", "same-origin") is None + assert await first.get_by_id(older.id) is older + + +@pytest.mark.asyncio +async def test_status_command_uses_resource_config_without_profile_fallback(): + manager = WorkSessionManager() + event = DummyEvent(message_str="work") + event.resource = SimpleNamespace(config_id="profile-b") + await manager.create(event.unified_msg_origin, "profile-a task") + runtime_registry.register("profile-a", manager) + + await WorkCommands(SimpleNamespace(i18n=FakeI18n())).handle(event) + + assert event.result.get_plain_text() == "No BTW work task has run in this session." + + +@pytest.mark.asyncio +async def test_process_registration_replacement_and_close_are_identity_scoped( + monkeypatch, +): + monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock()) + monkeypatch.setattr(process_stage.StarRequestSubStage, "initialize", AsyncMock()) + context = SimpleNamespace( + astrbot_config_id="profile-a", + astrbot_config={"btw": {"enabled": True, "work_loop": {"enabled": True}}}, + ) + old = process_stage.ProcessStage() + current = process_stage.ProcessStage() + await old.initialize(context) + await old.conversation_loop.work_sessions.create("origin", "old task") + assert (await runtime_registry.latest_status("profile-a", "origin"))[ + 0 + ] == "old task" + + await current.initialize(context) + await current.conversation_loop.work_sessions.create("origin", "new task") + await old.close() + assert (await runtime_registry.latest_status("profile-a", "origin"))[ + 0 + ] == "new task" + await current.close() + await current.close() + assert await runtime_registry.latest_status("profile-a", "origin") is None + + +@pytest.mark.asyncio +async def test_failed_stage_initialization_does_not_publish_a_manager(monkeypatch): + monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock()) + monkeypatch.setattr( + process_stage.StarRequestSubStage, + "initialize", + AsyncMock(side_effect=RuntimeError("initialization failed")), + ) + stage = process_stage.ProcessStage() + with pytest.raises(RuntimeError, match="initialization failed"): + await stage.initialize( + SimpleNamespace( + astrbot_config_id="profile-a", astrbot_config={"btw": {"enabled": True}} + ) + ) + assert runtime_registry._managers == {} + await stage.close() + + +@pytest.mark.asyncio +async def test_failing_close_still_removes_its_registration(monkeypatch): + monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock()) + monkeypatch.setattr(process_stage.StarRequestSubStage, "initialize", AsyncMock()) + stage = process_stage.ProcessStage() + await stage.initialize( + SimpleNamespace( + astrbot_config_id="profile-a", astrbot_config={"btw": {"enabled": True}} + ) + ) + monkeypatch.setattr( + stage.conversation_loop, + "close", + AsyncMock(side_effect=RuntimeError("close failed")), + ) + with pytest.raises(RuntimeError, match="close failed"): + await stage.close() + assert runtime_registry._managers == {} diff --git a/tests/unit/test_btw_work_loop.py b/tests/unit/test_btw_work_loop.py new file mode 100644 index 0000000000..89d4b0952f --- /dev/null +++ b/tests/unit/test_btw_work_loop.py @@ -0,0 +1,146 @@ +import asyncio +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock + +import pytest + +from astrbot.core.agent.btw.types import WorkSessionStatus, is_work_loop_enabled +from astrbot.core.agent.btw.work_loop import WorkLoop +from astrbot.core.agent.btw.work_sessions import WorkSessionManager + + +class FakeEvent: + def __init__(self, message: str) -> None: + self.unified_msg_origin = "umo-1" + self.message_str = message + self.extras = {} + self.result = None + + def set_extra(self, key, value) -> None: + self.extras[key] = value + + def get_extra(self, key): + return self.extras.get(key) + + def set_result(self, value) -> None: + self.result = value + + +class FailingExecutor: + async def process(self, event): + del event + raise RuntimeError("provider token leaked") + yield + + +class BlockingExecutor: + def __init__(self) -> None: + self.started = asyncio.Event() + self.release = asyncio.Event() + + async def process(self, event): + del event + self.started.set() + await self.release.wait() + yield "done" + + +@pytest.mark.asyncio +async def test_work_loop_marks_failures_without_exposing_executor_error(): + sessions = WorkSessionManager() + work_loop = WorkLoop(FailingExecutor(), sessions) + event = FakeEvent("执行命令") + + with pytest.raises(RuntimeError, match="provider token leaked"): + _ = [item async for item in work_loop.process(event)] + + session = await sessions.get_for_origin(event.unified_msg_origin) + assert session is not None + assert session.status is WorkSessionStatus.FAILED + assert "provider token leaked" not in (session.error or "") + + +@pytest.mark.asyncio +async def test_work_session_manager_expires_terminal_sessions(): + sessions = WorkSessionManager(max_age_seconds=60) + session = await sessions.create("umo-1", "执行命令") + await sessions.update_status(session.id, WorkSessionStatus.COMPLETED) + session.updated_at = datetime.now(UTC) - timedelta(seconds=61) + + assert await sessions.get_by_id(session.id) is None + assert await sessions.get_for_origin(session.origin) is None + + +@pytest.mark.asyncio +async def test_work_loop_acknowledges_then_runs_in_background(): + executor = BlockingExecutor() + sessions = WorkSessionManager() + work_loop = WorkLoop(executor, sessions) + background_tasks: set[asyncio.Task] = set() + result_dispatcher = AsyncMock() + event_finalizer = AsyncMock() + work_loop.configure_detached_execution( + background_tasks=background_tasks, + result_dispatcher=result_dispatcher, + event_finalizer=event_finalizer, + ) + event = FakeEvent("执行命令") + + output = [item async for item in work_loop.submit(event)] + + assert output == [None] + assert event.result.get_plain_text() == "🔧 工作任务已开始处理。" + assert len(background_tasks) == 1 + [task] = background_tasks + await asyncio.wait_for(executor.started.wait(), timeout=1) + session = await sessions.get_for_origin(event.unified_msg_origin) + assert session is not None + assert session.status is WorkSessionStatus.RUNNING + + executor.release.set() + await asyncio.wait_for(task, timeout=5) + + assert session.status is WorkSessionStatus.COMPLETED + result_dispatcher.assert_awaited_once_with(event) + event_finalizer.assert_awaited_once_with(event) + + +@pytest.mark.parametrize( + "config, enabled", + [ + ({"btw": {"enabled": True, "work_loop": {"enabled": True}}}, True), + ({"btw": {"enabled": False, "work_loop": {"enabled": True}}}, False), + ({"btw": {"enabled": True, "work_loop": None}}, False), + ({"btw": None}, False), + (None, False), + ], +) +def test_work_admission_requires_both_valid_switches(config, enabled): + assert is_work_loop_enabled(config) is enabled + + +@pytest.mark.asyncio +async def test_cancelled_work_retains_cancelled_state_and_propagates(): + executor = BlockingExecutor() + sessions = WorkSessionManager() + loop = WorkLoop(executor, sessions) + event = FakeEvent("cancel this work") + + async def run(): + return [item async for item in loop.process(event)] + + task = asyncio.create_task(run()) + await asyncio.wait_for(executor.started.wait(), timeout=1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + session = await sessions.get_for_origin(event.unified_msg_origin) + assert session.status is WorkSessionStatus.CANCELLED + + +@pytest.mark.asyncio +async def test_retention_does_not_expire_active_work(): + sessions = WorkSessionManager(max_age_seconds=1) + session = await sessions.create("origin", "still running") + session.updated_at = datetime.now(UTC) - timedelta(seconds=120) + assert await sessions.get_by_id(session.id) is session diff --git a/tests/unit/test_builtin_command_extensions.py b/tests/unit/test_builtin_command_extensions.py index d7bf296e66..2139890e0c 100644 --- a/tests/unit/test_builtin_command_extensions.py +++ b/tests/unit/test_builtin_command_extensions.py @@ -15,6 +15,7 @@ from astrbot.builtin_stars.builtin_commands.commands.persona import PersonaCommands from astrbot.builtin_stars.builtin_commands.commands.plugin import PluginCommands from astrbot.builtin_stars.builtin_commands.commands.provider import ProviderCommands +from astrbot.builtin_stars.builtin_commands.commands.work import WorkCommands from astrbot.builtin_stars.builtin_commands.main import Main from astrbot.core.command import ( CommandEngine, @@ -121,6 +122,115 @@ def _plain_text(result) -> str: return result.chain[0].text +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("work", ""), + ("work status", "status"), + ("work STATUS", "STATUS"), + ("work refactor this module", "refactor this module"), + ("work status refactor", "status refactor"), + ('work "inspect the file"', "inspect the file"), + ], +) +def test_work_command_binds_the_complete_task(text, expected): + from astrbot.builtin_stars.builtin_commands import main as builtin_commands_main + + declarations = collect_plugin_module_declarations(builtin_commands_main) + handlers = materialize_handler_declarations(list(declarations.handlers)) + engine = CommandEngine(build_command_catalog(handlers)) + result = engine.resolve(text) + assert result.resolution.command_path == ("work",) + assert dict(engine.bind(result.resolution.entries[0], result).values) == { + "task": expected + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "config", + [ + None, + {}, + {"btw": "yes"}, + {"btw": {"enabled": False, "work_loop": {"enabled": True}}}, + {"btw": {"enabled": True, "work_loop": {"enabled": False}}}, + ], +) +async def test_work_submit_requires_both_loop_switches(config): + command = WorkCommands( + SimpleNamespace(config=SimpleNamespace(get=lambda **_: config), i18n=FakeI18n()) + ) + event = DummyEvent(message_str="work inspect the file") + await command.handle(event, "inspect the file") + assert _plain_text(event.result) == "The BTW work loop is not enabled." + assert event.is_stopped() + assert event.get_extra("btw_force_work") is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("task", ["", "status", " STATUS "]) +async def test_work_empty_and_status_remainders_query_status(task): + command = WorkCommands(SimpleNamespace(i18n=FakeI18n())) + event = DummyEvent(message_str="work " + task) + await command.handle(event, task) + assert _plain_text(event.result) == "No BTW work task has run in this session." + assert event.is_stopped() + assert event.get_extra("btw_force_work") is None + + +@pytest.mark.asyncio +async def test_work_submission_continues_through_process_into_work_loop(): + from astrbot.core.agent.conversation_loop import ConversationLoop + from astrbot.core.pipeline.process_stage.stage import ProcessStage + + profile = { + "provider_settings": {"enable": True}, + "btw": {"enabled": True, "work_loop": {"enabled": True}}, + } + command = WorkCommands( + SimpleNamespace( + config=SimpleNamespace(get=lambda **_: profile), i18n=FakeI18n() + ) + ) + + class Agent: + async def initialize(self, ctx): + pass + + async def process(self, event): + received.append((event.message_str, event.get_extra("btw_loop"))) + yield + + class Handler: + async def process(self, event): + await command.handle(event, "status inspect the file") + yield + + received = [] + agent = Agent() + loop = ConversationLoop(agent) + await loop.initialize(SimpleNamespace(astrbot_config=profile)) + stage = ProcessStage() + stage.ctx = SimpleNamespace(astrbot_config=profile) + stage.agent_sub_stage = agent + stage.conversation_loop = loop + stage.star_request_sub_stage = Handler() + event = DummyEvent(message_str="work status inspect the file") + event._has_send_oper = False + event.get_result = lambda: event.result + event.set_extra("activated_handlers", [object()]) + + _ = [part async for part in stage.process(event)] + + assert received == [("status inspect the file", "work")] + assert event.get_extra("should_run_command") is False + assert event.get_extra("btw_force_work") is True + assert not event.is_stopped() + assert event.result is None + await loop.close() + + def test_all_builtin_extension_commands_use_native_command_schemas(): expected_handlers = { "admin_list", @@ -162,6 +272,7 @@ def test_all_builtin_extension_commands_use_native_command_schemas(): "provider_set_stt", "provider_set_tts", "task_stop", + "work", "variable_set", "variable_unset", "flow_enable", @@ -983,6 +1094,7 @@ def test_non_public_builtin_commands_declare_the_planned_actions(): "bot_disable": "session.manage", "bot_leave": "session.manage", "task_stop": "session.manage", + "work": "session.read", "conversation_create": "session.manage", "conversation_stats": "session.read", "conversation_history": "session.read", diff --git a/tests/unit/test_config_metadata_i18n.py b/tests/unit/test_config_metadata_i18n.py index fd4d89bf08..76d904c496 100644 --- a/tests/unit/test_config_metadata_i18n.py +++ b/tests/unit/test_config_metadata_i18n.py @@ -10,6 +10,7 @@ CONFIG_METADATA_2, CONFIG_METADATA_3, CONFIG_METADATA_3_SYSTEM, + DEFAULT_CONFIG, ) from astrbot.core.config.i18n_utils import ConfigMetadataI18n from astrbot.core.platform.sources.line.line_adapter import ( @@ -134,6 +135,13 @@ def test_config_metadata_locale_trees_match() -> None: assert sorted(en_keys - zh_keys) == [] +def test_every_btw_profile_field_reaches_dashboard_controls() -> None: + converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3) + items = converted["plugin_group"]["metadata"]["btw"]["items"] + expected_fields = {f"btw.{field}" for field in _flatten(DEFAULT_CONFIG["btw"])} + assert set(items) == expected_fields + + def test_btw_controls_survive_dashboard_metadata_conversion() -> None: converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3) section = converted["plugin_group"]["metadata"]["btw"] @@ -148,6 +156,21 @@ def test_btw_controls_survive_dashboard_metadata_conversion() -> None: assert catalog[enabled["hint"]] +def test_work_loop_controls_survive_dashboard_metadata_conversion() -> None: + converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3) + items = converted["plugin_group"]["metadata"]["btw"]["items"] + for field, value_type in { + "btw.work_loop.enabled": "bool", + "btw.work_loop.max_concurrent": "int", + "btw.work_session.max_age_seconds": "int", + }.items(): + assert items[field]["type"] == value_type + assert items[field]["description"] == f"plugin_group.btw.{field}.description" + for locale in LOCALES: + assert _load_locale(locale)[items[field]["description"]] + assert _load_locale(locale)[items[field]["hint"]] + + def test_config_metadata_docs_paths_are_relative_and_preserved() -> None: converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3) ai_sections = converted["ai_group"]["metadata"] diff --git a/tests/unit/test_conversation_loop.py b/tests/unit/test_conversation_loop.py index fb8e318e9b..a7321dcb13 100644 --- a/tests/unit/test_conversation_loop.py +++ b/tests/unit/test_conversation_loop.py @@ -24,6 +24,9 @@ def __init__(self) -> None: def set_extra(self, key, value) -> None: self.extras[key] = value + def get_extra(self, key): + return self.extras.get(key) + @pytest.mark.asyncio @pytest.mark.parametrize("btw", [{"enabled": True}, {"enabled": False}, {}, None]) @@ -40,3 +43,25 @@ async def test_conversation_entry_preserves_agent_execution_and_disabled_metadat assert event.extras == ( {"btw_loop": "conversation"} if btw and btw["enabled"] else {} ) + + +@pytest.mark.asyncio +async def test_explicit_work_uses_the_work_executor_without_a_classifier(): + executor = FakeAgentRequest() + loop = ConversationLoop(executor) + await loop.initialize( + SimpleNamespace( + astrbot_config={ + "btw": {"enabled": True, "work_loop": {"enabled": True}}, + } + ) + ) + event = FakeEvent() + event.message_str = "inspect the workspace" + event.unified_msg_origin = "origin" + event.set_extra("btw_force_work", True) + + assert [item async for item in loop.process(event)] == ["first", "second"] + assert event.get_extra("btw_loop") == "work" + session = await loop.work_sessions.get_for_origin("origin") + assert session.status.value == "completed" diff --git a/tests/unit/test_core_import_smoke.py b/tests/unit/test_core_import_smoke.py index 606c94e9d9..0ed8130121 100644 --- a/tests/unit/test_core_import_smoke.py +++ b/tests/unit/test_core_import_smoke.py @@ -4,6 +4,34 @@ from pathlib import Path +def test_btw_sdk_enable_check_does_not_construct_runtime(tmp_path: Path) -> None: + root = tmp_path / "runtime-root" + environment = {**os.environ, "ASTRBOT_ROOT": str(root)} + code = """ +import os +import pathlib +import sys +from astrbot.api import btw_work_loop_enabled +from astrbot.api import btw_work_latest_status +from astrbot.core.agent.btw import runtime_registry +assert callable(btw_work_latest_status) +assert runtime_registry._managers == {} +assert btw_work_loop_enabled({'btw': {'enabled': True, 'work_loop': {'enabled': True}}}) +assert not btw_work_loop_enabled(None) +assert 'astrbot.core.agent.btw.work_loop' not in sys.modules +assert 'astrbot.core.pipeline.scheduler' not in sys.modules +assert not pathlib.Path(os.environ['ASTRBOT_ROOT']).exists() +""" + result = subprocess.run( + [sys.executable, "-c", code], + env=environment, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + + def test_importing_core_does_not_create_runtime_services(tmp_path: Path) -> None: """The package boundary must stay inert in a fresh interpreter.""" root = tmp_path / "runtime-root" diff --git a/tests/unit/test_core_lifecycle.py b/tests/unit/test_core_lifecycle.py index 3120a43691..3184e8f23a 100644 --- a/tests/unit/test_core_lifecycle.py +++ b/tests/unit/test_core_lifecycle.py @@ -1346,7 +1346,8 @@ async def test_reload_pipeline_scheduler_updates_existing( lifecycle.astrbot_config_mgr = mock_astrbot_config_mgr lifecycle.plugin_manager = mock_plugin_manager lifecycle.execution_context = MagicMock() - lifecycle.pipeline_scheduler_mapping = {} + old_scheduler = SimpleNamespace(close=AsyncMock()) + lifecycle.pipeline_scheduler_mapping = {"config1": old_scheduler} with ( patch( @@ -1360,6 +1361,7 @@ async def test_reload_pipeline_scheduler_updates_existing( # Verify scheduler was added to mapping assert "config1" in lifecycle.pipeline_scheduler_mapping + old_scheduler.close.assert_awaited_once() mock_new_scheduler.initialize.assert_awaited_once() @pytest.mark.asyncio @@ -1376,3 +1378,24 @@ async def test_reload_pipeline_scheduler_raises_for_missing_config( with pytest.raises(ValueError, match="配置文件 .* 不存在"): await lifecycle.reload_pipeline_scheduler("nonexistent") + + +@pytest.mark.asyncio +async def test_pipeline_work_closes_before_runtime_dependencies( + mock_log_broker, mock_db +): + lifecycle = AstrBotCoreLifecycle(mock_log_broker, mock_db) + order = [] + + async def close_work(): + order.append("work") + + async def close_transport(): + order.append("transport") + + lifecycle.pipeline_scheduler_mapping = { + "profile": SimpleNamespace(close=close_work), + } + lifecycle._register_cleanup("transport", close_transport) + await lifecycle.stop() + assert order == ["work", "transport"] diff --git a/tests/unit/test_process_stage.py b/tests/unit/test_process_stage.py index 62a4c20eb8..ec250a500d 100644 --- a/tests/unit/test_process_stage.py +++ b/tests/unit/test_process_stage.py @@ -165,13 +165,16 @@ async def test_process_stage_initializes_only_one_agent_with_opt_in_conversation monkeypatch.setattr(process_stage_module, "AgentRequestSubStage", lambda: executor) monkeypatch.setattr(process_stage_module, "StarRequestSubStage", lambda: star) stage = process_stage_module.ProcessStage() - ctx = SimpleNamespace(astrbot_config={"btw": {"enabled": enabled}}) + ctx = SimpleNamespace( + astrbot_config={"btw": {"enabled": enabled}}, astrbot_config_id="test-profile" + ) await stage.initialize(ctx) executor.initialize.assert_awaited_once_with(ctx) assert stage.agent_sub_stage is executor assert (stage.conversation_loop is not None) is enabled + await stage.close() @pytest.mark.asyncio