"""The `skills` tool — find, read, write and propose skills and protocols. The catalog is the agent's own memory of *how* to do things. Skills are written by the agent freely (they are advice, and the user can edit them in settings); protocols are not: a protocol is a commitment, so the agent can only *propose* one and the user's click on "Добавить" is what creates the row (see ``navi.skills.store.SkillStore.create_proposal``). """ from __future__ import annotations import structlog from navi.skills import SkillError, SkillRef, SkillStore, parse_document from navi.tools._internal.base import ( Tool, ToolContext, ToolResult, current_session_id, current_user_id, current_user_role, ) log = structlog.get_logger() _KIND_HINT = "'skill' — a flexible approach with heuristics; 'protocol' — a strict ordered procedure." class SkillTool(Tool): name = "skills" description = ( "The catalog of skills and protocols — reusable methodologies this Navi has been " "taught, owned by you or shared by the hive.\n\n" "Call 'search' before improvising an approach to a recurring kind of task (home " f"automation, shutting down or bringing up infrastructure, workflows). {_KIND_HINT}\n\n" "Call 'load' with the id from a search result (or an exact name) to get the full " "instructions. A skill is advice: follow it as far as it fits. A protocol is a " "commitment: its steps are run in order with a check after each, and its escape " "hatches must be honoured when reality differs from the plan.\n\n" "Writing to the catalog:\n" "- 'create' and 'update' save a SKILL. Write one as soon as a way of working " "proved itself and is likely to come up again — you do not need permission, and " "a better version is a normal edit.\n" "- 'propose' stores a PROTOCOL for the user to accept. Never try to create or " "edit a protocol by another route: it becomes a catalog entry only when the user " "clicks 'Добавить' on the proposal card, which appears next to the tool result by " "itself — just say what the protocol does and wait. Pass 'replaces_id' to propose " "a new version of a protocol that already exists.\n\n" "Search costs one call and often replaces several turns of rediscovery — use it." ) parameters = { "type": "object", "properties": { "op": { "type": "string", "enum": ["search", "load", "create", "update", "propose"], "description": ( "search — find entries; load — read one entry in full; create — save a " "new skill; update — save a new version of a skill you own; propose — " "offer a protocol to the user." ), }, "query": { "type": "string", "description": ( "search: words to match against name, description, triggers and tags " "(e.g. 'energy saving', 'shutdown', 'backup')." ), }, "kind": { "type": "string", "enum": ["skill", "protocol"], "description": "search: only entries of this kind.", }, "tag": {"type": "string", "description": "search: require this tag."}, "area": { "type": "string", "enum": ["home", "infrastructure", "workflows", "general"], "description": "search: only this area.", }, "id": { "type": "string", "description": "load/update: the entry id from search.", }, "name": { "type": "string", "description": "load: the exact name, when you know it and have no id.", }, "content": { "type": "string", "description": ( "create/update/propose: the whole document — a YAML front matter block " "between '---' lines, then the markdown body. A protocol must declare " "steps, post_checks, on_deviation, on_blocked and on_already_done; a " "skill needs a name, a description and a body." ), }, "replaces_id": { "type": "string", "description": ( "propose: id of an existing protocol you own, when the proposal is a " "new version of it rather than a new entry." ), }, }, "required": ["op"], } def __init__(self, store: SkillStore) -> None: self._store = store async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: op = params.get("op") user_id = current_user_id.get(None) or (ctx.user_id if ctx else None) or "" if op == "search": return await self._search(params, user_id) if op == "load": return await self._load(params, user_id) if op == "create": return await self._create(params, user_id) if op == "update": return await self._update(params, user_id) if op == "propose": session_id = current_session_id.get(None) or (ctx.session_id if ctx else None) or "" return await self._propose(params, user_id, session_id) return ToolResult( success=False, output="", error=( f"Unknown op: {op!r}. Use 'search', 'load', 'create', 'update' or 'propose'." ), ) # ── reads ─────────────────────────────────────────────────────────────── async def _search(self, params: dict, user_id: str) -> ToolResult: try: entries = await self._store.list( user_id=user_id, kind=params.get("kind"), query=(params.get("query") or "").strip() or None, tag=(params.get("tag") or "").strip() or None, area=params.get("area"), ) except Exception as e: # the catalog being down must not look like "nothing found" log.warning("skills.search_failed", error=str(e)) return ToolResult(success=False, output="", error=f"Cannot search the catalog: {e}") if not entries: return ToolResult( success=True, output=( "No matching skills or protocols. Nothing is stored for this query yet — " "either it was never taught, or the wording differs: try one word " "('power', 'backup', 'shutdown') or drop the filters." ), ) lines = [f"{len(entries)} matching entries (personal first):"] lines.extend(SkillRef.of(e).line() for e in entries) lines.append("Load one with op='load' and its id.") return ToolResult(success=True, output="\n".join(lines)) async def _load(self, params: dict, user_id: str) -> ToolResult: entry_id = (params.get("id") or "").strip() name = (params.get("name") or "").strip() if not entry_id and not name: return ToolResult( success=False, output="", error="'id' or 'name' is required for 'load'", ) try: entry = await self._store.find( user_id=user_id, skill_id=entry_id or None, name=name or None, kind=params.get("kind"), ) except Exception as e: log.warning("skills.load_failed", error=str(e)) return ToolResult(success=False, output="", error=f"Cannot read the catalog: {e}") if entry is None: return ToolResult( success=False, output="", error=( f"No such entry ({entry_id or name!r}). Search first: " "op='search' with a word from its name or purpose." ), ) try: rendered = entry.doc except SkillError as e: return ToolResult( success=False, output="", error=f"The stored document is invalid and cannot be loaded: {e}", ) from navi.skills import render_for_agent return ToolResult( success=True, output=render_for_agent( rendered, entry_id=entry.id, version=entry.version, visibility=entry.visibility, ), metadata={ "skill_id": entry.id, "kind": entry.kind, "name": entry.name, "version": entry.version, }, ) # ── writes ────────────────────────────────────────────────────────────── async def _create(self, params: dict, user_id: str) -> ToolResult: content = params.get("content") or "" try: doc = parse_document(content) except SkillError as e: return ToolResult(success=False, output="", error=str(e)) if doc.kind == "protocol": return ToolResult( success=False, output="", error=( "A protocol is not created directly — propose it with op='propose' " "and render the card; it enters the catalog when the user accepts it." ), ) try: entry = await self._store.create( owner_id=user_id, content=content, edited_by=user_id, reason="written by the agent", ) except SkillError as e: return ToolResult(success=False, output="", error=str(e)) except Exception as e: log.warning("skills.create_failed", error=str(e)) return ToolResult(success=False, output="", error=f"Cannot write to the catalog: {e}") return ToolResult( success=True, output=( f"Saved skill {entry.name!r} (id={entry.id}, v{entry.version}, " f"area={entry.area}). The user can edit it in Settings → Скилы и протоколы." ), metadata={"skill_id": entry.id, "kind": entry.kind, "version": entry.version}, ) async def _update(self, params: dict, user_id: str) -> ToolResult: entry_id = (params.get("id") or "").strip() content = params.get("content") or "" if not entry_id: return ToolResult(success=False, output="", error="'id' is required for 'update'") try: current = await self._store.get(entry_id) except Exception as e: log.warning("skills.update_failed", error=str(e)) return ToolResult(success=False, output="", error=f"Cannot read the catalog: {e}") if current is None: return ToolResult(success=False, output="", error=f"No entry with id {entry_id!r}") if current.kind == "protocol": return ToolResult( success=False, output="", error=( "A protocol changes only through a proposal: call op='propose' with " f"replaces_id={entry_id!r} and let the user accept the new version." ), ) role = current_user_role.get("user") if current.owner_id != user_id and role != "admin": return ToolResult( success=False, output="", error="This entry belongs to someone else; only its owner or an admin can edit it.", ) try: entry = await self._store.update( entry_id, content=content, edited_by=user_id, reason="updated by the agent", ) except SkillError as e: return ToolResult(success=False, output="", error=str(e)) if entry is None: return ToolResult(success=False, output="", error=f"No entry with id {entry_id!r}") return ToolResult( success=True, output=( f"Updated skill {entry.name!r} to v{entry.version} (id={entry.id}). " "The previous version stays in the history." ), metadata={"skill_id": entry.id, "kind": entry.kind, "version": entry.version}, ) async def _propose(self, params: dict, user_id: str, session_id: str) -> ToolResult: content = params.get("content") or "" replaces_id = (params.get("replaces_id") or "").strip() or None try: proposal = await self._store.create_proposal( owner_id=user_id, content=content, session_id=session_id, replaces_id=replaces_id, ) except SkillError as e: return ToolResult(success=False, output="", error=str(e)) except Exception as e: log.warning("skills.propose_failed", error=str(e)) return ToolResult(success=False, output="", error=f"Cannot store the proposal: {e}") action = "replaces" if replaces_id else "adds" return ToolResult( success=True, output=( f"Proposal stored: {proposal.name!r} {action} a protocol " f"(proposal_id={proposal.id}). It is NOT in the catalog yet — the card " "beside this result carries the document and the 'Добавить' / 'Отклонить' " "buttons. Tell the user what the protocol does and that you need their " "decision; do not run it on your own." ), metadata={ "proposal_id": proposal.id, "kind": "protocol", "name": proposal.name, # Rendered by the webclient exactly like a navi_ui component. "ui_component": { "component": "skill_proposal", "payload": {"proposal_id": proposal.id}, }, }, )