production-proof-v116-high-utili-7255-3/agent.py

"""Production Proof v116: high-utility sales operations dashboard.

A deterministic full-stack A2A product: platform auth, managed Postgres,
MCP-callable typed tools, downloadable artifacts, and durable execution receipts.
"""
from __future__ import annotations

import csv
import hashlib
import io
import json
import os
import re
from contextlib import contextmanager
from datetime import UTC, datetime
from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator

import a2a_pack as a2a
from a2a_pack import (
    A2AAgent,
    AccountAccess,
    AgentDatabase,
    AgentDatabaseEnv,
    AgentDatabaseMigrations,
    AgentPlatformResources,
    PlatformUserAuth,
    Pricing,
    Resources,
    RunContext,
    State,
    WorkspaceAccess,
    WorkspaceMode,
)

DB_NAME = "production-proof-v116-high-utili-7255-3-data"
MAX_UPDATES = 24
MAX_TEXT_CHARS = 6000


class ProductionProofV116HighUtili72553Config(BaseModel):
    default_window_days: int = Field(default=30, ge=7, le=90)


class SalesUpdate(BaseModel):
    account: str = Field(min_length=1, max_length=80)
    owner: str = Field(min_length=1, max_length=80)
    stage: str = Field(min_length=1, max_length=40)
    amount_usd: float = Field(ge=0, le=10_000_000)
    forecast_category: Literal["commit", "best_case", "pipeline", "omitted"] = "pipeline"
    status: Literal["on_track", "at_risk", "blocked", "slipped", "won", "lost"] = "on_track"
    days_in_stage: int = Field(ge=0, le=365)
    update: str = Field(default="", max_length=400)

    @field_validator("account", "owner", "stage", "update")
    @classmethod
    def _clean_text(cls, value: str) -> str:
        return re.sub(r"\s+", " ", value.strip())


class BuildDashboardRequest(BaseModel):
    dashboard_name: str = Field(default="Weekly Sales Ops Pulse", min_length=3, max_length=80)
    segment: str = Field(default="Mid-market", min_length=1, max_length=80)
    period_label: str = Field(default="This week", min_length=1, max_length=60)
    min_exception_amount_usd: float = Field(default=25_000, ge=0, le=10_000_000)
    updates: list[SalesUpdate] = Field(default_factory=list, max_length=MAX_UPDATES)
    notes: str = Field(default="", max_length=MAX_TEXT_CHARS)

    @field_validator("dashboard_name", "segment", "period_label", "notes")
    @classmethod
    def _clean(cls, value: str) -> str:
        return re.sub(r"\s+", " ", value.strip())


class DashboardArtifact(BaseModel):
    name: str
    media_type: str
    uri: str
    size_bytes: int
    content: str


class DashboardResult(BaseModel):
    status: Literal["ok", "setup_required", "error"]
    dashboard_id: str
    receipt_id: str
    dashboard_data: list[dict[str, Any]] = Field(alias="dashboard-data")
    filters: dict[str, Any]
    trends: list[dict[str, Any]]
    exceptions: list[dict[str, Any]]
    history: list[dict[str, Any]]
    summary: str
    artifact: DashboardArtifact | None = None


class HistoryResult(BaseModel):
    status: Literal["ok", "setup_required", "error"]
    history: list[dict[str, Any]]


class DurableDashboardState(BaseModel):
    last_dashboard_id: str | None = None


