mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-02 02:30:16 +08:00
Compare commits
1 Commits
codex/rest
...
draft/work
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4483796c20 |
@@ -361,11 +361,19 @@ def _build_local_mode_prompt() -> str:
|
||||
if system_name.lower() == "windows"
|
||||
else "The runtime shell is Unix-like. Use POSIX-compatible shell commands."
|
||||
)
|
||||
return (
|
||||
"You have access to the host local environment and can execute shell commands and Python code. "
|
||||
f"Current operating system: {system_name}. "
|
||||
f"{shell_hint}"
|
||||
)
|
||||
lines = [
|
||||
"You have access to the host local environment and can execute shell commands and Python code.",
|
||||
f"Current operating system: {system_name}.",
|
||||
shell_hint,
|
||||
"",
|
||||
"You can write and modify the EXTRA_PROMPT.md file in the current workspace",
|
||||
"to customize your own system prompt instructions. This file will be automatically",
|
||||
"loaded and applied to your system prompt in subsequent conversations.",
|
||||
"",
|
||||
"When installing skills, unless explicitly specified otherwise, prefer installing",
|
||||
"them to the workspace/skills directory for better isolation and portability.",
|
||||
]
|
||||
return " ".join(lines)
|
||||
|
||||
|
||||
async def _ensure_persona_and_skills(
|
||||
|
||||
@@ -274,12 +274,17 @@ def build_skills_prompt(skills: list[SkillInfo]) -> str:
|
||||
|
||||
|
||||
class SkillManager:
|
||||
def __init__(self, skills_root: str | None = None) -> None:
|
||||
def __init__(
|
||||
self, skills_root: str | None = None, workspace_skills_root: str | None = None
|
||||
) -> None:
|
||||
self.skills_root = skills_root or get_astrbot_skills_path()
|
||||
self.workspace_skills_root = workspace_skills_root
|
||||
data_path = Path(get_astrbot_data_path())
|
||||
self.config_path = str(data_path / SKILLS_CONFIG_FILENAME)
|
||||
self.sandbox_skills_cache_path = str(data_path / SANDBOX_SKILLS_CACHE_FILENAME)
|
||||
os.makedirs(self.skills_root, exist_ok=True)
|
||||
if self.workspace_skills_root:
|
||||
os.makedirs(self.workspace_skills_root, exist_ok=True)
|
||||
|
||||
def _load_config(self) -> dict:
|
||||
if not os.path.exists(self.config_path):
|
||||
@@ -430,6 +435,34 @@ class SkillManager:
|
||||
sandbox_exists=sandbox_exists,
|
||||
)
|
||||
|
||||
# Scan workspace-local skills (if workspace_skills_root is set)
|
||||
if self.workspace_skills_root and os.path.isdir(self.workspace_skills_root):
|
||||
for entry in sorted(Path(self.workspace_skills_root).iterdir()):
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
skill_name = entry.name
|
||||
skill_md = _normalize_skill_markdown_path(entry)
|
||||
if skill_md is None:
|
||||
continue
|
||||
# Workspace skills are always active and workspace-local
|
||||
description = ""
|
||||
try:
|
||||
content = skill_md.read_text(encoding="utf-8")
|
||||
description = _parse_frontmatter_description(content)
|
||||
except Exception:
|
||||
description = ""
|
||||
path_str = str(skill_md).replace("\\", "/")
|
||||
skills_by_name[skill_name] = SkillInfo(
|
||||
name=skill_name,
|
||||
description=description,
|
||||
path=path_str,
|
||||
active=True,
|
||||
source_type="workspace_only",
|
||||
source_label="workspace",
|
||||
local_exists=True,
|
||||
sandbox_exists=False,
|
||||
)
|
||||
|
||||
if runtime == "sandbox":
|
||||
cache = self._load_sandbox_skills_cache()
|
||||
for item in cache.get("skills", []):
|
||||
@@ -541,6 +574,7 @@ class SkillManager:
|
||||
*,
|
||||
overwrite: bool = True,
|
||||
skill_name_hint: str | None = None,
|
||||
install_to_workspace: bool = False,
|
||||
) -> str:
|
||||
zip_path_obj = Path(zip_path)
|
||||
if not zip_path_obj.exists():
|
||||
@@ -548,6 +582,14 @@ class SkillManager:
|
||||
if not zipfile.is_zipfile(zip_path):
|
||||
raise ValueError("Uploaded file is not a valid zip archive.")
|
||||
|
||||
# Determine target skills root (global or workspace)
|
||||
if install_to_workspace:
|
||||
if not self.workspace_skills_root:
|
||||
raise ValueError("Workspace skills root not configured")
|
||||
target_skills_root = self.workspace_skills_root
|
||||
else:
|
||||
target_skills_root = self.skills_root
|
||||
|
||||
installed_skills = []
|
||||
|
||||
with zipfile.ZipFile(zip_path) as zf:
|
||||
@@ -605,7 +647,7 @@ class SkillManager:
|
||||
else:
|
||||
target_name = candidate_name
|
||||
|
||||
dest_dir = Path(self.skills_root) / target_name
|
||||
dest_dir = Path(target_skills_root) / target_name
|
||||
if dest_dir.exists():
|
||||
conflict_dirs.append(str(dest_dir))
|
||||
|
||||
@@ -638,7 +680,7 @@ class SkillManager:
|
||||
"SKILL.md not found in the root of the zip archive."
|
||||
)
|
||||
|
||||
dest_dir = Path(self.skills_root) / skill_name
|
||||
dest_dir = Path(target_skills_root) / skill_name
|
||||
if dest_dir.exists() and overwrite:
|
||||
shutil.rmtree(dest_dir)
|
||||
elif dest_dir.exists() and not overwrite:
|
||||
@@ -679,7 +721,7 @@ class SkillManager:
|
||||
if normalized_path is None:
|
||||
continue
|
||||
|
||||
dest_dir = Path(self.skills_root) / skill_name
|
||||
dest_dir = Path(target_skills_root) / skill_name
|
||||
if dest_dir.exists():
|
||||
if not overwrite:
|
||||
raise FileExistsError(
|
||||
|
||||
@@ -8,7 +8,12 @@ from astrbot.core.astr_agent_context import AstrAgentContext
|
||||
from astrbot.core.computer.computer_client import get_booter
|
||||
|
||||
from ..registry import builtin_tool
|
||||
from .util import check_admin_permission, is_local_runtime, workspace_root
|
||||
from .util import (
|
||||
check_admin_permission,
|
||||
init_workspace,
|
||||
is_local_runtime,
|
||||
workspace_root,
|
||||
)
|
||||
|
||||
_COMPUTER_RUNTIME_TOOL_CONFIG = {
|
||||
"provider_settings.computer_use_runtime": ("local", "sandbox"),
|
||||
@@ -61,10 +66,9 @@ class ExecuteShellTool(FunctionTool):
|
||||
try:
|
||||
cwd: str | None = None
|
||||
if is_local_runtime(context):
|
||||
current_workspace_root = workspace_root(
|
||||
current_workspace_root = init_workspace(
|
||||
context.context.event.unified_msg_origin
|
||||
)
|
||||
current_workspace_root.mkdir(parents=True, exist_ok=True)
|
||||
cwd = str(current_workspace_root)
|
||||
|
||||
result = await sb.shell.exec(
|
||||
|
||||
@@ -17,6 +17,35 @@ def workspace_root(umo: str) -> Path:
|
||||
return (Path(get_astrbot_workspaces_path()) / normalized_umo).resolve(strict=False)
|
||||
|
||||
|
||||
def init_workspace(umo: str) -> Path:
|
||||
"""Initialize workspace for local runtime.
|
||||
|
||||
Creates the workspace directory with:
|
||||
- EXTRA_PROMPT.md: for custom system prompt instructions
|
||||
- skills/: for workspace-local skills
|
||||
|
||||
Returns the workspace root path.
|
||||
"""
|
||||
root = workspace_root(umo)
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create EXTRA_PROMPT.md if not exists
|
||||
extra_prompt_path = root / "EXTRA_PROMPT.md"
|
||||
if not extra_prompt_path.exists():
|
||||
extra_prompt_path.write_text(
|
||||
"# System Extra Instructions\n\n"
|
||||
"Add your custom system prompt instructions here.\n"
|
||||
"These will be automatically loaded and applied to the agent's system prompt.\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
# Create skills directory if not exists
|
||||
skills_dir = root / "skills"
|
||||
skills_dir.mkdir(exist_ok=True)
|
||||
|
||||
return root
|
||||
|
||||
|
||||
def is_local_runtime(context: ContextWrapper[AstrAgentContext]) -> bool:
|
||||
cfg = context.context.context.get_config(
|
||||
umo=context.context.event.unified_msg_origin
|
||||
|
||||
Reference in New Issue
Block a user