a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,644 @@
-"""quote-judge-studio-v1 agent.
+"""QuoteJudge full-stack A2A agent.
 
-Starter stack:
+Deterministic quote comparison, per-user persistence in managed Postgres,
-  - DeepAgents for tool-calling orchestration
+browser-safe upload parsing, and production execution receipt storage.
-  - Caller-provided LLM credentials via ctx.llm
-  - 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
+import time
+import uuid
+from contextlib import contextmanager
+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,
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
-    WorkspaceAccess,
+    State,
-    WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+from a2a_pack.workspace import FileUpload, UploadedFile
+
+try:  # psycopg is installed in deployment and sandbox via requirements.txt.
+    import psycopg
+    from psycopg.rows import dict_row
+except Exception:  # pragma: no cover - import guard for card-only tooling
+    psycopg = None
+    dict_row = None
+
+
+MAX_QUOTES = 20
+MAX_VENDOR_LENGTH = 120
+MAX_UPLOADS = 5
+MAX_UPLOAD_BYTES = 128_000
+ALLOWED_UPLOAD_MEDIA_TYPES = {"text/csv", "text/plain", "application/json"}
+_DATABASE_NAME = "quote-judge-studio-v1-data"
+_LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {}
+_LOCAL_RECEIPTS: list[dict[str, Any]] = []
 
 
 class QuoteJudgeStudioV1Config(BaseModel):
-    pass
+    """No caller-configured settings are required."""
 
 
-SYSTEM_PROMPT = """\
+class QuoteInput(BaseModel):
-You are a compact tool-calling agent.
+    vendor: str = Field(..., min_length=1, max_length=MAX_VENDOR_LENGTH)
+    quantity: int = Field(..., gt=0, le=10_000_000)
+    unit_price: float = Field(..., ge=0, le=1_000_000_000)
+    delivery_days: int = Field(..., ge=0, le=3650)
+    warranty_months: int = Field(..., ge=0, le=600)
 
-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(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=100)
+    delivery: float = Field(..., ge=0, le=100)
+    warranty: float = Field(..., ge=0, le=100)
+
+    @field_validator("warranty")
+    @classmethod
+    def weights_must_be_positive(cls, value: float, info: Any) -> float:
+        values = info.data
+        total = float(values.get("price", 0)) + float(values.get("delivery", 0)) + float(value)
+        if total <= 0:
+            raise ValueError("at least one weight must be positive")
+        return value
+
+
+class BrowserQuoteDocument(BaseModel):
+    filename: str = Field(..., min_length=1, max_length=160)
+    media_type: str = Field(..., min_length=1, max_length=100)
+    data_base64: str = Field(..., min_length=1)
+
+    @field_validator("filename")
+    @classmethod
+    def safe_filename(cls, value: str) -> str:
+        cleaned = value.strip().replace("\\", "/").split("/")[-1]
+        if not cleaned or cleaned in {".", ".."}:
+            raise ValueError("filename is required")
+        return cleaned[:160]
+
+
+class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     name = "quote-judge-studio-v1"
-    description = "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty scoring, persists per-user comparisons, and serves a one-page product UI."
+    description = (
+        "QuoteJudge compares vendor quotes with weighted price, delivery, and "
+        "warranty scoring, persists per-user comparisons, and serves a one-page product UI."
+    )
     version = "0.1.0"
 
     config_model = QuoteJudgeStudioV1Config
-    auth_model = {{ auth_type }}
+    auth_model = PlatformUserAuth
-
+    state = State.DURABLE
-    # Hosted generated agents read the caller's saved LLM credential through
+    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300)
-    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
+    platform_resources = AgentPlatformResources(
-    # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
+        databases=(
-    # directly.
+            AgentDatabase(
-    llm_provisioning = LLMProvisioning.PLATFORM
+                name=_DATABASE_NAME,
+                scope="user",
+                access_mode="read_write",
+                env=AgentDatabaseEnv(url="DATABASE_URL"),
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            ),
+        )
+    )
     pricing = Pricing(
         price_per_call_usd=0.0,
-        caller_pays_llm=True,
+        caller_pays_llm=False,
-        notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
+        notes="Deterministic local quote scoring; no LLM credential is required.",
     )
-    workspace_access = WorkspaceAccess.dynamic(
+    tools_used = ("postgres", "mcp")
-        max_files=64,
-        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
-        require_reason=False,
-    )
-    tools_used = ("deepagents", "langchain")
 
-    @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=(
-        creds = ctx.llm
+            "Compare at least two vendor quotes using typed price, delivery, and warranty "
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+            "weights; persist the recommendation and a production receipt."
-        if not creds.api_key:
+        ),
-            return (
+        timeout_seconds=30,
-                "LLM key required. Add an LLM credential in Settings > LLM "
+        idempotent=True,
-                "credentials before running this agent; for local --invoke "
+        cost_class="cheap",
-                "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: 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.
+        tenant = _tenant_key(ctx)
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        cleaned_id = _clean_comparison_id(comparison_id)
-        from langchain.agents.middleware import wrap_model_call
+        started = _now_iso()
-        from langchain_core.tools import tool
+        input_payload = {
+            "comparison_id": cleaned_id,
+            "quotes": [quote.model_dump() for quote in quotes],
+            "weights": weights.model_dump(),
+        }
 
-        @tool
+        if len(quotes) < 2:
-        def text_stats(text: str) -> str:
+            result = {
-            """Return exact word, character, and line counts for text."""
+                "ok": False,
-            words = [part for part in text.split() if part.strip()]
+                "code": "at_least_two_quotes_required",
-            return json.dumps(
+                "message": "Add at least two vendor quotes before comparing.",
+                "comparison_id": cleaned_id,
+            }
+            await _persist_receipt(
+                tenant_key=tenant,
+                comparison_id=cleaned_id,
+                skill_name="compare_quotes",
+                input_payload=input_payload,
+                result_payload=result,
+                status="validation_error",
+                started_at=started,
+            )
+            return result
+        if len(quotes) > MAX_QUOTES:
+            result = {
+                "ok": False,
+                "code": "too_many_quotes",
+                "message": f"Compare at most {MAX_QUOTES} quotes at a time.",
+                "comparison_id": cleaned_id,
+            }
+            await _persist_receipt(
+                tenant_key=tenant,
+                comparison_id=cleaned_id,
+                skill_name="compare_quotes",
+                input_payload=input_payload,
+                result_payload=result,
+                status="validation_error",
+                started_at=started,
+            )
+            return result
+
+        result = _score_quotes(cleaned_id, quotes, weights)
+        await _save_comparison(tenant, result)
+        await _persist_receipt(
+            tenant_key=tenant,
+            comparison_id=cleaned_id,
+            skill_name="compare_quotes",
+            input_payload=input_payload,
+            result_payload=result,
+            status="ok",
+            started_at=started,
+        )
+        return result
+
+    @a2a.tool(
+        description="Reopen a saved QuoteJudge comparison for the authenticated user.",
+        timeout_seconds=15,
+        idempotent=True,
+        cost_class="cheap",
+    )
+    async def get_comparison(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+    ) -> dict[str, Any]:
+        tenant = _tenant_key(ctx)
+        cleaned_id = _clean_comparison_id(comparison_id)
+        started = _now_iso()
+        record = await _load_comparison(tenant, cleaned_id)
+        if record is None:
+            result = {
+                "ok": False,
+                "code": "comparison_not_found",
+                "message": "No saved comparison exists for this id and user.",
+                "comparison_id": cleaned_id,
+            }
+        else:
+            result = dict(record)
+            result["ok"] = True
+        await _persist_receipt(
+            tenant_key=tenant,
+            comparison_id=cleaned_id,
+            skill_name="get_comparison",
+            input_payload={"comparison_id": cleaned_id},
+            result_payload=result,
+            status="ok" if result.get("ok") else "not_found",
+            started_at=started,
+        )
+        return result
+
+    @a2a.tool(
+        description=(
+            "Browser upload bridge: parse bounded base64 JSON/CSV/text quote files, "
+            "compare extracted quotes, and persist the result."
+        ),
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="cheap",
+    )
+    async def compare_uploaded_quotes_browser(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        documents: list[BrowserQuoteDocument],
+        weights: QuoteWeights,
+    ) -> dict[str, Any]:
+        parsed = _parse_browser_documents(documents)
+        if not parsed["ok"]:
+            return parsed
+        return await self.compare_quotes(
+            ctx,
+            comparison_id=comparison_id,
+            quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]],
+            weights=weights,
+        )
+
+    @a2a.tool(
+        description=(
+            "External-client upload path: parse an uploaded CSV, JSON, or plain-text quote "
+            "file and compare extracted quotes."
+        ),
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="cheap",
+    )
+    async def compare_uploaded_quote_file(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        document: Annotated[
+            UploadedFile,
+            FileUpload(
+                accept=("text/csv", "text/plain", "application/json"),
+                max_bytes=MAX_UPLOAD_BYTES,
+                description="CSV, JSON, or text file containing vendor quote rows.",
+            ),
+        ],
+        weights: QuoteWeights,
+    ) -> dict[str, Any]:
+        if document.size_bytes > MAX_UPLOAD_BYTES:
+            return {
+                "ok": False,
+                "code": "upload_too_large",
+                "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller.",
+                "comparison_id": _clean_comparison_id(comparison_id),
+            }
+        if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
+            return {
+                "ok": False,
+                "code": "unsupported_media_type",
+                "message": "Upload must be CSV, JSON, or plain text.",
+                "comparison_id": _clean_comparison_id(comparison_id),
+            }
+        try:
+            view = await ctx.workspace.open_view(
+                purpose="Read uploaded vendor quote file",
+                hints=[document.path],
+                max_files=1,
+                reason="Parse the caller-uploaded quote file for comparison.",
+            )
+            raw = await view.read(document.path)
+        except Exception:
+            return {
+                "ok": False,
+                "code": "upload_read_failed",
+                "message": "Could not read the uploaded file from the granted workspace.",
+                "comparison_id": _clean_comparison_id(comparison_id),
+            }
+        parsed = _parse_document_bytes(document.filename, document.media_type, raw)
+        if not parsed["ok"]:
+            return parsed
+        return await self.compare_quotes(
+            ctx,
+            comparison_id=comparison_id,
+            quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]],
+            weights=weights,
+        )
+
+
+def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
+    auth = ctx.auth
+    ident = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
+    return f"user:{ident}"
+
+
+def _clean_comparison_id(value: str) -> str:
+    cleaned = str(value or "").strip()
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,79}", cleaned):
+        raise ValueError("comparison_id must be 1-80 safe characters")
+    return cleaned
+
+
+def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
+    quote_rows = [quote.model_dump() for quote in quotes]
+    for row in quote_rows:
+        row["subtotal"] = round(row["quantity"] * row["unit_price"], 2)
+
+    totals = [row["subtotal"] for row in quote_rows]
+    deliveries = [row["delivery_days"] for row in quote_rows]
+    warranties = [row["warranty_months"] for row in quote_rows]
+    total_weight = weights.price + weights.delivery + weights.warranty
+    normalized_weights = {
+        "price": weights.price / total_weight,
+        "delivery": weights.delivery / total_weight,
+        "warranty": weights.warranty / total_weight,
+    }
+
+    scored: list[dict[str, Any]] = []
+    for row in quote_rows:
+        price_score = _lower_is_better(row["subtotal"], totals)
+        delivery_score = _lower_is_better(row["delivery_days"], deliveries)
+        warranty_score = _higher_is_better(row["warranty_months"], warranties)
+        weighted_score = (
+            price_score * normalized_weights["price"]
+            + delivery_score * normalized_weights["delivery"]
+            + warranty_score * normalized_weights["warranty"]
+        )
+        scored.append(
             {
-                    "characters": len(text),
+                **row,
-                    "words": len(words),
+                "scores": {
-                    "lines": len(text.splitlines()) or 1,
+                    "price": round(price_score, 4),
+                    "delivery": round(delivery_score, 4),
+                    "warranty": round(warranty_score, 4),
+                    "weighted_total": round(weighted_score, 4),
+                },
             }
         )
 
-        @wrap_model_call
+    scored.sort(
-        async def log_model_call(request: Any, handler: Any) -> Any:
+        key=lambda item: (
-            messages = request.state.get("messages", [])
+            -item["scores"]["weighted_total"],
-            print(
+            item["subtotal"],
-                "[middleware] model_call "
+            item["delivery_days"],
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+            -item["warranty_months"],
+            item["vendor"].lower(),
         )
-            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]
+    baseline = min(scored, key=lambda item: item["subtotal"])
+    rationale_bits = [
+        f"{winner['vendor']} has the highest weighted score ({winner['scores']['weighted_total']:.2f}).",
+        f"Its normalized total is {winner['subtotal']:.2f} for {winner['quantity']} units.",
+    ]
+    if winner["delivery_days"] == min(deliveries):
+        rationale_bits.append("It also has the fastest delivery among the submitted quotes.")
+    if winner["warranty_months"] == max(warranties):
+        rationale_bits.append("It offers the strongest warranty among the submitted quotes.")
+    if winner["vendor"] == baseline["vendor"]:
+        rationale_bits.append("It is the lowest-price option under the selected quantity.")
+
+    return {
+        "ok": True,
+        "comparison_id": comparison_id,
+        "recommendation": {
+            "vendor": winner["vendor"],
+            "score": winner["scores"]["weighted_total"],
+            "normalized_total": winner["subtotal"],
+            "rationale": " ".join(rationale_bits),
+        },
+        "weights": {
+            "price": weights.price,
+            "delivery": weights.delivery,
+            "warranty": weights.warranty,
+            "normalized": {key: round(value, 4) for key, value in normalized_weights.items()},
+        },
+        "quotes": scored,
+        "created_at": _now_iso(),
+    }
 
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+def _lower_is_better(value: float, values: list[float]) -> float:
-    workspace = getattr(ctx, "_workspace", None)
+    min_value = min(values)
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+    max_value = max(values)
-    if not prefixes:
+    if max_value == min_value:
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+        return 1.0
-        prefixes = (outputs_prefix or "outputs/",)
+    return (max_value - value) / (max_value - min_value)
-    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, values: list[float]) -> float:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    min_value = min(values)
+    max_value = max(values)
+    if max_value == min_value:
+        return 1.0
+    return (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[BrowserQuoteDocument]) -> dict[str, Any]:
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+    if not documents:
-    files.
+        return {"ok": False, "code": "no_uploads", "message": "Upload at least one quote file."}
+    if len(documents) > MAX_UPLOADS:
+        return {"ok": False, "code": "too_many_uploads", "message": f"Upload at most {MAX_UPLOADS} files."}
+    quotes: list[dict[str, Any]] = []
+    for document in documents:
+        if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
+            return {"ok": False, "code": "unsupported_media_type", "message": "Upload must be CSV, JSON, or plain text."}
+        try:
+            raw = base64.b64decode(document.data_base64, validate=True)
+        except Exception:
+            return {"ok": False, "code": "invalid_base64", "message": "Uploaded file data is not valid base64."}
+        if len(raw) > MAX_UPLOAD_BYTES:
+            return {"ok": False, "code": "upload_too_large", "message": f"Each upload must be {MAX_UPLOAD_BYTES} bytes or smaller."}
+        parsed = _parse_document_bytes(document.filename, document.media_type, raw)
+        if not parsed["ok"]:
+            return parsed
+        quotes.extend(parsed["quotes"])
+    return {"ok": True, "quotes": quotes[:MAX_QUOTES]}
+
+
+def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]:
+    if len(raw) > MAX_UPLOAD_BYTES:
+        return {"ok": False, "code": "upload_too_large", "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller."}
+    text = raw.decode("utf-8-sig", errors="replace")
+    try:
+        if media_type == "application/json" or filename.lower().endswith(".json"):
+            data = json.loads(text)
+            rows = data.get("quotes", data) if isinstance(data, dict) else data
+            if not isinstance(rows, list):
+                raise ValueError("JSON must be a list or contain a quotes list")
+            quotes = [QuoteInput.model_validate(row).model_dump() for row in rows]
+        elif media_type == "text/csv" or filename.lower().endswith(".csv"):
+            reader = csv.DictReader(io.StringIO(text))
+            quotes = [QuoteInput.model_validate(_coerce_quote_row(row)).model_dump() for row in reader]
+        else:
+            quotes = _parse_plain_text_quotes(text)
+    except Exception as exc:
+        return {"ok": False, "code": "quote_parse_failed", "message": f"Could not parse quote upload: {exc}"}
+    if not quotes:
+        return {"ok": False, "code": "no_quotes_found", "message": "No quote rows were found in the upload."}
+    return {"ok": True, "quotes": quotes}
+
+
+def _coerce_quote_row(row: dict[str, Any]) -> dict[str, Any]:
+    aliases = {
+        "vendor": ["vendor", "supplier", "name"],
+        "quantity": ["quantity", "qty"],
+        "unit_price": ["unit_price", "price", "unit price", "unitPrice"],
+        "delivery_days": ["delivery_days", "delivery", "delivery days", "lead_time", "lead time"],
+        "warranty_months": ["warranty_months", "warranty", "warranty months"],
+    }
+    normalized = {str(k).strip().lower(): v for k, v in row.items()}
+    out: dict[str, Any] = {}
+    for target, names in aliases.items():
+        for name in names:
+            if name.lower() in normalized and normalized[name.lower()] not in (None, ""):
+                out[target] = normalized[name.lower()]
+                break
+    return out
+
+
+def _parse_plain_text_quotes(text: str) -> list[dict[str, Any]]:
+    quotes: list[dict[str, Any]] = []
+    for line in text.splitlines():
+        if not line.strip():
+            continue
+        parts = [part.strip() for part in re.split(r"[,;|]", line) if part.strip()]
+        if len(parts) >= 5:
+            quotes.append(
+                QuoteInput.model_validate(
+                    {
+                        "vendor": parts[0],
+                        "quantity": parts[1],
+                        "unit_price": parts[2],
+                        "delivery_days": parts[3],
+                        "warranty_months": parts[4],
+                    }
+                ).model_dump()
+            )
+    return quotes
+
+
+@contextmanager
+def _db_conn() -> Any:
+    url = os.environ.get("DATABASE_URL")
+    if not url or psycopg is None:
+        yield None
+        return
+    conn = psycopg.connect(url, row_factory=dict_row)
+    try:
+        yield conn
+        conn.commit()
+    finally:
+        conn.close()
+
+
+async def _save_comparison(tenant_key: str, payload: dict[str, Any]) -> None:
+    key = (tenant_key, payload["comparison_id"])
+    with _db_conn() as conn:
+        if conn is None:
+            _LOCAL_COMPARISONS[key] = dict(payload)
+            return
+        with conn.cursor() as cur:
+            cur.execute(
                 """
