diff --git a/interfaces/admin/skills_api.py b/interfaces/admin/skills_api.py index 27955a14..f6bdc11a 100644 --- a/interfaces/admin/skills_api.py +++ b/interfaces/admin/skills_api.py @@ -308,7 +308,18 @@ def include_skill_routes( status="ok" if out.ok else "fail", detail={"runtime_type": runtime_type, "detail": out.detail, "target_dir": out.target_dir}, ) - return {"ok": bool(out.ok), "result": {"name": out.name, "target_dir": out.target_dir, "detail": out.detail, "error_code": out.error_code, "retryable": bool(out.retryable)}} + return { + "ok": bool(out.ok), + "result": { + "name": out.name, + "target_dir": out.target_dir, + "detail": out.detail, + "error_code": out.error_code, + "retryable": bool(out.retryable), + "auto_enabled": bool(getattr(out, "auto_enabled", False)), + "binding_applied_roles": list(getattr(out, "binding_applied_roles", ()) or []), + }, + } @sk.get("/binding") def api_skills_binding_get(authorization: str | None = Header(default=None)) -> dict[str, Any]: @@ -489,6 +500,8 @@ def include_skill_routes( "detail": out.detail, "error_code": out.error_code, "retryable": bool(out.retryable), + "auto_enabled": bool(getattr(out, "auto_enabled", False)), + "binding_applied_roles": list(getattr(out, "binding_applied_roles", ()) or []), }, } @@ -553,6 +566,8 @@ def include_skill_routes( "detail": out.detail, "error_code": out.error_code, "retryable": bool(out.retryable), + "auto_enabled": bool(getattr(out, "auto_enabled", False)), + "binding_applied_roles": list(getattr(out, "binding_applied_roles", ()) or []), }, } diff --git a/runtime/skill_installer.py b/runtime/skill_installer.py index f126b6bd..a7abd3fd 100644 --- a/runtime/skill_installer.py +++ b/runtime/skill_installer.py @@ -12,6 +12,12 @@ from pathlib import Path from typing import Any from oclaw.runtime.skills_market import get_market_adapter +from oclaw.runtime.skill_role_binding import ( + SKILL_ROLE_BINDING_ENABLED_SETTING, + SKILL_ROLE_BINDING_KEY, + normalize_skill_role_binding, + ordered_binding_roles, +) from oclaw.runtime.skills import ( default_skills_root, discover_workspace_skill_manifests, @@ -20,6 +26,7 @@ from oclaw.runtime.skills import ( _DISABLED_SKILLS_KEY = "AIA_SKILL_DISABLED_NAMES" _AUTO_INSTALL_KEY = "AIA_SKILL_AUTO_INSTALL_ENABLED" +_AUTO_ENABLE_TRUSTED_KEY = "AIA_SKILL_AUTO_ENABLE_TRUSTED" @dataclass(frozen=True) @@ -30,6 +37,8 @@ class SkillInstallResult: detail: str = "" error_code: str = "" retryable: bool = False + auto_enabled: bool = False + binding_applied_roles: tuple[str, ...] = () def _classify_install_detail(detail: str) -> tuple[str, bool]: @@ -65,6 +74,17 @@ def skill_auto_install_enabled(store: Any) -> bool: return _truthy(raw) +def skill_auto_enable_trusted_enabled(store: Any) -> bool: + try: + raw = str(store.get_setting(_AUTO_ENABLE_TRUSTED_KEY) or "").strip() + except Exception: + raw = "" + # Default ON for workspace/auto-install lane; tenant can disable explicitly. + if not raw: + return True + return _truthy(raw) + + def _get_disabled_names(store: Any) -> set[str]: try: raw = str(store.get_setting(_DISABLED_SKILLS_KEY) or "").strip() @@ -86,6 +106,40 @@ def _set_disabled_names(store: Any, names: set[str]) -> None: store.set_setting(_DISABLED_SKILLS_KEY, json.dumps(vals, ensure_ascii=False)) +def _apply_auto_enable_binding(*, store: Any, skill_name: str, skills_root: str | Path | None = None) -> tuple[bool, tuple[str, ...]]: + nm = str(skill_name or "").strip() + if not nm: + return False, () + if not skill_auto_enable_trusted_enabled(store): + return False, () + roles = ordered_binding_roles() + valid = {str(m.name).strip() for m in discover_workspace_skill_manifests(skills_root) if str(m.name or "").strip()} + # If discovery lags for any reason, still include the newly created skill as valid candidate. + valid.add(nm) + try: + raw = str(store.get_setting(SKILL_ROLE_BINDING_KEY) or "").strip() + mapping_raw = json.loads(raw) if raw else {} + except Exception: + mapping_raw = {} + mapping = normalize_skill_role_binding( + mapping_raw=mapping_raw if isinstance(mapping_raw, dict) else {}, + valid_skill_names=valid, + available_roles=roles, + ) + changed = False + for role in roles: + cur = list(mapping.get(role) or []) + if nm not in cur: + cur.append(nm) + mapping[role] = cur + changed = True + if changed: + store.set_setting(SKILL_ROLE_BINDING_KEY, json.dumps(mapping, ensure_ascii=False)) + # Ensure role filter is active when we auto-attach to all roles. + store.set_setting(SKILL_ROLE_BINDING_ENABLED_SETTING, "1") + return True, tuple(roles) + + def list_skills_with_status(*, store: Any, skills_root: str | Path | None = None) -> list[dict[str, Any]]: disabled = _get_disabled_names(store) out: list[dict[str, Any]] = [] @@ -432,7 +486,7 @@ def create_workspace_skill( "permissions": {"fs_write": False, "net": False, "process": True}, }, } - return create_skill_from_template( + out = create_skill_from_template( store=store, name=nm, description=description, @@ -441,6 +495,19 @@ def create_workspace_skill( skills_root=target.parent, overwrite=True, ) + if not out.ok: + return out + auto_enabled, roles = _apply_auto_enable_binding(store=store, skill_name=nm, skills_root=root) + return SkillInstallResult( + ok=out.ok, + name=out.name, + target_dir=out.target_dir, + detail=out.detail, + error_code=out.error_code, + retryable=out.retryable, + auto_enabled=auto_enabled, + binding_applied_roles=roles, + ) def auto_install_skill_from_payload( @@ -458,7 +525,9 @@ def auto_install_skill_from_payload( md = payload.get("metadata_oclaw") md = dict(md) if isinstance(md, dict) else {} root = Path(skills_root).resolve() if skills_root else default_skills_root() - target = root / name + # Keep agent-authored/auto-installed skills isolated from primary managed skills. + workspace_root = root / "_workspace" + target = workspace_root / name before_disabled = _get_disabled_names(store) try: out = create_skill_from_template( @@ -467,14 +536,24 @@ def auto_install_skill_from_payload( description=description, body_markdown=body, metadata_oclaw=md, - skills_root=skills_root, + skills_root=workspace_root, overwrite=False, ) if bool(payload.get("force_error_for_test")): raise RuntimeError("forced_error_for_test") if not out.ok: return out - return out + auto_enabled, roles = _apply_auto_enable_binding(store=store, skill_name=name, skills_root=root) + return SkillInstallResult( + ok=out.ok, + name=out.name, + target_dir=out.target_dir, + detail=out.detail, + error_code=out.error_code, + retryable=out.retryable, + auto_enabled=auto_enabled, + binding_applied_roles=roles, + ) except Exception as exc: try: if target.exists() and target.is_dir(): diff --git a/runtime/skills/README.md b/runtime/skills/README.md index 4b1f19c3..f4ae595e 100644 --- a/runtime/skills/README.md +++ b/runtime/skills/README.md @@ -10,6 +10,16 @@ ## 示例结构 - `oclaw/runtime/skills//SKILL.md` +## 目录分层（重要） +- **主目录**：`oclaw/runtime/skills//` + 用于官方/手工管理的稳定技能（安装、维护、评审都在这层）。 +- **自写目录**：`oclaw/runtime/skills/_workspace//` + 用于 agent 自写/自动安装技能，和主目录隔离，避免混放。 + +说明： +- `auto_install_skill_from_payload` 产物默认落在 `_workspace` 下。 +- 这样可以把“生产稳定技能”和“实验/自写技能”分开治理，便于审计与回滚。 + ## 兼容说明 - 运行时优先读取 `oclaw/runtime/skills`。 - 若设置了环境变量 `AIA_SKILLS_ROOT`，以该变量为准。 diff --git a/tests/test_admin_skills_api.py b/tests/test_admin_skills_api.py index a4211884..74e15420 100644 --- a/tests/test_admin_skills_api.py +++ b/tests/test_admin_skills_api.py @@ -210,6 +210,9 @@ class AdminSkillsApiTests(unittest.TestCase): self.assertEqual(c.status_code, 200, c.text) body = c.json() or {} self.assertTrue(body.get("ok")) + result_create = body.get("result") if isinstance(body.get("result"), dict) else {} + self.assertIn("auto_enabled", result_create) + self.assertIn("binding_applied_roles", result_create) r1 = self.client.post( "/admin/api/skills/test-run", diff --git a/tests/test_skill_installer.py b/tests/test_skill_installer.py index 7f42438a..f61479e3 100644 --- a/tests/test_skill_installer.py +++ b/tests/test_skill_installer.py @@ -51,7 +51,33 @@ def test_auto_install_rollback_on_forced_error(tmp_path: Path) -> None: assert out.ok is False assert out.retryable is True assert out.error_code == "runtime_error" - assert not (root / "roll_me_back").exists() + assert not (root / "_workspace" / "roll_me_back").exists() + + +def test_auto_install_enables_binding_for_all_roles(tmp_path: Path) -> None: + db = tmp_path / "ops.sqlite" + store = SqliteStore(str(db)) + root = tmp_path / "skills" + out = auto_install_skill_from_payload( + store=store, + payload={ + "name": "auto_bound_skill", + "description": "demo", + "body_markdown": "x", + }, + skills_root=root, + ) + assert out.ok is True + assert out.auto_enabled is True + assert len(out.binding_applied_roles) >= 1 + assert (root / "_workspace" / "auto_bound_skill" / "SKILL.md").exists() + mapping_raw = str(store.get_setting("skill_role_binding") or "").strip() + assert mapping_raw + import json + + mapping = json.loads(mapping_raw) + assert isinstance(mapping, dict) + assert any("auto_bound_skill" in (mapping.get(k) or []) for k in mapping.keys()) def test_install_skill_from_registry_archive_file_url(tmp_path: Path) -> None: