Kairo-Jinkendo/backend/prompt_rendering.py

"""Single-mode prompt rendering without LLM execution."""

from __future__ import annotations

import json
import re
from typing import Any

from db import get_connection
from prompt_registry import load_active_prompt_version, load_prompt_definition, load_template_steps
from prompt_validation import PLACEHOLDER_PATTERN, PlaceholderValidationError, validate_placeholder_input
from services.audit import log_audit

PLACEHOLDER_SUB = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}")


class PromptRenderError(Exception):
    def __init__(self, message: str, *, code: str = "render_error"):
        super().__init__(message)
        self.code = code


def _render_template(template: str, values: dict[str, Any]) -> str:
    def replacer(match: re.Match[str]) -> str:
        key = match.group(1)
        if key not in values:
            return match.group(0)
        value = values[key]
        if isinstance(value, (dict, list)):
            return json.dumps(value, ensure_ascii=False)
        return str(value)

    return PLACEHOLDER_SUB.sub(replacer, template)


def _resolve_template_body(version: dict) -> tuple[str, str]:
    """Return (template_body, source) where source is 'body' or 'step'."""
    steps = load_template_steps(version["id"])
    template_steps = [s for s in steps if s["step_type"] == "template" and s["template_body"]]
    if template_steps:
        step = template_steps[0]
        return step["template_body"], f"step:{step['step_key']}"
    if version.get("body"):
        return version["body"], "body"
    raise PromptRenderError("No template body or template step found", code="no_template")


def _write_execution_log(
    *,
    definition: dict,
    version: dict,
    status: str,
    rendered_preview: str | None,
    input_summary: dict | None,
    error_message: str | None,
    user_id: str | None,
) -> None:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO prompt_execution_logs (
                    prompt_definition_id, prompt_version_id, execution_mode,
                    status, rendered_preview, input_summary, error_message, created_by
                )
                VALUES (%s::uuid, %s::uuid, %s, %s, %s, %s::jsonb, %s, %s::uuid)
                """,
                (
                    definition["id"],
                    version["id"],
                    definition["execution_mode"],
                    status,
                    rendered_preview,
                    json.dumps(input_summary) if input_summary else None,
                    error_message,
                    user_id,
                ),
            )
        conn.commit()
    finally:
        conn.close()


def render_prompt_single(
    prompt_key: str,
    *,
    values: dict[str, Any],
    user_id: str | None = None,
    log_execution: bool = True,
) -> dict[str, Any]:
    definition = load_prompt_definition(prompt_key)
    if not definition:
        raise PromptRenderError(f"Unknown prompt: {prompt_key}", code="not_found")
    if definition["status"] != "active":
        raise PromptRenderError(f"Prompt not active: {prompt_key}", code="inactive")

    execution_mode = definition["execution_mode"]
    if execution_mode in ("pipeline", "workflow"):
        raise PromptRenderError(
            f"execution_mode '{execution_mode}' is not implemented in AP0.4",
            code="mode_not_implemented",
        )

    version = load_active_prompt_version(definition)
    if not version:
        raise PromptRenderError("No active prompt version", code="no_version")

    template_body, template_source = _resolve_template_body(version)

    try:
        validation = validate_placeholder_input(
            template=template_body,
            context_kind=definition["context_kind"],
            values=values,
        )
        rendered = _render_template(template_body, values)
        result = {
            "prompt_key": prompt_key,
            "version": version["version"],
            "execution_mode": execution_mode,
            "template_source": template_source,
            "rendered_text": rendered,
            "validation": validation,
        }
        if log_execution:
            _write_execution_log(
                definition=definition,
                version=version,
                status="success",
                rendered_preview=rendered[:2000],
                input_summary={"keys": sorted(values.keys())},
                error_message=None,
                user_id=user_id,
            )
        return result
    except PlaceholderValidationError as exc:
        if log_execution:
            _write_execution_log(
                definition=definition,
                version=version,
                status="error",
                rendered_preview=None,
                input_summary={"keys": sorted(values.keys())},
                error_message=str(exc),
                user_id=user_id,
            )
        log_audit(
            "prompt.render.failed",
            user_id=user_id,
            details={
                "prompt_key": prompt_key,
                "missing": exc.missing,
                "type_errors": exc.type_errors,
                "unknown": exc.unknown,
            },
        )
        raise PromptRenderError(str(exc), code="validation_failed") from exc
    except PromptRenderError:
        raise
    except Exception as exc:
        if log_execution:
            _write_execution_log(
                definition=definition,
                version=version,
                status="error",
                rendered_preview=None,
                input_summary={"keys": sorted(values.keys())},
                error_message=str(exc),
                user_id=user_id,
            )
        raise PromptRenderError(str(exc), code="render_error") from exc