-    root = Path(__file__).parent / "skills"
+                INSERT INTO quote_comparisons
-    if not root.exists():
+                    (tenant_key, comparison_id, recommended_vendor, score, normalized_total, payload, updated_at)
-        return []
+                VALUES (%s, %s, %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]] = []
+                    recommended_vendor = EXCLUDED.recommended_vendor,
-    for path in root.rglob("*"):
+                    score = EXCLUDED.score,
-        if path.is_file():
+                    normalized_total = EXCLUDED.normalized_total,
-            rel = path.relative_to(root).as_posix()
+                    payload = EXCLUDED.payload,
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+                    updated_at = NOW()
-    if uploads:
+                """,
-        backend.upload_files(uploads)
+                (
-        return [runtime_skills_root]
+                    tenant_key,
-    return []
+                    payload["comparison_id"],
+                    payload["recommendation"]["vendor"],
+                    payload["recommendation"]["score"],
+                    payload["recommendation"]["normalized_total"],
+                    json.dumps(payload),
+                ),
+            )
 
 
-def _last_message_text(state: dict[str, Any]) -> str:
+async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None:
-    messages = state.get("messages") or []
+    key = (tenant_key, comparison_id)
-    if not messages:
+    with _db_conn() as conn:
-        return json.dumps(state, default=str)
+        if conn is None:
+            return _LOCAL_COMPARISONS.get(key)
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT payload
+                FROM quote_comparisons
+                WHERE tenant_key = %s AND comparison_id = %s
+                """,
+                (tenant_key, comparison_id),
+            )
+            row = cur.fetchone()
+    if not row:
+        return None
+    payload = row["payload"]
+    return payload if isinstance(payload, dict) else json.loads(payload)
 
