Gitea-Tools/webui/prompt_library.py

"""Canonical workflow prompt library for the internal web UI (#428)."""

from __future__ import annotations

import hashlib
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

_WORKFLOW_ROOT = Path("skills/llm-project-workflow/workflows")

_DEFAULT_PROMPT_RE = re.compile(
    r"\*\*Default task prompt:\*\*\s*\n+>\s*(.+?)(?=\n\n|\nDo not improvise)",
    re.DOTALL,
)
_FRONTMATTER_TASK_MODE_RE = re.compile(r"^task_mode:\s*(\S+)", re.MULTILINE)


@dataclass(frozen=True)
class PromptEntry:
    slug: str
    label: str
    prompt_text: str
    workflow_path: str
    task_mode: str | None
    workflow_hash: str | None
    source_note: str


def _repo_root() -> Path:
    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
    if override:
        return Path(override).resolve()
    return Path(__file__).resolve().parent.parent


def _workflow_file(path: str) -> Path:
    return _repo_root() / path


def _sha256_hex(content: str) -> str:
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def _read_workflow(path: str) -> tuple[str, str]:
    file_path = _workflow_file(path)
    text = file_path.read_text(encoding="utf-8")
    return text, _sha256_hex(text)


def _extract_default_prompt(markdown: str) -> str | None:
    match = _DEFAULT_PROMPT_RE.search(markdown)
    if not match:
        return None
    lines = [line.strip() for line in match.group(1).splitlines()]
    return " ".join(line for line in lines if line)


def _extract_task_mode(markdown: str) -> str | None:
    match = _FRONTMATTER_TASK_MODE_RE.search(markdown)
    return match.group(1) if match else None


def _entry_from_workflow(
    *,
    slug: str,
    label: str,
    workflow_path: str,
    prompt_override: str | None = None,
    source_note: str = "",
) -> PromptEntry:
    markdown, digest = _read_workflow(workflow_path)
    prompt_text = prompt_override or _extract_default_prompt(markdown)
    if not prompt_text:
        raise ValueError(f"No default task prompt found in {workflow_path}")
    return PromptEntry(
        slug=slug,
        label=label,
        prompt_text=prompt_text,
        workflow_path=workflow_path,
        task_mode=_extract_task_mode(markdown),
        workflow_hash=digest,
        source_note=source_note,
    )


def _static_entry(
    *,
    slug: str,
    label: str,
    prompt_text: str,
    workflow_path: str,
    source_note: str,
) -> PromptEntry:
    path = _workflow_file(workflow_path)
    digest = _sha256_hex(path.read_text(encoding="utf-8")) if path.is_file() else None
    markdown = path.read_text(encoding="utf-8") if path.is_file() else ""
    return PromptEntry(
        slug=slug,
        label=label,
        prompt_text=prompt_text,
        workflow_path=workflow_path,
        task_mode=_extract_task_mode(markdown) if markdown else None,
        workflow_hash=digest,
        source_note=source_note,
    )


def load_prompt_library() -> tuple[PromptEntry, ...]:
    """Load operator prompts derived from canonical workflows."""
    entries = (
        _entry_from_workflow(
            slug="review-pr",
            label="Review PR",
            workflow_path=str(_WORKFLOW_ROOT / "review-merge-pr.md"),
        ),
        _entry_from_workflow(
            slug="work-issue",
            label="Work issue",
            workflow_path=str(_WORKFLOW_ROOT / "work-issue.md"),
        ),
        _entry_from_workflow(
            slug="create-issue",
            label="Create issue",
            workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
        ),
        _static_entry(
            slug="comment-issue",
            label="Comment on issue",
            workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
            prompt_text=(
                "Comment on the target Gitea issue only if exact comment_issue "
                "capability is proven. Load the canonical create-issue workflow "
                "first and follow §16 (comment-on-existing issue rule). Include "
                "specific evidence; do not duplicate existing comments."
            ),
            source_note="Derived from create-issue.md §16; full policy remains in the workflow file.",
        ),
        _static_entry(
            slug="cleanup",
            label="Post-merge cleanup",
            workflow_path="skills/llm-project-workflow/templates/worktree-cleanup.md",
            prompt_text=(
                "Task: clean up branch/worktree for PR #<pr> / issue #<n> after merge. "
                "Confirm the merge on remote master before any deletion; never "
                "force-remove a dirty worktree."
            ),
            source_note="Full cleanup steps live in templates/worktree-cleanup.md.",
        ),
        _entry_from_workflow(
            slug="audit",
            label="Reconciliation audit",
            workflow_path=str(_WORKFLOW_ROOT / "reconcile-landed-pr.md"),
        ),
        _static_entry(
            slug="onboarding",
            label="Project onboarding",
            workflow_path="skills/llm-project-workflow/SKILL.md",
            prompt_text=(
                "Onboard this repository into the MCP Control Plane: prove identity "
                "and task capability, configure author/reviewer/reconciler profiles "
                "in separate namespaces, then complete the checklist at /projects. "
                "Canonical router: skills/llm-project-workflow/SKILL.md."
            ),
            source_note="Checklist details live in webui/data/projects.registry.json and /projects.",
        ),
    )
    return entries


def find_prompt(slug: str) -> PromptEntry | None:
    for entry in load_prompt_library():
        if entry.slug == slug:
            return entry
    return None


def prompt_to_dict(entry: PromptEntry) -> dict[str, Any]:
    return {
        "slug": entry.slug,
        "label": entry.label,
        "prompt_text": entry.prompt_text,
        "workflow_path": entry.workflow_path,
        "task_mode": entry.task_mode,
        "workflow_hash": entry.workflow_hash,
        "source_note": entry.source_note,
    }


def library_to_dict() -> dict[str, Any]:
    entries = load_prompt_library()
    return {
        "count": len(entries),
        "prompts": [prompt_to_dict(entry) for entry in entries],
    }