class ProductionProofV116HighUtili72553(
    A2AAgent[ProductionProofV116HighUtili72553Config, PlatformUserAuth]
):
    name = "production-proof-v116-high-utili-7255-3"
    description = (
        "One-page B2B sales ops dashboard that turns recurring operational "
        "updates into filters, trends, exceptions, durable history, receipts, "
        "downloadable outputs, and MCP-callable tools."
    )
    version = "0.1.0"

    config_model = ProductionProofV116HighUtili72553Config
    auth_model = PlatformUserAuth

    state = State.DURABLE
    state_model = DurableDashboardState
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic dashboard logic. Account trial funds exactly 3 platform skill calls, then BYOK is required by the platform.",
    )
    account_access = AccountAccess(required=True, platform_skill_calls=3, after_trial="byok")
    tools_used = ("postgres", "mcp", "artifacts", "scheduled-snapshot")
    capabilities = {
        "mcp": {"tools": ["build_dashboard", "list_dashboard_history", "run_scheduled_snapshot"]},
        "schedules": {"tool": "run_scheduled_snapshot", "idempotent": True},
        "artifacts": {"tool": "build_dashboard", "media_type": "text/csv"},
    }
    platform_resources = AgentPlatformResources(
        databases=(
            AgentDatabase(
                name=DB_NAME,
                scope="user",
                access_mode="read_write",
                env=AgentDatabaseEnv(url="DATABASE_URL"),
                migrations=AgentDatabaseMigrations(path="db/migrations"),
            ),
        )
    )

    @a2a.tool(
        description="Build and persist a one-page B2B sales operations dashboard with filters, trends, exceptions, history, receipt, and CSV artifact.",
        timeout_seconds=60,
        cost_class="standard",
    )
    async def build_dashboard(
        self,
        ctx: RunContext[PlatformUserAuth],
        request: BuildDashboardRequest,
    ) -> DashboardResult:
        tenant = _tenant_key(ctx)
        now = datetime.now(UTC)
        normalized = _with_default_updates(request)
        analysis = _analyze_updates(normalized)
        dashboard_id = _stable_id("dash", tenant, normalized.model_dump(mode="json"), now.isoformat(timespec="seconds"))
        receipt_id = _stable_id("rcpt", tenant, dashboard_id, ctx.task_id or "task")
        csv_text = _dashboard_csv(analysis["rows"])
        artifact_ref = await ctx.write_artifact(
            "sales-ops-dashboard.csv",
            csv_text.encode("utf-8"),
            "text/csv",
        )
        await ctx.emit_artifact(artifact_ref)

        history: list[dict[str, Any]] = []
        if not os.environ.get("DATABASE_URL"):
            return DashboardResult(
                status="setup_required",
                dashboard_id=dashboard_id,
                receipt_id=receipt_id,
                **{
                    "dashboard-data": analysis["rows"],
                    "filters": analysis["filters"],
                    "trends": analysis["trends"],
                    "exceptions": analysis["exceptions"],
                    "history": [],
                    "summary": "Managed Postgres is not configured yet; dashboard preview and artifact were generated without durable history.",
                    "artifact": _artifact_payload(artifact_ref, csv_text),
                },
            )

        payload = {
            "request": normalized.model_dump(mode="json"),
            "analysis": analysis,
            "artifact": {"name": artifact_ref.name, "uri": artifact_ref.uri, "media_type": artifact_ref.mime_type},
        }
        receipt_payload = {
            "receipt_id": receipt_id,
            "dashboard_id": dashboard_id,
            "skill": "build_dashboard",
            "status": "ok",
            "input_hash": _hash_json(normalized.model_dump(mode="json")),
            "artifact_name": artifact_ref.name,
            "created_at": now.isoformat(),
        }
        try:
            with _db() as conn:
                with conn.transaction():
                    _insert_dashboard(conn, tenant, dashboard_id, normalized, analysis, payload, now)
                    _insert_receipt(conn, tenant, receipt_id, dashboard_id, receipt_payload, now)
                    history = _fetch_history(conn, tenant, limit=6)
        except Exception as exc:  # noqa: BLE001 - return structured product error, never raw DB URL/secrets
            await ctx.emit_error("Could not persist dashboard history", code="db_persist_failed")
            return DashboardResult(
                status="error",
                dashboard_id=dashboard_id,
                receipt_id=receipt_id,
                **{
                    "dashboard-data": analysis["rows"],
                    "filters": analysis["filters"],
                    "trends": analysis["trends"],
                    "exceptions": analysis["exceptions"],
                    "history": [],
                    "summary": f"Dashboard generated, but durable history could not be saved ({type(exc).__name__}).",
                    "artifact": _artifact_payload(artifact_ref, csv_text),
                },
            )

        return DashboardResult(
            status="ok",
            dashboard_id=dashboard_id,
            receipt_id=receipt_id,
            **{
                "dashboard-data": analysis["rows"],
                "filters": analysis["filters"],
                "trends": analysis["trends"],
                "exceptions": analysis["exceptions"],
                "history": history,
                "summary": _summary(normalized, analysis),
                "artifact": _artifact_payload(artifact_ref, csv_text),
            },
        )

    @a2a.tool(
        description="List durable dashboard history for the signed-in user.",
        timeout_seconds=30,
        cost_class="cheap",
        idempotent=True,
    )
    async def list_dashboard_history(
        self,
        ctx: RunContext[PlatformUserAuth],
        limit: int = 10,
    ) -> HistoryResult:
        tenant = _tenant_key(ctx)
        bounded_limit = max(1, min(int(limit), 25))
        if not os.environ.get("DATABASE_URL"):
            return HistoryResult(status="setup_required", history=[])
        try:
            with _db() as conn:
                rows = _fetch_history(conn, tenant, bounded_limit)
        except Exception:  # noqa: BLE001
            return HistoryResult(status="error", history=[])
        return HistoryResult(status="ok", history=rows)

    @a2a.tool(
        description="Idempotent scheduled sales dashboard snapshot. Safe for recurring platform schedule/MCP calls using a stable run key.",
        timeout_seconds=60,
        cost_class="cheap",
        idempotent=True,
    )
    async def run_scheduled_snapshot(
        self,
        ctx: RunContext[PlatformUserAuth],
        run_key: str,
        segment: str = "Mid-market",
    ) -> DashboardResult:
        safe_key = re.sub(r"[^A-Za-z0-9_.:-]", "-", run_key.strip())[:80] or "manual"
        fixture = BuildDashboardRequest(
            dashboard_name=f"Scheduled Sales Ops Pulse {safe_key}",
            segment=segment[:80] or "Mid-market",
            period_label=safe_key,
            updates=_default_updates(),
        )
        return await self.build_dashboard(ctx, fixture)


