server/tools.py

"""
Launch Desk — Function Tools
=============================
Four @function_tool decorated tools for the Launch Planner agent.
Each does simple but realistic logic without external API calls.
"""

from __future__ import annotations

import re
from agents import function_tool


# ---------------------------------------------------------------------------
# 1. extract_tasks
# ---------------------------------------------------------------------------

@function_tool
def extract_tasks(brief: str) -> list[dict]:
    """Parse a product brief and extract a list of launch tasks.

    Each task includes a name, priority (high/medium/low), owner role,
    and a relative deadline expressed as days before launch.

    Args:
        brief: Free-form product brief text.

    Returns:
        A list of task dicts: {task, priority, owner, deadline}
    """
    if not brief or not brief.strip():
        return [
            {"task": "Define launch goals and metrics", "priority": "high",
             "owner": "Product", "deadline": "T-30 days"},
        ]

    text = brief.lower()

    # Heuristic keyword → task mapping. We scan the brief for domain keywords
    # and build a prioritised task list from whatever matches.
    task_templates: list[dict] = [
        {"keywords": ["api", "backend", "server", "database", "infra"],
         "task": "Freeze code and tag release candidate",
         "priority": "high", "owner": "Engineering", "deadline": "T-14 days"},
        {"keywords": ["qa", "test", "bug", "regression", "testing"],
         "task": "Complete QA regression and smoke testing",
         "priority": "high", "owner": "QA", "deadline": "T-7 days"},
        {"keywords": ["doc", "documentation", "guide", "readme", "manual"],
         "task": "Write user-facing documentation and API guides",
         "priority": "medium", "owner": "Technical Writing", "deadline": "T-10 days"},
        {"keywords": ["market", "marketing", "campaign", "ad", "social", "blog", "press"],
         "task": "Prepare marketing campaign assets and schedule posts",
         "priority": "high", "owner": "Marketing", "deadline": "T-5 days"},
        {"keywords": ["support", "customer", "help", "ticket", "oncall"],
         "task": "Staff support team and prepare runbooks",
         "priority": "medium", "owner": "Customer Success", "deadline": "T-3 days"},
        {"keywords": ["security", "compliance", "audit", "gdpr", "soc"],
         "task": "Complete security review and compliance sign-off",
         "priority": "high", "owner": "Security", "deadline": "T-21 days"},
        {"keywords": ["pricing", "billing", "plan", "tier", "subscription"],
         "task": "Finalise pricing tiers and billing configuration",
         "priority": "medium", "owner": "Finance", "deadline": "T-12 days"},
        {"keywords": ["analytics", "tracking", "metric", "dashboard", "kpi"],
         "task": "Set up analytics dashboards and success metrics",
         "priority": "low", "owner": "Data", "deadline": "T-8 days"},
        {"keywords": ["design", "ui", "ux", "figma", "mockup", "brand"],
         "task": "Finalise UI polish and brand assets",
         "priority": "medium", "owner": "Design", "deadline": "T-10 days"},
        {"keywords": ["legal", "contract", "terms", "privacy", "policy"],
         "task": "Review legal terms and privacy policy updates",
         "priority": "medium", "owner": "Legal", "deadline": "T-15 days"},
    ]

    tasks: list[dict] = []
    for tmpl in task_templates:
        if any(kw in text for kw in tmpl["keywords"]):
            tasks.append({
                "task": tmpl["task"],
                "priority": tmpl["priority"],
                "owner": tmpl["owner"],
                "deadline": tmpl["deadline"],
            })

    # Always include goal definition as the first task
    always = {
        "task": "Define launch goals and success metrics",
        "priority": "high",
        "owner": "Product",
        "deadline": "T-30 days",
    }
    if not any(t["task"] == always["task"] for t in tasks):
        tasks.insert(0, always)

    # Always include a launch-day verification task
    launch_day = {
        "task": "Launch-day verification and monitoring",
        "priority": "high",
        "owner": "Engineering",
        "deadline": "Launch day (T-0)",
    }
    if not any(t["task"] == launch_day["task"] for t in tasks):
        tasks.append(launch_day)

    return tasks


# ---------------------------------------------------------------------------
# 2. check_readiness
# ---------------------------------------------------------------------------

@function_tool
def check_readiness(tasks: str, launch_date: str) -> dict:
    """Check launch readiness against a five-point rubric.

    The rubric covers: code frozen, docs written, QA passed, marketing
    ready, and support staffed. We look for evidence of each in the
    provided task list text.

    Args:
        tasks: The extracted task list (as a string or JSON).
        launch_date: Target launch date in YYYY-MM-DD format.

    Returns:
        {ready: bool, score: int (0-100), gaps: list[str]}
    """
    text = (tasks or "").lower()

    rubric = [
        ("code frozen", ["code", "freeze", "tag", "release candidate", "engineering"]),
        ("docs written", ["doc", "documentation", "guide", "readme", "writing"]),
        ("QA passed", ["qa", "test", "regression", "smoke", "bug"]),
        ("marketing ready", ["market", "campaign", "social", "ad", "press", "blog"]),
        ("support staffed", ["support", "customer", "runbook", "oncall", "help"]),
    ]

    covered: list[str] = []
    gaps: list[str] = []

    for label, keywords in rubric:
        if any(kw in text for kw in keywords):
            covered.append(label)
        else:
            gaps.append(label)

    score = int((len(covered) / len(rubric)) * 100)
    ready = len(gaps) == 0 and score >= 100

    # If launch date is missing we flag it as a gap too
    if not launch_date or not launch_date.strip():
        gaps.append("launch date not set")
        score = max(0, score - 10)

    return {
        "ready": ready,
        "score": score,
        "gaps": gaps,
        "covered": covered,
    }


