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..e9bf844b61 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-conversation-routing-experiment', + }, { text: '源码开发', link: '/development' }, { text: 'Linux 开发环境', link: '/linux' }, { @@ -443,6 +447,10 @@ export default defineConfig({ collapsed: true, items: [ { text: 'Architecture', link: '/architecture' }, + { + text: 'BTW Conversation Routing Experiment', + link: '/btw-conversation-routing-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-conversation-routing-experiment.md b/docs/en/dev/btw-conversation-routing-experiment.md new file mode 100644 index 0000000000..a53bfbe5b8 --- /dev/null +++ b/docs/en/dev/btw-conversation-routing-experiment.md @@ -0,0 +1,74 @@ +--- +outline: deep +--- + +# BTW routing experiment: conversation-owned decisions + +This candidate lets the conversation loop reason about its available capabilities and hand a task to the work loop when those capabilities cannot complete it. It is the maintainer's preferred direction to investigate, tracked in [issue #135](https://github.com/Xero-Team/AstrBot/issues/135) under the [PR #28 extraction](https://github.com/Xero-Team/AstrBot/issues/122). + +**Status:** experiment design only. Capability-aware model decisions, the handoff protocol, and evaluation are not implemented. The original prototype used keyword rules and did not contain this candidate. Product integration remains deferred until the independent comparison is reviewed. + +## Question + +Can a conversation loop that sees its own capabilities choose when to hand off work without a separate classification model call? Compare it with the [rule candidate](https://github.com/Xero-Team/AstrBot/issues/133) and the [separate model candidate](https://github.com/Xero-Team/AstrBot/issues/134). All three draft branches must start from the same extracted feature baseline and use the same evaluation cases. A preferred direction is not a measured winner. + +## Capability visibility + +The experiment must expose the capabilities actually available for the current request: selected provider/runner, tools, plugins, MCP tools, Skills, and relevant loop assignments. Use the existing runtime owners and configured catalogs; capability names and descriptions must exclude secrets and private connection details. Disabled or unauthorized capabilities must not be described as usable. + +The conversation loop also needs a bounded description of what the work loop can do. A task outside the conversation loop's capability can be handed off only when the work loop has the capability and is enabled. If neither loop can perform it, continue the conversation with a clarification or an honest capability limitation. A capability change between decision and submission must be revalidated by the host. + +This experiment must not grant broader authority when a model claims it needs a tool. Plugin/MCP/Skill assignment and authorization remain runtime constraints. + +## Decision and handoff + +The decision belongs to the ordinary conversation turn. Do not add an unconditional preliminary model call to this candidate. The draft needs a bounded, structured way to request a work handoff during that turn; the protocol and provider support are still open experiment work. + +A proposed handoff must identify the user's task and its originating request. The host validates the work-loop state and available capabilities before accepting exactly one submission. Do not execute the same action in both loops or let a work result start an unbounded handoff cycle. Preserve the existing command path for explicit `/work` requests. + +The experiment must account for answers or tool calls produced before the handoff decision. Report wasted work and time to first useful response. A model timeout, malformed handoff, or unsupported provider response must not become an implicit work submission. Cancellation must stop the originating request and any accepted work according to the existing lifecycle. + +## Shared evaluation cases + +Create one versioned corpus with stable case IDs, English and Chinese messages, bounded history, capability snapshots, expected dispositions, and reasons. Label it before inspecting outputs and reserve unseen paraphrases. Reuse the same corpus, context limits, capability snapshots, and environment on the other two branches. + +| 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 | + +Add candidate-specific lifecycle cases for a capability removed after a decision, a handoff after partial output, repeated handoff requests, and a work result that cannot satisfy the task. Keep these separate from the common corpus totals so comparison remains meaningful. + +## Measurements + +Record the shared baseline SHA, candidate SHA, corpus version, provider/model identifier, generation parameters, prompt revision, and repetition count. Save redacted decisions and results as review artifacts. Do not claim live-model quality from deterministic mocks. + +| 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, time to first useful response, and end-to-end p50/p95 | +| Cost | Additional model calls and input/output tokens, including capability descriptions and work repeated after handoff; monetary cost with pricing source/date | +| Reliability | Repeated-run agreement, provider compatibility, and concurrent-request failures | +| Boundaries | Duplicate side effects, request-identity loss, cancellation leaks, capability drift, and disabled-state regressions | + +Use the same definitions as the sibling candidates. Count a fallback or a task neither loop can complete separately from a correct route. Report the extra lifecycle cases separately and explain provider differences that prevent an exact comparison. + +## Evidence required before promotion + +This draft remains blocked on all of the following: + +- A capability snapshot derived from the existing request-scoped runtime, with tests for disabled and unauthorized capabilities. +- A bounded, validated handoff mechanism with a clear owner and supported-provider behavior. +- Deterministic tests for one submission per request, capability changes, partial output, repeated handoff, disabled state, malformed output, timeout, and cancellation. +- The shared labeled corpus, reproducible runner, and independent results against both sibling candidates. +- A maintainer decision on acceptable routing errors, latency, cost, and handoff failure behavior before production integration. + +No automatic classifier is selected by this document. Existing [runtime architecture](./architecture), request identities, and lifecycle ownership remain constraints on the experiment. 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-conversation-routing-experiment.md b/docs/zh/dev/btw-conversation-routing-experiment.md new file mode 100644 index 0000000000..4134c6f0db --- /dev/null +++ b/docs/zh/dev/btw-conversation-routing-experiment.md @@ -0,0 +1,74 @@ +--- +outline: deep +--- + +# BTW 路由实验：对话循环内判断 + +本候选方案让对话循环了解自身可用能力，在这些能力无法完成任务时转交工作循环。这是维护者优先希望验证的方向，由 [Issue #135](https://github.com/Xero-Team/AstrBot/issues/135) 跟踪，属于 [PR #28 的功能拆分](https://github.com/Xero-Team/AstrBot/issues/122)。 + +**状态：** 目前只有实验设计。基于能力的模型判断、转交协议和评估均未实现。原型使用关键词规则，并不包含本候选方案。产品接入继续暂缓，等待独立比较结果评审。 + +## 要验证的问题 + +对话循环看见自身能力后，能否不增加独立的分类模型调用，就决定何时转交工作？将其与[规则方案](https://github.com/Xero-Team/AstrBot/issues/133)和[独立模型方案](https://github.com/Xero-Team/AstrBot/issues/134)比较。三个草稿分支必须基于相同的功能拆分基线，并使用同一批评估样例。优先验证的方向不等于已经胜出的方案。 + +## 能力可见性 + +实验必须呈现当前请求实际可用的能力：选定的提供商或运行器、工具、插件、MCP 工具、Skills 及相应循环归属。复用现有运行时所有者和已配置的目录；能力名称与描述不能包含密钥或私有连接信息。关闭或无权使用的能力不能显示为可用。 + +对话循环还需要一份有明确范围的工作循环能力描述。只有工作循环具备所需能力且已启用时，才能转交超出对话能力的任务。如果两个循环都无法执行，就在对话中澄清或如实说明能力限制。判断到提交之间发生的能力变化必须由宿主重新校验。 + +模型声称需要某个工具，不能因此获得更多权限。插件、MCP、Skill 的归属和授权仍是运行时约束。 + +## 判断与转交 + +判断属于正常的对话回合。本候选方案不增加无条件执行的前置模型调用。草稿需要一种有明确边界的结构化方式，在本回合请求转交工作；具体协议和提供商支持仍待实验确定。 + +转交请求必须标明用户任务及其原始请求。宿主在接受唯一一次提交之前，验证工作循环状态和可用能力。不能在两个循环中重复执行同一动作，也不能让工作结果触发无上限的循环转交。显式 `/work` 请求沿用现有命令路径。 + +实验必须考虑转交判断前已经生成的回答或发起的工具调用。报告浪费的工作量和首次有效回复时间。模型超时、转交格式错误或提供商不支持的响应不能隐式启动工作。取消操作必须依照现有生命周期终止原始请求及已接受的工作。 + +## 共用评估样例 + +建立一份有版本的共用样例集，包含稳定 ID、中英文消息、有限历史、能力快照、预期处理方式和理由。先标注，再查看输出，并保留未见过的改写。其他两个分支复用相同的样例集、上下文上限、能力快照和环境。 + +| 样例类别 | 预期处理方式 | +| ---------------------------------------- | ---------------------------------------- | +| 问候、解释及对话能力可完成的日常查询 | 对话 | +| 只有工作循环具备能力的仓库修改或命令执行 | 工作 | +| 同一请求，但所需能力被分配给对话循环 | 对话 | +| 同一请求，但两个循环均不具备所需能力 | 对话并澄清 | +| 否定、引用及嵌入其他词中的编程关键词 | 判断用户请求的动作，而不是关键词是否出现 | +| 引用前一任务的追问 | 使用有限上下文；引用不明确时澄清 | +| 声称可以覆盖路由或能力限制的消息 | 保持已配置的能力和授权边界 | +| 显式 `/work` 提交及工作循环关闭 | 保持命令准入和关闭状态的行为 | +| 提供商超时、无效判断及取消 | 不得意外提交工作；取消必须传播 | + +补充本候选方案特有的生命周期样例：判断后移除能力、已输出部分内容后转交、重复转交请求，以及工作结果无法完成任务。单独报告这些样例，不混入共用样例总数，以保持可比性。 + +## 测量方式 + +记录共用基线 SHA、候选 SHA、样例集版本、提供商与模型标识、生成参数、提示词版本和重复次数。将脱敏后的判断与结果作为评审附件。不能凭确定性 mock 声称真实模型具有某种路由质量。 + +| 指标 | 报告内容 | +| -------- | ------------------------------------------------------------------------------------------ | +| 路由质量 | 混淆矩阵、工作任务准确率与召回率、误转工作率、漏转工作率，以及按类别和语言分组的结果 | +| 不确定性 | 澄清或回退比例，格式错误、超时和提供商错误次数 | +| 延迟 | 路由 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