def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
    stable_id = ctx.auth.user_id or ctx.auth.sub
    if not stable_id:
        raise PermissionError("stable platform identity required")
    return f"user:{stable_id}"


@contextmanager
def _db():
    import psycopg

    conn = psycopg.connect(
        os.environ["DATABASE_URL"],
        options="-c statement_timeout=5000 -c lock_timeout=3000 -c idle_in_transaction_session_timeout=10000",
    )
    try:
        yield conn
    finally:
        conn.close()


def _default_updates() -> list[SalesUpdate]:
    return [
        SalesUpdate(account="Acme Manufacturing", owner="Priya", stage="Security review", amount_usd=84000, forecast_category="commit", status="blocked", days_in_stage=18, update="Waiting on vendor risk questionnaire."),
        SalesUpdate(account="Northstar Health", owner="Mateo", stage="Proposal", amount_usd=52000, forecast_category="best_case", status="at_risk", days_in_stage=11, update="Economic buyer asked for ROI proof."),
        SalesUpdate(account="Bluebird Logistics", owner="Priya", stage="Discovery", amount_usd=27000, forecast_category="pipeline", status="on_track", days_in_stage=4, update="Expansion pain confirmed."),
        SalesUpdate(account="Summit Retail", owner="Nora", stage="Procurement", amount_usd=125000, forecast_category="commit", status="slipped", days_in_stage=31, update="Legal redlines moved close date."),
    ]


def _with_default_updates(request: BuildDashboardRequest) -> BuildDashboardRequest:
    if request.updates:
        return request
    data = request.model_dump()
    data["updates"] = [item.model_dump() for item in _default_updates()]
    return BuildDashboardRequest.model_validate(data)


def _analyze_updates(request: BuildDashboardRequest) -> dict[str, Any]:
    updates = request.updates[:MAX_UPDATES]
    total_pipeline = round(sum(item.amount_usd for item in updates), 2)
    weighted = round(sum(item.amount_usd * _forecast_weight(item.forecast_category) for item in updates), 2)
    owners = sorted({item.owner for item in updates})
    stages = sorted({item.stage for item in updates})
    statuses = sorted({item.status for item in updates})
    rows: list[dict[str, Any]] = []
    exceptions: list[dict[str, Any]] = []
    for item in updates:
        risk_score = _risk_score(item)
        row = {
            "account": item.account,
            "owner": item.owner,
            "stage": item.stage,
            "amount_usd": round(item.amount_usd, 2),
            "forecast_category": item.forecast_category,
            "status": item.status,
            "days_in_stage": item.days_in_stage,
            "risk_score": risk_score,
            "update": item.update,
        }
        rows.append(row)
        if item.status in {"blocked", "slipped", "at_risk"} and item.amount_usd >= request.min_exception_amount_usd:
            exceptions.append(
                {
                    "account": item.account,
                    "owner": item.owner,
                    "reason": f"{item.status.replace('_', ' ')} deal worth ${item.amount_usd:,.0f}",
                    "recommended_action": _recommended_action(item),
                    "risk_score": risk_score,
                }
            )
    rows.sort(key=lambda row: (-row["risk_score"], -row["amount_usd"], row["account"]))
    exceptions.sort(key=lambda row: (-row["risk_score"], row["account"]))
    stage_totals: dict[str, float] = {}
    owner_totals: dict[str, float] = {}
    for item in updates:
        stage_totals[item.stage] = stage_totals.get(item.stage, 0.0) + item.amount_usd
        owner_totals[item.owner] = owner_totals.get(item.owner, 0.0) + item.amount_usd
    trends = [
        {"metric": "pipeline_total", "label": "Pipeline", "value": total_pipeline},
        {"metric": "weighted_pipeline", "label": "Weighted pipeline", "value": weighted},
        {"metric": "exception_count", "label": "Exceptions", "value": len(exceptions)},
        {"metric": "stale_deals", "label": "Deals >14 days in stage", "value": sum(1 for item in updates if item.days_in_stage > 14)},
    ]
    return {
        "rows": rows,
        "filters": {
            "segment": request.segment,
            "period_label": request.period_label,
            "owners": owners,
            "stages": stages,
            "statuses": statuses,
            "min_exception_amount_usd": request.min_exception_amount_usd,
        },
        "trends": trends,
        "exceptions": exceptions,
        "stage_totals": {key: round(value, 2) for key, value in sorted(stage_totals.items())},
        "owner_totals": {key: round(value, 2) for key, value in sorted(owner_totals.items())},
    }


