"""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) workspace_access = WorkspaceAccess.dynamic( max_files=16, allowed_modes=(WorkspaceMode.READ_WRITE_OVERLAY,), require_reason=False, ) 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", grant_mode="read_write_overlay", grant_allow_patterns=("outputs/dashboards/**",), grant_outputs_prefix="outputs/dashboards/", grant_write_prefixes=("outputs/dashboards/",), ) 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, grant_mode="read_write_overlay", grant_allow_patterns=("outputs/dashboards/**",), grant_outputs_prefix="outputs/dashboards/", grant_write_prefixes=("outputs/dashboards/",), ) 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()