prompt.py

"""Centralized prompt builders for all LLM calls."""

from __future__ import annotations

from typing import Any, Iterable, Mapping


def mission_planner_system_prompt() -> str:
    return (
        "You are an expert mission-aware perception planner for an intelligent sensing system. "
        "Your task is to translate a high-level mission directive and contextual cues into a "
        "prioritized set of semantic entities, objects, or events that the perception stack "
        "should sense.\n\n"

        "Think in terms of mission-relevant semantics rather than detector-specific classes. "
        "Objects may represent physical entities, behaviors, configurations, or latent threats. "
        "Do not constrain recommendations to any fixed taxonomy.\n\n"

        "For each recommendation, reason causally about how sensing this entity reduces mission "
        "uncertainty, enables downstream decisions, or serves as an early indicator of mission-relevant events. "
        "Favor entities that are information-dense and mission-critical.\n\n"

        "If the mission directive is abstract (e.g., tracking, monitoring, movement),"
        "you must decompose it into concrete, observable semantic entities or events"
        "that a perception system could plausibly sense.\n\n"

        "Ground every recommendation in observable perceptual evidence (visual, spatial, or temporal), "
        "but do not assume a specific detector architecture. "
        "Avoid hallucinations beyond the provided mission and cues."
    )


def mission_planner_user_prompt(
    mission: str,
    top_k: int,
    *,
    context: Mapping[str, Any] | None = None,
    cues: Mapping[str, Any] | None = None,
    pipeline_candidates: Iterable[str] | None = None,
    coco_catalog: str | None = None,
) -> str:
    mission_text = mission.strip() or "N/A"
    context_blob = _format_context_blob(context)
    cues_blob = _format_cues_blob(cues)
    candidate_text = (
        ", ".join(sorted({candidate.strip() for candidate in pipeline_candidates if candidate.strip()}))
        if pipeline_candidates
        else "Preselected by mission context."
    )
    coco_text = coco_catalog.strip() if coco_catalog else "COCO classes unavailable."

    return (
        f"Mission directive:\n{mission_text}\n\n"
        f"Structured context inputs:\n{context_blob}\n\n"
        f"Derived location/mission cues:\n{cues_blob}\n\n"
        f"Selectable pipeline IDs: {candidate_text}\n\n"
        "Downstream detector recognizes ONLY these COCO objects (use exact spelling):\n"
        f"{coco_text}\n\n"

        "Instructions:\n"
        "- Interpret the mission directive and contextual cues to determine what **semantic entities, "
        "objects, or events** are most critical for mission success.\n"

        "- If the mission is abstract (e.g., tracking, monitoring, movement, surveillance), you MUST "
        "decompose it into **concrete, observable entities or events** that a perception system could "
        "plausibly sense. Do not leave the mission at an abstract level.\n"

        "- Infer mission_type, location_type, time_of_day, and priority_level "
        "(bounded to allowed enums). Do NOT modify intel_flag, sensor_type, or compute_mode.\n"

        "- If multiple pipeline IDs are listed, select exactly ONE id from that list. "
        "If only one ID is given, respect it.\n"

        "- Never invent new pipelines or modify existing pipeline definitions.\n"

        "- Recommendations are NOT limited to static objects. They may include:\n"
        "  • dynamic entities (e.g., vehicles, aircraft, people)\n"
        "  • behaviors or motion patterns\n"
        "  • spatial configurations or interactions\n"
        "  • anomalous or unexpected activity\n"

        "- Each recommendation MUST correspond to a **specific, real-world, observable phenomenon**.\n"
        "- Do NOT use placeholder names such as 'Objective', 'Target', 'Entity', or generic labels.\n"
        "- Every entity name MUST be an exact string match to one of the provided COCO classes above. "
        "If the mission demands a concept outside this list, decompose it into observable COCO objects.\n"

        "- Provide at most "
        f"{top_k} "
        "recommendations, ordered by priority and expected information gain for the mission.\n"

        "- Scores MUST be floats in [0, 1] and represent **relative mission importance**, "
        "NOT detection confidence or likelihood.\n"

        "- For EACH recommendation, you MUST explain:\n"
        "  • why it matters to the mission\n"
        "  • what observable perceptual cues (visual, spatial, or temporal) would indicate its presence\n"
        "  • how sensing it informs downstream decision-making or reduces mission uncertainty\n\n"

        "Return JSON with the following schema ONLY (no extra text):\n"
        "{\n"
        '  "mission": "<possibly refined mission summary>",\n'
        '  "context": {\n'
        '      "mission_type": "<one of surveillance|tracking|threat_detection|safety_monitoring>",\n'
        '      "location_type": "<one of urban|suburban|rural|industrial|coastal|harbor|bridge|roadway|indoor|unknown>",\n'
        '      "time_of_day": "<day|night|null>",\n'
        '      "priority_level": "<routine|elevated|high|null>"\n'
        "  },\n"
        '  "pipeline": {"id": "<pipeline_id_from_list>", "reason": "<concise justification>"},\n'
        '  "entities": [\n'
        "    {\n"
        '      "name": "<semantic entity or event>",\n'
        '      "score": <float>,\n'
        '      "semantic_role": "<primary_target|threat_proxy|contextual_indicator|operational_constraint>",\n'
        '      "perceptual_cues": "<observable visual/spatial/temporal evidence>",\n'
        '      "rationale": "<mission-grounded justification>"\n'
        "    }\n"
        "  ]\n"
        "}"
    )



def _format_context_blob(context: Mapping[str, Any] | None) -> str:
    if not context:
        return "No additional context provided."

    lines = []
    mission_type = context.get("mission_type")
    location_type = context.get("location_type")
    time_of_day = context.get("time_of_day")

    if mission_type:
        lines.append(f"- Mission type: {mission_type}")
    if location_type:
        lines.append(f"- Location type: {location_type}")
    if time_of_day:
        lines.append(f"- Time of day: {time_of_day}")

    if not lines:
        return "Context provided but no structured cues supplied."
    return "\n".join(lines)


def _format_cues_blob(cues: Mapping[str, Any] | None) -> str:
    if not cues:
        return "No derived cues."
    lines = []
    for key, value in cues.items():
        if value is None:
            continue
        if isinstance(value, (list, tuple)):
            serialized = ", ".join(str(item) for item in value if item)
        else:
            serialized = str(value)
        if not serialized:
            continue
        lines.append(f"- {key.replace('_', ' ').capitalize()}: {serialized}")
    if not lines:
        return "Derived cues provided but empty."
    return "\n".join(lines)


def mission_summarizer_system_prompt() -> str:
    return (
        "You are a surveillance analyst producing brief situation reports. "
        "Review the provided mission context and detections, then respond with a concise summary "
        "(at most three short sentences) covering key findings only. "
        "If no detections appear, clearly state that fact. Avoid extra commentary or formatting."
    )


def mission_summarizer_user_prompt(payload_json: str) -> str:
    return (
        "Summarize this mission outcome succinctly (<=3 sentences, no bullet points):\n"
        f"{payload_json}"
    )