distinct_tools/skills.py

"""Skills: library members that produce an artifact rather than an answer.

A skill is not a different mechanism. It is a :class:`~distinct_tools.core.ToolSpec`
with ``kind="skill"``, registered in the same registry, approved by the same
operator policy, selected through the same per-run allowlist, and executed
behind the same broker with the same quotas, timeouts and byte caps. The only
difference is the shape of the result: a skill's output carries an ``artifact``
mapping, which the agent harness lifts out of the model's context and returns
with the run so the user's server session can store and offer it for download.

The artifact envelope is one shape everywhere::

    {"artifact": {"name": ..., "media_type": ..., "base64": ..., "size_bytes": ...}}

Like the local tools, nothing here opens a socket, reads a file, spawns a
process or consults a model. ``create_pdf`` builds the PDF bytes in memory from
its arguments; ``make_plan`` is a deterministic formatter. Both answer in one
call, which is the same viability criterion the local tool set applies below
7B.

The plan skill deserves its caveat stated rather than hidden:
``TOOL_CATEGORIES_REVIEW.md`` section 3 records that the one direct small-model
measurement found Llama-1B performing *worse* with its own plans than without.
``make_plan`` exists because a written, numbered plan is a deliverable a user
asked this network to produce, not because it improves the model. Its
description says when not to call it.

**The skill repository.** Beside the two skills written in this module there is
:mod:`distinct_skills`, a repository of skills held on disk in the Anthropic
Agent Skills format: a directory each, with a ``SKILL.md`` carrying the
frontmatter a human reads and the description the model reads, a ``tool.json``
carrying the argument schema, and the templates the output is rendered from.
:func:`install_skill_repository` turns one of those into exactly the same
``ToolSpec`` with ``kind="skill"`` that ``create_pdf`` is, in the same
registry, behind the same broker, refused by the same operator policy. There
is no parallel path and no second kind of skill.

Three properties are worth stating because the obvious alternative to each is
what goes wrong:

1. **The directory is read once, at install time.** ``load_repository`` runs
   when this module is imported, which is process start-up, and the result is
   held. A run never touches the disk. This is the same rule
   :mod:`distinct_tools.mcp` states for ``tools/list`` and for the same reason:
   ``ToolBroker`` snapshots the registry so a job's capabilities cannot change
   under it, and a definition re-read per call would defeat that from the other
   end. The cost is honest and worth naming: a skill edited while the worker is
   running is not picked up until it restarts.
2. **The definition is digest-pinned.** An edit to any file in a skill
   directory, template included, changes its digest and is refused at load
   rather than used. So the failure mode of a tampered file is a skill that
   does not install, not a skill that quietly does something else.
3. **A skill directory supplies data, never code.** ``tool.json`` names a
   handler and the name is resolved in a fixed table in
   :mod:`distinct_skills.handlers`. Importing a module out of the skill folder
   would make dropping in a directory equivalent to running arbitrary code,
   which is the property this repository refuses everywhere else.

Repository skills are deliberately **not** in
:data:`~distinct_tools.INSTALLABLE_SPECS` and therefore not in the ``local``
bundle. The bundle is what a single ``--tools local`` flag approves, and a
member that arrived from a directory rather than from this source file should
not join it without the operator seeing it by name. They are reachable, and
:func:`~distinct_tools.default_registry` installs them, but only for an
operator who wrote the exact ``id@version`` ref: the approval is read against
``INSTALLABLE_SPECS`` alone and the repository is then filtered through
``permitted_specs``, which expands no bundle. An MCP server's tools are
approved the same way and for the same reason.

Two consequences worth stating rather than leaving to be discovered.
``distinct_agent``'s ``--list-tools`` shows the repository as its own group,
so an operator can see what there is to name. The server's library picker
still reads ``INSTALLABLE_SPECS``, so a repository skill cannot be selected
from the web interface yet; that is a change to ``distinct_server``, which
this work does not own.
"""

from __future__ import annotations

import base64
import zlib
from collections.abc import Mapping, Sequence

from .core import Registry, ToolContext, ToolInputError, ToolRef, ToolSpec

try:
    from distinct_skills import (
        HANDLERS,
        SkillDefinition,
        SkillError,
        SkillInputError,
        describe,
        load_repository,
    )

    _IMPORT_FAILURE = ""
except ImportError as exc:  # pragma: no cover - both packages ship together
    # ``pyproject.toml`` ships ``distinct_skills`` alongside this package, so
    # this should not fire. It is kept because the two are separable: the tool
    # framework is useful without the skill repository, and a deployment that
    # carries only the framework should lose four skills rather than lose
    # every tool. An import error here would otherwise stop ``distinct_tools``
    # importing at all.
    _IMPORT_FAILURE = f"distinct_skills is not importable ({exc})"
    # Bound so that the explicit install path raises the error it promises
    # rather than a NameError about a missing exception class.
    SkillError = RuntimeError
    SkillInputError = ValueError

MAX_PDF_TEXT_CHARACTERS = 12_000
MAX_TITLE_CHARACTERS = 160
MAX_FILENAME_CHARACTERS = 80
MAX_PLAN_STEPS = 20
MAX_STEP_CHARACTERS = 400

_PAGE_WIDTH = 595  # A4 portrait, points
_PAGE_HEIGHT = 842
_MARGIN = 56
_BODY_SIZE = 11
_TITLE_SIZE = 18
_LEADING = 16
_CHARS_PER_LINE = 88
_LINES_PER_PAGE = int((_PAGE_HEIGHT - 2 * _MARGIN - 40) / _LEADING)


def _safe_filename(value: object, fallback: str) -> str:
    if not isinstance(value, str) or not value.strip():
        return fallback
    cleaned = "".join(
        ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_", ".", " ")
    ).strip()
    cleaned = cleaned.replace(" ", "-")[:MAX_FILENAME_CHARACTERS].strip(".") or fallback
    return cleaned


def _pdf_escape(text: str) -> bytes:
    """Escape a text run for a PDF literal string, in Latin-1.

    Characters outside Latin-1 are replaced rather than dropped, so the output
    is always well-formed even for input the base fonts cannot draw.
    """

    encoded = text.encode("latin-1", errors="replace")
    return (
        encoded.replace(b"\\", b"\\\\").replace(b"(", b"\\(").replace(b")", b"\\)")
    )


def _wrap(text: str, width: int) -> list[str]:
    lines: list[str] = []
    for raw_line in text.splitlines() or [""]:
        line = raw_line.rstrip()
        if not line:
            lines.append("")
            continue
        current = ""
        for word in line.split(" "):
            candidate = f"{current} {word}".strip()
            if len(candidate) <= width:
                current = candidate
                continue
            if current:
                lines.append(current)
            while len(word) > width:
                lines.append(word[:width])
                word = word[width:]
            current = word
        lines.append(current)
    return lines


def _build_pdf(title: str, body: str) -> bytes:
    """A complete, valid, dependency-free PDF: Helvetica, A4, flate-compressed."""

    wrapped = _wrap(body, _CHARS_PER_LINE)
    pages: list[list[str]] = []
    for index in range(0, max(1, len(wrapped)), _LINES_PER_PAGE):
        pages.append(wrapped[index : index + _LINES_PER_PAGE])

    content_streams: list[bytes] = []
    for page_number, page_lines in enumerate(pages):
        parts: list[bytes] = [b"BT\n"]
        cursor = _PAGE_HEIGHT - _MARGIN
        if page_number == 0 and title:
            parts.append(
                b"/F2 %d Tf\n1 0 0 1 %d %d Tm\n(%s) Tj\n"
                % (_TITLE_SIZE, _MARGIN, cursor, _pdf_escape(title))
            )
            cursor -= 30
        parts.append(b"/F1 %d Tf\n" % _BODY_SIZE)
        for line in page_lines:
            parts.append(
                b"1 0 0 1 %d %d Tm\n(%s) Tj\n" % (_MARGIN, cursor, _pdf_escape(line))
            )
            cursor -= _LEADING
        parts.append(b"ET")
        content_streams.append(zlib.compress(b"".join(parts)))

    objects: list[bytes] = []
    page_count = len(content_streams)
    # 1: catalog, 2: pages, 3: F1, 4: F2, then per page: page object, content.
    first_page_object = 5
    kids = b" ".join(
        b"%d 0 R" % (first_page_object + index * 2) for index in range(page_count)
    )
    objects.append(b"<< /Type /Catalog /Pages 2 0 R >>")
    objects.append(
        b"<< /Type /Pages /Kids [%s] /Count %d >>" % (kids, page_count)
    )
    objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
    objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica-Bold >>")
    for index, stream in enumerate(content_streams):
        content_object = first_page_object + index * 2 + 1
        objects.append(
            b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 %d %d] "
            b"/Resources << /Font << /F1 3 0 R /F2 4 0 R >> >> /Contents %d 0 R >>"
            % (_PAGE_WIDTH, _PAGE_HEIGHT, content_object)
        )
        objects.append(
            b"<< /Length %d /Filter /FlateDecode >>\nstream\n" % len(stream)
            + stream
            + b"\nendstream"
        )

    buffer = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
    offsets: list[int] = []
    for number, obj in enumerate(objects, start=1):
        offsets.append(len(buffer))
        buffer += b"%d 0 obj\n" % number + obj + b"\nendobj\n"
    xref_offset = len(buffer)
    buffer += b"xref\n0 %d\n0000000000 65535 f \n" % (len(objects) + 1)
    for offset in offsets:
        buffer += b"%010d 00000 n \n" % offset
    buffer += (
        b"trailer\n<< /Size %d /Root 1 0 R >>\nstartxref\n%d\n%%%%EOF\n"
        % (len(objects) + 1, xref_offset)
    )
    return bytes(buffer)


