resolutionflow/backend/app/services/resolution_note_generator.py

"""ResolutionNoteGeneratorService — drafts the structured Resolve note for a session.

Produces the four-section markdown that ships to the customer ticket (per
FLOWPILOT-MIGRATION.md Section 6.2):

    ## Problem
    ## What we confirmed
    ## Root cause
    ## Resolution

The output is the *draft* — engineers review and edit in the preview popover
before clicking Confirm & post (Phase 4). Caching is keyed on
`(session_id, ai_sessions.state_version)` per Section 5.5; the cache lives in
`preview_cache` and invalidates automatically when any fact / suggested fix /
script generation bumps the session's state_version.

Model: Sonnet (`resolution_note` action tier — quality matters because the
output is customer-facing). MCP intentionally disabled — this is a summary
of existing state, not a research task.

Sensitive parameter values in script_generations are redacted using the
script template's `parameters_schema` (`field_type: "password"`). Existing
ScriptTemplateEngine.redact_sensitive handles the substitution.
"""
from __future__ import annotations

import logging
from typing import Any
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.ai_provider import get_ai_provider
from app.core.config import settings
from app.models.ai_session import AISession
from app.models.script_template import ScriptGeneration, ScriptTemplate
from app.models.session_fact import SessionFact
from app.models.session_suggested_fix import SessionSuggestedFix
from app.services.preview_cache import preview_cache
from app.services.script_template_engine import ScriptTemplateEngine

logger = logging.getLogger(__name__)


_RESOLUTION_NOTE_SYSTEM_PROMPT = """\
You produce structured resolution notes for an MSP troubleshooting platform. \
The notes are posted as ticket notes in the customer's PSA, so they must read \
like a competent senior engineer summarized the work — not like an AI \
narration. Your output goes in front of paying customers.

Output exactly this markdown structure, no preamble, no closing remarks, no \
extra headings:

## Problem
<one short paragraph stating the issue the engineer worked on, derived from the \
session's intake/title and the incident header. Past tense. No "user reported" \
hedging — state the problem directly.>

## What we confirmed
<bulleted list of facts from the "What we know" section, each one a short line. \
Group similar facts together; do not invent connecting prose. If there are no \
facts, write "Nothing was confirmed." and skip to Root cause.>

## Root cause
<one short paragraph naming the root cause based on the active suggested fix \
and confirmed facts. If the suggested fix is low-confidence (<60%) or absent, \
say "Root cause not definitively isolated." and explain what is suspected based \
on facts.>

## Resolution
<one short paragraph describing the resolution applied. If a script ran during \
the session, mention it (e.g. "Cleared cached credentials via the \
clear-outlook-credentials script."). If no resolution has been performed yet, \
write "Resolution not yet applied — fix proposed: <fix title>." Pull verbatim \
script names and template references when available.>

Strict rules:
- Use ONLY the facts and state I provide. Never invent specifics that are not \
in the input.
- Do not include placeholder text like "TBD", "TODO", or empty bullets.
- Do not include the engineer's name, the AI's name, internal session IDs, or \
the session's chat transcript.
- Markdown headings exactly as shown (## level), no bolding the headings.
- No trailing whitespace, no double-blank lines, no horizontal rules.
"""


