a2a-source-edit: write agent.py

agent.py

@@ -1,8 +1,8 @@
 """QuoteJudge Studio full-stack A2A agent.
 
 Deterministic quote comparison product with platform-user tenant isolation,
-managed Postgres persistence, bounded browser uploads, typed external file
+managed Postgres persistence, database timeouts, bounded browser uploads,
-uploads, MCP-compatible tools, and execution receipt rows.
+typed external file uploads, MCP-compatible tools, and receipt persistence.
 """
 from __future__ import annotations
 
@@ -90,9 +90,7 @@ class BrowserQuoteDocument(BaseModel):
     def safe_media_type(cls, value: str) -> str:
         cleaned = value.split(";", 1)[0].strip().lower()
         if cleaned not in ALLOWED_UPLOAD_MEDIA_TYPES:
-            raise ValueError(
+            raise ValueError("unsupported media_type; use text/csv, text/plain, or application/json")
-                "unsupported media_type; use text/csv, text/plain, or application/json"
-            )
         return cleaned
 
 
@@ -107,7 +105,6 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     config_model = QuoteJudgeStudioV1Config
     auth_model = PlatformUserAuth
 
-    # Deterministic local logic only. No LLM credential is required.
     pricing = Pricing(
         price_per_call_usd=0.0,
         caller_pays_llm=False,
@@ -154,7 +151,6 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
         if validation is not None:
             await _persist_receipt_best_effort(ctx, "compare_quotes", comparison_id, "error", validation)
             return validation
-
         result = _score_quotes(comparison_id, quotes, weights)
         tenant = _tenant_key(ctx)
         try:
@@ -181,12 +177,7 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
         tenant = _tenant_key(ctx)
         clean_id = _clean_comparison_id(comparison_id)
         if clean_id is None:
-            return {
+            return {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id, "message": "comparison_id must be 1-120 URL-safe characters"}
-                "ok": False,
-                "code": "invalid_comparison_id",
-                "comparison_id": comparison_id,
-                "message": "comparison_id must be 1-120 URL-safe characters",
-            }
         try:
             record = await _load_comparison(tenant, clean_id)
             await _persist_receipt_best_effort(ctx, "get_comparison", clean_id, "ok", record or {"ok": False})
@@ -195,12 +186,7 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
         except Exception:  # noqa: BLE001
             return _temporary_db_error(clean_id)
         if record is None:
-            return {
+            return {"ok": False, "code": "comparison_not_found", "comparison_id": clean_id, "message": "No saved comparison exists for this user and id."}
-                "ok": False,
-                "code": "comparison_not_found",
-                "comparison_id": clean_id,
-                "message": "No saved comparison exists for this user and id.",
-            }
         return record
 
     @a2a.tool(
@@ -226,21 +212,27 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
         try:
             payload = ctx.workspace.read_bytes(document.path)  # type: ignore[attr-defined]
         except Exception:  # noqa: BLE001
-            return {
+            return {"ok": False, "code": "upload_read_failed", "comparison_id": comparison_id, "message": "Could not read the uploaded file from the invocation workspace."}
-                "ok": False,
+        parsed = _parse_uploaded_quotes(document.filename, document.media_type, payload)
-                "code": "upload_read_failed",
-                "comparison_id": comparison_id,
-                "message": "Could not read the uploaded file from the invocation workspace.",
-            }
-        parsed = _parse_uploaded_quotes(
-            filename=document.filename,
-            media_type=document.media_type,
-            data=payload,
-        )
         if not parsed["ok"]:
             return {**parsed, "comparison_id": comparison_id}
         return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
 
+    @a2a.tool(
+        description="Browser JSON bridge: accept one bounded base64 quote file, parse it, compare quotes, and persist the result.",
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def compare_quotes_upload(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        upload: BrowserQuoteDocument,
+        weights: QuoteWeights,
+    ) -> dict[str, Any]:
+        return await self.compare_quotes_browser_upload(ctx, comparison_id, weights, [upload])
+
     @a2a.tool(
         description="Browser JSON bridge: accept bounded base64 quote files, parse them, compare quotes, and persist the result.",
         timeout_seconds=30,
@@ -255,44 +247,22 @@ class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
         documents: list[BrowserQuoteDocument],
     ) -> dict[str, Any]:
         if not documents or len(documents) > MAX_BROWSER_UPLOADS:
-            return {
+            return {"ok": False, "code": "invalid_upload_count", "comparison_id": comparison_id, "message": f"Upload 1 to {MAX_BROWSER_UPLOADS} files."}
-                "ok": False,
-                "code": "invalid_upload_count",
-                "comparison_id": comparison_id,
-                "message": f"Upload 1 to {MAX_BROWSER_UPLOADS} files.",
-            }
         all_quotes: list[QuoteInput] = []
         for document in documents:
             try:
                 data = base64.b64decode(document.data_base64, validate=True)
             except Exception:  # noqa: BLE001
-                return {
+                return {"ok": False, "code": "invalid_base64_upload", "comparison_id": comparison_id, "message": f"{document.filename} is not valid base64."}
-                    "ok": False,
-                    "code": "invalid_base64_upload",
-                    "comparison_id": comparison_id,
-                    "message": f"{document.filename} is not valid base64.",
-                }
             if len(data) > MAX_UPLOAD_BYTES:
-                return {
+                return {"ok": False, "code": "upload_too_large", "comparison_id": comparison_id, "message": f"{document.filename} exceeds {MAX_UPLOAD_BYTES} bytes."}
-                    "ok": False,
+            parsed = _parse_uploaded_quotes(document.filename, document.media_type, data)
-                    "code": "upload_too_large",
-                    "comparison_id": comparison_id,
-                    "message": f"{document.filename} exceeds {MAX_UPLOAD_BYTES} bytes.",
-                }
-            parsed = _parse_uploaded_quotes(
-                filename=document.filename,
-                media_type=document.media_type,
-                data=data,
-            )
             if not parsed["ok"]:
                 return {**parsed, "comparison_id": comparison_id}
             all_quotes.extend(parsed["quotes"])
         return await self.compare_quotes(ctx, comparison_id, all_quotes, weights)
 
 
-# ---------- deterministic business logic ----------
-
-
 def _clean_comparison_id(value: str) -> str | None:
     cleaned = str(value or "").strip()
     if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}", cleaned):
@@ -300,91 +270,45 @@ def _clean_comparison_id(value: str) -> str | None:
     return cleaned
 
 
-def _validate_request(
+def _validate_request(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any] | None:
-    comparison_id: str,
-    quotes: list[QuoteInput],
-    weights: QuoteWeights,
-) -> dict[str, Any] | None:
     clean_id = _clean_comparison_id(comparison_id)
     if clean_id is None:
-        return {
+        return {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id, "message": "comparison_id must be 1-120 URL-safe characters"}
-            "ok": False,
-            "code": "invalid_comparison_id",
-            "comparison_id": comparison_id,
-            "message": "comparison_id must be 1-120 URL-safe characters",
-        }
     if len(quotes) < 2:
-        return {
+        return {"ok": False, "code": "at_least_two_quotes_required", "comparison_id": clean_id, "message": "Compare at least two vendor quotes."}
-            "ok": False,
-            "code": "at_least_two_quotes_required",
-            "comparison_id": clean_id,
-            "message": "Compare at least two vendor quotes.",
-        }
     if len(quotes) > MAX_QUOTES:
-        return {
+        return {"ok": False, "code": "too_many_quotes", "comparison_id": clean_id, "message": f"Compare at most {MAX_QUOTES} quotes at a time."}
-            "ok": False,
-            "code": "too_many_quotes",
-            "comparison_id": clean_id,
-            "message": f"Compare at most {MAX_QUOTES} quotes at a time.",
-        }
     if (weights.price + weights.delivery + weights.warranty) <= 0:
-        return {
+        return {"ok": False, "code": "weights_total_zero", "comparison_id": clean_id, "message": "At least one weight must be greater than zero."}
-            "ok": False,
-            "code": "weights_total_zero",
-            "comparison_id": clean_id,
-            "message": "At least one weight must be greater than zero.",
-        }
     return None
 
 
-def _score_quotes(
+def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
-    comparison_id: str,
-    quotes: list[QuoteInput],
-    weights: QuoteWeights,
-) -> dict[str, Any]:
     clean_id = _clean_comparison_id(comparison_id) or comparison_id
     min_unit_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) or 1
     weight_total = weights.price + weights.delivery + weights.warranty
-
     scored: list[dict[str, Any]] = []
     for quote in quotes:
-        price_score = min_unit_price / quote.unit_price if quote.unit_price else 0
+        price_score = min_unit_price / quote.unit_price
-        delivery_score = min_delivery / quote.delivery_days if quote.delivery_days else 0
+        delivery_score = min_delivery / quote.delivery_days
         warranty_score = quote.warranty_months / max_warranty if max_warranty else 0
-        weighted_score = (
+        weighted_score = (price_score * weights.price + delivery_score * weights.delivery + warranty_score * weights.warranty) / weight_total * 100
-            price_score * weights.price
-            + delivery_score * weights.delivery
-            + warranty_score * weights.warranty
-        ) / weight_total * 100
         total_price = Decimal(str(quote.unit_price)) * Decimal(quote.quantity)
-        scored.append(
+        rounded_score = _round(weighted_score)
-            {
+        scored.append({
             "vendor": quote.vendor,
             "unit_price": _money(quote.unit_price),
             "quantity": quote.quantity,
             "total_price": _money(total_price),
             "delivery_days": quote.delivery_days,
             "warranty_months": quote.warranty_months,
-                "score": _round(weighted_score),
+            "score": rounded_score,
-                "score_breakdown": {
+            "weighted_score": rounded_score,
-                    "price": _round(price_score * 100),
+            "score_breakdown": {"price": _round(price_score * 100), "delivery": _round(delivery_score * 100), "warranty": _round(warranty_score * 100)},
-                    "delivery": _round(delivery_score * 100),
+        })
-                    "warranty": _round(warranty_score * 100),
+    ranked = sorted(scored, key=lambda row: (-row["score"], row["total_price"], row["delivery_days"], -row["warranty_months"], row["vendor"].lower()))
-                },
-            }
-        )
-    ranked = sorted(
-        scored,
-        key=lambda row: (
-            -row["score"],
-            row["total_price"],
-            row["delivery_days"],
-            -row["warranty_months"],
-            row["vendor"].lower(),
-        ),
-    )
     winner = ranked[0]
     result = {
         "ok": True,
@@ -392,19 +316,13 @@ def _score_quotes(
         "recommendation": {
             "vendor": winner["vendor"],
             "score": winner["score"],
+            "weighted_score": winner["score"],
             "total_price": winner["total_price"],
             "delivery_days": winner["delivery_days"],
             "warranty_months": winner["warranty_months"],
-            "rationale": (
+            "rationale": f"{winner['vendor']} has the highest weighted score using price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}.",
-                f"{winner['vendor']} has the highest weighted score using "
-                f"price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}."
-            ),
-        },
-        "weights": {
-            "price": weights.price,
-            "delivery": weights.delivery,
-            "warranty": weights.warranty,
         },
+        "weights": {"price": weights.price, "delivery": weights.delivery, "warranty": weights.warranty},
         "quotes": scored,
         "ranked_quotes": ranked,
         "saved": True,
@@ -412,6 +330,7 @@ def _score_quotes(
         "generated_at": datetime.now(timezone.utc).isoformat(),
     }
     result["execution_receipt"] = _receipt_payload("compare_quotes", clean_id, result)
+    result["receipt"] = result["execution_receipt"]
     return result
 
 
@@ -424,22 +343,9 @@ def _round(value: Any) -> float:
 
 
 def _receipt_payload(skill_name: str, comparison_id: str, payload: dict[str, Any]) -> dict[str, Any]:
-    public_payload = {k: v for k, v in payload.items() if k != "execution_receipt"}
+    public_payload = {k: v for k, v in payload.items() if k not in {"execution_receipt", "receipt"}}
-    digest = hashlib.sha256(
+    digest = hashlib.sha256(json.dumps(public_payload, sort_keys=True, default=str).encode("utf-8")).hexdigest()
-        json.dumps(public_payload, sort_keys=True, default=str).encode("utf-8")
+    return {"agent": "quote-judge-studio-v1", "agent_version": VERSION, "skill": skill_name, "comparison_id": comparison_id, "payload_sha256": digest, "created_at": datetime.now(timezone.utc).isoformat(), "persisted": True}
-    ).hexdigest()
-    return {
-        "agent": "quote-judge-studio-v1",
-        "agent_version": VERSION,
-        "skill": skill_name,
-        "comparison_id": comparison_id,
-        "payload_sha256": digest,
-        "created_at": datetime.now(timezone.utc).isoformat(),
-        "persisted": True,
-    }
-
-
-# ---------- uploads ----------
 
 
 def _parse_uploaded_quotes(filename: str, media_type: str, data: bytes) -> dict[str, Any]:
@@ -459,11 +365,7 @@ def _parse_uploaded_quotes(filename: str, media_type: str, data: bytes) -> dict[
         else:
             quotes = _parse_csv_or_text(text)
     except Exception as exc:  # noqa: BLE001
-        return {
+        return {"ok": False, "code": "quote_parse_failed", "message": f"Could not parse quote upload: {str(exc)[:160]}"}
-            "ok": False,
-            "code": "quote_parse_failed",
-            "message": f"Could not parse quote upload: {str(exc)[:160]}",
-        }
     return {"ok": True, "quotes": quotes}
 
 
@@ -475,12 +377,7 @@ def _parse_csv_or_text(text: str) -> list[QuoteInput]:
     required = {"vendor", "unit_price", "quantity", "delivery_days", "warranty_months"}
     if reader.fieldnames and required.issubset({name.strip() for name in reader.fieldnames}):
         return [QuoteInput.model_validate({k.strip(): v for k, v in row.items()}) for row in reader]
-    raise ValueError(
+    raise ValueError("CSV must include vendor, unit_price, quantity, delivery_days, warranty_months columns")
-        "CSV must include vendor, unit_price, quantity, delivery_days, warranty_months columns"
-    )
-
-
-# ---------- tenant and database ----------
 
 
 def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
@@ -512,7 +409,6 @@ def _connect():
 
 async def _save_comparison(tenant_key: str, result: dict[str, Any]) -> None:
     import asyncio
-
     await asyncio.to_thread(_save_comparison_sync, tenant_key, result)
 
 
@@ -522,22 +418,22 @@ def _save_comparison_sync(tenant_key: str, result: dict[str, Any]) -> None:
         with conn.cursor() as cur:
             cur.execute(
                 """
-                INSERT INTO quote_comparisons (
+                INSERT INTO quote_judge_comparisons (
-                    tenant_key, comparison_id, recommendation_vendor, payload, updated_at
+                    tenant_key, comparison_id, recommendation_vendor, result_json, latest_receipt_id, updated_at
-                ) VALUES (%s, %s, %s, %s::jsonb, NOW())
+                ) VALUES (%s, %s, %s, %s::jsonb, %s, NOW())
                 ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
                     recommendation_vendor = EXCLUDED.recommendation_vendor,
-                    payload = EXCLUDED.payload,
+                    result_json = EXCLUDED.result_json,
+                    latest_receipt_id = EXCLUDED.latest_receipt_id,
                     updated_at = NOW()
                 """,
-                (tenant_key, result["comparison_id"], result["recommendation"]["vendor"], payload),
+                (tenant_key, result["comparison_id"], result["recommendation"]["vendor"], payload, result.get("execution_receipt", {}).get("payload_sha256")),
             )
         conn.commit()
 
 
 async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None:
     import asyncio
-
     return await asyncio.to_thread(_load_comparison_sync, tenant_key, comparison_id)
 
 
@@ -546,8 +442,8 @@ def _load_comparison_sync(tenant_key: str, comparison_id: str) -> dict[str, Any]
         with conn.cursor() as cur:
             cur.execute(
                 """
-                SELECT payload
+                SELECT result_json
-                  FROM quote_comparisons
+                  FROM quote_judge_comparisons
                  WHERE tenant_key = %s AND comparison_id = %s
                  LIMIT 1
                 """,
@@ -556,73 +452,39 @@ def _load_comparison_sync(tenant_key: str, comparison_id: str) -> dict[str, Any]
             row = cur.fetchone()
     if not row:
         return None
-    payload = row["payload"]
+    payload = row["result_json"]
-    if isinstance(payload, str):
+    return json.loads(payload) if isinstance(payload, str) else dict(payload)
-        return json.loads(payload)
-    return dict(payload)
 
 
-async def _persist_receipt_best_effort(
+async def _persist_receipt_best_effort(ctx: RunContext[PlatformUserAuth], skill_name: str, comparison_id: str, status: str, payload: dict[str, Any]) -> None:
-    ctx: RunContext[PlatformUserAuth],
-    skill_name: str,
-    comparison_id: str,
-    status: str,
-    payload: dict[str, Any],
-) -> None:
     import asyncio
-
     try:
         tenant = _tenant_key(ctx)
         receipt = _receipt_payload(skill_name, comparison_id, payload)
-        await asyncio.to_thread(
+        await asyncio.to_thread(_persist_receipt_sync, tenant, comparison_id, skill_name, status, getattr(ctx, "task_id", "") or "", receipt)
-            _persist_receipt_sync,
-            tenant,
-            comparison_id,
-            skill_name,
-            status,
-            getattr(ctx, "task_id", "") or "",
-            receipt,
-        )
     except Exception:  # noqa: BLE001
-        # Persistence of the comparison itself fails closed; duplicate receipt
-        # persistence must not mask the primary skill result.
         return
 
 
-def _persist_receipt_sync(
+def _persist_receipt_sync(tenant_key: str, comparison_id: str, skill_name: str, status: str, task_id: str, receipt: dict[str, Any]) -> None:
-    tenant_key: str,
+    receipt_json = json.dumps(receipt, default=str)
-    comparison_id: str,
-    skill_name: str,
-    status: str,
-    task_id: str,
-    receipt: dict[str, Any],
-) -> None:
     with _connect() as conn:
         with conn.cursor() as cur:
             cur.execute(
                 """
-                INSERT INTO quote_execution_receipts (
+                INSERT INTO quote_judge_receipts (
-                    tenant_key, comparison_id, skill_name, status, task_id, receipt
+                    receipt_id, tenant_key, comparison_id, skill_name, tool_name, status,
-                ) VALUES (%s, %s, %s, %s, %s, %s::jsonb)
+                    input_hash, result_hash, receipt_json, payload
+                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb)
                 """,
-                (tenant_key, comparison_id, skill_name, status, task_id, json.dumps(receipt, default=str)),
+                (receipt.get("payload_sha256"), tenant_key, comparison_id, skill_name, skill_name, status, receipt.get("payload_sha256"), receipt.get("payload_sha256"), receipt_json, receipt_json),
             )
         conn.commit()
 
 
 def _setup_required(comparison_id: str, message: str) -> dict[str, Any]:
-    return {
+    return {"ok": False, "code": "setup_required", "comparison_id": comparison_id, "message": message}
-        "ok": False,
-        "code": "setup_required",
-        "comparison_id": comparison_id,
-        "message": message,
-    }
 
 
 def _temporary_db_error(comparison_id: str) -> dict[str, Any]:
-    return {
+    return {"ok": False, "code": "database_unavailable", "comparison_id": comparison_id, "message": "QuoteJudge could not persist or read the comparison before the database timeout."}
-        "ok": False,
-        "code": "database_unavailable",
-        "comparison_id": comparison_id,
-        "message": "QuoteJudge could not persist or read the comparison before the database timeout.",
-    }