a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,565 @@
-"""quote-judge-studio-v1 agent.
+"""QuoteJudge Studio full-stack A2A agent.
 
-Starter stack:
+Deterministic, no-LLM quote comparison product:
-  - DeepAgents for tool-calling orchestration
+- platform-authenticated user tenancy;
-  - Caller-provided LLM credentials via ctx.llm
+- managed Postgres persistence;
-  - A tiny model-call middleware hook you can replace with tracing,
+- typed A2A/MCP tools for comparison, reload, browser uploads, API uploads;
-    routing, rate limits, or policy checks
+- production receipt records without secrets.
 """
 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 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,
+    FileUpload,
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
+    State,
+    UploadedFile,
     WorkspaceAccess,
     WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+
+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):
-    pass
+    """No user-editable config is required."""
 
 
-SYSTEM_PROMPT = """\
+class QuoteInput(BaseModel):
-You are a compact tool-calling agent.
+    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)]
 
-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: 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."
+    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 = {{ auth_type }}
+    auth_model = PlatformUserAuth
 
-    # Hosted generated agents read the caller's saved LLM credential through
+    state = State.DURABLE
-    # 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 quote scoring. No LLM credential is required.",
     )
+    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
     workspace_access = WorkspaceAccess.dynamic(
-        max_files=64,
+        max_files=8,
         allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
         require_reason=False,
+        max_total_size_bytes=512 * 1024,
     )
-    tools_used = ("deepagents", "langchain")
+    tools_used = ("postgres", "mcp", "packed-frontend")
-
+    platform_resources = AgentPlatformResources(
-    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
+        databases=(
-    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
+            AgentDatabase(
-        creds = ctx.llm
+                name="quote-judge-studio-v1-data",
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+                scope="user",
-        if not creds.api_key:
+                access_mode="read_write",
-            return (
+                env=AgentDatabaseEnv(url=DATABASE_ENV),
-                "LLM key required. Add an LLM credential in Settings > LLM "
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
-                "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 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],
-        ctx: RunContext[{{ auth_type }}],
+        comparison_id: Annotated[str, Field(min_length=1, max_length=120)],
-        creds: LLMCreds,
+        quotes: Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)],
-    ) -> 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.
+        await ctx.emit_progress("Validating quote comparison inputs")
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        tenant = _tenant_key(ctx)
-        from langchain.agents.middleware import wrap_model_call
+        result = _build_comparison_result(comparison_id, quotes, weights)
-        from langchain_core.tools import tool
+        if not result["ok"]:
+            await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "validation_error", {"code": result["code"]}, result)
+            return result
 
