script-build-and-framework-tmp/visualization/backend/app/prompt_projection.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from .sanitizer import sanitize


_MAX_PROMPT_CHARS = 60_000


def project_prompt(payload: dict[str, Any]) -> dict[str, Any]:
    """Build a truthful, lazy prompt view.

    Runtime events preserve the exact delegated task, but the application does
    not persist a system-prompt snapshot for each invocation.  The current
    prompt is therefore deliberately labelled as current configuration.
    """

    task = _text(payload.get("task"))
    prompt = _text(payload.get("promptContent"))
    truncated = bool(prompt and len(prompt) > _MAX_PROMPT_CHARS)
    if truncated:
        prompt = prompt[:_MAX_PROMPT_CHARS].rstrip() + "\n\n…（当前规则过长，已截断）"

    result: dict[str, Any] = {
        "promptRef": str(payload.get("promptRef") or ""),
        "actor": payload.get("actor") or {"role": "main", "label": "主 Agent"},
        "notices": [
            {
                "code": "PROMPT_NOT_SNAPSHOTTED",
                "message": "系统没有保存这次运行所用提示词的历史快照；下方展示当前提示词。",
            }
        ],
    }
    if mode_notice := _text(payload.get("modeNotice")):
        result["notices"].append(
            {"code": "CURRENT_MODE_CONTEXT", "message": mode_notice}
        )
    if task:
        result["exactTask"] = {
            "content": sanitize(task, max_text=_MAX_PROMPT_CHARS),
            "accuracy": "run-exact",
        }
    if prompt:
        result["systemPrompt"] = {
            "content": sanitize(prompt, max_text=_MAX_PROMPT_CHARS + 200),
            "source": payload.get("promptSource") or "current-file",
            "version": payload.get("promptVersion"),
            "accuracy": "current-not-run-snapshot",
            "capturedAt": datetime.now(timezone.utc).isoformat(),
            "truncated": truncated,
        }
    else:
        result["notices"].append(
            {
                "code": "PROMPT_UNAVAILABLE",
                "message": "当前提示词暂时不可读取。",
            }
        )
    if truncated:
        result["notices"].append(
            {"code": "PROMPT_TRUNCATED", "message": "当前提示词已按安全长度截断。"}
        )
    return sanitize(result, max_text=_MAX_PROMPT_CHARS + 400)


def _text(value: Any) -> str | None:
    if not isinstance(value, str):
        return None
    value = value.strip()
    return value or None