NexusOS/synapse/playbook_manager.py

from typing import List
from uuid import uuid4
from .playbooks.store import playbook_store, PlaybookItem

# Fields a caller can supply; anything else in a persist dict is ignored.
# `order` is deliberately excluded — see persist_playbook.
_EDITABLE_FIELDS = ("title", "goal", "instructions", "tags", "tools", "model")
_FIELD_DEFAULTS = {"title": "", "goal": "", "instructions": "", "tags": [], "tools": [], "model": ""}


def persist_playbook(data: dict, *, merge: bool) -> dict:
    """Write a playbook dict to the store, returning it as a plain dict.

    `merge=False` is today's behavior (main.py's HTTP form handlers): a full
    replace, with pydantic defaults for anything omitted. `merge=True` (used by
    the edit_playbook tool) instead keeps each field's *existing* value when the
    caller's dict doesn't supply it — a model calling with a partial argument
    set must not silently blank out the fields it didn't mention.

    `order` is never taken from `data` in merge mode: it is preserved from the
    existing playbook on update, or appended at the tail on create. Position 0
    is unconditionally the active system prompt (see get_main_playbook) — moving
    a playbook there is `make_main`'s job, never an accidental side effect of an
    ordinary field edit.
    """
    existing_id = str(data.get("id") or "")
    existing = playbook_store.get_playbook(existing_id) if existing_id else None

    if merge:
        fields = {}
        for key in _EDITABLE_FIELDS:
            if key in data and data[key] is not None:
                fields[key] = data[key]
            elif existing is not None:
                fields[key] = getattr(existing, key)
            else:
                fields[key] = _FIELD_DEFAULTS[key]
        if existing is None and not (fields["title"] and fields["goal"] and fields["instructions"]):
            raise ValueError("title, goal, and instructions are required to create a new playbook")
    else:
        fields = {key: data.get(key, _FIELD_DEFAULTS[key]) for key in _EDITABLE_FIELDS}

    order = existing.order if existing else data.get("order", len(playbook_store.all_playbooks()))
    item = PlaybookItem(id=existing_id or str(uuid4()), order=order, **fields)
    playbook_store.add_playbook(item)
    return item.model_dump()


def make_main(playbook_id: str) -> None:
    """Reorder so `playbook_id` is position 0 (the active system prompt)."""
    all_ids = [p.id for p in playbook_store.all_playbooks()]
    if playbook_id not in all_ids:
        raise ValueError(f"no playbook with id {playbook_id!r}")
    ordered = [playbook_id] + [pid for pid in all_ids if pid != playbook_id]
    playbook_store.reorder_playbooks(ordered)


class PlaybookManager:
    @classmethod
    def _all(cls) -> List[PlaybookItem]:
        """Return all playbooks sorted by order (position 0 is always main)."""
        return playbook_store.all_playbooks()

    @classmethod
    def get_main_playbook(cls) -> PlaybookItem | None:
        playbooks = cls._all()
        return playbooks[0] if playbooks else None

    @classmethod
    def get_context_playbooks(cls) -> List[PlaybookItem]:
        """All playbooks after the first — injected as reference context."""
        playbooks = cls._all()
        return playbooks[1:] if len(playbooks) > 1 else []

    @classmethod
    def get_system_prompt(cls) -> str:
        playbook = cls.get_main_playbook()
        if not playbook:
            return ""
        goal = (getattr(playbook, "goal", "") or "").strip()
        instructions = (getattr(playbook, "instructions", "") or "").strip()
        if goal and instructions:
            return f"{goal}\n\n{instructions}"
        return goal or instructions