a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,620 @@
-"""quote-judge-studio-v1 agent.
+"""QuoteJudge Studio full-stack A2A agent.
 
-Starter stack:
+Deterministic quote comparison product with platform-user tenant isolation,
-  - DeepAgents for tool-calling orchestration
+managed Postgres persistence, bounded browser uploads, typed external file
-  - Caller-provided LLM credentials via ctx.llm
+uploads, MCP-compatible tools, and execution receipt rows.
-  - A tiny model-call middleware hook you can replace with tracing,
-    routing, rate limits, or policy checks
 """
 from __future__ import annotations
 
+import base64
+import csv
+import hashlib
+import io
 import json
-from pathlib import Path
+import os
-from typing import Any
+import re
+from datetime import datetime, timezone
+from decimal import Decimal, ROUND_HALF_UP
+from typing import Annotated, Any
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    LLMProvisioning,
+    AgentDatabase,
-    {{ auth_type }},
+    AgentDatabaseEnv,
+    AgentDatabaseMigrations,
+    AgentPlatformResources,
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
-    WorkspaceAccess,
-    WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+from a2a_pack.workspace import FileUpload, UploadedFile
+
+DATABASE_NAME = "quote-judge-studio-v1-data"
+DB_CONNECT_TIMEOUT_SECONDS = 3
+DB_STATEMENT_TIMEOUT_MS = 5000
+DB_LOCK_TIMEOUT_MS = 3000
+MAX_QUOTES = 20
+MAX_UPLOAD_BYTES = 64_000
+MAX_BROWSER_UPLOADS = 3
+ALLOWED_UPLOAD_MEDIA_TYPES = ("text/csv", "text/plain", "application/json")
+VERSION = "0.1.0"
 
 
 class QuoteJudgeStudioV1Config(BaseModel):
     pass
 
 
-SYSTEM_PROMPT = """\
+class QuoteInput(BaseModel):
-You are a compact tool-calling agent.
+    vendor: str = Field(min_length=1, max_length=120)
+    unit_price: float = Field(gt=0, le=1_000_000)
+    quantity: int = Field(gt=0, le=10_000_000)
+    delivery_days: int = Field(gt=0, le=3650)
+    warranty_months: int = Field(ge=0, le=600)
 
