FabledGrimoire/src/scribe/services/moment_actions.py

"""Which actions reach which moment: shipped defaults plus each install's own (milestone 458 step 2).

The catalog (`services/moments.py`) says what the moments ARE. This module says
how the work gets there. One call can reach several moments: `kubectl apply`
is a run, a deliver and a reach outside the workspace all at once, and a rule
mounted on any of them should arrive.

AN ACTION is a tool plus an optional `match`:

- `match` empty — every call of that tool (`Edit` → work.change).
- A tool that runs a command (its input carries `command`) — how the command
  starts, tested against each segment of a compound line, so
  `cd app && make ship` reaches what `make ship` reaches. Leading `VAR=value`
  assignments are skipped; a word boundary is required, so `git push` does
  not match `git pushd`.
- Any other tool — `field=value` pairs, comma-separated, all of which must
  hold (`update_task` with `status=done`).

Tool names are compared without any MCP server prefix and without case:
`mcp__plugin_x__update_task` and `update_task` are one tool, because what an
install called its server is not something a mapping should depend on.

WHY THE DEFAULTS ARE CODE AND THE CORRECTIONS ARE ROWS

The defaults cover the actions every install shares — the harness's own tools,
Scribe's own tools, the commonest command shapes. They ship, and improve, with
the product. What no default can know is how one operator works, so an install
ADDS mappings and REMOVES defaults that misfire for it, in-session, through
`map_action` / `unmap_action`. A removal is stored rather than applied to the
defaults so it survives an upgrade that ships the same default again.

The concrete commands below are examples of reaching a moment, which is why
they may name particular tools when the moments themselves may not.
"""
from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from typing import Iterable

from sqlalchemy import select

from scribe.models import async_session
from scribe.models.moment_mapping import MomentMapping
from scribe.services import moments as catalog

logger = logging.getLogger(__name__)

ADD, REMOVE = "add", "remove"
EFFECTS = (ADD, REMOVE)
DEFAULT, INSTALL = "default", "install"

# The input field that makes a tool a command-runner, and the tools known to
# be one. The field drives MATCHING (any tool whose call carries a command is
# matched by prefix); the names drive VALIDATION, so a prefix-style match on a
# tool that takes no command is refused rather than stored to never fire.
COMMAND_FIELD = "command"
COMMAND_TOOLS = frozenset({"bash"})

# The harness's skill loader: loading a procedure reaches `skill.<its name>`,
# derived from the call rather than listed, since the names are the
# procedures' own.
SKILL_TOOL, SKILL_FIELD = "skill", "skill"

# A procedure also reaches the moment it is FOR (step 5): loading the
# reporting procedure is a report as surely as the reply that follows it.
#
# The product's own skills declare theirs in their SKILL.md frontmatter
# (`metadata: moments:`), and the declaration ships HERE as defaults, because
# the server never sees the plugin's files. `test_skill_moments` holds the two
# together. The plugin's name qualifies the skill, as the harness names it.
BUNDLED_SKILL_PLUGIN = "scribe"
BUNDLED_SKILL_MOMENTS: dict[str, tuple[str, ...]] = {
    "using-scribe": ("session.start",),
    "brainstorming": ("work.plan",),
    "writing-plans": ("work.plan",),
    "reusing-code": ("work.change",),
    "systematic-debugging": ("work.debug",),
    "verification": ("work.verify",),
    "shape-accounting": ("work.record",),
    "reporting-back": ("reply.report",),
}

# A stored process arrives as the skill `scribe-proc-<slug>`
# (scribe_sync_processes.sh). Its moments are the process's own field, read
# when it loads — not copied into the stub, which is written once a session
# and would go on firing a moment its process no longer declares.
PROCESS_SKILL_PREFIX = "scribe-proc-"
PROCESS_MOMENTS_FIELD = "moments"
PROCESS = "process"

_SEGMENT_SPLIT = re.compile(r"&&|\|\||[;|\n]")
_ENV_ASSIGN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=\S*\s+")


@dataclass(frozen=True)
class Action:
    tool: str
    match: str
    moment: str


