quote-judge-studio-v1/agent.py

"""QuoteJudge Studio full-stack A2A agent.

Deterministic, no-LLM quote comparison product:
- platform-authenticated user tenancy;
- managed Postgres persistence;
- typed A2A/MCP tools for comparison, reload, browser uploads, API uploads;
- production receipt records without secrets.
"""
from __future__ import annotations

import base64
import csv
import hashlib
import io
import json
import os
import re
from datetime import datetime, timezone
from typing import Annotated, Any

from pydantic import BaseModel, Field, field_validator

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

MAX_QUOTES = 25
MAX_BROWSER_DOCUMENTS = 4
MAX_BROWSER_DOCUMENT_BYTES = 64 * 1024
ACCEPTED_UPLOAD_TYPES = (
    "application/json",
    "text/csv",
    "text/plain",
)
DATABASE_ENV = "DATABASE_URL"


class QuoteJudgeStudioV1Config(BaseModel):
    """No user-editable config is required."""


class QuoteInput(BaseModel):
    vendor: Annotated[str, Field(min_length=1, max_length=160)]
    unit_price: Annotated[float, Field(gt=0, le=1_000_000)]
    quantity: Annotated[int, Field(gt=0, le=10_000_000)]
    delivery_days: Annotated[int, Field(ge=0, le=3650)]
    warranty_months: Annotated[int, Field(ge=0, le=600)]

    @field_validator("vendor")
    @classmethod
    def _clean_vendor(cls, value: str) -> str:
        cleaned = re.sub(r"\s+", " ", value).strip()
        if not cleaned:
            raise ValueError("vendor is required")
        return cleaned


class QuoteWeights(BaseModel):
    price: Annotated[float, Field(ge=0, le=100)] = 50
    delivery: Annotated[float, Field(ge=0, le=100)] = 30
    warranty: Annotated[float, Field(ge=0, le=100)] = 20


class BrowserDocument(BaseModel):
    filename: Annotated[str, Field(min_length=1, max_length=180)]
    media_type: Annotated[str, Field(min_length=1, max_length=100)]
    data_base64: Annotated[str, Field(min_length=1, max_length=90_000)]


class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
    name = "quote-judge-studio-v1"
    description = (
        "QuoteJudge compares vendor quotes with weighted price, delivery, "
        "and warranty scoring, persists recommendations, supports upload/paste "
        "workflows, and exposes typed A2A/MCP tools."
    )
    version = "0.1.0"

    config_model = QuoteJudgeStudioV1Config
    auth_model = PlatformUserAuth

    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic quote scoring. No LLM credential is required.",
    )
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
    workspace_access = WorkspaceAccess.dynamic(
        max_files=8,
        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
        require_reason=False,
        max_total_size_bytes=512 * 1024,
    )
    tools_used = ("postgres", "mcp", "packed-frontend")
    platform_resources = AgentPlatformResources(
        databases=(
            AgentDatabase(
                name="quote-judge-studio-v1-data",
                scope="user",
                access_mode="read_write",
                env=AgentDatabaseEnv(url=DATABASE_ENV),
                migrations=AgentDatabaseMigrations(path="db/migrations"),
            ),
        )
    )

    @a2a.tool(
        description=(
            "Compare at least two vendor quotes with price/delivery/warranty "
            "weights, persist the result, and return a recommendation."
        ),
        timeout_seconds=60,
        idempotent=True,
        cost_class="deterministic",
    )
    async def compare_quotes(
        self,
        ctx: RunContext[PlatformUserAuth],
        comparison_id: Annotated[str, Field(min_length=1, max_length=120)],
        quotes: Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)],
        weights: QuoteWeights,
    ) -> dict[str, Any]:
        await ctx.emit_progress("Validating quote comparison inputs")
        tenant = _tenant_key(ctx)
        result = _build_comparison_result(comparison_id, quotes, weights)
        if not result["ok"]:
            await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "validation_error", {"code": result["code"]}, result)
            return result

        await ctx.emit_progress("Persisting comparison and production receipt")
        await _save_comparison(tenant, comparison_id, result)
        receipt = await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "ok", _safe_input_payload(comparison_id, quotes, weights), result)
        result["receipt"] = receipt
        return result

    @a2a.tool(
        description="Reopen a previously persisted quote comparison by id.",
        timeout_seconds=30,
        idempotent=True,
        cost_class="deterministic",
    )
    async def get_comparison(
        self,
        ctx: RunContext[PlatformUserAuth],
        comparison_id: Annotated[str, Field(min_length=1, max_length=120)],
    ) -> dict[str, Any]:
        tenant = _tenant_key(ctx)
        record = await _load_comparison(tenant, comparison_id)
        if record is None:
            result = {
                "ok": False,
                "code": "comparison_not_found",
                "message": "No comparison was found for this id in your workspace.",
                "comparison_id": comparison_id,
            }
            await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "not_found", {"comparison_id": comparison_id}, result)
            return result
        await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "ok", {"comparison_id": comparison_id}, record)
        return record

    @a2a.tool(
        description="List recently saved quote comparisons for the signed-in user.",
        timeout_seconds=30,
        idempotent=True,
        cost_class="deterministic",
    )
    async def list_comparisons(
        self,
        ctx: RunContext[PlatformUserAuth],
        limit: Annotated[int, Field(ge=1, le=50)] = 10,
    ) -> dict[str, Any]:
        rows = await _list_comparisons(_tenant_key(ctx), limit)
        return {"ok": True, "comparisons": rows}

    @a2a.tool(
        description=(
            "Browser-safe upload bridge: accept bounded base64 JSON/CSV/text "
            "quote files, parse them, compare, persist, and return the result."
        ),
        timeout_seconds=60,
        idempotent=True,
        cost_class="deterministic",
    )
    async def compare_quotes_from_browser_upload(
        self,
        ctx: RunContext[PlatformUserAuth],
        comparison_id: Annotated[str, Field(min_length=1, max_length=120)],
        documents: Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_BROWSER_DOCUMENTS)],
        weights: QuoteWeights,
    ) -> dict[str, Any]:
        parsed = _parse_browser_documents(documents)
        if not parsed["ok"]:
            await _persist_receipt(ctx, _tenant_key(ctx), comparison_id, "compare_quotes_from_browser_upload", "validation_error", {"comparison_id": comparison_id}, parsed)
            return {**parsed, "comparison_id": comparison_id}
        return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights)

    @a2a.tool(
        description=(
            "External API upload path: accept a typed FileUpload containing "
            "JSON, CSV, or pasted text quote data and run the same comparison."
        ),
        timeout_seconds=60,
        idempotent=True,
        cost_class="deterministic",
        grant_mode="read_only",
        grant_allow_patterns=("**",),
    )
    async def compare_quotes_from_upload(
        self,
        ctx: RunContext[PlatformUserAuth],
        comparison_id: Annotated[str, Field(min_length=1, max_length=120)],
        quote_file: Annotated[
            UploadedFile,
            FileUpload(
                accept=ACCEPTED_UPLOAD_TYPES,
                max_bytes=MAX_BROWSER_DOCUMENT_BYTES,
                description="JSON list/object, CSV, or text containing vendor quotes.",
            ),
        ],
        weights: QuoteWeights,
    ) -> dict[str, Any]:
        if quote_file.size_bytes > MAX_BROWSER_DOCUMENT_BYTES:
            return _validation_error(
                "file_too_large",
                f"Upload is {quote_file.size_bytes} bytes; maximum is {MAX_BROWSER_DOCUMENT_BYTES} bytes.",
                comparison_id=comparison_id,
            )
        if quote_file.media_type not in ACCEPTED_UPLOAD_TYPES:
            return _validation_error(
                "unsupported_media_type",
                f"Unsupported media type {quote_file.media_type!r}. Use JSON, CSV, or plain text.",
                comparison_id=comparison_id,
            )
        reader = getattr(ctx.workspace, "read_bytes", None)
        if reader is None:
            return _validation_error("workspace_read_unavailable", "The runtime did not expose uploaded file bytes.", comparison_id=comparison_id)
        try:
            raw = reader(quote_file.path)
        except Exception:  # noqa: BLE001
            return _validation_error("file_read_failed", "Could not read the uploaded file from the granted workspace.", comparison_id=comparison_id)
        parsed = _parse_document_bytes(quote_file.filename, quote_file.media_type, raw)
        if not parsed["ok"]:
            return {**parsed, "comparison_id": comparison_id}
        return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights)


def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
    auth = ctx.auth
    stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
    return f"user:{stable}"


def _safe_input_payload(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
    return {
        "comparison_id": comparison_id,
        "quote_count": len(quotes),
        "vendors": [quote.vendor for quote in quotes],
        "weights": weights.model_dump(mode="json"),
    }


def _validation_error(code: str, message: str, *, comparison_id: str | None = None, details: list[dict[str, Any]] | None = None) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "ok": False,
        "code": code,
        "message": message,
        "validation_errors": details or [{"field": "quotes", "message": message}],
    }
    if comparison_id is not None:
        payload["comparison_id"] = comparison_id
    return payload


def _build_comparison_result(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
    if len(quotes) < 2:
        return _validation_error(
            "at_least_two_quotes_required",
            "Provide at least two vendor quotes before requesting a recommendation.",
            comparison_id=comparison_id,
            details=[{"field": "quotes", "message": "At least two quotes are required."}],
        )
    if len(quotes) > MAX_QUOTES:
        return _validation_error(
            "too_many_quotes",
            f"Provide no more than {MAX_QUOTES} quotes.",
            comparison_id=comparison_id,
        )
    total_weight = weights.price + weights.delivery + weights.warranty
    if total_weight <= 0:
        return _validation_error(
            "weights_must_sum_positive",
            "At least one of price, delivery, or warranty weight must be greater than zero.",
            comparison_id=comparison_id,
            details=[{"field": "weights", "message": "Weights must sum to a positive number."}],
        )

    min_price = min(q.unit_price for q in quotes)
    min_delivery = min(q.delivery_days for q in quotes)
    max_warranty = max(q.warranty_months for q in quotes)
    normalized_weights = {
        "price": weights.price / total_weight,
        "delivery": weights.delivery / total_weight,
        "warranty": weights.warranty / total_weight,
    }

    rows: list[dict[str, Any]] = []
    for quote in quotes:
        price_score = (min_price / quote.unit_price) * 100 if quote.unit_price else 0.0
        delivery_score = 100.0 if quote.delivery_days == 0 and min_delivery == 0 else ((min_delivery + 1) / (quote.delivery_days + 1)) * 100
        warranty_score = (quote.warranty_months / max_warranty) * 100 if max_warranty else 100.0
        weighted_score = (
            normalized_weights["price"] * price_score
            + normalized_weights["delivery"] * delivery_score
            + normalized_weights["warranty"] * warranty_score
        )
        total_price = quote.unit_price * quote.quantity
        rows.append(
            {
                "vendor": quote.vendor,
                "unit_price": round(quote.unit_price, 4),
                "quantity": quote.quantity,
                "total_price": round(total_price, 2),
                "delivery_days": quote.delivery_days,
                "warranty_months": quote.warranty_months,
                "scores": {
                    "price": round(price_score, 2),
                    "delivery": round(delivery_score, 2),
                    "warranty": round(warranty_score, 2),
                    "weighted_total": round(weighted_score, 2),
                },
            }
        )

    rows.sort(key=_quote_sort_key)
    winner = rows[0]
    recommendation = {
        "vendor": winner["vendor"],
        "score": winner["scores"]["weighted_total"],
        "reason": (
            f"{winner['vendor']} has the strongest weighted score "
            f"({winner['scores']['weighted_total']:.2f}) using price={weights.price:g}, "
            f"delivery={weights.delivery:g}, warranty={weights.warranty:g}."
        ),
    }
    return {
        "ok": True,
        "comparison_id": comparison_id,
        "recommendation": recommendation,
        "weights": {
            "price": weights.price,
            "delivery": weights.delivery,
            "warranty": weights.warranty,
            "normalized": {key: round(value, 4) for key, value in normalized_weights.items()},
        },
        "quote_count": len(quotes),
        "comparison_table": rows,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }


def _quote_sort_key(item: dict[str, Any]) -> tuple[Any, ...]:
    return (
        -item["scores"]["weighted_total"],
        item["total_price"],
        item["delivery_days"],
        item["vendor"].lower(),
    )


def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
    quotes: list[QuoteInput] = []
    errors: list[dict[str, Any]] = []
    for index, doc in enumerate(documents):
        if doc.media_type not in ACCEPTED_UPLOAD_TYPES:
            errors.append({"field": f"documents[{index}].media_type", "message": "Use JSON, CSV, or plain text."})
            continue
        try:
            raw = base64.b64decode(doc.data_base64, validate=True)
        except Exception:  # noqa: BLE001
            errors.append({"field": f"documents[{index}].data_base64", "message": "Invalid base64 payload."})
            continue
        if len(raw) > MAX_BROWSER_DOCUMENT_BYTES:
            errors.append({"field": f"documents[{index}]", "message": f"File exceeds {MAX_BROWSER_DOCUMENT_BYTES} bytes."})
            continue
        parsed = _parse_document_bytes(doc.filename, doc.media_type, raw)
        if parsed["ok"]:
            quotes.extend(parsed["quotes"])
        else:
            errors.extend(parsed.get("validation_errors") or [])
    if errors:
        return _validation_error("upload_parse_failed", "One or more uploaded quote files could not be parsed.", details=errors)
    return {"ok": True, "quotes": quotes[:MAX_QUOTES]}


def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]:
    if len(raw) > MAX_BROWSER_DOCUMENT_BYTES:
        return _validation_error("file_too_large", f"{filename} exceeds the {MAX_BROWSER_DOCUMENT_BYTES} byte limit.")
    text = raw.decode("utf-8", errors="replace").strip()
    if not text:
        return _validation_error("empty_upload", "The uploaded quote file was empty.")
    try:
        if media_type == "application/json" or filename.lower().endswith(".json"):
            data = json.loads(text)
            items = data.get("quotes") if isinstance(data, dict) else data
            if not isinstance(items, list):
                return _validation_error("invalid_json_quotes", "JSON must be a list of quotes or an object with a quotes list.")
            return {"ok": True, "quotes": [QuoteInput.model_validate(item) for item in items]}
        if media_type == "text/csv" or filename.lower().endswith(".csv"):
            reader = csv.DictReader(io.StringIO(text))
            return {"ok": True, "quotes": [QuoteInput.model_validate(row) for row in reader]}
        return {"ok": True, "quotes": _parse_loose_text_quotes(text)}
    except Exception as exc:  # noqa: BLE001
        return _validation_error("quote_parse_failed", f"Could not parse quote data: {type(exc).__name__}.")


def _parse_loose_text_quotes(text: str) -> list[QuoteInput]:
    quotes: list[QuoteInput] = []
    for line in text.splitlines():
        parts = [part.strip() for part in re.split(r"[,|;]\s*", line) if part.strip()]
        if len(parts) < 5 or parts[0].lower() in {"vendor", "supplier"}:
            continue
        quotes.append(
            QuoteInput(
                vendor=parts[0],
                unit_price=float(parts[1]),
                quantity=int(float(parts[2])),
                delivery_days=int(float(parts[3])),
                warranty_months=int(float(parts[4])),
            )
        )
    if not quotes:
        raise ValueError("No quote rows found. Expected vendor, unit_price, quantity, delivery_days, warranty_months.")
    return quotes


def _db_url() -> str:
    return os.environ.get(DATABASE_ENV, "").strip()


def _connect():
    url = _db_url()
    if not url:
        raise RuntimeError("managed database is not configured")
    import psycopg

    return psycopg.connect(url, autocommit=True)


async def _save_comparison(tenant_key: str, comparison_id: str, result: dict[str, Any]) -> None:
    with _connect() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO quote_comparisons (
                    tenant_key, comparison_id, recommendation_vendor, recommended_vendor,
                    result, payload, weighted_score, score, normalized_total
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
                    recommendation_vendor = EXCLUDED.recommendation_vendor,
                    recommended_vendor = EXCLUDED.recommended_vendor,
                    result = EXCLUDED.result,
                    payload = EXCLUDED.payload,
                    weighted_score = EXCLUDED.weighted_score,
                    score = EXCLUDED.score,
                    normalized_total = EXCLUDED.normalized_total,
                    updated_at = NOW()
                """,
                (
                    tenant_key,
                    comparison_id,
                    result["recommendation"]["vendor"],
                    result["recommendation"]["vendor"],
                    json.dumps(result),
                    json.dumps(result),
                    result["recommendation"]["score"],
                    result["recommendation"]["score"],
                    result["recommendation"]["score"],
                ),
            )


async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None:
    with _connect() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT COALESCE(result, payload) FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s",
                (tenant_key, comparison_id),
            )
            row = cur.fetchone()
    if row is None:
        return None
    value = row[0]
    if isinstance(value, str):
        return json.loads(value)
    return value


async def _list_comparisons(tenant_key: str, limit: int) -> list[dict[str, Any]]:
    with _connect() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT comparison_id, COALESCE(recommendation_vendor, recommended_vendor), updated_at, COALESCE(result, payload)
                FROM quote_comparisons
                WHERE tenant_key = %s
                ORDER BY updated_at DESC
                LIMIT %s
                """,
                (tenant_key, limit),
            )
            rows = cur.fetchall()
    output: list[dict[str, Any]] = []
    for comparison_id, recommendation_vendor, updated_at, result in rows:
        output.append(
            {
                "comparison_id": comparison_id,
                "recommendation_vendor": recommendation_vendor,
                "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
                "quote_count": (result or {}).get("quote_count") if isinstance(result, dict) else None,
            }
        )
    return output


