"""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