a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,747 @@
-"""quote-judge-studio-v1 agent.
+"""QuoteJudge Studio v1: deterministic quote comparison product agent.
 
-Starter stack:
+This agent intentionally does not call an LLM. It exposes a compact typed A2A
-  - DeepAgents for tool-calling orchestration
+surface, a packed React app, managed Postgres persistence, MCP-compatible tools,
-  - Caller-provided LLM credentials via ctx.llm
+and bounded upload paths for both browser JSON and external FileUpload clients.
-  - A tiny model-call middleware hook you can replace with tracing,
-    routing, rate limits, or policy checks
 """
 from __future__ import annotations
 
+import base64
+import binascii
+import csv
+import hashlib
+import io
 import json
-from pathlib import Path
+import os
-from typing import Any
+import re
+import time
+from datetime import UTC, datetime
+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,
     RunContext,
+    UploadedFile,
     WorkspaceAccess,
     WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+
+DATABASE_NAME = "quote-judge-studio-v1-data"
+MAX_QUOTES = 20
+MAX_UPLOAD_FILES = 2
+MAX_UPLOAD_BYTES = 64 * 1024
+MAX_BASE64_CHARS = ((MAX_UPLOAD_BYTES + 2) // 3) * 4 + 4
+ALLOWED_UPLOAD_MEDIA_TYPES = {
+    "application/json",
+    "text/json",
+    "text/csv",
+    "application/csv",
+    "text/plain",
+    "application/octet-stream",
+}
+_COMPARISON_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,79}$")
+_LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {}
+_LOCAL_RECEIPTS: list[dict[str, Any]] = []
 
 
 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(ge=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 = " ".join(str(value).strip().split())
-
+        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=1000)
+    delivery: float = Field(ge=0, le=1000)
+    warranty: float = Field(ge=0, le=1000)
+
+
+class BrowserDocument(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_BASE64_CHARS)
+
+
+QuoteList = Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)]
+DocumentList = Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_UPLOAD_FILES)]
+ComparisonId = Annotated[str, Field(min_length=1, max_length=80)]
+
+
+class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     name = "quote-judge-studio-v1"
-    description = "QuoteJudge Studio compares vendor quotes with weighted price, delivery, and warranty scoring, persists results, supports uploads, and exposes MCP-compatible A2A tools."
+    description = (
+        "QuoteJudge Studio compares vendor quotes with weighted price, delivery, "
+        "and warranty scoring, persists results, supports bounded uploads, and "
+        "exposes typed MCP-compatible A2A 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
-    # 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 and managed Postgres persistence; no LLM required.",
     )
     workspace_access = WorkspaceAccess.dynamic(
-        max_files=64,
+        max_files=4,
-        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
+        allowed_modes=(WorkspaceMode.READ_ONLY,),
         require_reason=False,
+        max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_UPLOAD_FILES,
     )
-    tools_used = ("deepagents", "langchain")
+    platform_resources = AgentPlatformResources(
+        databases=(
+            AgentDatabase(
+                name=DATABASE_NAME,
+                scope="user",
+                access_mode="read_write",
+                env=AgentDatabaseEnv(url="DATABASE_URL"),
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            ),
+        )
+    )
+    tools_used = ("postgres", "mcp", "file-upload")
 
-    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
+    @a2a.tool(
-    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
+        description="Compare at least two vendor quotes with price/delivery/warranty weights and persist a production receipt.",
-        creds = ctx.llm
+        timeout_seconds=30,
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+        cost_class="deterministic",
-        if not creds.api_key:
-            return (
-                "LLM key required. Add an LLM credential in Settings > LLM "
-                "credentials before running this agent; for local --invoke "
-                "runs set AGENT_LLM_KEY."
     )
-        graph = self._build_deep_agent(ctx=ctx, creds=creds)
+    async def compare_quotes(
-        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(
         self,
-        *,
+        ctx: RunContext[PlatformUserAuth],
-        ctx: RunContext[{{ auth_type }}],
+        comparison_id: ComparisonId,
-        creds: LLMCreds,
+        quotes: QuoteList,
-    ) -> 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.
+        tenant = _tenant_key(ctx)
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        validation = _validate_compare_inputs(comparison_id, quotes, weights)
-        from langchain.agents.middleware import wrap_model_call
+        if validation is not None:
-        from langchain_core.tools import tool
+            await _persist_receipt(ctx, tenant, "compare_quotes", {"comparison_id": comparison_id}, validation)
+            return validation
 
-        @tool
+        result = _score_quotes(comparison_id, quotes, weights)
-        def text_stats(text: str) -> str:
+        result["tenant_scope"] = "user"
-            """Return exact word, character, and line counts for text."""
+        result["receipt"] = await _persist_receipt(
-            words = [part for part in text.split() if part.strip()]
+            ctx,
-            return json.dumps(
+            tenant,
+            "compare_quotes",
             {
-                    "characters": len(text),
+                "comparison_id": comparison_id,
-                    "words": len(words),
+                "quote_count": len(quotes),
-                    "lines": len(text.splitlines()) or 1,
+                "weights": weights.model_dump(mode="json"),
+            },
+            result,
+        )
+        saved = await _save_comparison(tenant, result)
+        if not saved["ok"]:
+            return {
+                "ok": False,
+                "code": saved["code"],
+                "message": saved["message"],
+                "comparison_id": comparison_id,
+                "action": "Retry after the managed database is available, or contact the agent owner.",
+            }
+        await ctx.emit_progress(f"saved comparison {comparison_id}")
+        return result
+
+    @a2a.tool(
+        description="Reopen a persisted QuoteJudge comparison by id.",
+        timeout_seconds=15,
+        cost_class="deterministic",
+        idempotent=True,
+    )
+    async def get_comparison(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: ComparisonId,
+    ) -> dict[str, Any]:
+        tenant = _tenant_key(ctx)
+        if not _valid_comparison_id(comparison_id):
+            return _validation_error(
+                comparison_id,
+                "invalid_comparison_id",
+                "comparison_id must start with a letter or number and contain only letters, numbers, dash, underscore, dot, or colon.",
+            )
+        result = await _load_comparison(tenant, comparison_id)
+        if result is None:
+            return {
+                "ok": False,
+                "code": "comparison_not_found",
+                "comparison_id": comparison_id,
+                "message": "No saved comparison exists for this signed-in user and comparison_id.",
+                "action": "Run compare_quotes first, then call get_comparison with the same comparison_id.",
+            }
+        return result
+
+    @a2a.tool(
+        description="Compatibility browser upload bridge: parse bounded base64 JSON/CSV quote files, compare, persist, and receipt the result.",
+        timeout_seconds=30,
+        cost_class="deterministic",
+    )
+    async def upload_quotes(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: ComparisonId,
+        documents: DocumentList,
+        weights: QuoteWeights,
+    ) -> dict[str, Any]:
+        parsed = _parse_browser_documents(documents)
+        if not parsed["ok"]:
+            await _persist_receipt(ctx, _tenant_key(ctx), "upload_quotes", {"comparison_id": comparison_id}, parsed)
+            return parsed | {"comparison_id": comparison_id}
+        return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
+
+    @a2a.tool(
+        description="Compatibility external FileUpload tool: parse one bounded JSON/CSV quote file, compare, persist, and receipt the result.",
+        timeout_seconds=30,
+        cost_class="deterministic",
+    )
+    async def upload_quote_file(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: ComparisonId,
+        file: Annotated[
+            UploadedFile,
+            FileUpload(
+                accept=tuple(sorted(ALLOWED_UPLOAD_MEDIA_TYPES)),
+                max_bytes=MAX_UPLOAD_BYTES,
+                description="JSON or CSV quote file with vendor, unit_price, quantity, delivery_days, warranty_months.",
+            ),
+        ],
+        weights: QuoteWeights,
+    ) -> dict[str, Any]:
+        if file.size_bytes > MAX_UPLOAD_BYTES:
+            return _validation_error(
+                comparison_id,
+                "upload_too_large",
+                f"Upload is {file.size_bytes} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
+            )
+        if file.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
+            return _validation_error(
+                comparison_id,
+                "unsupported_media_type",
+                "Upload must be JSON, CSV, or plain text containing JSON/CSV quote data.",
+            )
+        try:
+            reader = getattr(ctx.workspace, "read_bytes", None)
+            if reader is None:
+                raise RuntimeError("workspace reader unavailable")
+            data = reader(file.path)
+        except Exception:  # noqa: BLE001
+            return _validation_error(
+                comparison_id,
+                "upload_unreadable",
+                "The uploaded file could not be read from the granted workspace. Re-upload the file and retry.",
+            )
+        if len(data) > MAX_UPLOAD_BYTES:
+            return _validation_error(
+                comparison_id,
+                "upload_too_large",
+                f"Decoded upload is {len(data)} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
+            )
+        parsed = _parse_quote_bytes(data, filename=file.filename, media_type=file.media_type)
+        if not parsed["ok"]:
+            return parsed | {"comparison_id": comparison_id}
+        return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
+
+
+def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
+    auth = ctx.auth
+    stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
+    return f"user:{stable}"
+
+
+def _valid_comparison_id(value: str) -> bool:
+    return bool(_COMPARISON_ID_RE.fullmatch(str(value or "").strip()))
+
+
+def _validate_compare_inputs(
+    comparison_id: str,
+    quotes: list[QuoteInput],
+    weights: QuoteWeights,
+) -> dict[str, Any] | None:
+    if not _valid_comparison_id(comparison_id):
+        return _validation_error(
+            comparison_id,
+            "invalid_comparison_id",
+            "comparison_id must start with a letter or number and contain only letters, numbers, dash, underscore, dot, or colon.",
+        )
+    if len(quotes) < 2:
+        return _validation_error(
+            comparison_id,
+            "at_least_two_quotes_required",
+            "QuoteJudge needs at least two vendor quotes to make a comparison.",
+        )
+    if len({quote.vendor.casefold() for quote in quotes}) < 2:
+        return _validation_error(
+            comparison_id,
+            "at_least_two_vendors_required",
+            "Provide quotes from at least two distinct vendors.",
+        )
+    total_weight = weights.price + weights.delivery + weights.warranty
+    if total_weight <= 0:
+        return _validation_error(
+            comparison_id,
+            "positive_weight_required",
+            "At least one of price, delivery, or warranty weight must be greater than zero.",
+        )
+    return None
+
+
+def _validation_error(comparison_id: str, code: str, message: str) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "code": code,
+        "comparison_id": comparison_id,
+        "message": message,
+        "action": "Correct the input and retry the same tool call.",
+    }
+
+
+def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
+    rows = [
+        {
+            **quote.model_dump(mode="json"),
+            "total_price": round(quote.unit_price * quote.quantity, 4),
+        }
+        for quote in quotes
+    ]
+    price_values = [row["total_price"] for row in rows]
+    delivery_values = [row["delivery_days"] for row in rows]
+    warranty_values = [row["warranty_months"] for row in rows]
+    total_weight = weights.price + weights.delivery + weights.warranty
+
+    scored: list[dict[str, Any]] = []
+    for row in rows:
+        price_score = _lower_is_better(row["total_price"], min(price_values), max(price_values))
+        delivery_score = _lower_is_better(row["delivery_days"], min(delivery_values), max(delivery_values))
+        warranty_score = _higher_is_better(row["warranty_months"], min(warranty_values), max(warranty_values))
+        weighted_score = (
+            price_score * weights.price
+            + delivery_score * weights.delivery
+            + warranty_score * weights.warranty
+        ) / total_weight
+        scored.append(
+            {
+                **row,
+                "scores": {
+                    "price": round(price_score, 2),
+                    "delivery": round(delivery_score, 2),
+                    "warranty": round(warranty_score, 2),
+                    "weighted_total": round(weighted_score, 2),
+                },
             }
         )
-
+    scored.sort(
-        @wrap_model_call
+        key=lambda row: (
-        async def log_model_call(request: Any, handler: Any) -> Any:
+            -row["scores"]["weighted_total"],
-            messages = request.state.get("messages", [])
+            row["total_price"],
-            print(
+            row["delivery_days"],
-                "[middleware] model_call "
+            -row["warranty_months"],
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+            row["vendor"].casefold(),
         )
-            return await handler(request)
-
-        backend = ctx.workspace_backend()
-        skill_sources = _seed_runtime_skills(backend, ctx)
-        # create_a2a_deep_agent resolves provider:model strings with
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
-        # provider-specific extra body, and runtime model overrides.
-        return create_a2a_deep_agent(
-            ctx,
-            creds=creds,
-            backend=backend,
-            skills=skill_sources or None,
-            tools=[text_stats],
-            middleware=[log_model_call],
-            system_prompt=SYSTEM_PROMPT,
     )
+    winner = scored[0]
+    now = datetime.now(UTC).isoformat()
+    return {
+        "ok": True,
+        "comparison_id": comparison_id,
+        "recommendation": {
+            "vendor": winner["vendor"],
+            "score": winner["scores"]["weighted_total"],
+            "reason": (
+                f"{winner['vendor']} has the strongest weighted score using "
+                f"price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}."
+            ),
+        },
+        "weights": weights.model_dump(mode="json"),
+        "quote_count": len(quotes),
+        "quotes": scored,
+        "created_at": now,
+        "updated_at": now,
+    }
 
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+def _lower_is_better(value: float, min_value: float, max_value: float) -> float:
-    workspace = getattr(ctx, "_workspace", None)
+    if max_value == min_value:
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+        return 100.0
-    if not prefixes:
+    return max(0.0, min(100.0, 100.0 * (max_value - value) / (max_value - min_value)))
-        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 _higher_is_better(value: float, min_value: float, max_value: float) -> float:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    if max_value == min_value:
+        return 100.0
+    return max(0.0, min(100.0, 100.0 * (value - min_value) / (max_value - min_value)))
 
-    DeepAgents loads skills from its backend, while source-controlled
+
-    ``skills/`` folders live in the image. This bridge lets generated agents
+def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+    all_quotes: list[QuoteInput] = []
-    files.
+    for document in documents:
+        filename = _safe_filename(document.filename)
+        media_type = document.media_type.strip().lower()
+        if media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
+            return {
+                "ok": False,
+                "code": "unsupported_media_type",
+                "message": "Upload must be JSON, CSV, or plain text containing JSON/CSV quote data.",
+                "action": "Upload a .json or .csv file exported from your quote sheet.",
+            }
+        raw_b64 = document.data_base64.strip()
+        if len(raw_b64) > MAX_BASE64_CHARS:
+            return {
+                "ok": False,
+                "code": "upload_too_large",
+                "message": f"{filename} exceeds the {MAX_UPLOAD_BYTES} byte decoded upload limit.",
+                "action": "Trim the file or paste the quotes directly into the form.",
+            }
+        try:
+            data = base64.b64decode(raw_b64, validate=True)
+        except (binascii.Error, ValueError):
+            return {
+                "ok": False,
+                "code": "invalid_base64",
+                "message": f"{filename} is not valid base64 data.",
+                "action": "Re-select the file in the browser and retry the upload.",
+            }
+        if len(data) > MAX_UPLOAD_BYTES:
+            return {
+                "ok": False,
+                "code": "upload_too_large",
+                "message": f"{filename} decoded to {len(data)} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
+                "action": "Trim the file or paste the quotes directly into the form.",
+            }
+        parsed = _parse_quote_bytes(data, filename=filename, media_type=media_type)
+        if not parsed["ok"]:
+            return parsed
+        all_quotes.extend(parsed["quotes"])
+    if len(all_quotes) > MAX_QUOTES:
+        return {
+            "ok": False,
+            "code": "too_many_quotes",
+            "message": f"At most {MAX_QUOTES} quotes can be compared in one call.",
+            "action": "Split the quote set into smaller comparisons.",
+        }
+    return {"ok": True, "quotes": all_quotes}
+
+
+def _safe_filename(value: str) -> str:
+    cleaned = re.sub(r"[^A-Za-z0-9._ -]", "_", str(value or "upload").strip())[:160]
+    return cleaned or "upload"
+
+
+def _parse_quote_bytes(data: bytes, *, filename: str, media_type: str) -> dict[str, Any]:
+    try:
+        text = data.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        return {
+            "ok": False,
+            "code": "upload_not_utf8",
+            "message": f"{filename} must be UTF-8 encoded JSON or CSV text.",
+            "action": "Export the file as UTF-8 CSV or JSON and retry.",
+        }
+    if media_type in {"application/json", "text/json"} or filename.lower().endswith(".json"):
+        return _parse_json_quotes(text, filename=filename)
+    return _parse_csv_quotes(text, filename=filename)
+
+
+def _parse_json_quotes(text: str, *, filename: str) -> dict[str, Any]:
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError as exc:
+        return {
+            "ok": False,
+            "code": "invalid_json",
+            "message": f"{filename} is not valid JSON: {exc.msg}.",
+            "action": "Use a JSON array of quote objects, or an object with a quotes array.",
+        }
+    raw_quotes = payload.get("quotes") if isinstance(payload, dict) else payload
+    if not isinstance(raw_quotes, list):
+        return {
+            "ok": False,
+            "code": "quotes_array_required",
+            "message": f"{filename} must contain a quotes array.",
+            "action": "Provide [{vendor, unit_price, quantity, delivery_days, warranty_months}, ...].",
+        }
+    return _coerce_quotes(raw_quotes, source=filename)
+
+
+def _parse_csv_quotes(text: str, *, filename: str) -> dict[str, Any]:
+    try:
+        rows = list(csv.DictReader(io.StringIO(text)))
+    except csv.Error as exc:
+        return {
+            "ok": False,
+            "code": "invalid_csv",
+            "message": f"{filename} is not valid CSV: {exc}.",
+            "action": "Use a header row with vendor, unit_price, quantity, delivery_days, warranty_months.",
+        }
+    if not rows:
+        return {
+            "ok": False,
+            "code": "no_quotes_found",
+            "message": f"{filename} did not contain any quote rows.",
+            "action": "Add at least two vendor quote rows.",
+        }
+    return _coerce_quotes(rows, source=filename)
+
+
+def _coerce_quotes(raw_quotes: list[Any], *, source: str) -> dict[str, Any]:
+    if len(raw_quotes) > MAX_QUOTES:
+        return {
+            "ok": False,
+            "code": "too_many_quotes",
+            "message": f"{source} contains {len(raw_quotes)} quotes; maximum is {MAX_QUOTES}.",
+            "action": "Split the file into smaller comparisons.",
+        }
+    quotes: list[QuoteInput] = []
+    errors: list[str] = []
+    for idx, item in enumerate(raw_quotes, start=1):
+        if not isinstance(item, dict):
+            errors.append(f"row {idx}: expected an object")
+            continue
+        try:
+            quotes.append(QuoteInput.model_validate(item))
+        except Exception as exc:  # noqa: BLE001
+            errors.append(f"row {idx}: {exc}")
+    if errors:
+        return {
+            "ok": False,
+            "code": "invalid_quote_rows",
+            "message": "One or more quote rows are invalid.",
+            "errors": errors[:5],
+            "action": "Fix the listed rows and retry.",
+        }
+    return {"ok": True, "quotes": quotes}
+
+
+async def _save_comparison(tenant: str, result: dict[str, Any]) -> dict[str, Any]:
+    db_url = os.environ.get("DATABASE_URL")
+    key = (tenant, result["comparison_id"])
+    if not db_url:
+        _LOCAL_COMPARISONS[key] = json.loads(json.dumps(result, default=str))
+        return {"ok": True}
+    try:
+        import psycopg
+        from psycopg.types.json import Jsonb
+
+        with psycopg.connect(db_url, autocommit=True) as conn:
+            _ensure_schema(conn)
+            conn.execute(
                 """
