openworker/coworker/personas/manifest.py

"""Persona manifest — parse + validate a persona definition.

Format: YAML frontmatter (identity + capability declaration) followed by a markdown body that
is the system prompt. `persona ⊇ skill` — the same frontmatter-markdown shape as SKILL.md, with
more structured fields. Parsing is strict: an invalid manifest raises ``ManifestError`` rather
than silently producing a broken persona (a third-party persona must fail loudly).
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional

import yaml

# Persona ids become directory names under the managed install area (and registry keys), so
# they are restricted to a filesystem-safe slug on every OS: no path separators or `..`
# (traversal), no `:*?"<>|` (invalid on Windows), bounded length.
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")

VALID_FAMILIES = {"code", "knowledge"}  # legacy key, shimmed in parse()
VALID_TEAM = {"lead", "worker"}
VALID_MODES = {"discuss", "plan", "interactive", "custom", "auto"}
VALID_REC_KINDS = {"connector", "mcp"}
VALID_REC_TIERS = {"core", "optional"}
VALID_GROUPS = {"general", "security"}


class ManifestError(ValueError):
    """A persona manifest is malformed or references unknown capabilities/values."""


@dataclass
class Recommendation:
    """A connection a persona recommends, surfaced in the per-session connections drawer. ``ref`` is a
    connector id or an MCP server name; ``reason`` is the value it unlocks; ``tier`` ranks it. Not
    validated against shipped connectors — a persona may recommend one we don't ship yet.
    """

    kind: str  # "connector" | "mcp"
    ref: str
    reason: str = ""
    tier: str = "optional"  # "core" | "optional"


@dataclass
class PersonaManifest:
    id: str
    name: str
    system_prompt: str
    icon: str = ""
    tagline: str = ""
    description: str = ""
    tools: list[str] = field(default_factory=list)
    # Workspace/toolset traits (workspace-scratch-design.md — replaces the old
    # family/workspace pair). requires_folder: the composer/engine gate on a
    # user-picked primary folder. subagents: explorer fan-out. scheduling:
    # scheduled tasks + self-wake (defaults to the opposite of requires_folder
    # when the manifest is silent — folder personas fan out instead).
    requires_folder: bool = False
    subagents: bool = False
    scheduling: bool = True
    messaging: bool = False
    # Connector grant (OPE-93): False = none, a tuple = allowlist of connector ids
    # (session exposes declared ∩ connected), True = every connected connector — the
    # `all` sentinel, reserved for built-in general personas. Coarser grants leaked
    # undeclared tools (browser, email) into security sessions; undeclared = absent.
    connectors: bool | tuple[str, ...] = False
    # Team identity (agent-teams design, third/fourth pass): "lead" = coordinates a
    # team (gets the board coordination verbs + gates; consent copy says "can create
    # and direct worker coworkers"); "worker" = purpose-built to work under a lead
    # (board worker verbs, no ask_user-shaped prompt); None = solo-only. Solo
    # personas are NOT team-eligible — team-awareness changes who the prompt talks
    # to, so staffing fails closed on personas without the trait.
    team: Optional[str] = None
    default_permission_mode: str = "interactive"
    recommended_models: list[str] = field(default_factory=list)
    skills: list[str] = field(default_factory=list)
    mcp: list[str] = field(default_factory=list)
    # Sharing v1 (OPE-7): the author's version string ("1", "1.2", "2026-08"…). Purely
    # informational provenance — with folder/git distribution there is no authoritative
    # update channel, so this drives the "replaces vN" note on re-install, nothing more.
    version: str = ""
    recommends: list[Recommendation] = field(default_factory=list)
    # Distribution decision, not a maturity claim (owner, 2026-08-21): ships:false
    # coworkers exist in the codebase but are absent from release builds — internal
    # builds opt them in via OPENWORKER_UNSHIPPED=1.
    ships: bool = True
    # Settings-page grouping ("general" | "security"). Cosmetic — grouping never
    # gates behavior, so a third-party persona claiming "security" is harmless.
    group: str = "general"
    builtin: bool = False
    source: Optional[str] = (
        None  # where it was loaded from (path / url), for provenance
    )

    def to_agent(self):
        """Materialize the runtime Agent (prompt + catalog-expanded tools + traits)."""
        from ..agents.base import Agent
        from ..catalog import expand

        tool_ids = list(self.tools)
        factory = (lambda ctx: expand(tool_ids, ctx)) if tool_ids else None
        return Agent(
            name=self.id,
            title=self.name,
            system_prompt=self.system_prompt,
            tool_factory=factory,
            requires_folder=self.requires_folder,
            subagents=self.subagents,
            scheduling=self.scheduling,
            messaging=self.messaging,
            connectors=self.connectors,
            team=self.team,
        )


def _connectors(
    persona_id: str,
    raw: Any,
    recommends: list[Recommendation],
    builtin: bool,
) -> bool | tuple[str, ...]:
    """Parse the connector grant (OPE-93). Fail closed at every ambiguity.

    - list → explicit allowlist (the normal case).
    - "all" → every connected connector; reserved for BUILT-IN general personas — a
      shared bundle claiming it is exactly the trust violation the allowlist exists
      to prevent, so third-party loads reject it.
    - legacy `true` (pre-allowlist manifests) → the connector refs the manifest already
      recommends (author intent); no recommends → no grant.
    - recommends must stay within the grant: a recommendation the coworker can't use is
      author drift, surfaced at load rather than at the user's consent screen.
    """
    if raw is None or raw is False:
        declared: bool | tuple[str, ...] = False
    elif raw is True:
        refs = {r.ref for r in recommends if r.kind == "connector"}
        declared = tuple(sorted(refs)) if refs else False
    elif isinstance(raw, str):
        if raw.strip().lower() != "all":
            raise ManifestError(
                f"{persona_id}: `connectors` must be a list of connector ids or 'all'"
            )
        if not builtin:
            raise ManifestError(
                f"{persona_id}: `connectors: all` is reserved for built-in coworkers — "
                "declare the specific connectors this coworker uses"
            )
        declared = True
    elif isinstance(raw, list):
        declared = tuple(
            dict.fromkeys(s for s in (str(x).strip() for x in raw) if s)
        )
    else:
        raise ManifestError(
            f"{persona_id}: `connectors` must be a list of connector ids or 'all'"
        )

    if declared is not True:
        granted = set(declared or ())
        for r in recommends:
            if r.kind == "connector" and r.ref not in granted:
                raise ManifestError(
                    f"{persona_id}: recommends connector '{r.ref}' but does not declare "
                    "it in `connectors` — a recommendation must stay within the grant"
                )
    return declared


def _split_frontmatter(text: str) -> tuple[dict[str, Any], str]:
    if not text.startswith("---"):
        raise ManifestError("manifest must start with a YAML frontmatter block (---)")
    end = text.find("\n---", 3)
    if end == -1:
        raise ManifestError("unterminated frontmatter block (missing closing ---)")
    raw = text[3:end]
    body = text[end + 4 :].lstrip("\n")
    try:
        meta = yaml.safe_load(raw) or {}
    except yaml.YAMLError as e:  # pragma: no cover - exercised via parse error path
        raise ManifestError(f"invalid YAML frontmatter: {e}") from e
    if not isinstance(meta, dict):
        raise ManifestError("frontmatter must be a mapping of key: value")
    return meta, body


def _slugify(stem: str) -> str:
    """Normalize a filename stem into the persona-id charset (used only for ids derived
    from filenames; explicit `id:` values must already be valid)."""
    slug = re.sub(r"[^a-z0-9_-]+", "-", stem.strip().lower()).strip("-_")[:64]
    return slug if _ID_RE.match(slug) else ""


def _strlist(meta: dict, key: str) -> list[str]:
    val = meta.get(key, [])
    if val is None:
        return []
    if isinstance(val, str):
        return [v.strip() for v in val.split(",") if v.strip()]
    if isinstance(val, list):
        return [str(v).strip() for v in val if str(v).strip()]
    raise ManifestError(f"`{key}` must be a list or comma-separated string")


def _recommends(persona_id: str, meta: dict) -> list[Recommendation]:
    raw = meta.get("recommends")
    if raw is None:
        return []
    if not isinstance(raw, list):
        raise ManifestError(f"persona {persona_id!r}: `recommends` must be a list")
    out: list[Recommendation] = []
    for item in raw:
        if not isinstance(item, dict):
            raise ManifestError(
                f"persona {persona_id!r}: each `recommends` item must be a mapping"
            )
        if "connector" in item:
            kind, ref = "connector", str(item.get("connector") or "").strip()
        elif "mcp" in item:
            kind, ref = "mcp", str(item.get("mcp") or "").strip()
        else:
            raise ManifestError(
                f"persona {persona_id!r}: each `recommends` item needs a `connector:` or `mcp:` key"
            )
        if not ref:
            raise ManifestError(
                f"persona {persona_id!r}: a `recommends` item has an empty {kind}"
            )
        tier = str(item.get("tier", "optional")).strip().lower()
        if tier not in VALID_REC_TIERS:
            raise ManifestError(
                f"persona {persona_id!r}: recommend tier must be one of {sorted(VALID_REC_TIERS)}"
            )
        out.append(
            Recommendation(
                kind=kind,
                ref=ref,
                reason=str(item.get("reason", "")).strip(),
                tier=tier,
            )
        )
    return out


def parse_manifest(
    text: str,
    *,
    fallback_id: Optional[str] = None,
    builtin: bool = False,
    source: Optional[str] = None,
) -> PersonaManifest:
    meta, body = _split_frontmatter(text)

    explicit_id = str(meta.get("id") or "").strip()
    if explicit_id:
        persona_id = explicit_id
        if not _ID_RE.match(persona_id):
            raise ManifestError(
                f"persona id {persona_id!r} is invalid: lowercase letters, digits, '-' or '_' "
                "only, starting with a letter/digit, max 64 chars (ids become directory names)"
            )
    else:
        # Derived from the filename: normalize it into the id charset instead of erroring,
        # so `My Persona.md` without an explicit id still installs (as `my-persona`).
        persona_id = _slugify(str(fallback_id or ""))
        if not persona_id:
            raise ManifestError(
                "manifest needs an `id` (or a filename to derive one from)"
            )
    if not body.strip():
        raise ManifestError(f"persona {persona_id!r} has no body (the system prompt)")

    # Workspace/toolset traits (workspace-scratch-design.md). Legacy shim: pre-trait
    # bundles declared `family: code|knowledge` (and a dead `workspace:` enum, ignored
    # here) — when the new keys are absent, `family: code` maps to the folder-gated
    # profile so an old bundle keeps its gate. New keys always win.
    legacy_family = str(meta.get("family", "")).strip().lower()
    if legacy_family and legacy_family not in VALID_FAMILIES:
        raise ManifestError(
            f"persona {persona_id!r}: family (legacy) must be one of {sorted(VALID_FAMILIES)}"
        )
    legacy_code = legacy_family == "code"
    requires_folder = bool(meta.get("requires_folder", legacy_code))
    subagents = bool(meta.get("subagents", legacy_code))
    # Folder personas fan out to explorers instead of scheduling — the silent default
    # mirrors that split; either can be declared explicitly.
    scheduling = bool(meta.get("scheduling", not requires_folder))

    mode = str(meta.get("default_permission_mode", "interactive")).strip().lower()
    if mode not in VALID_MODES:
        raise ManifestError(
            f"persona {persona_id!r}: default_permission_mode must be one of {sorted(VALID_MODES)}"
        )

    group = str(meta.get("group", "general") or "general").strip().lower()
    if group not in VALID_GROUPS:
        raise ManifestError(
            f"persona {persona_id!r}: group must be one of {sorted(VALID_GROUPS)}"
        )

    team_raw = str(meta.get("team", "") or "").strip().lower()
    if team_raw and team_raw not in VALID_TEAM:
        raise ManifestError(
            f"persona {persona_id!r}: team must be one of {sorted(VALID_TEAM)}"
            " (omit for a solo coworker)"
        )

    tools = _strlist(meta, "tools")
    _validate_tools(persona_id, tools)
    recommends = _recommends(persona_id, meta)
    connectors = _connectors(persona_id, meta.get("connectors"), recommends, builtin)

    return PersonaManifest(
        id=persona_id,
        name=str(meta.get("name") or persona_id).strip(),
        system_prompt=body.strip(),
        icon=str(meta.get("icon", "")).strip(),
        tagline=str(meta.get("tagline", "")).strip(),
        description=str(meta.get("description", "")).strip(),
        tools=tools,
        requires_folder=requires_folder,
        subagents=subagents,
        scheduling=scheduling,
        messaging=bool(meta.get("messaging", False)),
        connectors=connectors,
        team=team_raw or None,
        default_permission_mode=mode,
        recommended_models=_strlist(meta, "recommended_models"),
        skills=_strlist(meta, "skills"),
        mcp=_strlist(meta, "mcp"),
        version=str(meta.get("version", "") or "").strip(),
        recommends=recommends,
        ships=bool(meta.get("ships", True)),
        group=group,
        builtin=builtin,
        source=source,
    )


def _validate_tools(persona_id: str, tools: list[str]) -> None:
    # Imported here to avoid a module-load cycle (catalog imports agents.base).
    from ..catalog import CATALOG

    unknown = [t for t in tools if t not in CATALOG]
    if unknown:
        raise ManifestError(
            f"persona {persona_id!r} references unknown tool capabilities: {unknown}. "
            f"Known: {sorted(CATALOG)}"
        )


def load_manifest_file(path: str | Path, *, builtin: bool = False) -> PersonaManifest:
    p = Path(path)
    return parse_manifest(
        p.read_text(encoding="utf-8"),
        fallback_id=p.stem,
        builtin=builtin,
        source=str(p),
    )