def _create_pdf(arguments: Mapping[str, object], context: ToolContext) -> Mapping[str, object]:
    del context
    known = frozenset({"title", "body", "filename"})
    unknown = sorted(set(map(str, arguments)) - known)
    if unknown:
        raise ToolInputError(f"unsupported arguments: {', '.join(unknown)}")
    body = arguments.get("body")
    if not isinstance(body, str) or not body.strip():
        raise ToolInputError("body must be a non-empty string")
    if len(body) > MAX_PDF_TEXT_CHARACTERS:
        raise ToolInputError(
            f"body must be at most {MAX_PDF_TEXT_CHARACTERS:,} characters"
        )
    title = arguments.get("title")
    if title is not None and not isinstance(title, str):
        raise ToolInputError("title must be a string")
    title = (title or "").strip()[:MAX_TITLE_CHARACTERS]
    name = _safe_filename(arguments.get("filename"), "document")
    if not name.lower().endswith(".pdf"):
        name = f"{name}.pdf"
    payload = _build_pdf(title, body)
    return {
        "created": True,
        "pages": max(1, (len(_wrap(body, _CHARS_PER_LINE)) + _LINES_PER_PAGE - 1) // _LINES_PER_PAGE),
        "artifact": {
            "name": name,
            "media_type": "application/pdf",
            "base64": base64.b64encode(payload).decode("ascii"),
            "size_bytes": len(payload),
        },
    }


def _make_plan(arguments: Mapping[str, object], context: ToolContext) -> Mapping[str, object]:
    del context
    known = frozenset({"goal", "steps"})
    unknown = sorted(set(map(str, arguments)) - known)
    if unknown:
        raise ToolInputError(f"unsupported arguments: {', '.join(unknown)}")
    goal = arguments.get("goal")
    if not isinstance(goal, str) or not goal.strip():
        raise ToolInputError("goal must be a non-empty string")
    if len(goal) > MAX_STEP_CHARACTERS:
        raise ToolInputError(f"goal must be at most {MAX_STEP_CHARACTERS} characters")
    steps = arguments.get("steps")
    if not isinstance(steps, (list, tuple)) or not steps:
        raise ToolInputError("steps must be a non-empty list of strings")
    if len(steps) > MAX_PLAN_STEPS:
        raise ToolInputError(f"steps must contain at most {MAX_PLAN_STEPS} entries")
    cleaned: list[str] = []
    for step in steps:
        if not isinstance(step, str) or not step.strip():
            raise ToolInputError("every step must be a non-empty string")
        if len(step) > MAX_STEP_CHARACTERS:
            raise ToolInputError(
                f"each step must be at most {MAX_STEP_CHARACTERS} characters"
            )
        cleaned.append(" ".join(step.split()))
    lines = [f"Plan: {' '.join(goal.split())}", ""]
    lines += [f"{index}. {step}" for index, step in enumerate(cleaned, start=1)]
    text = "\n".join(lines)
    return {
        "plan": text,
        "step_count": len(cleaned),
        "artifact": {
            "name": "plan.txt",
            "media_type": "text/plain",
            "base64": base64.b64encode(text.encode("utf-8")).decode("ascii"),
            "size_bytes": len(text.encode("utf-8")),
        },
    }


SKILL_SPECS: tuple[ToolSpec, ...] = (
    ToolSpec(
        tool_id="create_pdf",
        version="1",
        kind="skill",
        description=(
            "Create a PDF document from text you already have, and return it as a "
            "downloadable file for the user. Arguments: body (required text, at most "
            "12,000 characters), title (optional heading), filename (optional). Call "
            "this only when the user asked for a document or a file; do not call it "
            "to answer an ordinary question, and do not call it twice for the same "
            "document."
        ),
        input_schema={
            "type": "object",
            "additionalProperties": False,
            "properties": {
                "title": {"type": "string", "description": "optional document title"},
                "body": {"type": "string", "description": "the document text"},
                "filename": {"type": "string", "description": "optional file name"},
            },
            "required": ["body"],
        },
    ),
    ToolSpec(
        tool_id="make_plan",
        version="1",
        kind="skill",
        description=(
            "Record a numbered plan as a small text artifact the user can keep. "
            "Arguments: goal (one sentence) and steps (a list of short strings, at "
            "most 20). Call this only when the user asked for a plan as a "
            "deliverable; do not call it to organise your own reasoning, and never "
            "call it more than once per request."
        ),
        input_schema={
            "type": "object",
            "additionalProperties": False,
            "properties": {
                "goal": {"type": "string", "description": "what the plan achieves"},
                "steps": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "ordered steps, each one action",
                },
            },
            "required": ["goal", "steps"],
        },
    ),
)


def register_skills(registry: Registry, *, only: frozenset | None = None) -> None:
    """Register the skill set explicitly, mirroring ``register_local_tools``.

    This registers the two skills written in this module. The repository on
    disk is installed by :func:`install_skill_repository`, which is a separate
    call because it is a separate decision: these two ship approved by the
    ``local`` bundle, and a definition read from a directory should not.
    """

    handlers = {
        "create_pdf": _create_pdf,
        "make_plan": _make_plan,
    }
    for spec in SKILL_SPECS:
        if only is not None and spec.ref not in only:
            continue
        registry.register(spec, handlers[spec.tool_id])


# --------------------------------------------------------------------------
# The on-disk repository: read once, pinned, installed by an explicit call
# --------------------------------------------------------------------------


def _spec_for(skill: SkillDefinition) -> ToolSpec:
    """Turn a loaded skill into an ordinary spec, with nothing added.

    ``origin`` is carried through rather than forced to ``builtin``. The
    packaged repository is this repository's own text and says so; a directory
    an operator supplied says something else, and ``ToolSpec.third_party``
    then reports true, which is the flag a harness uses to decide whether a
    description needs fencing before it reaches the prompt.
    """

    return ToolSpec(
        tool_id=skill.tool_id,
        version=skill.version,
        kind="skill",
        description=skill.description,
        input_schema=skill.input_schema,
        origin=skill.origin,
    )


def _bind(skill: SkillDefinition):
    """Adapt a repository handler to the broker's handler signature.

    The repository knows nothing about this framework, deliberately, so its
    handlers take their own arguments and raise their own input error. The
    translation happens here, at the one boundary, rather than by having the
    repository import the framework and risk an import cycle.
    """

    try:
        handler = HANDLERS[skill.handler]
    except KeyError as exc:
        # Reachable only for a definition loaded without the handler table to
        # check it against. Failing here means failing at install time, which
        # is where an unresolvable name should stop rather than mid-run.
        raise SkillError(
            f"skill {skill.name!r} names the handler {skill.handler!r}, which this "
            "repository does not define"
        ) from exc

    def call(arguments: Mapping[str, object], context: ToolContext) -> object:
        del context  # no deadline, no hosts: these are pure transformations
        try:
            return handler(arguments, skill)
        except SkillInputError as exc:
            raise ToolInputError(str(exc)) from exc

    call.__name__ = f"skill_{skill.tool_id}"
    return call