def _defaults() -> tuple[Action, ...]:
    out: list[Action] = []

    def on(moment: str, tool: str, *matches: str) -> None:
        for m in matches or ("",):
            out.append(Action(tool, m, moment))

    # The harness's own tools.
    on("work.change", "Edit")
    on("work.change", "Write")
    on("work.change", "MultiEdit")
    on("work.change", "NotebookEdit")
    on("work.run", "Bash")
    on("work.delegate", "Task")
    on("work.delegate", "Agent")
    on("work.plan", "EnterPlanMode")
    on("work.plan", "ExitPlanMode")
    on("reply.ask", "AskUserQuestion")
    for skill, reached in BUNDLED_SKILL_MOMENTS.items():
        for moment in reached:
            on(moment, "Skill", f"{SKILL_FIELD}={BUNDLED_SKILL_PLUGIN}:{skill}")

    # Scribe's own tools — every install has these, so their moments ship.
    on("work.start", "update_task", "status=in_progress")
    on("work.finish", "update_task", "status=done", "status=cancelled")
    on("work.finish", "update_milestone", "status=done")
    on("work.plan", "start_planning")
    on("work.plan", "create_milestone")
    for tool in ("create_note", "create_lesson", "create_rule",
                 "create_project_rule", "create_preference", "create_snippet",
                 "create_process"):
        on("work.record", tool)

    # The commonest command shapes. An install's own (`make ship`,
    # `./deploy.sh`) are what map_action is for.
    on("work.deliver", "Bash",
       "git push", "git merge", "gh pr merge", "gh release create",
       "docker push", "kubectl apply", "helm install", "helm upgrade",
       "terraform apply", "npm publish", "twine upload", "cargo publish")
    on("work.verify", "Bash",
       "pytest", "python -m pytest", "npm test", "npm run test", "yarn test",
       "pnpm test", "go test", "go vet", "cargo test", "make test",
       "make check", "ruff check", "mypy", "terraform plan",
       "terraform validate")
    on("env.reach", "Bash",
       "ssh", "scp", "curl", "wget", "kubectl", "helm")
    return tuple(out)


DEFAULT_ACTIONS: tuple[Action, ...] = _defaults()


# ── matching ──────────────────────────────────────────────────────────────

def tool_key(name: str) -> str:
    """The tool's bare, lowercased name: no MCP server prefix."""
    clean = (name or "").strip()
    if clean.startswith("mcp__"):
        clean = clean.rsplit("__", 1)[-1]
    return clean.lower()


def _squash(text: str) -> str:
    return " ".join((text or "").split())


def _segments(command: str) -> list[str]:
    out = []
    for part in _SEGMENT_SPLIT.split(command or ""):
        seg = part.strip().lstrip("(").strip()
        while _ENV_ASSIGN.match(seg):
            seg = _ENV_ASSIGN.sub("", seg, count=1)
        seg = _squash(seg)
        if seg:
            out.append(seg)
    return out


def _pairs(match: str) -> dict[str, str] | None:
    """`a=b, c=d` → {"a": "b", "c": "d"}; None when it is not that shape."""
    out: dict[str, str] = {}
    for part in (match or "").split(","):
        key, sep, value = part.partition("=")
        if not sep or not key.strip():
            return None
        out[key.strip().lower()] = value.strip().lower()
    return out or None


def action_matches(action: Action, tool: str, tool_input: dict | None) -> bool:
    if tool_key(action.tool) != tool_key(tool):
        return False
    if not action.match:
        return True
    tool_input = tool_input or {}
    command = tool_input.get(COMMAND_FIELD)
    if isinstance(command, str):
        want = _squash(action.match)
        return any(seg == want or seg.startswith(want + " ")
                   for seg in _segments(command))
    pairs = _pairs(action.match)
    if pairs is None:
        return False
    given = {str(k).lower(): v for k, v in tool_input.items()}
    return all(
        str(given.get(k, "")).strip().lower() == v for k, v in pairs.items()
    )


def effective_actions(mappings: Iterable) -> list[tuple[Action, str]]:
    """The defaults this install has not removed, then its own additions.

    `mappings` are MomentMapping rows, or anything with the same four fields.
    """
    removed: set[tuple[str, str, str]] = set()
    added: list[Action] = []
    for m in mappings:
        key = (tool_key(m.tool), _squash(m.match), m.moment)
        if m.effect == REMOVE:
            removed.add(key)
        elif m.effect == ADD:
            added.append(Action(m.tool, _squash(m.match), m.moment))
    out = [
        (a, DEFAULT) for a in DEFAULT_ACTIONS
        if (tool_key(a.tool), a.match, a.moment) not in removed
    ]
    out.extend((a, INSTALL) for a in added)
    return out


