"""QuoteJudge full-stack A2A agent. Deterministic quote comparison, per-user persistence in managed Postgres, browser-safe upload parsing, and production execution receipt storage. """ from __future__ import annotations import base64 import csv import hashlib import io import json import os import re import uuid from contextlib import contextmanager 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, PlatformUserAuth, Pricing, Resources, RunContext, State, WorkspaceAccess, WorkspaceMode, ) from a2a_pack.workspace import FileUpload, UploadedFile try: # psycopg is installed in deployment and sandbox via requirements.txt. import psycopg from psycopg.rows import dict_row except Exception: # pragma: no cover - import guard for card-only tooling psycopg = None dict_row = None MAX_QUOTES = 20 MAX_VENDOR_LENGTH = 120 MAX_UPLOADS = 5 MAX_UPLOAD_BYTES = 128_000 MAX_UPLOAD_BASE64_CHARS = ((MAX_UPLOAD_BYTES + 2) // 3) * 4 ALLOWED_UPLOAD_MEDIA_TYPES = {"text/csv", "text/plain", "application/json"} _DATABASE_NAME = "quote-judge-studio-v1-data" _LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {} _LOCAL_RECEIPTS: list[dict[str, Any]] = [] class QuoteJudgeStudioV1Config(BaseModel): """No caller-configured settings are required.""" class QuoteInput(BaseModel): vendor: str = Field(..., min_length=1, max_length=MAX_VENDOR_LENGTH) quantity: int = Field(..., gt=0, le=10_000_000) unit_price: float = Field(..., ge=0, le=1_000_000_000) delivery_days: int = Field(..., ge=0, le=3650) warranty_months: int = Field(..., ge=0, le=600) @field_validator("vendor") @classmethod def clean_vendor(cls, value: str) -> str: cleaned = " ".join(value.strip().split()) if not cleaned: raise ValueError("vendor is required") return cleaned class QuoteWeights(BaseModel): price: float = Field(..., ge=0, le=100) delivery: float = Field(..., ge=0, le=100) warranty: float = Field(..., ge=0, le=100) class BrowserQuoteDocument(BaseModel): filename: str = Field(..., min_length=1, max_length=160) media_type: str = Field(..., min_length=1, max_length=100) data_base64: str = Field(..., min_length=1, max_length=MAX_UPLOAD_BASE64_CHARS) @field_validator("filename") @classmethod def safe_filename(cls, value: str) -> str: cleaned = value.strip().replace("\\", "/").split("/")[-1] if not cleaned or cleaned in {".", ".."}: raise ValueError("filename is required") return cleaned[:160] class QuoteJudgeState(BaseModel): """Durable state is stored in managed Postgres; this schema advertises that fact.""" persistence: str = "managed_postgres" class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]): name = "quote-judge-studio-v1" description = ( "QuoteJudge compares vendor quotes with weighted price, delivery, and " "warranty scoring, persists per-user comparisons, and serves a one-page product UI." ) version = "0.1.4" config_model = QuoteJudgeStudioV1Config auth_model = PlatformUserAuth state = State.DURABLE state_model = QuoteJudgeState resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300) platform_resources = AgentPlatformResources( databases=( AgentDatabase( name=_DATABASE_NAME, scope="user", access_mode="read_write", env=AgentDatabaseEnv(url="DATABASE_URL"), migrations=AgentDatabaseMigrations(path="db/migrations"), ), ) ) pricing = Pricing( price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic local quote scoring; no LLM credential is required.", ) workspace_access = WorkspaceAccess.dynamic( max_files=MAX_UPLOADS, allowed_modes=(WorkspaceMode.READ_ONLY,), require_reason=False, max_total_size_bytes=MAX_UPLOADS * MAX_UPLOAD_BYTES, ) tools_used = ("postgres", "mcp") @a2a.tool( description=( "Compare at least two vendor quotes using typed price, delivery, and warranty " "weights; persist the recommendation and a production receipt." ), timeout_seconds=30, idempotent=False, cost_class="cheap", ) async def compare_quotes( self, ctx: RunContext[PlatformUserAuth], comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights, ) -> dict[str, Any]: tenant = _tenant_key(ctx) cleaned_id = _clean_comparison_id(comparison_id) started = _now_iso() input_payload = { "comparison_id": cleaned_id, "quotes": [quote.model_dump() for quote in quotes], "weights": weights.model_dump(), } if weights.price + weights.delivery + weights.warranty <= 0: result = { "ok": False, "code": "positive_weight_required", "message": "Set at least one weight above zero before comparing.", "comparison_id": cleaned_id, } await _persist_receipt( tenant_key=tenant, comparison_id=cleaned_id, skill_name="compare_quotes", input_payload=input_payload, result_payload=result, status="validation_error", started_at=started, ) return result if len(quotes) < 2: result = { "ok": False, "code": "at_least_two_quotes_required", "message": "Add at least two vendor quotes before comparing.", "comparison_id": cleaned_id, } await _persist_receipt( tenant_key=tenant, comparison_id=cleaned_id, skill_name="compare_quotes", input_payload=input_payload, result_payload=result, status="validation_error", started_at=started, ) return result if len(quotes) > MAX_QUOTES: result = { "ok": False, "code": "too_many_quotes", "message": f"Compare at most {MAX_QUOTES} quotes at a time.", "comparison_id": cleaned_id, } await _persist_receipt( tenant_key=tenant, comparison_id=cleaned_id, skill_name="compare_quotes", input_payload=input_payload, result_payload=result, status="validation_error", started_at=started, ) return result result = _score_quotes(cleaned_id, quotes, weights) await _save_comparison(tenant, result) await _persist_receipt( tenant_key=tenant, comparison_id=cleaned_id, skill_name="compare_quotes", input_payload=input_payload, result_payload=result, status="ok", started_at=started, ) return result @a2a.tool( description="Reopen a saved QuoteJudge comparison for the authenticated user.", timeout_seconds=15, idempotent=False, cost_class="cheap", ) async def get_comparison( self, ctx: RunContext[PlatformUserAuth], comparison_id: str, ) -> dict[str, Any]: tenant = _tenant_key(ctx) cleaned_id = _clean_comparison_id(comparison_id) started = _now_iso() record = await _load_comparison(tenant, cleaned_id) if record is None: result = { "ok": False, "code": "comparison_not_found", "message": "No saved comparison exists for this id and user.", "comparison_id": cleaned_id, } else: result = dict(record) result["ok"] = True await _persist_receipt( tenant_key=tenant, comparison_id=cleaned_id, skill_name="get_comparison", input_payload={"comparison_id": cleaned_id}, result_payload=result, status="ok" if result.get("ok") else "not_found", started_at=started, ) return result @a2a.tool( description=( "Browser upload bridge: parse bounded base64 JSON/CSV/text quote files, " "compare extracted quotes, and persist the result." ), timeout_seconds=30, idempotent=False, cost_class="cheap", ) async def compare_uploaded_quotes_browser( self, ctx: RunContext[PlatformUserAuth], comparison_id: str, documents: list[BrowserQuoteDocument], weights: QuoteWeights, ) -> dict[str, Any]: parsed = _parse_browser_documents(documents) if not parsed["ok"]: return parsed return await self.compare_quotes( ctx, comparison_id=comparison_id, quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]], weights=weights, ) @a2a.tool( description=( "External-client upload path: parse an uploaded CSV, JSON, or plain-text quote " "file and compare extracted quotes." ), timeout_seconds=30, idempotent=False, cost_class="cheap", ) async def compare_uploaded_quote_file( self, ctx: RunContext[PlatformUserAuth], comparison_id: str, document: Annotated[ UploadedFile, FileUpload( accept=("text/csv", "text/plain", "application/json"), max_bytes=MAX_UPLOAD_BYTES, description="CSV, JSON, or text file containing vendor quote rows.", ), ], weights: QuoteWeights, ) -> dict[str, Any]: if document.size_bytes > MAX_UPLOAD_BYTES: return { "ok": False, "code": "upload_too_large", "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller.", "comparison_id": _clean_comparison_id(comparison_id), } if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: return { "ok": False, "code": "unsupported_media_type", "message": "Upload must be CSV, JSON, or plain text.", "comparison_id": _clean_comparison_id(comparison_id), } try: reader = getattr(ctx.workspace, "read_bytes") raw = reader(document.path) except Exception: return { "ok": False, "code": "upload_read_failed", "message": "Could not read the uploaded file from the granted workspace.", "comparison_id": _clean_comparison_id(comparison_id), } parsed = _parse_document_bytes(document.filename, document.media_type, raw) if not parsed["ok"]: return parsed return await self.compare_quotes( ctx, comparison_id=comparison_id, quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]], weights=weights, ) def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: auth = ctx.auth ident = auth.user_id if auth.user_id is not None else (auth.sub or auth.email) if ident is None: raise ValueError("authenticated user identity is required") return f"user:{ident}" def _clean_comparison_id(value: str) -> str: cleaned = str(value or "").strip() if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,79}", cleaned): raise ValueError("comparison_id must be 1-80 safe characters") return cleaned def _score_sort_key(row: dict[str, Any]) -> tuple[float, float, int, int, str]: return ( -float(row["scores"]["weighted_total"]), float(row["subtotal"]), int(row["delivery_days"]), -int(row["warranty_months"]), str(row["vendor"]).lower(), ) def _subtotal_key(row: dict[str, Any]) -> float: return float(row["subtotal"]) def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: quote_rows = [quote.model_dump() for quote in quotes] for row in quote_rows: row["subtotal"] = round(row["quantity"] * row["unit_price"], 2) totals = [row["subtotal"] for row in quote_rows] deliveries = [row["delivery_days"] for row in quote_rows] warranties = [row["warranty_months"] for row in quote_rows] total_weight = weights.price + weights.delivery + weights.warranty normalized_weights = { "price": weights.price / total_weight, "delivery": weights.delivery / total_weight, "warranty": weights.warranty / total_weight, } scored: list[dict[str, Any]] = [] for row in quote_rows: price_score = _lower_is_better(row["subtotal"], totals) delivery_score = _lower_is_better(row["delivery_days"], deliveries) warranty_score = _higher_is_better(row["warranty_months"], warranties) weighted_score = ( price_score * normalized_weights["price"] + delivery_score * normalized_weights["delivery"] + warranty_score * normalized_weights["warranty"] ) scored.append( { **row, "scores": { "price": round(price_score, 4), "delivery": round(delivery_score, 4), "warranty": round(warranty_score, 4), "weighted_total": round(weighted_score, 4), }, } ) scored.sort(key=_score_sort_key) winner = scored[0] baseline = min(scored, key=_subtotal_key) rationale_bits = [ f"{winner['vendor']} has the highest weighted score ({winner['scores']['weighted_total']:.2f}).", f"Its normalized total is {winner['subtotal']:.2f} for {winner['quantity']} units.", ] if winner["delivery_days"] == min(deliveries): rationale_bits.append("It also has the fastest delivery among the submitted quotes.") if winner["warranty_months"] == max(warranties): rationale_bits.append("It offers the strongest warranty among the submitted quotes.") if winner["vendor"] == baseline["vendor"]: rationale_bits.append("It is the lowest-price option under the selected quantity.") return { "ok": True, "comparison_id": comparison_id, "recommendation": { "vendor": winner["vendor"], "score": winner["scores"]["weighted_total"], "normalized_total": winner["subtotal"], "rationale": " ".join(rationale_bits), }, "weights": { "price": weights.price, "delivery": weights.delivery, "warranty": weights.warranty, "normalized": {key: round(value, 4) for key, value in normalized_weights.items()}, }, "quotes": scored, "created_at": _now_iso(), } def _lower_is_better(value: float, values: list[float]) -> float: min_value = min(values) max_value = max(values) if max_value == min_value: return 1.0 return (max_value - value) / (max_value - min_value) def _higher_is_better(value: float, values: list[float]) -> float: min_value = min(values) max_value = max(values) if max_value == min_value: return 1.0 return (value - min_value) / (max_value - min_value) def _parse_browser_documents(documents: list[BrowserQuoteDocument]) -> dict[str, Any]: if not documents: return {"ok": False, "code": "no_uploads", "message": "Upload at least one quote file."} if len(documents) > MAX_UPLOADS: return {"ok": False, "code": "too_many_uploads", "message": f"Upload at most {MAX_UPLOADS} files."} quotes: list[dict[str, Any]] = [] for document in documents: if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: return {"ok": False, "code": "unsupported_media_type", "message": "Upload must be CSV, JSON, or plain text."} try: raw = base64.b64decode(document.data_base64, validate=True) except Exception: return {"ok": False, "code": "invalid_base64", "message": "Uploaded file data is not valid base64."} if len(raw) > MAX_UPLOAD_BYTES: return {"ok": False, "code": "upload_too_large", "message": f"Each upload must be {MAX_UPLOAD_BYTES} bytes or smaller."} parsed = _parse_document_bytes(document.filename, document.media_type, raw) if not parsed["ok"]: return parsed quotes.extend(parsed["quotes"]) 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_UPLOAD_BYTES: return {"ok": False, "code": "upload_too_large", "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller."} text = raw.decode("utf-8-sig", errors="replace") try: if media_type == "application/json" or filename.lower().endswith(".json"): data = json.loads(text) rows = data.get("quotes", data) if isinstance(data, dict) else data if not isinstance(rows, list): raise ValueError("JSON must be a list or contain a quotes list") quotes = [QuoteInput.model_validate(row).model_dump() for row in rows] elif media_type == "text/csv" or filename.lower().endswith(".csv"): reader = csv.DictReader(io.StringIO(text)) quotes = [QuoteInput.model_validate(_coerce_quote_row(row)).model_dump() for row in reader] else: quotes = _parse_plain_text_quotes(text) except Exception as exc: return {"ok": False, "code": "quote_parse_failed", "message": f"Could not parse quote upload: {exc}"} if not quotes: return {"ok": False, "code": "no_quotes_found", "message": "No quote rows were found in the upload."} return {"ok": True, "quotes": quotes} def _coerce_quote_row(row: dict[str, Any]) -> dict[str, Any]: aliases = { "vendor": ["vendor", "supplier", "name"], "quantity": ["quantity", "qty"], "unit_price": ["unit_price", "price", "unit price", "unitPrice"], "delivery_days": ["delivery_days", "delivery", "delivery days", "lead_time", "lead time"], "warranty_months": ["warranty_months", "warranty", "warranty months"], } normalized = {str(k).strip().lower(): v for k, v in row.items()} out: dict[str, Any] = {} for target, names in aliases.items(): for name in names: if name.lower() in normalized and normalized[name.lower()] not in (None, ""): out[target] = normalized[name.lower()] break return out def _parse_plain_text_quotes(text: str) -> list[dict[str, Any]]: quotes: list[dict[str, Any]] = [] for line in text.splitlines(): if not line.strip(): continue parts = [part.strip() for part in re.split(r"[,;|]", line) if part.strip()] if len(parts) >= 5: quotes.append( QuoteInput.model_validate( { "vendor": parts[0], "quantity": parts[1], "unit_price": parts[2], "delivery_days": parts[3], "warranty_months": parts[4], } ).model_dump() ) return quotes @contextmanager def _db_conn() -> Any: url = os.environ.get("DATABASE_URL") if not url or psycopg is None: yield None return conn = psycopg.connect(url, row_factory=dict_row) try: yield conn conn.commit() finally: conn.close() async def _save_comparison(tenant_key: str, payload: dict[str, Any]) -> None: key = (tenant_key, payload["comparison_id"]) with _db_conn() as conn: if conn is None: _LOCAL_COMPARISONS[key] = dict(payload) return with conn.cursor() as cur: cur.execute( """ INSERT INTO quote_comparisons (tenant_key, comparison_id, recommended_vendor, score, normalized_total, payload, updated_at) VALUES (%s, %s, %s, %s, %s, %s, NOW()) ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET recommended_vendor = EXCLUDED.recommended_vendor, score = EXCLUDED.score, normalized_total = EXCLUDED.normalized_total, payload = EXCLUDED.payload, updated_at = NOW() """, ( tenant_key, payload["comparison_id"], payload["recommendation"]["vendor"], payload["recommendation"]["score"], payload["recommendation"]["normalized_total"], json.dumps(payload), ), ) async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None: key = (tenant_key, comparison_id) with _db_conn() as conn: if conn is None: return _LOCAL_COMPARISONS.get(key) with conn.cursor() as cur: cur.execute( """ SELECT payload FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s """, (tenant_key, comparison_id), ) row = cur.fetchone() if not row: return None payload = row["payload"] return payload if isinstance(payload, dict) else json.loads(payload) async def _persist_receipt( *, tenant_key: str, comparison_id: str, skill_name: str, input_payload: dict[str, Any], result_payload: dict[str, Any], status: str, started_at: str, ) -> dict[str, Any]: receipt = { "receipt_id": str(uuid.uuid4()), "agent": QuoteJudgeStudioV1.name, "agent_version": QuoteJudgeStudioV1.version, "skill": skill_name, "comparison_id": comparison_id, "status": status, "started_at": started_at, "completed_at": _now_iso(), "input_hash": _hash_json(input_payload), "result_hash": _hash_json(result_payload), } with _db_conn() as conn: if conn is None: _LOCAL_RECEIPTS.append({"tenant_key": tenant_key, **receipt}) return receipt with conn.cursor() as cur: cur.execute( """ INSERT INTO quote_execution_receipts (tenant_key, comparison_id, receipt_id, skill_name, status, input_hash, result_hash, payload, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW()) """, ( tenant_key, comparison_id, receipt["receipt_id"], skill_name, status, receipt["input_hash"], receipt["result_hash"], json.dumps(receipt), ), ) return receipt def _hash_json(payload: dict[str, Any]) -> str: encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat()