"""Host-side skill lifecycle behaviour. The dispatcher remains the policy boundary (permissions, audit, UI activity, soft failures). This service owns the skill domain itself so retrieval and filesystem mutation are visible in one class instead of being scattered across ``HostDispatcher``. """ from __future__ import annotations import shutil from pathlib import Path from typing import Any from openai4s import execution_principal from openai4s.config import Config from openai4s.host import resource_allowlist from openai4s.skills_loader import SkillLoader, SkillVersionService class SkillService: """Retrieve and manage Code-as-Action skill directories.""" def __init__(self, cfg: Config) -> None: self.cfg = cfg self.loader = SkillLoader(cfg=cfg) self.versions = SkillVersionService(cfg) self.project_id: str | None = None self.session_id: str | None = None #: Tri-state: None inherits, [] denies everything, a list is exactly #: those. Enforced on every read path below, because until now it was #: stored on the specialist, inherited through delegation, merged into #: the spec — and never consulted by anything. D5 deferred exactly this #: to P1-B and hid the UI meanwhile, so that no lock was displayed that #: was not enforced. self._allowed_skills: object = None def set_scope( self, *, project_id: str | None = None, session_id: str | None = None, ) -> None: """Retarget prompt/search/read/bootstrap to one capability snapshot.""" self.loader = self.loader.scoped( project_id=project_id, session_id=session_id, ) self.project_id = str(project_id or "").strip() or None self.session_id = str(session_id or "").strip() or None def set_allowed_skills(self, allowed: object) -> None: """Restrict every read path to these skills. Only narrows. Composed through `resource_allowlist.narrow`, so calling this twice -- which is what a delegation chain does -- can never widen. A child that passes `None` inherits the parent's restriction rather than clearing it; otherwise delegating would be the way out of every allowlist. """ from openai4s.host import resource_allowlist self._allowed_skills = resource_allowlist.narrow(self._allowed_skills, allowed) def _permits(self, name: str) -> bool: from openai4s.host import resource_allowlist return resource_allowlist.permits(self._allowed_skills, name) def _filter(self, rows: list) -> list: """Drop rows naming a skill this specialist may not reach. Applied to the catalogue and to search, not only to load: a name the agent can see is a name it will ask for, and listing something it cannot open is both a leak and a dead end. The report's exit criterion is that an unlisted resource is unavailable in the UI, the prompt, the catalogue *and* direct Host RPC -- four surfaces, and the catalogue is the one that feeds the prompt. """ from openai4s.host import resource_allowlist if resource_allowlist.normalise(self._allowed_skills) is None: return rows out = [] for row in rows: name = ( row.get("name") if isinstance(row, dict) else getattr(row, "name", "") ) if self._permits(str(name or "")): out.append(row) return out def _writable_scope(self, skill: Any) -> tuple[str, str | None]: """Return the (scope, project_id) that owns a discovered skill on disk. ``discover()`` gives project skills precedence over personal ones, so a project-scoped edit/delete must target the project scope. Hardcoding ``personal`` here would write to (or look up) a shadowed personal copy the loader never surfaces, leaving the real project skill untouched. """ if getattr(skill, "source", "") == "project" and self.project_id: return "project", self.project_id return "personal", None def _authorize_mutation( self, scope: str, project_id: str | None ) -> execution_principal.Principal: """Apply team authorization in addition to interactive approval. The permission broker answers whether the person running this turn consents to an agent mutation. It does not make that person an administrator. Both the personal Skill directory and a project Skill overlay are executable inputs to later agent runs: a project member's model must not be able to rewrite the recipe or ``kernel.py`` that a peer will load merely because the member may make an ordinary project edit through the authenticated Web API. Host-originated Skill writes therefore require an administrator in team mode. Human project authoring remains on the separate HTTP service and its project-member authorization boundary. ``resolve()`` is deliberately used instead of reading an optional principal: an unpropagated team identity must fail closed. Off team mode it returns the explicit single-user principal, preserving the existing local authoring contract. """ principal = execution_principal.resolve() if principal.is_admin: return principal target = "project" if scope == "project" and project_id else "personal" raise PermissionError( f"{target} Skill mutation through Host requires a team administrator" ) def load(self, name: str | dict) -> dict: """Load full guidance, with the historical fuzzy-name fallback.""" if isinstance(name, dict): name = name.get("name", "") self.loader.discover() # Guarded fuzzy resolution, over the PERMITTED set. Resolving against # the whole corpus and refusing afterwards turns "you misspelled your # own skill" into "no such skill" for a child with an allowlist, and # the unguarded single-best-match fallback handed back an unrelated # collection recipe under the name the caller asked for. skill = self.loader.resolve(str(name), permits=self._permits) if skill is None or not self._permits(getattr(skill, "name", "")): # Same answer for "does not exist" and "not yours". A distinct # refusal would confirm the skill is there, which is most of what # an enumerator wants from a name it guessed. return {"error": f"no such skill: {name!r}"} try: content = (skill.root / "SKILL.md").read_text("utf-8") except Exception: # noqa: BLE001 - loader doc is the compatibility fallback content = getattr(skill, "doc", "") or "" network = getattr(skill, "network", None) payload = { "name": skill.name, "origin": skill.origin, "description": skill.description, "content": content, "version": getattr(skill, "version", "") or "", "document_sha256": getattr(skill, "document_sha256", "") or "", } if network is not None: payload["capabilities"] = network.public_dict() payload["network_manifest_digest"] = network.digest return payload def search(self, spec: dict) -> list: self.loader.discover() # The allowlist is applied inside the ranking, between scoring and the # limit slice, NOT to the sliced result. Scoring still happens over the # whole corpus, so a permitted skill keeps the rank it earned -- but # filtering a global top-5 afterwards is how a Specialist got nothing: # with 561 collection recipes in the same lexical index, none of the # two skills it is allowed appear in the top 5 for "protein structure # prediction and design pipeline", where before the import it got # `proteinmpnn`. `_filter` stays as the belt-and-braces pass so this # method still cannot return a row `_permits` would reject. return self._filter( self.loader.search( spec.get("query", ""), limit=int(spec.get("limit", 5)), permits=self._permits, ) ) def system_context(self) -> str: """The prompt block, filtered by this session's skill allowlist. Lives here rather than on the loader because this is the object that holds `_allowed_skills`. `HostDispatcher.skill_loader` returns the raw loader, and `Agent` bound *that* for the system prompt -- so the filter was armed on the service while the prompt was rendered from the corpus beside it. """ self.loader.discover() return self.loader.system_context( only=resource_allowlist.normalise(self._allowed_skills) ) def bootstrap_code(self) -> str: """The in-kernel sidecar gate, closed over this session's allowlist. Same reason as `system_context`: the allowlist lives on this object and the generator lives on the loader, so a caller holding the loader got a gate that knew only about `disabled`. """ return self.loader.bootstrap_code( allowed=resource_allowlist.normalise(self._allowed_skills) ) def list(self) -> list: self.loader.discover() return self._filter(self.loader.catalog()) def get(self, name: str) -> dict: self.loader.discover() skill = self.loader.get(name) if skill is None or not self._permits(getattr(skill, "name", "")): raise KeyError(f"no such skill: {name!r}") return { "name": skill.name, "origin": skill.origin, "description": skill.description, "has_kernel": skill.has_kernel, "read_only": skill.read_only, "sidecar_gate": skill.sidecar_gate(), } def read(self, spec: dict) -> str: self.loader.discover() # The one that actually hands back file contents, so it is the one an # allowlist most has to cover. Gated on the requested name rather than # on a resolved skill: `loader.read` does its own lookup, and checking # a name the loader might resolve differently is how a gate ends up # guarding a different object from the one that gets returned. if not self._permits(str(spec.get("name") or "")): raise KeyError(f"no such skill: {spec.get('name')!r}") return self.loader.read(spec["name"], spec.get("path", "SKILL.md")) def edit(self, spec: dict) -> dict: name = spec["name"] # The allowlist gated the three READ paths and none of the three write # paths, so a child restricted to `["a"]` could not read skill `b` and # could overwrite, publish or delete it. That is the worse half: the # parent goes on to *execute* the recipe a restricted child rewrote, so # an unreadable Skill was a writable one. A name that is not permitted # does not exist for this caller, and it must not be creatable either — # authoring a new name outside the allowlist would be the same escape # with an extra step. if not self._permits(str(name or "")): raise KeyError(f"no such skill: {name!r}") relative = spec.get("path", "SKILL.md") content = spec.get("content", "") old_string = spec.get("old_string") self.loader.discover() existing = self.loader.get(name, include_disabled=True) if existing is not None and existing.read_only: raise PermissionError( f"skill {name!r} origin={existing.origin} is read-only" ) scope, project_id = self._writable_scope(existing) principal = self._authorize_mutation(scope, project_id) candidate_directory = ( existing.root.name if existing is not None else self.versions.slug(name) ) directory_collision = self.loader.bundled_directory_collision( candidate_directory ) if directory_collision is not None: raise PermissionError( f"skill directory {candidate_directory!r} collides with read-only " f"bundled directory {directory_collision}" ) if relative == "SKILL.md" and old_string is None: self._reject_bundled_name_collision(content, fallback=name) if existing is not None: root = existing.root else: user_directory = self.loader.user_skills_dir() if user_directory.is_symlink(): raise ValueError("unsafe user skill directory") user_directory.mkdir(parents=True, exist_ok=True) user_directory = user_directory.resolve() candidate = user_directory / self.versions.slug(name) if candidate.is_symlink(): raise ValueError(f"unsafe skill path: {name!r}") root = candidate.resolve() if root == user_directory or not root.is_relative_to(user_directory): raise ValueError(f"unsafe skill name: {name!r}") target = self._safe_path(root, relative) files = self.versions.read_package(root) if (root / "SKILL.md").exists() else {} if "SKILL.md" not in files and relative != "SKILL.md": files["SKILL.md"] = ( f"---\nname: {name}\ndescription: (draft)\norigin: draft\n---\n" f"# Skill: {name}\n" ).encode("utf-8") if old_string is None: updated_content = content mode = "overwrite" else: if relative not in files: raise FileNotFoundError(f"{relative} does not exist for str_replace") current = files[relative].decode("utf-8") if old_string not in current: raise ValueError("old_string not found in file") updated_content = current.replace(old_string, content, 1) mode = "str_replace" if relative == "SKILL.md": self._reject_bundled_name_collision( updated_content, fallback=existing.name if existing is not None else name, ) files[relative] = updated_content.encode("utf-8") self.versions.install( existing.name if existing is not None else name, files, event="upgraded" if existing is not None else "installed", slug=root.name, scope=scope, project_id=project_id, require_sidecar_gate=False, metadata={ "source": "host_skills_edit", "path": relative, # This bit is written by the trusted Host only after the # authorization check above. The human HTTP rollback service # can therefore distinguish a post-fix administrator-authored # recipe from an un-attributed legacy Host version without # exposing a user id in version history. "authorized_admin": principal.is_admin, }, ) result: dict[str, Any] = { "ok": True, "mode": mode, "path": str(target), } if target.name == "kernel.py": self.loader.discover() skill = self.loader.get(name, include_disabled=True) result["sidecar_gate"] = ( skill.sidecar_gate() if skill else {"ok": True, "error": None} ) return result def publish(self, name: str) -> dict: if not self._permits(str(name or "")): raise KeyError(f"no such skill: {name!r}") self.loader.discover() skill = self.loader.get(name, include_disabled=True) if skill is None: raise KeyError(f"no such skill: {name!r}") if skill.read_only: raise PermissionError(f"skill {name!r} is read-only") scope, project_id = self._writable_scope(skill) self._authorize_mutation(scope, project_id) self.versions.publish( name, scope=scope, project_id=project_id, slug=skill.root.name, ) return { "ok": True, "origin": "personal" if scope == "personal" else skill.origin, } def delete(self, name: str) -> dict: if not self._permits(str(name or "")): raise KeyError(f"no such skill: {name!r}") self.loader.discover() skill = self.loader.get(name, include_disabled=True) if skill is None: raise KeyError(f"no such skill: {name!r}") if skill.read_only: raise PermissionError(f"skill {name!r} is read-only") scope, project_id = self._writable_scope(skill) self._authorize_mutation(scope, project_id) installation = self.versions.repository.get_installation( skill.name, scope=scope, scope_id=str(project_id or "") if scope == "project" else "", ) if installation is not None and installation.get("active_version_id"): self.versions.delete(skill.name, scope=scope, project_id=project_id) else: shutil.rmtree(skill.root) return {"ok": True, "deleted": name} def _version_request(self, spec: str | dict) -> tuple[str, str, str | None, int]: if isinstance(spec, dict): name = str(spec.get("name") or "").strip() scope = str(spec.get("scope") or "personal").strip().lower() requested_project = str(spec.get("project_id") or "").strip() or None limit = int(spec.get("limit") or 200) else: name = str(spec or "").strip() scope = "personal" requested_project = None limit = 200 if not name: raise ValueError("skill name is required") if scope not in {"personal", "project"}: raise ValueError("skill scope must be 'personal' or 'project'") if scope == "personal": if requested_project: raise ValueError("personal Skill scope cannot have project_id") project_id = None else: project_id = requested_project or self.project_id if not project_id: raise PermissionError("project Skill scope requires an active project") if self.project_id and project_id != self.project_id: raise PermissionError("project Skill scope cannot cross projects") return name, scope, project_id, max(1, min(limit, 200)) def status(self, spec: str | dict) -> dict: """Return safe active-version metadata for one exact writable scope.""" name, scope, project_id, _limit = self._version_request(spec) bundled = self.loader.bundled_name_collision(name) if bundled is not None: return { "name": bundled.name, "scope": "bundled", "scope_id": "", "installed": True, "active": True, "active_version_id": None, "manifest": { "origin": bundled.origin, "document_sha256": bundled.document_sha256, "sidecar": { "present": bundled.has_kernel, "sha256": bundled.sidecar_sha256, "gate": bundled.sidecar_gate(), }, }, "read_only": True, "rollback_available": False, } return { **self.versions.status( name, scope=scope, project_id=project_id, ), "read_only": False, } def history(self, name: str | dict, *, limit: int = 200) -> dict: """Expose immutable lifecycle records to trusted orchestration code.""" spec = name if isinstance(name, dict) else {"name": name, "limit": limit} skill_name, scope, project_id, resolved_limit = self._version_request(spec) if self.loader.bundled_name_collision(skill_name) is not None: raise PermissionError(f"skill {skill_name!r} is bundled and read-only") return self.versions.history( skill_name, scope=scope, project_id=project_id, limit=resolved_limit, ) def rollback(self, name: str | dict, version_id: str | None = None) -> dict: spec = ( name if isinstance(name, dict) else {"name": name, "version_id": version_id, "scope": "personal"} ) skill_name, scope, project_id, _limit = self._version_request(spec) self._authorize_mutation(scope, project_id) target_version = str(spec.get("version_id") or "").strip() if not target_version: raise ValueError("skill version_id is required") if self.loader.bundled_name_collision(skill_name) is not None: raise PermissionError(f"skill {skill_name!r} is bundled and read-only") self.loader.discover() result = self.versions.rollback( skill_name, target_version, scope=scope, project_id=project_id, ) self.loader.discover() return result def _reject_bundled_name_collision(self, content: str, *, fallback: str) -> None: metadata, _body = self.loader.parse_document(content) declared_name = metadata.get("name") or fallback collision = self.loader.bundled_name_collision(declared_name) if collision is not None: raise PermissionError( f"skill name {declared_name!r} collides with read-only bundled " f"skill {collision.name!r}" ) @staticmethod def _safe_path(root: Path, relative: str) -> Path: root = root.resolve() target = (root / relative).resolve() if root != target and root not in target.parents: raise ValueError(f"path escapes skill dir: {relative!r}") return target __all__ = ["SkillService"]