-Use the text_stats tool when the user asks about text, counts, summaries,
+    @field_validator("vendor")
-or anything where exact length/word numbers would help. Mention tool results
+    @classmethod
-briefly instead of dumping raw JSON.
+    def clean_vendor(cls, value: str) -> str:
-"""
+        cleaned = re.sub(r"\s+", " ", value).strip()
-
+        if not cleaned:
-RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/"
+            raise ValueError("vendor is required")
-DEEPAGENTS_RECURSION_LIMIT = 500
+        return cleaned
 
 
-class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
+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=120)
+    data_base64: str = Field(min_length=1, max_length=MAX_UPLOAD_BYTES * 2)
+
+    @field_validator("filename")
+    @classmethod
+    def safe_filename(cls, value: str) -> str:
+        cleaned = value.replace("\\", "/").split("/")[-1].strip()
+        if not cleaned or cleaned in {".", ".."}:
+            raise ValueError("filename is required")
+        return cleaned[:160]
+
+    @field_validator("media_type")
+    @classmethod
+    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(
+                "unsupported media_type; use text/csv, text/plain, or application/json"
+            )
+        return cleaned
+
+
+class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     name = "quote-judge-studio-v1"
-    description = "One-page QuoteJudge full-stack startup for comparing vendor quotes with managed persistence."
+    description = (
-    version = "0.1.0"
+        "Compare vendor quotes by price, delivery, and warranty; persist the "
+        "winner per signed-in A2A Cloud user; reopen saved comparisons."
+    )
+    version = VERSION
 
     config_model = QuoteJudgeStudioV1Config
-    auth_model = {{ auth_type }}
+    auth_model = PlatformUserAuth
 
-    # Hosted generated agents read the caller's saved LLM credential through
+    # Deterministic local logic only. No LLM credential is required.
-    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
-    # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
-    # directly.
-    llm_provisioning = LLMProvisioning.PLATFORM
     pricing = Pricing(
         price_per_call_usd=0.0,
-        caller_pays_llm=True,
+        caller_pays_llm=False,
-        notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
+        notes="Deterministic QuoteJudge comparison; no LLM is called.",
     )
-    workspace_access = WorkspaceAccess.dynamic(
+    resources = Resources(cpu="250m", memory="512Mi", max_runtime_seconds=120)
-        max_files=64,
+    tools_used = ("postgres", "managed-postgres", "mcp", "packed-frontend")
-        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
+    platform_resources = AgentPlatformResources(
-        require_reason=False,
+        databases=(
-    )
+            AgentDatabase(
-    tools_used = ("deepagents", "langchain")
+                name=DATABASE_NAME,
-
+                provider="neon",
-    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
+                engine="postgres",
-    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
+                scope="user",
-        creds = ctx.llm
+                branch="main",
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+                access_mode="read_write",
-        if not creds.api_key:
+                env=AgentDatabaseEnv(url="DATABASE_URL"),
-            return (
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
-                "LLM key required. Add an LLM credential in Settings > LLM "
+                scale_to_zero=True,
-                "credentials before running this agent; for local --invoke "
+            ),
-                "runs set AGENT_LLM_KEY."
-            )
-        graph = self._build_deep_agent(ctx=ctx, creds=creds)
-        state = await graph.ainvoke(
-            {"messages": [{"role": "user", "content": prompt}]},
-            config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
         )
-        await ctx.emit_progress("deepagent finished")
+    )
-        return _last_message_text(state)
 
-    def _build_deep_agent(
+    @a2a.tool(
+        description="Compare two or more vendor quotes, persist the result for this platform user, and return the recommendation.",
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def compare_quotes(
         self,
-        *,
+        ctx: RunContext[PlatformUserAuth],
-        ctx: RunContext[{{ auth_type }}],
+        comparison_id: str,
-        creds: LLMCreds,
+        quotes: list[QuoteInput],
-    ) -> Any:
+        weights: QuoteWeights,
-        # Lazy imports keep `a2a card` usable before local dependencies are
+    ) -> dict[str, Any]:
-        # installed. `a2a deploy` installs requirements.txt during the build.
+        validation = _validate_request(comparison_id, quotes, weights)
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        if validation is not None:
-        from langchain.agents.middleware import wrap_model_call
+            await _persist_receipt_best_effort(ctx, "compare_quotes", comparison_id, "error", validation)
-        from langchain_core.tools import tool
+            return validation
 
-        @tool
+        result = _score_quotes(comparison_id, quotes, weights)
-        def text_stats(text: str) -> str:
+        tenant = _tenant_key(ctx)
-            """Return exact word, character, and line counts for text."""
+        try:
-            words = [part for part in text.split() if part.strip()]
+            await _save_comparison(tenant, result)
-            return json.dumps(
+            await _persist_receipt_best_effort(ctx, "compare_quotes", comparison_id, "ok", result)
-                {
+        except RuntimeError as exc:
-                    "characters": len(text),
+            return _setup_required(comparison_id, str(exc))
-                    "words": len(words),
+        except Exception:  # noqa: BLE001
-                    "lines": len(text.splitlines()) or 1,
+            return _temporary_db_error(comparison_id)
-                }
+        await ctx.emit_progress(f"saved comparison {comparison_id}")
-            )
+        return result
 
-        @wrap_model_call
+    @a2a.tool(
-        async def log_model_call(request: Any, handler: Any) -> Any:
+        description="Reload a saved quote comparison for the current signed-in platform user.",
-            messages = request.state.get("messages", [])
+        timeout_seconds=20,
-            print(
+        idempotent=True,
-                "[middleware] model_call "
+        cost_class="deterministic",
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+    )
-            )
+    async def get_comparison(
-            return await handler(request)
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+    ) -> dict[str, Any]:
+        tenant = _tenant_key(ctx)
+        clean_id = _clean_comparison_id(comparison_id)
+        if clean_id is None:
+            return {
+                "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})
+        except RuntimeError as exc:
+            return _setup_required(clean_id, str(exc))
+        except Exception:  # noqa: BLE001
+            return _temporary_db_error(clean_id)
+        if record is None:
+            return {
+                "ok": False,
+                "code": "comparison_not_found",
+                "comparison_id": clean_id,
+                "message": "No saved comparison exists for this user and id.",
+            }
+        return record
 
-        backend = ctx.workspace_backend()
+    @a2a.tool(
-        skill_sources = _seed_runtime_skills(backend, ctx)
+        description="Compare quotes from an external client FileUpload (CSV, JSON, or plain text) and persist the result.",
-        # create_a2a_deep_agent resolves provider:model strings with
+        timeout_seconds=30,
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
+        idempotent=True,
-        # provider-specific extra body, and runtime model overrides.
+        cost_class="deterministic",
-        return create_a2a_deep_agent(
+    )
-            ctx,
+    async def compare_quotes_file(
-            creds=creds,
+        self,
-            backend=backend,
+        ctx: RunContext[PlatformUserAuth],
-            skills=skill_sources or None,
+        comparison_id: str,
-            tools=[text_stats],
+        weights: QuoteWeights,
-            middleware=[log_model_call],
+        document: Annotated[
-            system_prompt=SYSTEM_PROMPT,
+            UploadedFile,
+            FileUpload(
+                accept=ALLOWED_UPLOAD_MEDIA_TYPES,
+                max_bytes=MAX_UPLOAD_BYTES,
+                description="CSV/JSON/text quote file with vendor, unit_price, quantity, delivery_days, warranty_months.",
+            ),
+        ],
+    ) -> dict[str, Any]:
+        try:
+            payload = ctx.workspace.read_bytes(document.path)  # type: ignore[attr-defined]
+        except Exception:  # noqa: BLE001
+            return {
+                "ok": False,
+                "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 bounded base64 quote files, parse them, compare quotes, and persist the result.",
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def compare_quotes_browser_upload(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        weights: QuoteWeights,
+        documents: list[BrowserQuoteDocument],
+    ) -> dict[str, Any]:
+        if not documents or len(documents) > MAX_BROWSER_UPLOADS:
+            return {
+                "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 {
+                    "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 {
+                    "ok": False,
+                    "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)
 
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+# ---------- deterministic business logic ----------
-    workspace = getattr(ctx, "_workspace", None)
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
-    if not prefixes:
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
-        prefixes = (outputs_prefix or "outputs/",)
-    prefix = str(prefixes[0]).strip("/")
-    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
 
 
-def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
+def _clean_comparison_id(value: str) -> str | None:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    cleaned = str(value or "").strip()
-
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}", cleaned):
-    DeepAgents loads skills from its backend, while source-controlled
+        return None
-    ``skills/`` folders live in the image. This bridge lets generated agents
+    return cleaned
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
-    files.
-    """
-    root = Path(__file__).parent / "skills"
-    if not root.exists():
-        return []
-    runtime_skills_root = _runtime_skills_root(ctx)
-    uploads: list[tuple[str, bytes]] = []
-    for path in root.rglob("*"):
-        if path.is_file():
-            rel = path.relative_to(root).as_posix()
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
-    if uploads:
-        backend.upload_files(uploads)
-        return [runtime_skills_root]
-    return []
 
 
-def _last_message_text(state: dict[str, Any]) -> str:
+def _validate_request(
-    messages = state.get("messages") or []
+    comparison_id: str,
-    if not messages:
+    quotes: list[QuoteInput],
-        return json.dumps(state, default=str)
+    weights: QuoteWeights,
+) -> dict[str, Any] | None:
+    clean_id = _clean_comparison_id(comparison_id)
+    if clean_id is None:
+        return {
+            "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 {
+            "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 {
+            "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 {
+            "ok": False,
+            "code": "weights_total_zero",
+            "comparison_id": clean_id,
+            "message": "At least one weight must be greater than zero.",
+        }
+    return None
 
