src/agents/planner/prompt.py

"""Builds the planner LLM human-message content.

The system prompt (`config/prompts/planner.md`) carries the role, invariants,
and planning principles. This module assembles the per-call human content:
business context + condensed catalog + available tools + constraints + the
few-shot examples + the question (+ the prior error on retry).

Few-shot examples are rendered from `examples.py` (which builds them from the
real `TaskList` schema) so they cannot drift from the output contract.
"""

from __future__ import annotations

from .contracts import BusinessContext, ToolRegistry
from .examples import render_examples
from .inputs import CatalogSummary, Constraints


def render_business_context(context: BusinessContext) -> str:
    lines = [
        f"Project: {context.project_id} (industry: {context.industry}, "
        f"context completeness: {context.completeness})",
        f"Business: {context.business_description}",
        f"Scale & scope: {context.scale_and_scope}",
    ]
    if context.key_terms:
        lines.append("Key terms:")
        lines.extend(f"  - {kt.term}: {kt.meaning}" for kt in context.key_terms)
    if context.data_overview:
        lines.append("Data overview:")
        lines.extend(
            f"  - {n.table_name}: {n.row_represents}" for n in context.data_overview
        )
    if context.data_column_notes:
        lines.append("Column notes:")
        lines.extend(
            f"  - {n.column_name}: {n.meaning}" for n in context.data_column_notes
        )
    if context.whats_normal:
        lines.append(f"What's normal: {context.whats_normal}")
    if context.recent_events:
        lines.append(f"Recent events: {context.recent_events}")
    if context.things_to_watch_for:
        lines.append(f"Things to watch for: {context.things_to_watch_for}")
    if context.open_questions:
        lines.append("Known open questions:")
        lines.extend(f"  - {q}" for q in context.open_questions)
    return "\n".join(lines)


def render_registry(tools: ToolRegistry) -> str:
    if not tools.tools:
        return "(no tools available)"
    blocks: list[str] = []
    for spec in tools.tools:
        # Render EVERY allowed arg (from input_schema.properties), flagging which
        # are required. The validator (check #8) rejects any arg outside this set,
        # so the planner must see the full set — not just the required subset — or
        # it invents args like `limit`/`order_by` on a tool that doesn't take them
        # and fails validation.
        props = spec.input_schema.get("properties", {})
        required = set(spec.input_schema.get("required", []))
        args = (
            ", ".join(
                f"{name}{' (required)' if name in required else ''}" for name in props
            )
            or "(none)"
        )
        blocks.append(
            f"- {spec.name} (category: {spec.category}, returns: {spec.output_kind})\n"
            f"  args: {args}\n"
            f"  {spec.description}"
        )
    return "\n".join(blocks)


def render_constraints(constraints: Constraints) -> str:
    lines = [
        f"- max_tasks: {constraints.max_tasks}",
        f"- modeling_allowed: {constraints.modeling_allowed} "
        "(no modeling tools exist in v1 — do not emit modeling tasks)",
        f"- row_budget: {constraints.row_budget}",
    ]
    if constraints.token_budget is not None:
        lines.append(f"- token_budget: {constraints.token_budget}")
    if constraints.time_budget_seconds is not None:
        lines.append(f"- time_budget_seconds: {constraints.time_budget_seconds}")
    return "\n".join(lines)


def build_planner_prompt(
    context: BusinessContext,
    catalog: CatalogSummary,
    tools: ToolRegistry,
    query: str,
    constraints: Constraints,
    previous_errors: list[str] | None = None,
    reply_language: str | None = None,
) -> str:
    """Return the human-message content for the planner LLM.

    The system prompt (`config/prompts/planner.md`) is loaded separately by
    `PlannerService`. `previous_errors` is the full history of prior validation
    failures (oldest first) so a retry fixes ALL of them at once instead of fixing
    one and reintroducing an earlier one. `reply_language` (the pipeline's detected
    user language) is surfaced so the only user-facing free text the planner emits
    — `infeasible_reason` — comes back in that language instead of always English.
    """
    sections = [
        f"# Business context\n\n{render_business_context(context)}",
        f"# Catalog\n\n{catalog.render()}",
        f"# Available tools\n\n{render_registry(tools)}",
        f"# Constraints\n\n{render_constraints(constraints)}",
        f"# Examples\n\n{render_examples()}",
        f"# Question\n\n{query}",
    ]
    if reply_language:
        sections.append(
            f"# Reply language\n\n"
            f"The user writes in {reply_language}. If (and only if) you return an "
            f"infeasible plan, write `infeasible_reason` in {reply_language} — it is "
            f"shown to the user verbatim. The plan structure itself is unaffected."
        )
    if previous_errors:
        joined = "\n".join(f"- {err}" for err in previous_errors)
        sections.append(
            "# Previous attempts failed validation\n\n"
            "Fix ALL of these. Do NOT reintroduce an earlier error while fixing a "
            "newer one:\n"
            f"{joined}\n\n"
            "Re-read the tools' `args` lists and the QueryIR filter rules above, "
            "then emit a corrected TaskList."
        )
    return "\n\n".join(sections)