Blame for tiny_computer_agent/skills.py

"""Skills: a verified episode, written down once, replayed forever.

A skill is the cheapest brain there is - no weights, no prompt, no model - and it is exactly
as trustworthy as the run that produced it, because :func:`record_skill` refuses to record
anything that was not *verified*. That is the whole bargain of this project in one object:
the model is only ever needed to make a choice the first time; after that the harness owns it.

A skill stores *labels*, never coordinates, so it re-grounds itself on whatever screen it
meets - the same trick that lets a 256M model pick "Save" from a menu.

    library = SkillLibrary(Path("skills.json"))
    skill = record_skill(loop.run("click 'Save' to save the document"), library=library)
    library.save()

    loop = AgentLoop(screen=..., brain=SkillBrain(library), executor=DryRunExecutor())
    loop.run("click 'Save' to save the document")      # no model, no OCR guesswork
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Sequence

from .actions import Action
from .brain import BrainError, GoalUnreachable
from .diff import ChangeVerifier
from .loop import Episode, Step
from .observe import Element, Observation, normalise_label

SKILLS_FORMAT = "tiny-computer-agent/skills"
SKILLS_VERSION = 1
DEFAULT_SKILLS_PATH = Path.home() / ".tiny-computer-agent" / "skills.json"
_STOPWORDS = frozenset({"the", "a", "an", "to", "in", "on", "of", "and", "then", "click", "press"})


def slug(text: str, limit: int = 48) -> str:
    """A stable, readable name for a goal: 'click Save to save the document' -> 'save-document'."""
    words = [w for w in re.split(r"[^a-z0-9]+", text.lower()) if w and w not in _STOPWORDS]
    unique: list[str] = []
    for word in words:
        if word not in unique:
            unique.append(word)
    return "-".join(unique)[:limit] or "skill"


def _tokens(text: str) -> frozenset[str]:
    return frozenset(w for w in re.split(r"[^a-z0-9]+", text.lower()) if w and w not in _STOPWORDS)


@dataclass(frozen=True)
class SkillStep:
    """One action of a skill, with the label it resolved to at record time."""

    action: Action
    aimed_at: str = ""
    """The label the action was grounded on - the thing that has to still be on screen."""

    def describe(self) -> str:
        if self.aimed_at:
            return f"{self.action.describe()}  [{self.aimed_at}]"
        return self.action.describe()

    def to_dict(self) -> dict[str, Any]:
        return {"action": self.action.to_dict(), "aimed_at": self.aimed_at}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "SkillStep":
        return cls(action=Action(**data["action"]), aimed_at=data.get("aimed_at", ""))


@dataclass(frozen=True)
class Skill:
    """A named, verified action sequence for a goal."""

    name: str
    goal: str
    steps: tuple[SkillStep, ...]
    verified: bool = False
    verify_text: str = ""
    verify_gone: str = ""
    verify_appeared: str = ""
    recorded: str = ""
    notes: str = ""

    def __post_init__(self) -> None:
        if not self.steps:
            raise ValueError("a skill with no steps is not a skill")

    def actions(self) -> tuple[Action, ...]:
        return tuple(step.action for step in self.steps)

    def preconditions(self) -> tuple[str, ...]:
        """The *entry* condition: the labels the first step needs.

        Not every label in the skill - a skill that opens a menu and then clicks an item would
        never qualify otherwise, because the item does not exist until the menu is open. Later
        steps are grounded as they arrive, and a label that has vanished raises rather than
        clicking at where the button used to be.
        """
        first = self.steps[0]
        return (first.aimed_at,) if first.aimed_at else ()

    def score(self, goal: str) -> float:
        """How well a goal matches this skill: 1.0 for the same words, else a token overlap."""
        if normalise_label(goal) == normalise_label(self.goal):
            return 1.0
        wanted = _tokens(goal)
        if not wanted:
            return 0.0
        return len(wanted & _tokens(self.goal)) / len(wanted)

    def applies(self, goal: str, observation: Observation, *, threshold: float = 0.6) -> bool:
        """Right goal *and* every precondition still on screen - otherwise it would aim at nothing."""
        if self.score(goal) < threshold:
            return False
        return all(observation.find(label) is not None for label in self.preconditions())

    def verification(self) -> str:
        """The check, as one line, for a table or a console."""
        if self.verify_text:
            return f"text {self.verify_text!r} appears"
        bits = []
        if self.verify_appeared:
            bits.append(f"{self.verify_appeared!r} appears")
        if self.verify_gone:
            bits.append(f"{self.verify_gone!r} goes away")
        return " and ".join(bits) or "(unverified)"

    def verifier(self) -> Callable[[Observation], bool] | None:
        """The check that made this skill trustworthy, as a verifier the loop can reuse."""
        if self.verify_text:
            wanted = normalise_label(self.verify_text)

            def by_text(observation: Observation) -> bool:
                return wanted in normalise_label(observation.text)

            return by_text
        if self.verify_gone or self.verify_appeared:
            return ChangeVerifier(appeared=(self.verify_appeared,) if self.verify_appeared else (),
                                  disappeared=(self.verify_gone,) if self.verify_gone else ())
        return None

    def line(self) -> str:
        mark = "verified" if self.verified else "UNVERIFIED"
        return f"{self.name:28s} {len(self.steps)} step(s)  {mark}  {self.goal[:52]}"

    def to_dict(self) -> dict[str, Any]:
        return {"name": self.name, "goal": self.goal, "verified": self.verified,
                "verify_text": self.verify_text, "verify_gone": self.verify_gone,
                "verify_appeared": self.verify_appeared, "recorded": self.recorded,
                "notes": self.notes, "steps": [s.to_dict() for s in self.steps]}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Skill":
        return cls(name=data["name"], goal=data.get("goal", ""), verified=bool(data.get("verified")),
                   verify_text=data.get("verify_text", ""), verify_gone=data.get("verify_gone", ""),
                   verify_appeared=data.get("verify_appeared", ""), recorded=data.get("recorded", ""),
                   notes=data.get("notes", ""),
                   steps=tuple(SkillStep.from_dict(s) for s in data.get("steps", ())))


def record_skill(episode: Episode, *, name: str | None = None, library: "SkillLibrary | None" = None,
                 verify_text: str = "", verify_gone: str = "", verify_appeared: str = "",
                 require_verified: bool = True, overwrite: bool = True) -> Skill | None:
    """Turn a finished episode into a reusable skill - or refuse, loudly, if it earned it.

    Returns ``None`` when the episode is not something worth repeating: no steps, no actions,
    or (unless ``require_verified`` is False) not a verified success. Recording an unverified
    run would turn a guess into a library others trust, which is the one thing this project
    does not do.
    """
    evidence = (episode.stop_reason.startswith("verified")
                or bool(verify_text or verify_gone or verify_appeared))
    if require_verified and not (episode.success and evidence):
        # A brain reporting "done" is a claim, not evidence - the project's own rule. Recording
        # that would turn a hunch into a library other runs trust, so it is refused here.
        return None
    steps: list[SkillStep] = []
    for step in episode.steps:
        action = step.action
        if action is None or action.kind == "done":
            continue
        aimed = action.label or ""
        steps.append(SkillStep(action=action, aimed_at=aimed))
    if not steps:
        return None
    skill = Skill(name=name or slug(episode.goal), goal=episode.goal, steps=tuple(steps),
                  verified=bool(episode.success), verify_text=verify_text,
                  verify_gone=verify_gone, verify_appeared=verify_appeared,
                  recorded=datetime.now(timezone.utc).isoformat(timespec="seconds"),
                  notes=f"stop_reason={episode.stop_reason}")
    if library is not None:
        library.add(skill, overwrite=overwrite)
    return skill


@dataclass
class SkillLibrary:
    """A portable, diffable set of skills - JSON on disk, labels inside, no weights anywhere."""

    path: Path | None = None
    skills: dict[str, Skill] = field(default_factory=dict)

    def add(self, skill: Skill, *, overwrite: bool = True) -> Skill:
        if skill.name in self.skills and not overwrite:
            raise KeyError(f"a skill named {skill.name!r} is already in the library")
        self.skills[skill.name] = skill
        return skill

    def remove(self, name: str) -> bool:
        return self.skills.pop(name, None) is not None

    def get(self, name: str) -> Skill | None:
        return self.skills.get(name)

    def names(self) -> tuple[str, ...]:
        return tuple(sorted(self.skills))

    def __len__(self) -> int:
        return len(self.skills)

    def __iter__(self) -> Iterator[Skill]:
        return iter(self.skills.values())

    def match(self, goal: str, observation: Observation, *,
              threshold: float = 0.6) -> Skill | None:
        """The best applicable skill for a goal, or None. Never returns a skill aimed at nothing."""
        candidates = [s for s in self.skills.values() if s.applies(goal, observation, threshold=threshold)]
        if not candidates:
            return None
        candidates.sort(key=lambda s: (s.score(goal), s.recorded, s.name), reverse=True)
        return candidates[0]

    def to_rows(self) -> list[dict[str, Any]]:
        return [{"name": s.name, "steps": len(s.steps), "verified": s.verified,
                 "goal": s.goal, "recorded": s.recorded, "needs": ", ".join(s.preconditions())}
                for s in sorted(self.skills.values(), key=lambda s: s.name)]

    def to_dict(self) -> dict[str, Any]:
        return {"format": SKILLS_FORMAT, "version": SKILLS_VERSION,
                "skills": [s.to_dict() for s in sorted(self.skills.values(), key=lambda s: s.name)]}

    def save(self, path: Path | str | None = None) -> Path:
        target = Path(path or self.path or DEFAULT_SKILLS_PATH)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(json.dumps(self.to_dict(), indent=2) + "\n")
        self.path = target
        return target

    @classmethod
    def load(cls, path: Path | str | None = None, *, missing_ok: bool = True) -> "SkillLibrary":
        source = Path(path or DEFAULT_SKILLS_PATH)
        if not source.exists():
            if missing_ok:
                return cls(path=source)
            raise FileNotFoundError(source)
        data = json.loads(source.read_text())
        if not isinstance(data, dict) or "skills" not in data:
            raise ValueError(f"{source} is not a skill library")
        library = cls(path=source)
        for entry in data["skills"]:
            library.add(Skill.from_dict(entry))
        return library


class SkillBrain:
    """A brain that answers from a skill library, and says so when it cannot.

    It never invents an action: it re-grounds the next step of the matched skill on the current
    screen and, if that step's label is gone, raises instead of clicking at where the button used
    to be. With a ``fallback`` brain it defers the *choice* the first time and reuses it after.
    """

    def __init__(self, library: SkillLibrary, fallback: Any = None, *, threshold: float = 0.6,
                 name: str = "skills") -> None:
        self.library = library
        self.fallback = fallback
        self.threshold = threshold
        self.name = name
        self._skill: Skill | None = None
        self._goal = ""
        self._cursor = 0

    @property
    def skill(self) -> Skill | None:
        """The skill currently being replayed, if any."""
        return self._skill

    @property
    def cursor(self) -> int:
        return self._cursor

    @property
    def progress(self) -> str:
        if self._skill is None:
            return "-"
        return f"{self._cursor}/{len(self._skill.steps)}"

    def reset(self) -> "SkillBrain":
        self._skill, self._goal, self._cursor = None, "", 0
        return self

    def decide(self, goal: str, observation: Observation) -> Action:
        if self._skill is None or goal != self._goal:
            self._skill = self.library.match(goal, observation, threshold=self.threshold)
            self._goal, self._cursor = goal, 0
        if self._skill is None:
            if self.fallback is not None:
                return self.fallback.decide(goal, observation)
            raise GoalUnreachable(f"no skill matches {goal!r} (the library has {len(self.library)})")
        if self._cursor >= len(self._skill.steps):
            return Action("done", reason=f"skill {self._skill.name!r} finished")
        step = self._skill.steps[self._cursor]
        self._cursor += 1
        if not step.aimed_at:
            return step.action
        found = observation.find(step.aimed_at)
        if found is None:
            raise BrainError(f"skill {self._skill.name!r} step {self._cursor} needs "
                             f"{step.aimed_at!r}, which is not on this screen")
        # re-ground on the label the screen actually shows now, not the one recorded months ago
        return Action(step.action.kind, label=found.label,
                      text=step.action.text, keys=step.action.keys,
                      amount=step.action.amount, seconds=step.action.seconds,
                      reason=f"skill {self._skill.name!r} step {self._cursor}")