LLeMbas/src/lembas/services/library/skills.py

"""Skills: named instructions the model can choose to follow.

Two fields carry the design.

`description` is what gets injected -- one line per skill, for every skill --
and is therefore the only thing the model has to go on when deciding whether a
skill is relevant. A description that does not say *when* to use the skill makes
it invisible in practice.

`body` is fetched only when the model decides to use it. That split is what
makes a hundred skills affordable: the index costs a line each, the instructions
cost nothing until wanted.

**A model may rewrite its own skills**, which is the point -- it is how it
learns a procedure once instead of being told every time. The safety story is
not a gate but a record: every write snapshots what was there first, so a change
can be read and undone. A skill written after reading a hostile web page is a
real risk, and the honest mitigation is that it is visible, attributed and
revertible rather than that it was somehow prevented.
"""

from __future__ import annotations

import logging
import re

from sqlalchemy import select
from sqlalchemy.orm import Session as DBSession

from lembas.db.models import AUTHOR_MODEL, AUTHOR_USER, Skill, SkillRevision, User
from lembas.services import sharing
from lembas.services.library.fts import search_ids

log = logging.getLogger(__name__)

INDEX = "skills_fts"

# A name the model can quote back without getting it wrong.
SKILL_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{1,60}$")

MAX_DESCRIPTION_CHARS = 400
MAX_BODY_CHARS = 20_000

# The index goes into every request, so it has a ceiling like memory does.
MAX_INDEX_SKILLS = 60


class SkillError(Exception):
    """A rejected skill write, with a message fit for the model or the user."""


def slugify(name: str) -> str:
    cleaned = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower()).strip("-")
    return cleaned[:60]


def visible(db: DBSession, user: User | None):
    return select(Skill).where(sharing.visible_to(Skill, user))


def get(db: DBSession, skill_id: str, user: User | None) -> Skill | None:
    skill = db.get(Skill, skill_id)
    if skill is None or not sharing.can_read(db, skill, user):
        return None
    return skill


def by_name(db: DBSession, name: str, user: User | None) -> Skill | None:
    """Look one up the way the model refers to it."""
    if user is None:
        return None
    return db.scalar(visible(db, user).where(Skill.name == slugify(name)))


def enabled_for(db: DBSession, user: User | None) -> list[Skill]:
    """Skills that should appear in the index, oldest first for a stable order."""
    if user is None:
        return []
    return list(
        db.scalars(
            visible(db, user)
            .where(Skill.enabled.is_(True))
            .order_by(Skill.name)
            .limit(MAX_INDEX_SKILLS)
        )
    )


def search(db: DBSession, user: User | None, needle: str, *, limit: int = 10) -> list[Skill]:
    hits = search_ids(db, INDEX, needle, limit=limit * 4)
    if not hits:
        return []
    order = {hit.id: position for position, hit in enumerate(hits)}
    rows = list(db.scalars(visible(db, user).where(Skill.id.in_(list(order)))))
    rows.sort(key=lambda skill: order.get(skill.id, len(order)))
    return rows[:limit]


def snapshot(db: DBSession, skill: Skill, *, author: str, note: str = "") -> SkillRevision:
    """Record what a skill looked like before it is changed."""
    revision = SkillRevision(
        skill_id=skill.id,
        description=skill.description,
        body=skill.body,
        author=author,
        note=note[:200],
    )
    db.add(revision)
    return revision


def create(
    db: DBSession,
    *,
    owner: User,
    name: str,
    description: str,
    body: str,
    author: str = AUTHOR_USER,
) -> Skill:
    slug = slugify(name)
    if not SKILL_NAME_PATTERN.match(slug):
        raise SkillError(
            "A skill name must be two or more letters, numbers or hyphens, "
            "such as 'weekly-report'."
        )
    if by_name(db, slug, owner) is not None:
        raise SkillError(f"A skill called {slug!r} already exists. Edit it instead.")
    if not description.strip():
        raise SkillError(
            "A skill needs a description saying when to use it — it is the only "
            "thing shown until the skill is opened."
        )

    skill = Skill(
        owner_id=owner.id,
        name=slug,
        description=description.strip()[:MAX_DESCRIPTION_CHARS],
        body=body.strip()[:MAX_BODY_CHARS],
        author=author if author in (AUTHOR_USER, AUTHOR_MODEL) else AUTHOR_USER,
    )
    db.add(skill)
    db.commit()
    log.info("skill %r created by %s", slug, author)
    return skill


def update(
    db: DBSession,
    skill: Skill,
    *,
    description: str | None = None,
    body: str | None = None,
    enabled: bool | None = None,
    author: str = AUTHOR_USER,
    note: str = "",
) -> Skill:
    """Change a skill, keeping what it was.

    The snapshot happens before the change and in the same transaction, so
    there is no window where a skill has been rewritten with no record of what
    it used to say.
    """
    changing = (description is not None and description.strip() != skill.description) or (
        body is not None and body.strip() != skill.body
    )
    if changing:
        snapshot(db, skill, author=author, note=note)

    if description is not None and description.strip():
        skill.description = description.strip()[:MAX_DESCRIPTION_CHARS]
    if body is not None:
        skill.body = body.strip()[:MAX_BODY_CHARS]
    if enabled is not None:
        skill.enabled = enabled
    if changing:
        skill.author = author if author in (AUTHOR_USER, AUTHOR_MODEL) else skill.author

    db.commit()
    return skill


def revert(db: DBSession, skill: Skill, revision: SkillRevision, *, author: str) -> Skill:
    """Put a skill back to an earlier revision.

    The revert is itself a change, so the current state is snapshotted first --
    going back is undoable too.
    """
    snapshot(db, skill, author=author, note="before revert")
    skill.description = revision.description
    skill.body = revision.body
    db.commit()
    return skill


def delete(db: DBSession, skill: Skill) -> None:
    sharing.forget_resource(db, skill)
    db.delete(skill)
    db.commit()


def index_block(db: DBSession, user: User | None) -> str:
    """The one-line-per-skill listing that goes into the prompt."""
    skills = enabled_for(db, user)
    if not skills:
        return ""
    return "\n".join(f"- {skill.name}: {skill.description}" for skill in skills)