def skill_name(tool: str, tool_input: dict | None) -> str:
    """The procedure a skill-loader call loads, lowercased; "" for any other call."""
    if tool_key(tool) != SKILL_TOOL:
        return ""
    return str((tool_input or {}).get(SKILL_FIELD) or "").strip().lower()


def declared_moments(data: dict | None) -> list[str]:
    """The moments a stored process declares — the valid ones, in its order.

    Read defensively: the field is written through require_moments, but a
    moment the catalog later drops must stop firing, not break the load.
    """
    raw = (data or {}).get(PROCESS_MOMENTS_FIELD) or []
    if isinstance(raw, str):
        raw = [raw]
    return [m for m in dict.fromkeys(str(n).strip().lower() for n in raw)
            if catalog.is_moment(m)]


def resolve(
    tool: str, tool_input: dict | None, mappings: Iterable = (),
    declared: Iterable[str] = (),
) -> list[dict]:
    """Every moment this call reaches, each with the action that reached it.

    One entry per moment, in catalog order. When two actions reach the same
    moment the more specific one — the longer match — is the one named, since
    "reached by `git push`" says more than "reached by Bash".

    `declared` are the moments the procedure this call loads says it is for
    (a stored process's own field); each is reached by the load itself.
    """
    best: dict[str, dict] = {}
    for action, via in effective_actions(mappings):
        if not action_matches(action, tool, tool_input):
            continue
        held = best.get(action.moment)
        if held is None or len(action.match) > len(held["match"]):
            best[action.moment] = {
                "moment": action.moment, "tool": action.tool,
                "match": action.match, "via": via,
            }

    name = skill_name(tool, tool_input)
    if name:
        moment = catalog.SKILL_PREFIX + name
        if catalog.is_moment(moment):
            best[moment] = {"moment": moment, "tool": tool,
                            "match": f"{SKILL_FIELD}={name}", "via": DEFAULT}
        for moment in declared:
            best.setdefault(moment, {"moment": moment, "tool": tool,
                                     "match": f"{SKILL_FIELD}={name}", "via": PROCESS})

    order = {name: i for i, name in enumerate(catalog.MOMENTS)}
    return sorted(best.values(), key=lambda h: (order.get(h["moment"], len(order)), h["moment"]))


def _sample_input(tool: str, match: str) -> dict:
    """A call that this action would describe — for reporting what it reaches."""
    if not match:
        return {}
    if tool_key(tool) in COMMAND_TOOLS:
        return {COMMAND_FIELD: match}
    return dict(_pairs(match) or {})


def _clean(tool: str, match: str, moment: str) -> tuple[str, str, str]:
    """Validate a mapping's three parts, or refuse with what would work."""
    bare = (tool or "").strip()
    if bare.startswith("mcp__"):
        bare = bare.rsplit("__", 1)[-1]
    if not bare:
        raise ValueError("tool is required — the tool as the harness names it, e.g. Bash or update_task")
    clean_match = _squash(match)
    if clean_match and tool_key(bare) not in COMMAND_TOOLS and _pairs(clean_match) is None:
        raise ValueError(
            f"{bare} does not run a command, so `match` names its arguments as "
            f"field=value pairs (e.g. status=done), not {clean_match!r}. Leave "
            f"it empty to map every {bare} call."
        )
    return bare, clean_match, catalog.require_moment(moment)


def _is_default(tool: str, match: str, moment: str) -> bool:
    return any(
        tool_key(a.tool) == tool_key(tool) and a.match == match and a.moment == moment
        for a in DEFAULT_ACTIONS
    )


# ── the install's own ─────────────────────────────────────────────────────

async def list_mappings(user_id: int) -> list[MomentMapping]:
    async with async_session() as session:
        rows = await session.execute(
            select(MomentMapping)
            .where(MomentMapping.user_id == user_id)
            .order_by(MomentMapping.id)
        )
        return list(rows.scalars().all())


async def moments_for(user_id: int, tool: str, tool_input: dict | None) -> list[dict]:
    """The moments a call reaches for this user.

    Fails OPEN to the defaults: an unreadable mapping table must cost the
    install its corrections, not every moment.
    """
    try:
        mappings = await list_mappings(user_id)
    except Exception:
        logger.warning("moment mappings unreadable; using the defaults", exc_info=True)
        mappings = []
    name = skill_name(tool, tool_input)
    declared = (await process_moments(user_id, name)
                if name.startswith(PROCESS_SKILL_PREFIX) else [])
    return resolve(tool, tool_input, mappings, declared)