def _forecast_weight(category: str) -> float:
    return {"commit": 0.9, "best_case": 0.55, "pipeline": 0.25, "omitted": 0.0}.get(category, 0.25)


def _risk_score(item: SalesUpdate) -> int:
    score = {"blocked": 90, "slipped": 75, "at_risk": 65, "on_track": 25, "won": 5, "lost": 5}[item.status]
    if item.days_in_stage > 21:
        score += 15
    elif item.days_in_stage > 14:
        score += 8
    if item.amount_usd >= 100_000:
        score += 10
    return min(score, 100)


def _recommended_action(item: SalesUpdate) -> str:
    if item.status == "blocked":
        return "Assign an owner to remove the blocker within 24 hours."
    if item.status == "slipped":
        return "Confirm the new close plan and next mutual action."
    return "Book an executive or value-proof follow-up before the next forecast call."


def _summary(request: BuildDashboardRequest, analysis: dict[str, Any]) -> str:
    total = next(item["value"] for item in analysis["trends"] if item["metric"] == "pipeline_total")
    return (
        f"{request.dashboard_name} has {len(analysis['rows'])} opportunities, "
        f"${total:,.0f} in pipeline, and {len(analysis['exceptions'])} prioritized exceptions."
    )


def _dashboard_csv(rows: list[dict[str, Any]]) -> str:
    output = io.StringIO()
    fieldnames = ["account", "owner", "stage", "amount_usd", "forecast_category", "status", "days_in_stage", "risk_score", "update"]
    writer = csv.DictWriter(output, fieldnames=fieldnames)
    writer.writeheader()
    for row in rows:
        writer.writerow({key: row.get(key, "") for key in fieldnames})
    return output.getvalue()


def _artifact_payload(ref: Any, content: str) -> DashboardArtifact:
    return DashboardArtifact(
        name=ref.name,
        media_type=ref.mime_type,
        uri=ref.uri,
        size_bytes=ref.size_bytes,
        content=content[:120_000],
    )


def _insert_dashboard(conn: Any, tenant: str, dashboard_id: str, request: BuildDashboardRequest, analysis: dict[str, Any], payload: dict[str, Any], now: datetime) -> None:
    conn.execute(
        """
        INSERT INTO sales_dashboards
            (tenant_key, dashboard_id, dashboard_name, segment, period_label, pipeline_total_usd, exception_count, payload, created_at)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s)
        """,
        (
            tenant,
            dashboard_id,
            request.dashboard_name,
            request.segment,
            request.period_label,
            next(item["value"] for item in analysis["trends"] if item["metric"] == "pipeline_total"),
            len(analysis["exceptions"]),
            json.dumps(payload),
            now,
        ),
    )


def _insert_receipt(conn: Any, tenant: str, receipt_id: str, dashboard_id: str, payload: dict[str, Any], now: datetime) -> None:
    conn.execute(
        """
        INSERT INTO execution_receipts
            (tenant_key, receipt_id, dashboard_id, skill_name, status, payload, created_at)
        VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s)
        ON CONFLICT (tenant_key, receipt_id) DO UPDATE SET
            status = EXCLUDED.status,
            payload = EXCLUDED.payload
        """,
        (tenant, receipt_id, dashboard_id, "build_dashboard", "ok", json.dumps(payload), now),
    )


def _fetch_history(conn: Any, tenant: str, limit: int) -> list[dict[str, Any]]:
    cur = conn.execute(
        """
        SELECT dashboard_id, dashboard_name, segment, period_label, pipeline_total_usd, exception_count, created_at
        FROM sales_dashboards
        WHERE tenant_key = %s
        ORDER BY created_at DESC
        LIMIT %s
        """,
        (tenant, limit),
    )
    rows = []
    for row in cur.fetchall():
        rows.append(
            {
                "dashboard_id": row[0],
                "dashboard_name": row[1],
                "segment": row[2],
                "period_label": row[3],
                "pipeline_total_usd": float(row[4]),
                "exception_count": int(row[5]),
                "created_at": row[6].isoformat() if hasattr(row[6], "isoformat") else str(row[6]),
            }
        )
    return rows


def _stable_id(prefix: str, *parts: Any) -> str:
    digest = hashlib.sha256("|".join(json.dumps(part, sort_keys=True, default=str) for part in parts).encode("utf-8")).hexdigest()[:16]
    return f"{prefix}_{digest}"


def _hash_json(value: Any) -> str:
    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode("utf-8")).hexdigest()