feat: 接入 Shipyard Neo 自迭代 Skill 闭环与管理能力#5028
Merged
RC-CHN merged 54 commits intoAstrBotDevs:masterfrom Mar 2, 2026
Merged
Conversation
dosubot
bot
added
the
size:XXL
Closed
5 tasks
|
Documentation Updates 1 document(s) were updated by changes in this PR: pr4697的改动View Changes@@ -67,6 +67,9 @@
- `PYTHON_TOOL`：在沙盒环境中执行 Python 代码
- `FILE_UPLOAD_TOOL`：上传文件到沙盒环境
- `FILE_DOWNLOAD_TOOL`：从沙盒环境下载文件
+ - `BROWSER_EXECUTE_TOOL`（仅 `shipyard_neo`）：在沙盒环境中执行浏览器自动化命令
+ - `BROWSER_BATCH_EXECUTE_TOOL`（仅 `shipyard_neo`）：批量执行浏览器命令
+ - `RUN_BROWSER_SKILL_TOOL`（仅 `shipyard_neo`）：运行已发布的浏览器技能
- **`runtime="local"`** 或 **`runtime="local_sandboxed"`** 提供以下工具：
- `LOCAL_EXECUTE_SHELL_TOOL`：在本地环境中执行 Shell 命令
@@ -76,6 +79,23 @@
- 在 `local_sandboxed` 模式下，文件写入受限于工作区目录（`~/.astrbot/workspace/<session>`）
这些工具在 SubAgent handoff 场景下可正常使用，与主 Agent 运行时动态挂载的工具保持一致。
+
+**Shipyard Neo 运行时（PR #5028）**
+
+[PR #5028](https://github.com/AstrBotDevs/AstrBot/pull/5028) 新增了 `shipyard_neo` 沙箱运行时选项，基于 Shipyard Neo（Bay）架构，提供浏览器自动化和技能自迭代能力：
+
+- **默认运行时**：`shipyard_neo` 现为默认沙箱运行时（`provider_settings.sandbox.booter` 默认值从 `shipyard` 改为 `shipyard_neo`）
+- **能力支持**：python、shell、filesystem、browser（取决于所选 profile）
+- **Profile 自动选择**：系统会查询 Bay 服务的可用 profile 并自动选择能力最多的 profile（优先包含 browser 能力），用户也可显式指定
+- **自动启动模式**：当 `shipyard_neo_endpoint` 为空或设为 `__auto__` 时，系统会自动启动 Bay 容器（类似 Boxlite 自动启动 Ship 容器）
+- **凭证自动发现**：如果 `shipyard_neo_access_token` 未配置，系统会自动从 Bay 的 `credentials.json` 文件中读取 API 密钥（搜索路径：`BAY_DATA_DIR` 环境变量、mono-repo 相对路径 `../pkgs/bay/`、当前工作目录）
+
+**配置项（provider_settings.sandbox）**：
+
+- `shipyard_neo_endpoint`：Bay 服务的 API 地址，默认 `http://127.0.0.1:8114`（留空或设为 `__auto__` 启用自动启动模式）
+- `shipyard_neo_access_token`：Bay 的 API 密钥（`sk-bay-...`），留空时自动从 `credentials.json` 发现
+- `shipyard_neo_profile`：沙箱 profile 名称，如 `python-default`（留空时自动选择能力最多的 profile）
+- `shipyard_neo_ttl`：沙箱生存时间（秒），默认 3600
**本地沙箱增强模式（local_sandboxed）**
@@ -290,6 +310,51 @@
- 动态注册工具时需确保配置同步更新
- 后台任务需正确设置 `is_background_task: true`
+#### Neo 技能生命周期工具（PR #5028）
+
+[PR #5028](https://github.com/AstrBotDevs/AstrBot/pull/5028) 新增了一套技能生命周期管理工具，支持在 `shipyard_neo` 运行时下进行技能自迭代闭环（执行取证 → payload/candidate → evaluate → promote → stable 回写本地 `SKILL.md`）。这些工具仅在使用 `shipyard_neo` 运行时且用户为管理员时可用。
+
+**浏览器自动化工具**：
+
+- `astrbot_execute_browser`：在沙盒中执行单条浏览器自动化命令
+- `astrbot_execute_browser_batch`：批量执行浏览器命令
+- `astrbot_run_browser_skill`：运行已发布的浏览器技能（通过 skill_key 调用）
+
+**执行历史与标注**：
+
+- `astrbot_get_execution_history`：获取沙箱执行历史（支持按执行类型、成功状态、标签等过滤）
+- `astrbot_annotate_execution`：为执行历史记录添加描述、标签、备注
+
+**技能自迭代工作流（三步法）**：
+
+1. **步骤 1 - 创建 Payload**：
+ - `astrbot_create_skill_payload`：创建不可变的技能 payload 内容（包含 `skill_markdown`、inputs、outputs、meta 等），返回 `payload_ref`
+ - `astrbot_get_skill_payload`：通过 `payload_ref` 获取 payload 详情
+
+2. **步骤 2 - 创建 Candidate**：
+ - `astrbot_create_skill_candidate`：通过绑定执行证据（`source_execution_ids`）和技能标识（`skill_key`）创建候选版本，可选关联 `payload_ref`
+ - `astrbot_list_skill_candidates`：列出技能候选版本（支持按状态、skill_key 过滤）
+
+3. **步骤 3 - 评估与发布**：
+ - `astrbot_evaluate_skill_candidate`：评估候选版本（标记 passed/failed、打分、附带评估报告）
+ - `astrbot_promote_skill_candidate`：将候选版本提升为 canary 或 stable 发布
+ - 当 `stage=stable` 且 `sync_to_local=true` 时，系统会自动将 `payload.skill_markdown` 回写到本地 `data/skills/<local_skill_name>/SKILL.md`
+ - 回写失败时自动执行 rollback，确保 Neo 真源与本地可用状态一致
+ - `astrbot_list_skill_releases`：列出技能发布版本
+ - `astrbot_rollback_skill_release`：回滚技能发布版本
+ - `astrbot_sync_skill_release`：手动同步 stable 发布到本地 SKILL.md
+
+**Neo 真源语义与技能同步**：
+
+- Neo 作为技能版本的"真源"（single source of truth），stable 发布会自动回写到本地 `SKILL.md`
+- 系统维护 `data/skills/neo_skill_map.json` 映射文件，记录 `skill_key` 与本地技能名称、release_id、candidate_id、payload_ref 的对应关系
+- 回写后会自动激活本地技能，并最佳努力刷新活跃沙箱中的技能内容
+- 技能名称规范化：Neo skill_key 会转换为合法的本地技能名称（`neo_<normalized_key>`），确保唯一性
+
+**技能同步行为变更**：
+
+PR #5028 修改了 `_sync_skills_to_sandbox` 的同步策略，从增量同步（`unzip -n`）改为覆盖同步，确保沙箱中的技能不会因增量策略而长期陈旧。
+
---
### 5. UI 和 API 支持
@@ -313,6 +378,32 @@
#### API 支持
新增 CronRoute 等 API 路由，支持通过 API 管理定时任务。
+
+#### Neo 技能管理 UI（PR #5028）
+
+Dashboard Skills 页面新增 Neo 技能视图切换和管理能力：
+
+- **Local / Neo 视图切换**：支持在本地技能（Local Skills）和 Neo 托管技能（Neo Skills）之间切换
+- **Neo Candidates 列表**：显示技能候选版本，支持状态过滤（pending、evaluated、promoted 等）
+- **Neo Releases 列表**：显示技能发布版本，支持 stage 过滤（canary、stable）
+- **关键动作**：
+ - Evaluate：评估候选版本
+ - Promote：将候选版本提升为 canary 或 stable 发布
+ - Rollback：回滚发布版本
+ - Sync：手动同步 stable 发布到本地 SKILL.md
+ - View Payload：查看技能 payload 详情
+
+**Neo Skills API 路由**：
+
+- `GET /api/skills/neo/candidates`：列出技能候选版本
+- `GET /api/skills/neo/releases`：列出技能发布版本
+- `GET /api/skills/neo/payload`：获取 payload 详情
+- `POST /api/skills/neo/evaluate`：评估候选版本
+- `POST /api/skills/neo/promote`：提升候选版本为发布
+- `POST /api/skills/neo/rollback`：回滚发布版本
+- `POST /api/skills/neo/sync`：同步发布到本地
+
+中英文 i18n 文案已全面覆盖 Neo 技能管理功能。
---
How did I do? Any feedback? |
w31r4
changed the title
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了 4 个问题，并给出了一些整体反馈：
- 你在
pyproject.toml中添加了对shipyard-neo-sdk的直接引用，但没有在requirements.txt中添加；如果你的部署或开发流程仍依赖requirements.txt，建议也在其中添加相同依赖，以保持两个清单文件的一致性。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- You added `shipyard-neo-sdk` as a direct reference in `pyproject.toml` but not in `requirements.txt`; if your deployment or dev workflows still rely on `requirements.txt`, consider adding the same dependency there to keep the two manifests consistent.
## Individual Comments
### Comment 1
<location> `astrbot/dashboard/routes/skills.py:304-313` </location>
<code_context>
+ sync_json: dict[str, Any] | None = None
+ rollback_json: dict[str, Any] | None = None
+ if stage == "stable" and sync_to_local:
+ sync_mgr = NeoSkillSyncManager()
+ try:
+ sync_result = await sync_mgr.sync_release(
</code_context>
<issue_to_address>
**suggestion (performance):** Stable promotion triggers skills sync twice (inside sync manager and again after), which is redundant.
When `stage == "stable" and sync_to_local`, `NeoSkillSyncManager().sync_release(...)` already triggers `sync_skills_to_active_sandboxes()`, and this route then calls `sync_skills_to_active_sandboxes()` again in the best-effort block. That means the same change is synced twice in a row. Consider relying on just one of these (either the manager’s call or the explicit call here) to avoid redundant work.
Suggested implementation:
```python
sync_json = None
did_sync_to_local = False
if stage == "stable" and sync_to_local:
sync_mgr = NeoSkillSyncManager()
try:
sync_result = await sync_mgr.sync_release(
client,
release_id=str(release_json.get("id", "")),
require_stable=True,
)
did_sync_to_local = True
sync_json = {
"skill_key": sync_result.skill_key,
"local_skill_name": sync_result.local_skill_name,
"release_id": sync_result.release_id,
"candidate_id": sync_result.candidate_id,
"payload_ref": sync_result.payload_ref,
```
To fully avoid the double sync, you should also adjust the later "best-effort" sync call in this route:
1. Locate the block that currently calls `sync_skills_to_active_sandboxes()` (likely in a `try/except` "best-effort" section after the snippet you provided).
2. Wrap that call so it only runs when `did_sync_to_local` is `False`, for example:
```python
if not did_sync_to_local:
await sync_skills_to_active_sandboxes()
```
This way, when `stage == "stable" and sync_to_local` and `NeoSkillSyncManager().sync_release(...)` has already performed the sync, the route will skip the redundant second call.
</issue_to_address>
### Comment 2
<location> `astrbot/core/computer/booters/base.py:16-22` </location>
<code_context>
@property
def shell(self) -> ShellComponent: ...
+ @property
+ def browser(self) -> BrowserComponent:
+ raise NotImplementedError(
+ f"{self.__class__.__name__} does not support browser capability."
+ )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** The base `browser` property raising `NotImplementedError` conflicts with `getattr(..., "browser", None)` checks in callers.
Because callers like `_get_browser_component` use `getattr(booter, "browser", None)` for capability detection, this property will raise instead of returning `None`, so the `if browser is None` path never executes. While the outer `try/except` masks the exception, it makes feature detection brittle and hides the clearer error message. Consider either returning `None` in the base implementation and overriding it in concrete booters, or updating callers to explicitly handle `NotImplementedError` when probing for `browser` support.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/skills.py:171` </location>
<code_context>
logger.error(traceback.format_exc())
return Response().error(str(e)).__dict__
+
+ async def get_neo_candidates(self):
+ try:
+ endpoint, access_token = self._get_neo_client_config()
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared helpers for Neo client usage and sync result formatting to simplify the SkillsRoute handlers and remove duplication.
You can keep all behavior but reduce complexity/duplication inside `SkillsRoute` by factoring out the repeated “get config → open BayClient → call API → shape response” pattern and the repeated sync-result shaping.
### 1. Centralize BayClient construction and error handling
The Neo endpoints all do:
- `_get_neo_client_config`
- import `BayClient`
- `async with BayClient(...) as client: ...`
- try/except with identical logging + error response
You can move this into a small helper to remove repetition and keep each handler focused on *what* it does rather than *how* it talks to Neo:
```python
from shipyard_neo import BayClient # move import to top-level
class SkillsRoute(Route):
...
async def _with_neo_client(self, func):
try:
endpoint, access_token = self._get_neo_client_config()
async with BayClient(
endpoint_url=endpoint,
access_token=access_token,
) as client:
return await func(client)
except Exception as e:
logger.error(traceback.format_exc())
return Response().error(str(e)).__dict__
```
Usage example (applies to `get_neo_candidates`, `get_neo_releases`, `get_neo_payload`, `evaluate_neo_candidate`, `promote_neo_candidate`, `rollback_neo_release`, `sync_neo_release`):
```python
async def get_neo_candidates(self):
status = request.args.get("status")
skill_key = request.args.get("skill_key")
limit = int(request.args.get("limit", 100))
offset = int(request.args.get("offset", 0))
async def _do(client):
candidates = await client.skills.list_candidates(
status=status,
skill_key=skill_key,
limit=limit,
offset=offset,
)
return Response().ok(_to_jsonable(candidates)).__dict__
return await self._with_neo_client(_do)
```
This keeps route methods short and removes boilerplate.
### 2. Factor out the sync result shaping
`promote_neo_candidate` and `sync_neo_release` both build the same dict from `NeoSkillSyncResult`:
```python
{
"skill_key": result.skill_key,
"local_skill_name": result.local_skill_name,
"release_id": result.release_id,
"candidate_id": result.candidate_id,
"payload_ref": result.payload_ref,
"map_path": result.map_path,
"synced_at": result.synced_at,
}
```
You can extract a tiny helper to avoid drift and keep both endpoints consistent:
```python
def _sync_result_to_dict(self, result):
return {
"skill_key": result.skill_key,
"local_skill_name": result.local_skill_name,
"release_id": result.release_id,
"candidate_id": result.candidate_id,
"payload_ref": result.payload_ref,
"map_path": result.map_path,
"synced_at": result.synced_at,
}
```
Then:
```python
# in promote_neo_candidate
sync_json = None
if stage == "stable" and sync_to_local:
sync_result = await sync_mgr.sync_release(...)
sync_json = self._sync_result_to_dict(sync_result)
# in sync_neo_release
result = await sync_mgr.sync_release(...)
return Response().ok(self._sync_result_to_dict(result)).__dict__
```
These two small extractions reduce duplication and make the route methods materially simpler without changing any behavior or moving to a full-blown service layer.
</issue_to_address>
### Comment 4
<location> `astrbot/core/computer/tools/neo_skills.py:51` </location>
<code_context>
+ return browser
+
+
+@dataclass
+class BrowserExecTool(FunctionTool):
+ name: str = "astrbot_execute_browser"
</code_context>
<issue_to_address>
**issue (complexity):** Consider introducing a shared Neo tool base class and moving promote/sync/rollback orchestration into NeoSkillSyncManager to eliminate repeated admin/client/error/JSON boilerplate and centralize lifecycle logic.
You can reduce the complexity and duplication meaningfully without changing behavior by introducing a shared base and centralizing the promotion/sync/rollback orchestration.
### 1. Factor out a common Neo tool base
All tools repeat the same pattern:
- ` _ensure_admin`
- `_get_neo_context`
- try/except → call SDK → JSON-ify or format error
You can keep behavior identical but move the plumbing into a base class so each tool only defines its parameters and the actual SDK call.
```python
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
@dataclass
class NeoSkillToolBase(FunctionTool):
error_prefix: str = "Error"
async def _run(
self,
context: ContextWrapper[AstrAgentContext],
neo_call: Callable[[Any, Any], Awaitable[Any]],
error_action: str,
) -> ToolExecResult:
if err := _ensure_admin(context):
return err
try:
client, sandbox = await _get_neo_context(context)
result = await neo_call(client, sandbox)
return _to_json_text(result)
except Exception as e:
return f"{self.error_prefix} {error_action}: {str(e)}"
```
Then each tool becomes much smaller; for example, `ListSkillReleasesTool`:
```python
@dataclass
class ListSkillReleasesTool(NeoSkillToolBase):
name: str = "astrbot_list_skill_releases"
description: str = "List skill releases."
parameters: dict = field(
default_factory=lambda: {
"type": "object",
"properties": {
"skill_key": {"type": "string"},
"active_only": {"type": "boolean", "default": False},
"stage": {"type": "string"},
"limit": {"type": "integer", "default": 100},
"offset": {"type": "integer", "default": 0},
},
"required": [],
}
)
async def call(
self,
context: ContextWrapper[AstrAgentContext],
skill_key: str | None = None,
active_only: bool = False,
stage: str | None = None,
limit: int = 100,
offset: int = 0,
) -> ToolExecResult:
return await self._run(
context,
lambda client, _sandbox: client.skills.list_releases(
skill_key=skill_key,
active_only=active_only,
stage=stage,
limit=limit,
offset=offset,
),
error_action="listing skill releases",
)
```
You can gradually migrate each tool to this pattern; it keeps the same signatures and behavior but removes repeated admin/client/error/JSON boilerplate.
### 2. Centralize promote/sync/rollback orchestration
`PromoteSkillCandidateTool` contains orchestration logic that’s already conceptually part of `NeoSkillSyncManager`/lifecycle concerns. You can move the flow into a helper function/service and keep the tool thin.
For example, add a method to `NeoSkillSyncManager` (or a small service next to it):
```python
class NeoSkillSyncManager:
# existing methods...
async def promote_with_optional_sync(
self,
client: Any,
candidate_id: str,
stage: str,
sync_to_local: bool,
) -> dict[str, Any]:
release = await client.skills.promote_candidate(candidate_id, stage=stage)
release_json = _to_jsonable(release)
sync_json: dict[str, Any] | None = None
rollback_json: dict[str, Any] | None = None
if stage == "stable" and sync_to_local:
try:
sync_result = await self.sync_release(
client,
release_id=str(release_json.get("id", "")),
require_stable=True,
)
sync_json = {
"skill_key": sync_result.skill_key,
"local_skill_name": sync_result.local_skill_name,
"release_id": sync_result.release_id,
"candidate_id": sync_result.candidate_id,
"payload_ref": sync_result.payload_ref,
"map_path": sync_result.map_path,
"synced_at": sync_result.synced_at,
}
except Exception as sync_err:
try:
rollback = await client.skills.rollback_release(
str(release_json.get("id", ""))
)
rollback_json = _to_jsonable(rollback)
except Exception as rollback_err:
raise RuntimeError(
"stable release sync failed and auto rollback also failed; "
f"sync_error={sync_err}; rollback_error={rollback_err}"
) from rollback_err
raise RuntimeError(
"stable release sync failed; auto rollback succeeded; "
f"sync_error={sync_err}; rollback={_to_json_text(rollback_json)}"
) from sync_err
return {
"release": release_json,
"sync": sync_json,
"rollback": rollback_json,
}
```
Then `PromoteSkillCandidateTool.call` becomes:
```python
@dataclass
class PromoteSkillCandidateTool(NeoSkillToolBase):
name: str = "astrbot_promote_skill_candidate"
description: str = "Promote one candidate to release stage (canary/stable)."
# parameters unchanged ...
async def call(
self,
context: ContextWrapper[AstrAgentContext],
candidate_id: str,
stage: str = "canary",
sync_to_local: bool = True,
) -> ToolExecResult:
if stage not in {"canary", "stable"}:
return "Error promoting skill candidate: stage must be canary or stable."
if err := _ensure_admin(context):
return err
try:
client, _sandbox = await _get_neo_context(context)
sync_mgr = NeoSkillSyncManager()
result = await sync_mgr.promote_with_optional_sync(
client=client,
candidate_id=candidate_id,
stage=stage,
sync_to_local=sync_to_local,
)
return _to_json_text(result)
except Exception as e:
return f"Error promoting skill candidate: {str(e)}"
```
This keeps all current behavior (including rollback semantics and error messages, if you preserve them in the manager) but:
- Removes duplicated orchestration from the tool.
- Gives you a single place (`NeoSkillSyncManager`) to align behavior between tools and HTTP routes.
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续的评审。
Original comment in English
Hey - I've found 4 issues, and left some high level feedback:
- You added
shipyard-neo-sdkas a direct reference inpyproject.tomlbut not inrequirements.txt; if your deployment or dev workflows still rely onrequirements.txt, consider adding the same dependency there to keep the two manifests consistent.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- You added `shipyard-neo-sdk` as a direct reference in `pyproject.toml` but not in `requirements.txt`; if your deployment or dev workflows still rely on `requirements.txt`, consider adding the same dependency there to keep the two manifests consistent.
## Individual Comments
### Comment 1
<location> `astrbot/dashboard/routes/skills.py:304-313` </location>
<code_context>
+ sync_json: dict[str, Any] | None = None
+ rollback_json: dict[str, Any] | None = None
+ if stage == "stable" and sync_to_local:
+ sync_mgr = NeoSkillSyncManager()
+ try:
+ sync_result = await sync_mgr.sync_release(
</code_context>
<issue_to_address>
**suggestion (performance):** Stable promotion triggers skills sync twice (inside sync manager and again after), which is redundant.
When `stage == "stable" and sync_to_local`, `NeoSkillSyncManager().sync_release(...)` already triggers `sync_skills_to_active_sandboxes()`, and this route then calls `sync_skills_to_active_sandboxes()` again in the best-effort block. That means the same change is synced twice in a row. Consider relying on just one of these (either the manager’s call or the explicit call here) to avoid redundant work.
Suggested implementation:
```python
sync_json = None
did_sync_to_local = False
if stage == "stable" and sync_to_local:
sync_mgr = NeoSkillSyncManager()
try:
sync_result = await sync_mgr.sync_release(
client,
release_id=str(release_json.get("id", "")),
require_stable=True,
)
did_sync_to_local = True
sync_json = {
"skill_key": sync_result.skill_key,
"local_skill_name": sync_result.local_skill_name,
"release_id": sync_result.release_id,
"candidate_id": sync_result.candidate_id,
"payload_ref": sync_result.payload_ref,
```
To fully avoid the double sync, you should also adjust the later "best-effort" sync call in this route:
1. Locate the block that currently calls `sync_skills_to_active_sandboxes()` (likely in a `try/except` "best-effort" section after the snippet you provided).
2. Wrap that call so it only runs when `did_sync_to_local` is `False`, for example:
```python
if not did_sync_to_local:
await sync_skills_to_active_sandboxes()
```
This way, when `stage == "stable" and sync_to_local` and `NeoSkillSyncManager().sync_release(...)` has already performed the sync, the route will skip the redundant second call.
</issue_to_address>
### Comment 2
<location> `astrbot/core/computer/booters/base.py:16-22` </location>
<code_context>
@property
def shell(self) -> ShellComponent: ...
+ @property
+ def browser(self) -> BrowserComponent:
+ raise NotImplementedError(
+ f"{self.__class__.__name__} does not support browser capability."
+ )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** The base `browser` property raising `NotImplementedError` conflicts with `getattr(..., "browser", None)` checks in callers.
Because callers like `_get_browser_component` use `getattr(booter, "browser", None)` for capability detection, this property will raise instead of returning `None`, so the `if browser is None` path never executes. While the outer `try/except` masks the exception, it makes feature detection brittle and hides the clearer error message. Consider either returning `None` in the base implementation and overriding it in concrete booters, or updating callers to explicitly handle `NotImplementedError` when probing for `browser` support.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/skills.py:171` </location>
<code_context>
logger.error(traceback.format_exc())
return Response().error(str(e)).__dict__
+
+ async def get_neo_candidates(self):
+ try:
+ endpoint, access_token = self._get_neo_client_config()
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared helpers for Neo client usage and sync result formatting to simplify the SkillsRoute handlers and remove duplication.
You can keep all behavior but reduce complexity/duplication inside `SkillsRoute` by factoring out the repeated “get config → open BayClient → call API → shape response” pattern and the repeated sync-result shaping.
### 1. Centralize BayClient construction and error handling
The Neo endpoints all do:
- `_get_neo_client_config`
- import `BayClient`
- `async with BayClient(...) as client: ...`
- try/except with identical logging + error response
You can move this into a small helper to remove repetition and keep each handler focused on *what* it does rather than *how* it talks to Neo:
```python
from shipyard_neo import BayClient # move import to top-level
class SkillsRoute(Route):
...
async def _with_neo_client(self, func):
try:
endpoint, access_token = self._get_neo_client_config()
async with BayClient(
endpoint_url=endpoint,
access_token=access_token,
) as client:
return await func(client)
except Exception as e:
logger.error(traceback.format_exc())
return Response().error(str(e)).__dict__
```
Usage example (applies to `get_neo_candidates`, `get_neo_releases`, `get_neo_payload`, `evaluate_neo_candidate`, `promote_neo_candidate`, `rollback_neo_release`, `sync_neo_release`):
```python
async def get_neo_candidates(self):
status = request.args.get("status")
skill_key = request.args.get("skill_key")
limit = int(request.args.get("limit", 100))
offset = int(request.args.get("offset", 0))
async def _do(client):
candidates = await client.skills.list_candidates(
status=status,
skill_key=skill_key,
limit=limit,
offset=offset,
)
return Response().ok(_to_jsonable(candidates)).__dict__
return await self._with_neo_client(_do)
```
This keeps route methods short and removes boilerplate.
### 2. Factor out the sync result shaping
`promote_neo_candidate` and `sync_neo_release` both build the same dict from `NeoSkillSyncResult`:
```python
{
"skill_key": result.skill_key,
"local_skill_name": result.local_skill_name,
"release_id": result.release_id,
"candidate_id": result.candidate_id,
"payload_ref": result.payload_ref,
"map_path": result.map_path,
"synced_at": result.synced_at,
}
```
You can extract a tiny helper to avoid drift and keep both endpoints consistent:
```python
def _sync_result_to_dict(self, result):
return {
"skill_key": result.skill_key,
"local_skill_name": result.local_skill_name,
"release_id": result.release_id,
"candidate_id": result.candidate_id,
"payload_ref": result.payload_ref,
"map_path": result.map_path,
"synced_at": result.synced_at,
}
```
Then:
```python
# in promote_neo_candidate
sync_json = None
if stage == "stable" and sync_to_local:
sync_result = await sync_mgr.sync_release(...)
sync_json = self._sync_result_to_dict(sync_result)
# in sync_neo_release
result = await sync_mgr.sync_release(...)
return Response().ok(self._sync_result_to_dict(result)).__dict__
```
These two small extractions reduce duplication and make the route methods materially simpler without changing any behavior or moving to a full-blown service layer.
</issue_to_address>
### Comment 4
<location> `astrbot/core/computer/tools/neo_skills.py:51` </location>
<code_context>
+ return browser
+
+
+@dataclass
+class BrowserExecTool(FunctionTool):
+ name: str = "astrbot_execute_browser"
</code_context>
<issue_to_address>
**issue (complexity):** Consider introducing a shared Neo tool base class and moving promote/sync/rollback orchestration into NeoSkillSyncManager to eliminate repeated admin/client/error/JSON boilerplate and centralize lifecycle logic.
You can reduce the complexity and duplication meaningfully without changing behavior by introducing a shared base and centralizing the promotion/sync/rollback orchestration.
### 1. Factor out a common Neo tool base
All tools repeat the same pattern:
- ` _ensure_admin`
- `_get_neo_context`
- try/except → call SDK → JSON-ify or format error
You can keep behavior identical but move the plumbing into a base class so each tool only defines its parameters and the actual SDK call.
```python
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
@dataclass
class NeoSkillToolBase(FunctionTool):
error_prefix: str = "Error"
async def _run(
self,
context: ContextWrapper[AstrAgentContext],
neo_call: Callable[[Any, Any], Awaitable[Any]],
error_action: str,
) -> ToolExecResult:
if err := _ensure_admin(context):
return err
try:
client, sandbox = await _get_neo_context(context)
result = await neo_call(client, sandbox)
return _to_json_text(result)
except Exception as e:
return f"{self.error_prefix} {error_action}: {str(e)}"
```
Then each tool becomes much smaller; for example, `ListSkillReleasesTool`:
```python
@dataclass
class ListSkillReleasesTool(NeoSkillToolBase):
name: str = "astrbot_list_skill_releases"
description: str = "List skill releases."
parameters: dict = field(
default_factory=lambda: {
"type": "object",
"properties": {
"skill_key": {"type": "string"},
"active_only": {"type": "boolean", "default": False},
"stage": {"type": "string"},
"limit": {"type": "integer", "default": 100},
"offset": {"type": "integer", "default": 0},
},
"required": [],
}
)
async def call(
self,
context: ContextWrapper[AstrAgentContext],
skill_key: str | None = None,
active_only: bool = False,
stage: str | None = None,
limit: int = 100,
offset: int = 0,
) -> ToolExecResult:
return await self._run(
context,
lambda client, _sandbox: client.skills.list_releases(
skill_key=skill_key,
active_only=active_only,
stage=stage,
limit=limit,
offset=offset,
),
error_action="listing skill releases",
)
```
You can gradually migrate each tool to this pattern; it keeps the same signatures and behavior but removes repeated admin/client/error/JSON boilerplate.
### 2. Centralize promote/sync/rollback orchestration
`PromoteSkillCandidateTool` contains orchestration logic that’s already conceptually part of `NeoSkillSyncManager`/lifecycle concerns. You can move the flow into a helper function/service and keep the tool thin.
For example, add a method to `NeoSkillSyncManager` (or a small service next to it):
```python
class NeoSkillSyncManager:
# existing methods...
async def promote_with_optional_sync(
self,
client: Any,
candidate_id: str,
stage: str,
sync_to_local: bool,
) -> dict[str, Any]:
release = await client.skills.promote_candidate(candidate_id, stage=stage)
release_json = _to_jsonable(release)
sync_json: dict[str, Any] | None = None
rollback_json: dict[str, Any] | None = None
if stage == "stable" and sync_to_local:
try:
sync_result = await self.sync_release(
client,
release_id=str(release_json.get("id", "")),
require_stable=True,
)
sync_json = {
"skill_key": sync_result.skill_key,
"local_skill_name": sync_result.local_skill_name,
"release_id": sync_result.release_id,
"candidate_id": sync_result.candidate_id,
"payload_ref": sync_result.payload_ref,
"map_path": sync_result.map_path,
"synced_at": sync_result.synced_at,
}
except Exception as sync_err:
try:
rollback = await client.skills.rollback_release(
str(release_json.get("id", ""))
)
rollback_json = _to_jsonable(rollback)
except Exception as rollback_err:
raise RuntimeError(
"stable release sync failed and auto rollback also failed; "
f"sync_error={sync_err}; rollback_error={rollback_err}"
) from rollback_err
raise RuntimeError(
"stable release sync failed; auto rollback succeeded; "
f"sync_error={sync_err}; rollback={_to_json_text(rollback_json)}"
) from sync_err
return {
"release": release_json,
"sync": sync_json,
"rollback": rollback_json,
}
```
Then `PromoteSkillCandidateTool.call` becomes:
```python
@dataclass
class PromoteSkillCandidateTool(NeoSkillToolBase):
name: str = "astrbot_promote_skill_candidate"
description: str = "Promote one candidate to release stage (canary/stable)."
# parameters unchanged ...
async def call(
self,
context: ContextWrapper[AstrAgentContext],
candidate_id: str,
stage: str = "canary",
sync_to_local: bool = True,
) -> ToolExecResult:
if stage not in {"canary", "stable"}:
return "Error promoting skill candidate: stage must be canary or stable."
if err := _ensure_admin(context):
return err
try:
client, _sandbox = await _get_neo_context(context)
sync_mgr = NeoSkillSyncManager()
result = await sync_mgr.promote_with_optional_sync(
client=client,
candidate_id=candidate_id,
stage=stage,
sync_to_local=sync_to_local,
)
return _to_json_text(result)
except Exception as e:
return f"Error promoting skill candidate: {str(e)}"
```
This keeps all current behavior (including rollback semantics and error messages, if you preserve them in the manager) but:
- Removes duplicated orchestration from the tool.
- Gives you a single place (`NeoSkillSyncManager`) to align behavior between tools and HTTP routes.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
dosubot
bot
added
area:core
w31r4
changed the title
w31r4
changed the title
Contributor
Author
|
补充说明（仅针对最近一个提交）： 本次修补主要解决“本地上传 skill 与 ship/gull 预置 skill 的共存与同步一致性”问题：
本提交已本地验证：
|
…tions Add [Computer] prefixed INFO logs to: - shipyard_neo.py: shutdown, upload_file, download_file, available - shipyard.py: shutdown, upload_file, download_file, available - boxlite.py: upload_file success path - computer_client.py: sync_skills_to_active_sandboxes, _sync_skills_to_sandbox Improves traceability of sandbox lifecycle events.
Add _log_computer_config_changes() to detect and log modifications to computer_use_runtime and sandbox.* keys when saving config via Dashboard. Sensitive fields (tokens/secrets) are masked in log output.
Add scripts/start-with-neo.sh: one-click launcher that auto-generates Bay config.yaml (anonymous mode, host_port), pulls Ship image, starts Bay (port 8114) with health check, then starts AstrBot in foreground. Ctrl+C stops both services. Supports BAY_PORT env var override.
- Endpoint hint: mention default port 8114 - Access Token hint: mention sk-bay-* format and credentials.json auto-discovery - Updated in default.py, zh-CN, and en-US i18n files
- Generated config uses allow_anonymous: false (triggers auto-provision) - Set BAY_DATA_DIR so credentials.json writes to pkgs/bay/ - Add read_bay_credentials() to extract auto-generated key after boot - Display API key in config hints for easy AstrBot setup
19 new tests in test_computer_config.py: - TestDiscoverBayCredentials (9 tests): env priority, cwd fallback, missing file, empty key, malformed JSON, endpoint mismatch, slash normalization - TestLogComputerConfigChanges (10 tests): runtime change, sandbox key change, token masking, empty token label, missing provider_settings, add/remove keys
Document make pr-test-neo and make pr-test-full commands for local CI-equivalent verification before submitting PRs.
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Also add shipyard-neo-sdk dependency for neo support
- Rewrite build_skills_prompt() with structured numbered rules and markdown formatting for better LLM comprehension - Sanitize example_path with _SAFE_PATH_RE before embedding in system prompt to prevent prompt injection via crafted skill paths - Add docstring to _parse_frontmatter_description() - Remove debug print(top_dirs) from install_skill_from_zip() - Remove stale commented-out SANDBOX_SKILLS_ROOT line
- Resolve skills root via Path.resolve() so LLM prompts always reference absolute paths regardless of sandbox cwd - Use resolved path in skill metadata for reliable cat/head commands - Add DRY cross-reference comment for frontmatter parser - Remove dead skills_root_abs field from JSON output (no consumer) - Remove unnecessary os import and fake resolve/abspath branch
17 tests covering: - ShipyardNeoBooter.capabilities property (tuple, immutability, pre/post boot) - _apply_sandbox_tools conditional browser tool registration - _resolve_profile smart selection (user-specified, browser preference, API error fallback, empty profiles, auth error pass-through) - ComputerBooter base class defaults
11 tests covering: - _parse_frontmatter_description: standard, description-only, empty, missing delimiter, quoted values - build_skills_prompt: format, absolute path in example, progressive disclosure rules, absence of legacy custom fields - SkillManager.list_skills: local frontmatter parsing, sandbox cache description passthrough
append a Shipyard Neo-specific system prompt note for filesystem tool calls so paths are provided relative to the workspace root. this prevents models from prepending `/workspace` and causing tool path resolution failures
Normalize release stage values before stability checks so enum-like objects and mixed-case strings are handled consistently. When stable sync fails, treat "no previous release exists" during auto-rollback as a skipped rollback instead of raising a secondary runtime error
Normalize SKILL.md content during sync so each file includes name and description metadata in a frontmatter block. Preserve existing frontmatter values when present, derive description from markdown content when missing, and fallback to a default description to keep metadata complete and consistent.
Add backend routes to delete neo candidates and releases with optional reason support and demo mode protection. Expose delete actions in the Skills dashboard for candidate and release rows, refresh data after success, and add localized success/failure messages in en-US and zh-CN.
Disable the Neo mode toggle unless runtime is sandbox with shipyard_neo configured, and show a warning when Neo is unavailable. Also avoid loading Neo data when the environment is not compatible and fall back to local mode to prevent invalid requests and confusion.
Sanitize the `name` field in `SKILL.md` frontmatter to remove newlines and control characters. This prevents potential frontmatter injection vulnerabilities where a malicious skill name could introduce arbitrary YAML fields or corrupt the file structure. - Modified `_ensure_skill_frontmatter` in `astrbot/core/skills/neo_skill_sync.py` to normalize whitespace in `name`. - Ensured `name` is cast to string before splitting to handle non-string inputs safely.
Sanitized the name field in SKILL.md frontmatter within astrbot/core/skills/neo_skill_sync.py. This prevents potential frontmatter injection vulnerabilities by removing newlines and control characters from the skill name. Verified the fix with a reproduction script and ensured existing tests pass.
Feat/neo skill self iteration
Add explicit Neo lifecycle instructions to the main agent prompt so skill creation and updates follow payload -> candidate -> promotion instead of direct local folder writes. Clarify lifecycle tool descriptions and parameter semantics, including skill_key/source_execution_ids usage and stable release sync_to_local behavior, to reduce ambiguity and improve consistent skill publishing.
keep sandbox skill cache in sync when deleting a skill from disk. this prevents stale entries in the UI when no sandbox session is active to refresh runtime cache
default sandbox skill paths to /workspace/skills/<name>/SKILL.md when loading config and when exposing sandbox paths. preserve cached sandbox paths when available to avoid losing resolved locations for existing skills.
expose skill source metadata and sandbox cache status in the skills API response so the dashboard can distinguish local, sandbox-only, and synced skills. prevent enabling, disabling, or deleting sandbox-only preset skills in both backend guards and UI actions to avoid invalid local operations. add source badges, discovery-pending hinting for sandbox runtime, and new i18n strings for source labels and readonly warnings.
Soulter
approved these changes
Feb 28, 2026
…1r4/AstrBot into feat/neo-skill-self-iteration
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Modifications / 改动点
本 PR 在 AstrBot 中接入 Shipyard Neo，并落地“人工触发的 Skill 自迭代闭环”（执行取证 -> payload/candidate -> evaluate -> promote -> stable 回写本地
SKILL.md）。主要改动如下：
Sandbox 运行时基线
ShipyardNeoBooter，支持python/shell/filesystem/browser。computer_client增加shipyard_neo分支，保留旧shipyard兼容路径。_sync_skills_to_sandbox改为覆盖同步，避免旧unzip -n导致技能长期陈旧。配置与依赖
provider_settings.sandbox.booter增加shipyard_neo，并默认使用shipyard_neo。shipyard_neo_endpointshipyard_neo_access_tokenshipyard_neo_profileshipyard_neo_ttlshipyard-neo-sdk依赖，并同步更新requirements.txt与pyproject.toml。Agent 工具（仅
shipyard_neo挂载）astrbot_execute_browserastrbot_execute_browser_batchastrbot_run_browser_skillastrbot_get_execution_historyastrbot_annotate_executionastrbot_create_skill_payloadastrbot_get_skill_payloadastrbot_create_skill_candidateastrbot_list_skill_candidatesastrbot_evaluate_skill_candidateastrbot_promote_skill_candidateastrbot_list_skill_releasesastrbot_rollback_skill_releaseastrbot_sync_skill_releasestable 回写管理（Neo 真源语义）
NeoSkillSyncManager：data/skills/<local_skill_name>/SKILL.md。data/skills/neo_skill_map.json。stage=stable回写失败时自动执行 rollback，保证 Neo 真源与本地可用状态一致。Dashboard API 与 WebUI
GET /api/skills/neo/candidatesGET /api/skills/neo/releasesGET /api/skills/neo/payloadPOST /api/skills/neo/evaluatePOST /api/skills/neo/promotePOST /api/skills/neo/rollbackPOST /api/skills/neo/syncLocal / Neo视图切换、过滤和关键动作（evaluate/promote/rollback/sync/payload 查看）。This is NOT a breaking change. / 这不是一个破坏性变更。
Screenshots or Test Results / 运行截图或测试结果
验证步骤：
cd AstrBot uv run pytest -q \ tests/test_neo_skill_sync.py \ tests/test_neo_skill_tools.py \ tests/test_dashboard.py::test_neo_skills_routes \ tests/test_dashboard.py::test_auth_login运行结果：
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。