-        @tool
+        await ctx.emit_progress("Persisting comparison and production receipt")
-        def text_stats(text: str) -> str:
+        await _save_comparison(tenant, comparison_id, result)
-            """Return exact word, character, and line counts for text."""
+        receipt = await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "ok", _safe_input_payload(comparison_id, quotes, weights), result)
-            words = [part for part in text.split() if part.strip()]
+        result["receipt"] = receipt
-            return json.dumps(
+        return result
-                {
+
-                    "characters": len(text),
+    @a2a.tool(
-                    "words": len(words),
+        description="Reopen a previously persisted quote comparison by id.",
-                    "lines": len(text.splitlines()) or 1,
+        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:
-        @wrap_model_call
+            return _validation_error(
-        async def log_model_call(request: Any, handler: Any) -> Any:
+                "unsupported_media_type",
-            messages = request.state.get("messages", [])
+                f"Unsupported media type {quote_file.media_type!r}. Use JSON, CSV, or plain text.",
-            print(
+                comparison_id=comparison_id,
-                "[middleware] model_call "
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
             )
-            return await handler(request)
+        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)
 
-        backend = ctx.workspace_backend()
+
-        skill_sources = _seed_runtime_skills(backend, ctx)
+def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
-        # create_a2a_deep_agent resolves provider:model strings with
+    auth = ctx.auth
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
+    stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
-        # provider-specific extra body, and runtime model overrides.
+    return f"user:{stable}"
-        return create_a2a_deep_agent(
+
-            ctx,
+
-            creds=creds,
+def _safe_input_payload(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
-            backend=backend,
+    return {
-            skills=skill_sources or None,
+        "comparison_id": comparison_id,
-            tools=[text_stats],
+        "quote_count": len(quotes),
-            middleware=[log_model_call],
+        "vendors": [quote.vendor for quote in quotes],
-            system_prompt=SYSTEM_PROMPT,
+        "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,
+    }
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+    rows: list[dict[str, Any]] = []
-    workspace = getattr(ctx, "_workspace", None)
+    for quote in quotes:
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+        price_score = (min_price / quote.unit_price) * 100 if quote.unit_price else 0.0
-    if not prefixes:
+        delivery_score = 100.0 if quote.delivery_days == 0 and min_delivery == 0 else ((min_delivery + 1) / (quote.delivery_days + 1)) * 100
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+        warranty_score = (quote.warranty_months / max_warranty) * 100 if max_warranty else 100.0
-        prefixes = (outputs_prefix or "outputs/",)
+        weighted_score = (
-    prefix = str(prefixes[0]).strip("/")
+            normalized_weights["price"] * price_score
-    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+            + 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=lambda row: (-row["scores"]["weighted_total"], row["total_price"], row["delivery_days"], row["vendor"].lower()))
+    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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
+def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    quotes: list[QuoteInput] = []
-
+    errors: list[dict[str, Any]] = []
-    DeepAgents loads skills from its backend, while source-controlled
+    for index, doc in enumerate(documents):
-    ``skills/`` folders live in the image. This bridge lets generated agents
+        if doc.media_type not in ACCEPTED_UPLOAD_TYPES:
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+            errors.append({"field": f"documents[{index}].media_type", "message": "Use JSON, CSV, or plain text."})
-    files.
+            continue
-    """
+        try:
-    root = Path(__file__).parent / "skills"
+            raw = base64.b64decode(doc.data_base64, validate=True)
-    if not root.exists():
+        except Exception:  # noqa: BLE001
-        return []
+            errors.append({"field": f"documents[{index}].data_base64", "message": "Invalid base64 payload."})
-    runtime_skills_root = _runtime_skills_root(ctx)
+            continue
-    uploads: list[tuple[str, bytes]] = []
+        if len(raw) > MAX_BROWSER_DOCUMENT_BYTES:
-    for path in root.rglob("*"):
+            errors.append({"field": f"documents[{index}]", "message": f"File exceeds {MAX_BROWSER_DOCUMENT_BYTES} bytes."})
-        if path.is_file():
+            continue
-            rel = path.relative_to(root).as_posix()
+        parsed = _parse_document_bytes(doc.filename, doc.media_type, raw)
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+        if parsed["ok"]:
-    if uploads:
+            quotes.extend(parsed["quotes"])
-        backend.upload_files(uploads)
+        else:
-        return [runtime_skills_root]
+            errors.extend(parsed.get("validation_errors") or [])
-    return []
+    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 _last_message_text(state: dict[str, Any]) -> str:
+def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]:
-    messages = state.get("messages") or []
+    if len(raw) > MAX_BROWSER_DOCUMENT_BYTES:
-    if not messages:
+        return _validation_error("file_too_large", f"{filename} exceeds the {MAX_BROWSER_DOCUMENT_BYTES} byte limit.")
-        return json.dumps(state, default=str)
+    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__}.")
 
-    content = getattr(messages[-1], "content", None)
+
-    if isinstance(content, str):
+def _parse_loose_text_quotes(text: str) -> list[QuoteInput]:
-        return content
+    quotes: list[QuoteInput] = []
-    if isinstance(content, list):
+    for line in text.splitlines():
-        parts: list[str] = []
+        parts = [part.strip() for part in re.split(r"[,|;]\s*", line) if part.strip()]
-        for item in content:
+        if len(parts) < 5 or parts[0].lower() in {"vendor", "supplier"}:
-            if isinstance(item, dict):
+            continue
-                text = item.get("text") or item.get("content")
+        quotes.append(
-                if text:
+            QuoteInput(
-                    parts.append(str(text))
+                vendor=parts[0],
-            elif item:
+                unit_price=float(parts[1]),
-                parts.append(str(item))
+                quantity=int(float(parts[2])),
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+                delivery_days=int(float(parts[3])),
-    return str(content or messages[-1])
+                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, result)
+                VALUES (%s, %s, %s, %s)
+                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
+                    recommendation_vendor = EXCLUDED.recommendation_vendor,
+                    result = EXCLUDED.result,
+                    updated_at = NOW()
+                """,
+                (tenant_key, comparison_id, result["recommendation"]["vendor"], json.dumps(result)),
+            )
+
+
+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 result 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, recommendation_vendor, updated_at, result
+                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()