# ---------------------------------------------------------------------------
# 3. generate_checklist
# ---------------------------------------------------------------------------

@function_tool
def generate_checklist(tasks: str) -> list[dict]:
    """Generate per-owner checklists from a task list.

    Parses the tasks string and groups items by owner, producing a
    checklist for each owner with concrete action items.

    Args:
        tasks: The extracted task list (string or JSON).

    Returns:
        A list of {owner, items: [str]} dicts.
    """
    # Try to parse as JSON list of dicts; fall back to line parsing
    import json

    task_items: list[dict] = []
    stripped = (tasks or "").strip()

    if stripped:
        try:
            parsed = json.loads(stripped)
            if isinstance(parsed, list):
                task_items = parsed
        except (json.JSONDecodeError, TypeError):
            # Fall back: split by newline, treat each line as a task
            for line in stripped.splitlines():
                line = line.strip().strip("-").strip()
                if line:
                    task_items.append({"task": line, "owner": "General"})

    # If nothing parsed, provide a default checklist
    if not task_items:
        task_items = [
            {"task": "Define launch goals and metrics", "owner": "Product"},
            {"task": "Freeze code and tag release candidate", "owner": "Engineering"},
            {"task": "Complete QA regression testing", "owner": "QA"},
            {"task": "Write documentation", "owner": "Technical Writing"},
            {"task": "Prepare marketing campaign", "owner": "Marketing"},
            {"task": "Staff support team", "owner": "Customer Success"},
        ]

    # Group tasks by owner
    by_owner: dict[str, list[str]] = {}
    for t in task_items:
        owner = t.get("owner", "General")
        task_name = t.get("task", str(t))
        by_owner.setdefault(owner, []).append(task_name)

    # Build owner checklists with concrete action items
    checklists: list[dict] = []
    for owner, task_names in by_owner.items():
        items: list[str] = []
        for tn in task_names:
            items.append(f"Confirm: {tn}")
            items.append(f"Sign off: {tn}")
        # Add a shared pre-launch check for every owner
        items.append("Review final launch checklist with launch lead")
        checklists.append({"owner": owner, "items": items})

    # Ensure there is always a Launch Lead checklist
    if not any(c["owner"] == "Launch Lead" for c in checklists):
        checklists.append({
            "owner": "Launch Lead",
            "items": [
                "Confirm all owners have signed off",
                "Verify go/no-go criteria are met",
                "Schedule launch-day standup",
                "Monitor dashboards during launch window",
            ],
        })

    return checklists


# ---------------------------------------------------------------------------
# 4. draft_copy
# ---------------------------------------------------------------------------

@function_tool
def draft_copy(channel: str, product: str, audience: str) -> dict:
    """Draft marketing copy for a specific channel.

    Generates a headline, body, and call-to-action tailored to the
    channel, product, and audience.

    Args:
        channel: Marketing channel (e.g. "email", "twitter", "blog",
                 "linkedin", "landing_page").
        product: Product name or short description.
        audience: Target audience description.

    Returns:
        {headline: str, body: str, cta: str}
    """
    ch = (channel or "").lower().strip()
    prod = (product or "").strip() or "our new product"
    aud = (audience or "").strip() or "your team"

    # Channel-specific copy templates
    if ch in ("twitter", "x", "tweet"):
        headline = f"🚀 {prod} is live — built for {aud}."
        body = (
            f"We just shipped {prod}, designed from the ground up for {aud}. "
            "Faster, simpler, and ready for production today. "
            "Check it out 👇"
        )
        cta = "Try it now →"

    elif ch in ("email", "newsletter", "drip"):
        headline = f"Introducing {prod} — now available for {aud}"
        body = (
            f"Hi there,\n\n"
            f"We're excited to announce {prod}, purpose-built for {aud}. "
            "This release brings a refined experience, improved performance, "
            "and the features you've been asking for.\n\n"
            "Here's what's new:\n"
            "  • Streamlined workflow that saves you time\n"
            "  • Built-in reliability for production use\n"
            "  • Thoughtful design choices throughout\n\n"
            "We'd love your feedback as you try it out."
        )
        cta = "Get started today"

    elif ch in ("blog", "blog post", "announcement"):
        headline = f"{prod}: A new standard for {aud}"
        body = (
            f"Today we're launching {prod}, our most refined release yet. "
            f"Built specifically for {aud}, it combines the features you need "
            "with the reliability you expect.\n\n"
            "In this post we'll walk through what's new, why we built it, "
            "and how to get started in under five minutes."
        )
        cta = "Read the full announcement"

    elif ch in ("linkedin", "li"):
        headline = f"We just launched {prod} — here's why it matters for {aud}"
        body = (
            f"After months of iteration, {prod} is now generally available. "
            f"We built it for {aud} who need a solution that's both powerful "
            "and dependable. Here's what went into it and what it means for you."
        )
        cta = "Learn more"

    elif ch in ("landing", "landing page", "landing_page", "homepage"):
        headline = f"{prod} — purpose-built for {aud}"
        body = (
            f"{prod} gives {aud} the tools they need to move faster, "
            "with the reliability required for real production workloads. "
            "No noise, no bloat — just what matters."
        )
        cta = "Start building for free"

    else:
        # Generic fallback for unknown channels
        headline = f"Introducing {prod}"
        body = (
            f"{prod} is now available. Designed for {aud}, it delivers "
            "the features and reliability you need to succeed."
        )
        cta = "Learn more"

    return {
        "headline": headline,
        "body": body,
        "cta": cta,
        "channel": ch or "general",
    }