-    root = Path(__file__).parent / "skills"
+                INSERT INTO quote_judge_comparisons
-    if not root.exists():
+                    (tenant_key, comparison_id, recommendation_vendor, result_json, updated_at)
-        return []
+                VALUES (%s, %s, %s, %s, NOW())
-    runtime_skills_root = _runtime_skills_root(ctx)
+                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
-    uploads: list[tuple[str, bytes]] = []
+                    recommendation_vendor = EXCLUDED.recommendation_vendor,
-    for path in root.rglob("*"):
+                    result_json = EXCLUDED.result_json,
-        if path.is_file():
+                    updated_at = NOW()
-            rel = path.relative_to(root).as_posix()
+                """,
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+                (
-    if uploads:
+                    tenant,
-        backend.upload_files(uploads)
+                    result["comparison_id"],
-        return [runtime_skills_root]
+                    result["recommendation"]["vendor"],
-    return []
+                    Jsonb(result),
+                ),
+            )
+            conn.execute(
+                """
+                INSERT INTO app_records (tenant_key, record_kind, payload, updated_at)
+                VALUES (%s, 'quote_comparison', %s, NOW())
+                """,
+                (tenant, Jsonb(result)),
+            )
+        return {"ok": True}
+    except Exception:  # noqa: BLE001
+        return {
+            "ok": False,
+            "code": "database_unavailable",
+            "message": "The managed Postgres database could not save this comparison.",
+        }
 
 
-def _last_message_text(state: dict[str, Any]) -> str:
+async def _load_comparison(tenant: str, comparison_id: str) -> dict[str, Any] | None:
-    messages = state.get("messages") or []
+    key = (tenant, comparison_id)
-    if not messages:
+    db_url = os.environ.get("DATABASE_URL")
-        return json.dumps(state, default=str)
+    if not db_url:
+        return _LOCAL_COMPARISONS.get(key)
+    try:
+        import psycopg
 
-    content = getattr(messages[-1], "content", None)
+        with psycopg.connect(db_url) as conn:
-    if isinstance(content, str):
+            _ensure_schema(conn)
-        return content
+            row = conn.execute(
-    if isinstance(content, list):
+                """
-        parts: list[str] = []
+                SELECT result_json FROM quote_judge_comparisons
-        for item in content:
+                WHERE tenant_key = %s AND comparison_id = %s
-            if isinstance(item, dict):
+                ORDER BY updated_at DESC LIMIT 1
-                text = item.get("text") or item.get("content")
+                """,
-                if text:
+                (tenant, comparison_id),
-                    parts.append(str(text))
+            ).fetchone()
-            elif item:
+            if row and row[0]:
-                parts.append(str(item))
+                return dict(row[0])
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+            row = conn.execute(
-    return str(content or messages[-1])
+                """
+                SELECT payload FROM app_records
+                WHERE tenant_key = %s
+                  AND record_kind IN ('quote_comparison', 'comparison', 'quotejudge_comparison')
+                  AND payload->>'comparison_id' = %s
+                ORDER BY updated_at DESC LIMIT 1
+                """,
+                (tenant, comparison_id),
+            ).fetchone()
+            if row and row[0]:
+                return dict(row[0])
+            return _load_legacy_comparison(conn, tenant, comparison_id)
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def _ensure_schema(conn: Any) -> None:
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS app_records (
+            id BIGSERIAL PRIMARY KEY,
+            tenant_key TEXT NOT NULL,
+            record_kind TEXT NOT NULL,
+            payload JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+        """
+    )
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS quote_judge_comparisons (
+            id BIGSERIAL PRIMARY KEY,
+            tenant_key TEXT NOT NULL,
+            comparison_id TEXT NOT NULL,
+            recommendation_vendor TEXT NOT NULL,
+            result_json JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            UNIQUE (tenant_key, comparison_id)
+        )
+        """
+    )
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS quote_judge_receipts (
+            id BIGSERIAL PRIMARY KEY,
+            tenant_key TEXT NOT NULL,
+            comparison_id TEXT,
+            skill_name TEXT NOT NULL,
+            receipt_json JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+        """
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS app_records_tenant_kind_idx ON app_records (tenant_key, record_kind, created_at DESC)"
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS quote_judge_receipts_tenant_idx ON quote_judge_receipts (tenant_key, created_at DESC)"
+    )
+
+
+def _load_legacy_comparison(conn: Any, tenant: str, comparison_id: str) -> dict[str, Any] | None:
+    for table in ("quote_comparisons", "comparisons"):
+        columns = _table_columns(conn, table)
+        if not columns or "tenant_key" not in columns or "comparison_id" not in columns:
+            continue
+        for json_col in ("result_json", "payload", "result", "data"):
+            if json_col not in columns:
+                continue
+            row = conn.execute(
+                f"SELECT {json_col} FROM {table} WHERE tenant_key = %s AND comparison_id = %s LIMIT 1",
+                (tenant, comparison_id),
+            ).fetchone()
+            if row and isinstance(row[0], dict):
+                return dict(row[0])
+    return None
+
+
+def _table_columns(conn: Any, table: str) -> set[str]:
+    rows = conn.execute(
+        """
+        SELECT column_name FROM information_schema.columns
+        WHERE table_schema = current_schema() AND table_name = %s
+        """,
+        (table,),
+    ).fetchall()
+    return {str(row[0]) for row in rows}
+
+
+async def _persist_receipt(
+    ctx: RunContext[PlatformUserAuth],
+    tenant: str,
+    skill_name: str,
+    inputs: dict[str, Any],
+    result: dict[str, Any],
+) -> dict[str, Any]:
+    comparison_id = str(inputs.get("comparison_id") or result.get("comparison_id") or "")
+    receipt = {
+        "receipt_type": "quote_judge_execution",
+        "agent": QuoteJudgeStudioV1.name,
+        "agent_version": QuoteJudgeStudioV1.version,
+        "skill_name": skill_name,
+        "comparison_id": comparison_id,
+        "tenant_scope": "user",
+        "task_id": getattr(ctx, "task_id", ""),
+        "created_at": datetime.now(UTC).isoformat(),
+        "input_hash": _hash_json(inputs),
+        "result_hash": _hash_json({k: v for k, v in result.items() if k != "receipt"}),
+        "status": "ok" if result.get("ok") else "validation_error",
+    }
+    receipt["receipt_id"] = _hash_json(receipt)[:32]
+    db_url = os.environ.get("DATABASE_URL")
+    if not db_url:
+        _LOCAL_RECEIPTS.append(receipt)
+        return {"persisted": True, "receipt_id": receipt["receipt_id"]}
+    try:
+        import psycopg
+        from psycopg.types.json import Jsonb
+
+        with psycopg.connect(db_url, autocommit=True) as conn:
+            _ensure_schema(conn)
+            conn.execute(
+                """
+                INSERT INTO quote_judge_receipts
+                    (tenant_key, comparison_id, skill_name, receipt_json)
+                VALUES (%s, %s, %s, %s)
+                """,
+                (tenant, comparison_id or None, skill_name, Jsonb(receipt)),
+            )
+        return {"persisted": True, "receipt_id": receipt["receipt_id"]}
+    except Exception:  # noqa: BLE001
+        return {"persisted": False, "receipt_id": receipt["receipt_id"], "warning": "receipt_database_write_failed"}
+
+
+def _hash_json(payload: dict[str, Any]) -> str:
+    data = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+    return hashlib.sha256(data).hexdigest()