async def process_moments(user_id: int, skill: str) -> list[str]:
    """The moments the stored process behind `scribe-proc-<slug>` declares.

    The slug is resolved through the same manifest the sync script wrote the
    skill from, so the two agree on which process a slug names. Fails open to
    none: a process that cannot be read still loads, it just reaches no more
    than its own `skill.<name>`.
    """
    from scribe.services import notes as notes_svc
    from scribe.services import plugin_context

    slug = skill[len(PROCESS_SKILL_PREFIX):]
    try:
        manifest = await plugin_context.build_process_manifest(user_id)
        entry = next((p for p in manifest.get("processes", []) if p.get("slug") == slug), None)
        if entry is None:
            return []
        loaded = await notes_svc.get_note_for_user(user_id, int(entry["id"]))
        note = loaded[0] if loaded else None
        return declared_moments(note.data if note is not None else None)
    except Exception:
        logger.debug("process moments for %s unreadable", skill, exc_info=True)
        return []


async def _find(session, user_id: int, tool: str, match: str, moment: str):
    rows = await session.execute(
        select(MomentMapping).where(
            MomentMapping.user_id == user_id,
            MomentMapping.moment == moment,
            MomentMapping.match == match,
        )
    )
    return next((r for r in rows.scalars().all() if tool_key(r.tool) == tool_key(tool)), None)


async def _reaches(user_id: int, tool: str, match: str) -> list[dict]:
    return resolve(tool, _sample_input(tool, match), await list_mappings(user_id))


async def map_action(
    user_id: int, tool: str, match: str, moment: str,
    *, reason: str = "", actor: str = "model",
) -> dict:
    """Make an action reach a moment for this install.

    Mapping a shipped default this install had removed restores it; mapping
    one already in force changes nothing and says so.
    """
    tool, match, moment = _clean(tool, match, moment)
    reason = (reason or "").strip()
    async with async_session() as session:
        row = await _find(session, user_id, tool, match, moment)
        if _is_default(tool, match, moment):
            if row is not None and row.effect == REMOVE:
                await session.delete(row)
                await session.commit()
                change = "restored the shipped default"
            else:
                change = "already a shipped default — nothing to add"
        elif row is not None:
            row.reason = reason or row.reason
            row.actor = actor
            await session.commit()
            change = "already mapped — reason updated" if reason else "already mapped"
        else:
            session.add(MomentMapping(
                user_id=user_id, tool=tool, match=match, moment=moment,
                effect=ADD, reason=reason, actor=actor,
            ))
            await session.commit()
            change = "mapped"
    return {
        "change": change, "tool": tool, "match": match, "moment": moment,
        "now_reaches": await _reaches(user_id, tool, match),
    }


async def unmap_action(
    user_id: int, tool: str, match: str, moment: str,
    *, reason: str = "", actor: str = "model",
) -> dict:
    """Stop an action reaching a moment for this install.

    An install's own mapping is deleted. A shipped default is switched off by
    a stored removal, so the next release does not switch it back on.
    """
    tool, match, moment = _clean(tool, match, moment)
    reason = (reason or "").strip()
    async with async_session() as session:
        row = await _find(session, user_id, tool, match, moment)
        if row is not None and row.effect == ADD:
            await session.delete(row)
            await session.commit()
            change = "removed this install's mapping"
        elif _is_default(tool, match, moment):
            if row is None:
                session.add(MomentMapping(
                    user_id=user_id, tool=tool, match=match, moment=moment,
                    effect=REMOVE, reason=reason, actor=actor,
                ))
                await session.commit()
                change = "switched off the shipped default"
            else:
                change = "the shipped default was already off"
        else:
            raise ValueError(
                f"nothing maps {tool}{' ' + repr(match) if match else ''} onto "
                f"{moment}, so there is nothing to remove. list_moments shows "
                f"which actions reach each moment."
            )
    return {
        "change": change, "tool": tool, "match": match, "moment": moment,
        "now_reaches": await _reaches(user_id, tool, match),
    }


async def actions_by_moment(user_id: int) -> dict:
    """Each moment's actions in force, and the defaults this install removed."""
    mappings = await list_mappings(user_id)
    by_moment: dict[str, list[dict]] = {}
    for action, via in effective_actions(mappings):
        by_moment.setdefault(action.moment, []).append(
            {"tool": action.tool, "match": action.match, "via": via}
        )
    removed = [m.to_dict() for m in mappings if m.effect == REMOVE]
    return {"actions": by_moment, "removed_defaults": removed}