runtime/tools.py

from __future__ import annotations

from collections.abc import Callable

from .models import ToolResult
from .permissions import PermissionBroker
from .store import ContextStore


class ToolRegistry:
    def __init__(self, store: ContextStore, permissions: PermissionBroker) -> None:
        self.store = store
        self.permissions = permissions
        self._tools: dict[str, tuple[str, Callable[..., ToolResult]]] = {
            "search_context": ("context.read", self.search_context),
            "summarize_screen": ("screen.read", self.summarize_screen),
            "create_reminder": ("reminders.write", self.create_reminder),
            "open_document": ("documents.open", self.open_document),
        }

    def execute(self, name: str, request_id: str, **kwargs) -> ToolResult:
        permission, fn = self._tools[name]
        allowed = self.permissions.check(permission)
        self.store.audit(request_id, permission, name, "allowed" if allowed else "denied")
        if not allowed:
            return ToolResult(name=name, ok=False, content=f"Permission denied: {permission}")
        return fn(**kwargs)

    def search_context(self, query: str, limit: int = 5) -> ToolResult:
        matches = self.store.search(query, limit=limit)
        if not matches:
            return ToolResult("search_context", True, "No matching local context found.", {"matches": []})
        bullets = [
            f"- {m['title']} ({m['app']}): {m['body']}"
            for m in matches
        ]
        return ToolResult("search_context", True, "\n".join(bullets), {"matches": matches})

    def summarize_screen(self) -> ToolResult:
        matches = self.store.search("screen current profiling performance", limit=1)
        if not matches:
            return ToolResult("summarize_screen", True, "No screen snapshot is available.", {"matches": []})
        item = matches[0]
        return ToolResult(
            "summarize_screen",
            True,
            f"Your current screen appears to be {item['title']}. Key detail: {item['body']}",
            {"matches": matches},
        )

    def create_reminder(self, text: str, due_hint: str = "later") -> ToolResult:
        reminder = self.store.create_reminder(text, due_hint)
        return ToolResult(
            "create_reminder",
            True,
            f"Created reminder #{reminder['id']}: {reminder['text']} ({reminder['due_hint']}).",
            {"reminder": reminder},
        )

    def open_document(self, query: str) -> ToolResult:
        matches = [m for m in self.store.search(query, limit=5) if m["kind"] == "document"]
        if not matches:
            return ToolResult("open_document", True, "I could not find a matching local document.", {"matches": []})
        doc = matches[0]
        return ToolResult(
            "open_document",
            True,
            f"Opened {doc['title']} from {doc['source']}. Summary: {doc['body']}",
            {"document": doc},
        )