class ResolutionNoteGeneratorService:
    """Generates and caches the four-section Resolve note markdown."""

    KIND = "resolution_note"

    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def generate_or_get_cached(
        self, session_id: UUID, *, force: bool = False,
    ) -> dict[str, Any]:
        """Return the preview for the session.

        Reads `(KIND, session_id, state_version)` from the in-process cache;
        on miss, generates fresh markdown and stores under the same key.
        `force=True` bypasses the cache and refreshes the cached entry.

        Returns `{"markdown": str, "target_ticket_ref": str | None,
                   "state_version": int, "from_cache": bool}`.
        """
        session = await self._load_session(session_id)
        cached = preview_cache.get(self.KIND, session.id, session.state_version) if not force else None
        if cached is not None:
            return {**cached, "from_cache": True}

        markdown = await self._render(session)
        target = self._target_ticket_ref(session)
        payload = {
            "markdown": markdown,
            "target_ticket_ref": target,
            "state_version": session.state_version,
        }
        preview_cache.set(self.KIND, session.id, session.state_version, payload)
        return {**payload, "from_cache": False}

    # ── Internals ─────────────────────────────────────────────────────────

    async def _load_session(self, session_id: UUID) -> AISession:
        result = await self.db.execute(
            select(AISession).where(AISession.id == session_id)
        )
        session = result.scalar_one_or_none()
        if session is None:
            raise ValueError(f"Session {session_id} not found")
        return session

    async def _render(self, session: AISession) -> str:
        """Build the prompt input bundle, call the model, return markdown."""
        facts = await self._load_facts(session.id)
        active_fix = await self._load_active_fix(session.id)
        gens = await self._load_redacted_generations(session.id)

        bundle = self._build_input_bundle(session, facts, active_fix, gens)

        model = settings.get_model_for_action("resolution_note")
        provider = get_ai_provider(model=model)

        # Cache the system prompt — identical across every preview call for
        # every session. Per-session bundle is in the user message, uncached.
        system_blocks: list[dict[str, Any]] = [
            {
                "type": "text",
                "text": _RESOLUTION_NOTE_SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},
                # cacheable: identical across every resolution-note preview call
            },
        ]

        try:
            text, _in, _out = await provider.generate_text(
                system_prompt=system_blocks,
                messages=[{"role": "user", "content": bundle}],
                max_tokens=1200,
            )
        except Exception:
            logger.exception("Resolution note generation failed for session %s", session.id)
            raise
        return text.strip()

    async def _load_facts(self, session_id: UUID) -> list[SessionFact]:
        result = await self.db.execute(
            select(SessionFact)
            .where(
                SessionFact.session_id == session_id,
                SessionFact.deleted_at.is_(None),
            )
            .order_by(SessionFact.created_at.asc())
        )
        return list(result.scalars().all())

    async def _load_active_fix(self, session_id: UUID) -> SessionSuggestedFix | None:
        result = await self.db.execute(
            select(SessionSuggestedFix)
            .where(
                SessionSuggestedFix.session_id == session_id,
                SessionSuggestedFix.superseded_at.is_(None),
            )
            .order_by(SessionSuggestedFix.created_at.desc())
        )
        return result.scalars().first()

    async def _load_redacted_generations(
        self, session_id: UUID
    ) -> list[dict[str, Any]]:
        """Pull script_generations for the session, redacting password params.

        Password fields are inferred from the linked template's
        `parameters_schema` (`field_type: "password"`). The existing
        ScriptTemplateEngine.redact_sensitive handles the substitution.
        """
        result = await self.db.execute(
            select(ScriptGeneration)
            .where(ScriptGeneration.ai_session_id == session_id)
            .order_by(ScriptGeneration.created_at.asc())
        )
        gens = list(result.scalars().all())
        if not gens:
            return []

        template_ids = {g.template_id for g in gens}
        tpl_result = await self.db.execute(
            select(ScriptTemplate).where(ScriptTemplate.id.in_(template_ids))
        )
        templates_by_id = {t.id: t for t in tpl_result.scalars().all()}

        engine = ScriptTemplateEngine()
        out: list[dict[str, Any]] = []
        for g in gens:
            tpl = templates_by_id.get(g.template_id)
            sensitive_keys = self._sensitive_keys_from_schema(
                (tpl.parameters_schema if tpl else {}) or {}
            )
            redacted_params = engine.redact_sensitive(g.parameters_used or {}, sensitive_keys)
            out.append({
                "template_name": tpl.name if tpl else "(unknown template)",
                "template_slug": tpl.slug if tpl else None,
                "parameters_used": redacted_params,
                "created_at": g.created_at.isoformat(),
            })
        return out

    @staticmethod
    def _sensitive_keys_from_schema(schema: dict[str, Any]) -> set[str]:
        """Extract password-typed parameter keys from a template's schema.

        The schema shape is `{"parameters": [{"key": "...", "field_type": "password", ...}]}`
        per the existing Script Generator convention. Tolerate both that shape
        and the simpler `{"key": {"field_type": "password"}}` form.
        """
        keys: set[str] = set()
        params = schema.get("parameters") if isinstance(schema, dict) else None
        if isinstance(params, list):
            for p in params:
                if isinstance(p, dict) and p.get("field_type") == "password":
                    k = p.get("key") or p.get("variable_name")
                    if isinstance(k, str):
                        keys.add(k)
        elif isinstance(schema, dict):
            for k, v in schema.items():
                if isinstance(v, dict) and v.get("field_type") == "password":
                    keys.add(k)
        return keys

    @staticmethod
    def _target_ticket_ref(session: AISession) -> str | None:
        """Display ref for the linked PSA ticket, e.g. 'CW #48291'.

        ConnectWise is the only PSA wired today (per the Phase 1 constraint),
        so a CW prefix is reasonable. Other PSAs will need provider-aware
        formatting in Phase 4.
        """
        if not session.psa_ticket_id:
            return None
        return f"CW #{session.psa_ticket_id}"

    @staticmethod
    def _build_input_bundle(
        session: AISession,
        facts: list[SessionFact],
        active_fix: SessionSuggestedFix | None,
        generations: list[dict[str, Any]],
    ) -> str:
        """Compose the structured input the LLM sees for one preview call."""
        lines: list[str] = []
        lines.append("# Session context")
        lines.append(f"Title: {session.title or '(untitled)'}")
        if session.problem_summary:
            lines.append(f"Problem summary: {session.problem_summary}")
        if session.problem_domain:
            lines.append(f"Domain: {session.problem_domain}")
        intake_text = (session.intake_content or {}).get("text") if isinstance(session.intake_content, dict) else None
        if intake_text:
            lines.append(f"Intake message: {intake_text}")
        if session.psa_ticket_id:
            lines.append(f"Linked PSA ticket: CW #{session.psa_ticket_id}")

        lines.append("")
        lines.append("# Confirmed facts (What we know)")
        if not facts:
            lines.append("(none)")
        else:
            for f in facts:
                tag = f.source_type
                summary = f" — {f.source_summary}" if f.source_summary else ""
                lines.append(f"- [{tag}] {f.text}{summary}")

        lines.append("")
        lines.append("# Active suggested fix")
        if active_fix is None:
            lines.append("(no active suggested fix)")
        else:
            lines.append(f"Title: {active_fix.title}")
            lines.append(f"Confidence: {active_fix.confidence_pct}%")
            lines.append(f"Description: {active_fix.description}")
            if active_fix.user_decision:
                lines.append(f"Engineer decision: {active_fix.user_decision}")

        lines.append("")
        lines.append("# Scripts run during the session (passwords redacted)")
        if not generations:
            lines.append("(none)")
        else:
            for g in generations:
                lines.append(f"- {g['template_name']} (slug={g['template_slug']})")
                if g["parameters_used"]:
                    lines.append(f"  parameters: {g['parameters_used']}")

        lines.append("")
        lines.append(
            "Produce the four-section resolution note now. Use only the input above."
        )
        return "\n".join(lines)