-    content = getattr(messages[-1], "content", None)
+
-    if isinstance(content, str):
+async def _persist_receipt(
-        return content
+    *,
-    if isinstance(content, list):
+    tenant_key: str,
-        parts: list[str] = []
+    comparison_id: str,
-        for item in content:
+    skill_name: str,
-            if isinstance(item, dict):
+    input_payload: dict[str, Any],
-                text = item.get("text") or item.get("content")
+    result_payload: dict[str, Any],
-                if text:
+    status: str,
-                    parts.append(str(text))
+    started_at: str,
-            elif item:
+) -> dict[str, Any]:
-                parts.append(str(item))
+    receipt = {
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+        "receipt_id": str(uuid.uuid4()),
-    return str(content or messages[-1])
+        "agent": QuoteJudgeStudioV1.name,
+        "agent_version": QuoteJudgeStudioV1.version,
+        "skill": skill_name,
+        "comparison_id": comparison_id,
+        "status": status,
+        "started_at": started_at,
+        "completed_at": _now_iso(),
+        "input_hash": _hash_json(input_payload),
+        "result_hash": _hash_json(result_payload),
+    }
+    with _db_conn() as conn:
+        if conn is None:
+            _LOCAL_RECEIPTS.append({"tenant_key": tenant_key, **receipt})
+            return receipt
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                INSERT INTO quote_execution_receipts
+                    (tenant_key, comparison_id, receipt_id, skill_name, status, input_hash,
+                     result_hash, payload, created_at)
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW())
+                """,
+                (
+                    tenant_key,
+                    comparison_id,
+                    receipt["receipt_id"],
+                    skill_name,
+                    status,
+                    receipt["input_hash"],
+                    receipt["result_hash"],
+                    json.dumps(receipt),
+                ),
+            )
+    return receipt
+
+
+def _hash_json(payload: dict[str, Any]) -> str:
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()