def _load_packaged_repository() -> tuple[tuple, tuple, str]:
    """Read the packaged repository once, and never fail an import over it.

    A broken or tampered repository must not stop the worker starting, because
    the rest of the tool framework is unaffected by it and a worker that
    refuses to boot is a worse outcome than a worker missing four skills. It
    must also not be silent, so the reason is kept and
    :func:`describe_skill_repository` prints it. An explicit
    :func:`install_skill_repository` call raises instead: a caller who asked
    for the repository is entitled to the exception.
    """

    if _IMPORT_FAILURE:
        return (), (), _IMPORT_FAILURE
    try:
        skills = load_repository()
        specs = tuple(_spec_for(skill) for skill in skills)
    except (SkillError, ValueError, OSError) as exc:
        return (), (), f"the packaged skill repository was refused: {exc}"
    return skills, specs, f"{len(skills)} skill(s) loaded and digest-checked"


#: The packaged repository, read at import time, which is process start-up and
#: therefore install time. Empty when it could not be read; the reason is in
#: :data:`REPOSITORY_STATUS`.
REPOSITORY_SKILLS, REPOSITORY_SPECS, REPOSITORY_STATUS = _load_packaged_repository()


def install_skill_repository(
    registry: Registry,
    *,
    only: frozenset[ToolRef] | None = None,
    skills: Sequence[SkillDefinition] | None = None,
) -> tuple[ToolSpec, ...]:
    """Register repository skills into an ordinary registry.

    ``only`` filters by exact ``ToolRef``, exactly as ``register_skills`` does,
    so the caller passes whatever the operator approved and nothing else
    appears. Approval itself is unchanged and happens elsewhere: the operator
    policy decides, and :class:`ToolBroker` refuses anything unapproved before
    the job starts.

    ``skills`` overrides what is installed, for an operator running their own
    directory: they call :func:`distinct_skills.load_repository` with their own
    root and pins and pass the result. Loading stays in one place that way, and
    the disk read is visible at the call site rather than hidden in here.

    :func:`~distinct_tools.default_registry` already does all of this for an
    operator who named a ref. A caller building a registry by hand should copy
    the shape it uses, and the first line most of all::

        # INSTALLABLE_SPECS and nothing else: `local` expands over this list.
        policy = load_policy(INSTALLABLE_SPECS)   # operator named status_update@1
        approved = frozenset(spec.ref for spec in policy.permitted_specs(INSTALLABLE_SPECS))
        repository = frozenset(spec.ref for spec in policy.permitted_specs(REPOSITORY_SPECS))
        registry = Registry()
        register_local_tools(registry, only=approved)
        register_skills(registry, only=approved)
        install_skill_repository(registry, only=repository)
        # The run's own allowlist is the third gate and normally comes from
        # the server's JobSpec; approving both halves is the standalone case.
        job = JobPolicy(allowed_tools=approved | repository)
        broker = ToolBroker(registry, job, job_id=..., operator_policy=policy)

    An earlier version of this docstring passed ``(*INSTALLABLE_SPECS,
    *REPOSITORY_SPECS)`` to ``load_policy``, which reads as harmless and is
    not: the bundle name ``local`` expands over whatever that list contains, so
    it would have quietly put every skill on disk inside one flag. The
    two-step above, approve narrowly then filter by ref, is the difference.

    Raises rather than installing part of a set, and raises rather than
    installing nothing quietly when the packaged repository could not be read.
    """

    if not isinstance(registry, Registry):
        raise TypeError("registry must be a Registry")
    if skills is None:
        if not REPOSITORY_SKILLS:
            raise SkillError(REPOSITORY_STATUS or "no skill repository is available")
        skills = REPOSITORY_SKILLS

    chosen = []
    for skill in skills:
        spec = _spec_for(skill)
        if only is not None and spec.ref not in only:
            continue
        chosen.append((spec, skill))

    # Every spec is built before anything is registered, so a malformed member
    # leaves the registry as it was rather than half filled.
    for spec, skill in chosen:
        registry.register(spec, _bind(skill))
    return tuple(spec for spec, _ in chosen)


def describe_skill_repository() -> str:
    """Render the packaged repository for an operator to read before approving."""

    if not REPOSITORY_SKILLS:
        return f"No repository skills are installed: {REPOSITORY_STATUS}"
    return describe(REPOSITORY_SKILLS)


__all__ = [
    "REPOSITORY_SKILLS",
    "REPOSITORY_SPECS",
    "REPOSITORY_STATUS",
    "SKILL_SPECS",
    "describe_skill_repository",
    "install_skill_repository",
    "register_skills",
]