async def _persist_receipt(
    ctx: RunContext[PlatformUserAuth],
    tenant_key: str,
    comparison_id: str,
    tool_name: str,
    status: str,
    inputs: dict[str, Any],
    result: dict[str, Any],
) -> dict[str, Any]:
    payload = {
        "agent": QuoteJudgeStudioV1.name,
        "version": QuoteJudgeStudioV1.version,
        "tool": tool_name,
        "status": status,
        "comparison_id": comparison_id,
        "caller_scope": "platform_user",
        "task_id": getattr(ctx, "task_id", ""),
        "input_hash": _hash_json(inputs),
        "result_hash": _hash_json(_receipt_safe_result(result)),
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    payload["receipt_id"] = "qjr_" + _hash_json(payload)[:24]
    try:
        with _connect() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO quote_judge_receipts
                        (receipt_id, tenant_key, comparison_id, tool_name, status, input_hash, payload)
                    VALUES (%s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (receipt_id) DO NOTHING
                    """,
                    (payload["receipt_id"], tenant_key, comparison_id, tool_name, status, payload["input_hash"], json.dumps(payload)),
                )
    except Exception as exc:  # noqa: BLE001
        await ctx.emit_error("Receipt persistence failed; comparison result was still computed.", code="receipt_persist_failed")
        return {"persisted": False, "code": "receipt_persist_failed", "message": type(exc).__name__}
    return {"persisted": True, "receipt_id": payload["receipt_id"], "input_hash": payload["input_hash"]}


def _receipt_safe_result(result: dict[str, Any]) -> dict[str, Any]:
    return {
        "ok": result.get("ok"),
        "comparison_id": result.get("comparison_id"),
        "recommendation": result.get("recommendation"),
        "code": result.get("code"),
    }


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