-    content = getattr(messages[-1], "content", None)
+
-    if isinstance(content, str):
+def _score_quotes(
-        return content
+    comparison_id: str,
-    if isinstance(content, list):
+    quotes: list[QuoteInput],
-        parts: list[str] = []
+    weights: QuoteWeights,
-        for item in content:
+) -> dict[str, Any]:
-            if isinstance(item, dict):
+    clean_id = _clean_comparison_id(comparison_id) or comparison_id
-                text = item.get("text") or item.get("content")
+    min_unit_price = min(q.unit_price for q in quotes)
-                if text:
+    min_delivery = min(q.delivery_days for q in quotes)
-                    parts.append(str(text))
+    max_warranty = max(q.warranty_months for q in quotes) or 1
-            elif item:
+    weight_total = weights.price + weights.delivery + weights.warranty
-                parts.append(str(item))
+
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+    scored: list[dict[str, Any]] = []
-    return str(content or messages[-1])
+    for quote in quotes:
+        price_score = min_unit_price / quote.unit_price if quote.unit_price else 0
+        delivery_score = min_delivery / quote.delivery_days if quote.delivery_days else 0
+        warranty_score = quote.warranty_months / max_warranty if max_warranty else 0
+        weighted_score = (
+            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(
+            {
+                "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_breakdown": {
+                    "price": _round(price_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(),
+        ),
+    )
+    winner = ranked[0]
+    result = {
+        "ok": True,
+        "comparison_id": clean_id,
+        "recommendation": {
+            "vendor": winner["vendor"],
+            "score": winner["score"],
+            "total_price": winner["total_price"],
+            "delivery_days": winner["delivery_days"],
+            "warranty_months": winner["warranty_months"],
+            "rationale": (
+                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,
+        },
+        "quotes": scored,
+        "ranked_quotes": ranked,
+        "saved": True,
+        "version": VERSION,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+    }
+    result["execution_receipt"] = _receipt_payload("compare_quotes", clean_id, result)
+    return result
+
+
+def _money(value: Any) -> float:
+    return float(Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
+
+
+def _round(value: Any) -> float:
+    return float(Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
+
+
+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"}
+    digest = hashlib.sha256(
+        json.dumps(public_payload, sort_keys=True, default=str).encode("utf-8")
+    ).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]:
+    safe_media_type = media_type.split(";", 1)[0].strip().lower()
+    if safe_media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
+        return {"ok": False, "code": "unsupported_upload_media_type"}
+    if len(data) > MAX_UPLOAD_BYTES:
+        return {"ok": False, "code": "upload_too_large"}
+    text = data.decode("utf-8-sig", errors="replace")
+    try:
+        if safe_media_type == "application/json" or filename.lower().endswith(".json"):
+            raw = json.loads(text)
+            rows = raw.get("quotes", raw) if isinstance(raw, dict) else raw
+            if not isinstance(rows, list):
+                raise ValueError("JSON upload must be a list or {'quotes': [...]} object")
+            quotes = [QuoteInput.model_validate(row) for row in rows]
+        else:
+            quotes = _parse_csv_or_text(text)
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "ok": False,
+            "code": "quote_parse_failed",
+            "message": f"Could not parse quote upload: {str(exc)[:160]}",
+        }
+    return {"ok": True, "quotes": quotes}
+
+
+def _parse_csv_or_text(text: str) -> list[QuoteInput]:
+    sample = text.strip()
+    if not sample:
+        raise ValueError("file is empty")
+    reader = csv.DictReader(io.StringIO(sample))
+    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(
+        "CSV must include vendor, unit_price, quantity, delivery_days, warranty_months columns"
+    )
+
+
+# ---------- tenant and database ----------
+
+
+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)
+    if not stable:
+        raise RuntimeError("authenticated platform user is required")
+    return f"user:{stable}"
+
+
+def _database_url() -> str:
+    url = os.environ.get("DATABASE_URL", "").strip()
+    if not url:
+        raise RuntimeError("Managed Postgres is not configured; DATABASE_URL is missing.")
+    return url
+
+
+def _connect():
+    import psycopg
+    from psycopg.rows import dict_row
+
+    return psycopg.connect(
+        _database_url(),
+        connect_timeout=DB_CONNECT_TIMEOUT_SECONDS,
+        row_factory=dict_row,
+        options=f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} -c lock_timeout={DB_LOCK_TIMEOUT_MS}",
+    )
+
+
+async def _save_comparison(tenant_key: str, result: dict[str, Any]) -> None:
+    import asyncio
+
+    await asyncio.to_thread(_save_comparison_sync, tenant_key, result)
+
+
+def _save_comparison_sync(tenant_key: str, result: dict[str, Any]) -> None:
+    payload = json.dumps(result, default=str)
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                INSERT INTO quote_comparisons (
+                    tenant_key, comparison_id, recommendation_vendor, payload, updated_at
+                ) VALUES (%s, %s, %s, %s::jsonb, NOW())
+                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
+                    recommendation_vendor = EXCLUDED.recommendation_vendor,
+                    payload = EXCLUDED.payload,
+                    updated_at = NOW()
+                """,
+                (tenant_key, result["comparison_id"], result["recommendation"]["vendor"], payload),
+            )
+        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)
+
+
+def _load_comparison_sync(tenant_key: str, comparison_id: str) -> dict[str, Any] | None:
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT payload
+                  FROM quote_comparisons
+                 WHERE tenant_key = %s AND comparison_id = %s
+                 LIMIT 1
+                """,
+                (tenant_key, comparison_id),
+            )
+            row = cur.fetchone()
+    if not row:
+        return None
+    payload = row["payload"]
+    if isinstance(payload, str):
+        return json.loads(payload)
+    return dict(payload)
+
+
+async def _persist_receipt_best_effort(
+    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(
+            _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(
+    tenant_key: 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 (
+                    tenant_key, comparison_id, skill_name, status, task_id, receipt
+                ) VALUES (%s, %s, %s, %s, %s, %s::jsonb)
+                """,
+                (tenant_key, comparison_id, skill_name, status, task_id, json.dumps(receipt, default=str)),
+            )
+        conn.commit()
+
+
+def _setup_required(comparison_id: str, message: str) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "code": "setup_required",
+        "comparison_id": comparison_id,
+        "message": message,
+    }
+
+
+def _temporary_db_error(comparison_id: str) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "code": "database_unavailable",
+        "comparison_id": comparison_id,
+        "message": "QuoteJudge could not persist or read the comparison before the database timeout.",
+    }