| """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): |
| |
| |
| 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") |
| |
| 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}") |
|
|
