a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,496 @@
-"""quote-judge-studio-v1 agent.
+"""QuoteJudge Studio v1: deterministic quote comparison with user-scoped persistence."""
-
-Starter stack:
-  - DeepAgents for tool-calling orchestration
-  - 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 uuid
+from datetime import datetime, timezone
+from typing import Annotated, Any
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    LLMProvisioning,
+    AgentDatabase,
-    {{ auth_type }},
+    AgentDatabaseEnv,
+    AgentDatabaseMigrations,
+    AgentPlatformResources,
+    FileUpload,
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
-    WorkspaceAccess,
+    State,
-    WorkspaceMode,
+    UploadedFile,
 )
-from a2a_pack.context import LLMCreds
+
+
+MAX_QUOTES = 25
+MAX_VENDOR_LEN = 120
+MAX_COMPARISON_ID_LEN = 80
+MAX_UPLOAD_BYTES = 64 * 1024
+ACCEPTED_UPLOAD_MEDIA_TYPES = ("text/csv", "text/plain", "application/json")
+DATABASE_NAME = "quote-judge-studio-v1-data"
 
 
 class QuoteJudgeStudioV1Config(BaseModel):
-    pass
+    """No caller-provided configuration is required."""
 
 
-SYSTEM_PROMPT = """\
+class QuoteInput(BaseModel):
-You are a compact tool-calling agent.
+    vendor: str = Field(..., min_length=1, max_length=MAX_VENDOR_LEN)
+    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 = 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 WeightInput(BaseModel):
+    price: float = Field(..., ge=0, le=1000)
+    delivery: float = Field(..., ge=0, le=1000)
+    warranty: float = Field(..., ge=0, le=1000)
+
+    @model_validator(mode="after")
+    def require_positive_total(self) -> "WeightInput":
+        if self.price + self.delivery + self.warranty <= 0:
+            raise ValueError("at least one weight must be greater than zero")
+        return self
+
+
+class BrowserQuoteUpload(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, max_length=((MAX_UPLOAD_BYTES + 2) // 3) * 4 + 16)
+
+    @field_validator("filename")
+    @classmethod
+    def clean_filename(cls, value: str) -> str:
+        cleaned = value.replace("\\", "/").split("/")[-1].strip()
+        if not cleaned or cleaned in {".", ".."}:
+            raise ValueError("filename is required")
+        return cleaned
+
+
+class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     name = "quote-judge-studio-v1"
-    description = "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty scoring, persists recommendations per platform user, and serves a one-page React product UI."
+    description = (
+        "QuoteJudge compares vendor quotes with weighted price, delivery, and "
+        "warranty scoring, persists recommendations per platform user, and "
+        "serves a one-page React product UI."
+    )
     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; no LLM credential required.",
     )
-    workspace_access = WorkspaceAccess.dynamic(
+    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
-        max_files=64,
+    state = State.DURABLE
-        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
+    platform_resources = AgentPlatformResources(
-        require_reason=False,
+        databases=(
+            AgentDatabase(
+                name=DATABASE_NAME,
+                scope="user",
+                access_mode="read_write",
+                env=AgentDatabaseEnv(url="DATABASE_URL"),
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            ),
         )
-    tools_used = ("deepagents", "langchain")
+    )
+    capabilities = {
+        "product": {
+            "name": "QuoteJudge",
+            "primary_workflow": "Upload or paste vendor quotes, weight price/delivery/warranty, compare them, save the result, and reopen it.",
+            "uploads": True,
+            "receipts_persisted": True,
+            "database_scope": "user",
+        }
+    }
 
-    @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 two or more vendor quotes with weighted price, delivery, and warranty scoring; persist the result for the signed-in user.",
-        creds = ctx.llm
+        timeout_seconds=30,
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+        idempotent=True,
-        if not creds.api_key:
+        cost_class="deterministic",
-            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: str,
-        creds: LLMCreds,
+        quotes: list[QuoteInput],
-    ) -> Any:
+        weights: WeightInput,
-        # 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 = _validate_comparison_id(comparison_id)
-        from langchain.agents.middleware import wrap_model_call
+        validation_error = _validate_quotes(quotes)
-        from langchain_core.tools import tool
+        if validation_error is not None:
+            result = _error(cleaned_id, validation_error, _error_message(validation_error))
+            await _persist_receipt(ctx, tenant, "compare_quotes", cleaned_id, {"quotes": len(quotes), "weights": weights.model_dump()}, result)
+            return result
 
-        @tool
+        result = _build_comparison(cleaned_id, quotes, weights)
-        def text_stats(text: str) -> str:
+        await _upsert_comparison(tenant, result)
-            """Return exact word, character, and line counts for text."""
+        await _persist_receipt(ctx, tenant, "compare_quotes", cleaned_id, {"quotes": [q.model_dump() for q in quotes], "weights": weights.model_dump()}, result)
-            words = [part for part in text.split() if part.strip()]
+        await ctx.emit_progress(f"Saved comparison {cleaned_id}: {result['recommendation']['vendor']} recommended")
-            return json.dumps(
+        return result
+
+    @a2a.tool(
+        description="Reload a previously saved quote comparison for the signed-in user.",
+        timeout_seconds=15,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def get_comparison(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+    ) -> dict[str, Any]:
+        tenant = _tenant_key(ctx)
+        cleaned_id = _validate_comparison_id(comparison_id)
+        result = await _load_comparison(tenant, cleaned_id)
+        if result is None:
+            result = _error(cleaned_id, "comparison_not_found", "No comparison with that id exists for this signed-in user.")
+        await _persist_receipt(ctx, tenant, "get_comparison", cleaned_id, {"comparison_id": cleaned_id}, result)
+        return result
+
+    @a2a.tool(
+        description="Browser upload bridge: decode a bounded base64 CSV/JSON quote file, compare quotes, and persist the result.",
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def compare_quotes_upload(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        upload: BrowserQuoteUpload,
+        weights: WeightInput,
+    ) -> dict[str, Any]:
+        decoded = _decode_browser_upload(upload)
+        if not decoded["ok"]:
+            cleaned_id = _safe_comparison_id(comparison_id)
+            result = _error(cleaned_id, decoded["code"], decoded["message"])
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_upload", cleaned_id, {"filename": upload.filename, "media_type": upload.media_type}, result)
+            return result
+        quotes_or_error = _parse_quote_bytes(decoded["bytes"], decoded["media_type"])
+        if isinstance(quotes_or_error, dict):
+            cleaned_id = _safe_comparison_id(comparison_id)
+            result = _error(cleaned_id, quotes_or_error["code"], quotes_or_error["message"])
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_upload", cleaned_id, {"filename": upload.filename}, result)
+            return result
+        return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=quotes_or_error, weights=weights)
+
+    @a2a.tool(
+        description="External typed FileUpload path: read an uploaded CSV/JSON quote file from the caller workspace, compare quotes, and persist the result.",
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="deterministic",
+    )
+    async def compare_quotes_file(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        comparison_id: str,
+        document: Annotated[
+            UploadedFile,
+            FileUpload(
+                accept=ACCEPTED_UPLOAD_MEDIA_TYPES,
+                max_bytes=MAX_UPLOAD_BYTES,
+                description="CSV or JSON file containing quote rows with vendor, unit_price, quantity, delivery_days, and warranty_months.",
+            ),
+        ],
+        weights: WeightInput,
+    ) -> dict[str, Any]:
+        if document.size_bytes > MAX_UPLOAD_BYTES:
+            result = _error(_safe_comparison_id(comparison_id), "upload_too_large", f"Upload must be at most {MAX_UPLOAD_BYTES} bytes.")
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_file", _safe_comparison_id(comparison_id), {"filename": document.filename}, result)
+            return result
+        if document.media_type not in ACCEPTED_UPLOAD_MEDIA_TYPES:
+            result = _error(_safe_comparison_id(comparison_id), "unsupported_media_type", "Upload must be CSV, plain text CSV, or JSON.")
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_file", _safe_comparison_id(comparison_id), {"filename": document.filename}, result)
+            return result
+        try:
+            data = ctx.workspace.read_bytes(document.path)
+        except Exception:
+            result = _error(_safe_comparison_id(comparison_id), "upload_read_failed", "Could not read the uploaded file from the granted workspace.")
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_file", _safe_comparison_id(comparison_id), {"filename": document.filename}, result)
+            return result
+        quotes_or_error = _parse_quote_bytes(data[: MAX_UPLOAD_BYTES + 1], document.media_type)
+        if isinstance(quotes_or_error, dict):
+            result = _error(_safe_comparison_id(comparison_id), quotes_or_error["code"], quotes_or_error["message"])
+            await _persist_receipt(ctx, _tenant_key(ctx), "compare_quotes_file", _safe_comparison_id(comparison_id), {"filename": document.filename}, result)
+            return result
+        return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=quotes_or_error, weights=weights)
+
+
+def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
+    auth = ctx.auth
+    principal = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
+    if not principal:
+        raise ValueError("platform user identity is required")
+    return f"user:{principal}"
+
+
+def _safe_comparison_id(value: str) -> str:
+    try:
+        return _validate_comparison_id(value)
+    except Exception:
+        return "invalid"
+
+
+def _validate_comparison_id(value: str) -> str:
+    cleaned = str(value or "").strip()
+    if not cleaned:
+        raise ValueError("comparison_id is required")
+    if len(cleaned) > MAX_COMPARISON_ID_LEN or 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 _validate_quotes(quotes: list[QuoteInput]) -> str | None:
+    if len(quotes) < 2:
+        return "at_least_two_quotes_required"
+    if len(quotes) > MAX_QUOTES:
+        return "too_many_quotes"
+    vendors = [q.vendor.casefold() for q in quotes]
+    if len(vendors) != len(set(vendors)):
+        return "duplicate_vendor"
+    quantities = {q.quantity for q in quotes}
+    if len(quantities) != 1:
+        return "quantity_mismatch"
+    return None
+
+
+def _error_message(code: str) -> str:
+    return {
+        "at_least_two_quotes_required": "At least two quotes are required to make a recommendation.",
+        "too_many_quotes": f"Compare at most {MAX_QUOTES} quotes at a time.",
+        "duplicate_vendor": "Each quote must have a unique vendor name.",
+        "quantity_mismatch": "All quotes must use the same quantity for a fair comparison.",
+    }.get(code, "The quote input is invalid.")
+
+
+def _score_lower_is_better(value: float, minimum: float, maximum: float) -> float:
+    if maximum == minimum:
+        return 100.0
+    return ((maximum - value) / (maximum - minimum)) * 100.0
+
+
+def _score_higher_is_better(value: float, minimum: float, maximum: float) -> float:
+    if maximum == minimum:
+        return 100.0
+    return ((value - minimum) / (maximum - minimum)) * 100.0
+
+
+def _build_comparison(comparison_id: str, quotes: list[QuoteInput], weights: WeightInput) -> dict[str, Any]:
+    prices = [q.unit_price for q in quotes]
+    deliveries = [q.delivery_days for q in quotes]
+    warranties = [q.warranty_months for q in quotes]
+    total_weight = weights.price + weights.delivery + weights.warranty
+    rows: list[dict[str, Any]] = []
+    for quote in quotes:
+        components = {
+            "price": round(_score_lower_is_better(quote.unit_price, min(prices), max(prices)), 4),
+            "delivery": round(_score_lower_is_better(quote.delivery_days, min(deliveries), max(deliveries)), 4),
+            "warranty": round(_score_higher_is_better(quote.warranty_months, min(warranties), max(warranties)), 4),
+        }
+        weighted_score = (
+            components["price"] * weights.price
+            + components["delivery"] * weights.delivery
+            + components["warranty"] * weights.warranty
+        ) / total_weight
+        total_price = quote.unit_price * quote.quantity
+        rows.append(
             {
-                    "characters": len(text),
+                **quote.model_dump(),
-                    "words": len(words),
+                "total_price": round(total_price, 2),
-                    "lines": len(text.splitlines()) or 1,
+                "component_scores": components,
+                "weighted_score": round(weighted_score, 4),
             }
         )
-
+    rows.sort(key=lambda item: (-item["weighted_score"], item["total_price"], item["delivery_days"], -item["warranty_months"], item["vendor"].casefold()))
-        @wrap_model_call
+    winner = rows[0]
-        async def log_model_call(request: Any, handler: Any) -> Any:
+    return {
-            messages = request.state.get("messages", [])
+        "ok": True,
-            print(
+        "comparison_id": comparison_id,
-                "[middleware] model_call "
+        "recommendation": {
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+            "vendor": winner["vendor"],
-            )
+            "weighted_score": winner["weighted_score"],
-            return await handler(request)
+            "rationale": _rationale(winner, weights),
-
+        },
-        backend = ctx.workspace_backend()
+        "weights": weights.model_dump(),
-        skill_sources = _seed_runtime_skills(backend, ctx)
+        "quotes": rows,
-        # create_a2a_deep_agent resolves provider:model strings with
+        "created_at": _now_iso(),
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
+        "receipt": {"persisted": True, "tool": "compare_quotes"},
-        # 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,
-        )
 
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+def _rationale(winner: dict[str, Any], weights: WeightInput) -> str:
-    workspace = getattr(ctx, "_workspace", None)
+    strengths: list[str] = []
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+    if winner["component_scores"]["price"] >= 99:
-    if not prefixes:
+        strengths.append("best price")
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+    if winner["component_scores"]["delivery"] >= 99:
-        prefixes = (outputs_prefix or "outputs/",)
+        strengths.append("fastest delivery")
-    prefix = str(prefixes[0]).strip("/")
+    if winner["component_scores"]["warranty"] >= 99:
-    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+        strengths.append("strongest warranty")
+    basis = ", ".join(strengths) if strengths else "best weighted balance"
+    return f"{winner['vendor']} has the highest weighted score using price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}; key advantage: {basis}."
 
 
-def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
+def _error(comparison_id: str, code: str, message: str) -> dict[str, Any]:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    return {"ok": False, "comparison_id": comparison_id, "code": code, "message": message}
 
-    DeepAgents loads skills from its backend, while source-controlled
+
-    ``skills/`` folders live in the image. This bridge lets generated agents
+def _now_iso() -> str:
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
-    files.
+
+
+def _decode_browser_upload(upload: BrowserQuoteUpload) -> dict[str, Any]:
+    if upload.media_type not in ACCEPTED_UPLOAD_MEDIA_TYPES:
+        return {"ok": False, "code": "unsupported_media_type", "message": "Upload must be CSV, plain text CSV, or JSON."}
+    try:
+        data = base64.b64decode(upload.data_base64, validate=True)
+    except Exception:
+        return {"ok": False, "code": "invalid_base64", "message": "Upload data_base64 must be valid base64."}
+    if len(data) > MAX_UPLOAD_BYTES:
+        return {"ok": False, "code": "upload_too_large", "message": f"Upload must be at most {MAX_UPLOAD_BYTES} bytes."}
+    if not data:
+        return {"ok": False, "code": "empty_upload", "message": "Upload file is empty."}
+    return {"ok": True, "bytes": data, "media_type": upload.media_type}
+
+
+def _parse_quote_bytes(data: bytes, media_type: str) -> list[QuoteInput] | dict[str, str]:
+    if len(data) > MAX_UPLOAD_BYTES:
+        return {"code": "upload_too_large", "message": f"Upload must be at most {MAX_UPLOAD_BYTES} bytes."}
+    try:
+        text = data.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        return {"code": "invalid_text_encoding", "message": "Upload must be UTF-8 text."}
+    try:
+        if media_type == "application/json":
+            raw = json.loads(text)
+            rows = raw.get("quotes") if isinstance(raw, dict) else raw
+            if not isinstance(rows, list):
+                return {"code": "invalid_json_quotes", "message": "JSON upload must be an array or an object with a quotes array."}
+            return [QuoteInput.model_validate(item) for item in rows]
+        reader = csv.DictReader(io.StringIO(text))
+        required = {"vendor", "unit_price", "quantity", "delivery_days", "warranty_months"}
+        if not reader.fieldnames or not required <= set(reader.fieldnames):
+            return {"code": "invalid_csv_headers", "message": "CSV must include vendor, unit_price, quantity, delivery_days, and warranty_months headers."}
+        rows = []
+        for row in reader:
+            if len(rows) >= MAX_QUOTES + 1:
+                break
+            rows.append(QuoteInput.model_validate(row))
+        return rows
+    except Exception as exc:
+        return {"code": "quote_parse_failed", "message": f"Could not parse quote rows: {type(exc).__name__}."}
+
+
+def _db_url() -> str | None:
+    return os.environ.get("DATABASE_URL")
+
+
+def _connect() -> Any:
+    import psycopg
+
+    url = _db_url()
+    if not url:
+        raise RuntimeError("DATABASE_URL is not configured for managed Postgres persistence")
+    return psycopg.connect(url, autocommit=True)
+
+
+async def _upsert_comparison(tenant_key: str, result: dict[str, Any]) -> None:
+    with _connect() as conn:
+        conn.execute(
             """
-    root = Path(__file__).parent / "skills"
+            INSERT INTO quote_comparisons (tenant_key, comparison_id, recommendation_vendor, weighted_score, payload, updated_at)
-    if not root.exists():
+            VALUES (%s, %s, %s, %s, %s::jsonb, NOW())
-        return []
+            ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
-    runtime_skills_root = _runtime_skills_root(ctx)
+                recommendation_vendor = EXCLUDED.recommendation_vendor,
-    uploads: list[tuple[str, bytes]] = []
+                weighted_score = EXCLUDED.weighted_score,
-    for path in root.rglob("*"):
+                payload = EXCLUDED.payload,
-        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_key,
-        backend.upload_files(uploads)
+                result["comparison_id"],
-        return [runtime_skills_root]
+                result["recommendation"]["vendor"],
-    return []
+                result["recommendation"]["weighted_score"],
+                json.dumps(result, separators=(",", ":")),
+            ),
+        )
 
 
-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 []
+    with _connect() as conn:
-    if not messages:
+        row = conn.execute(
-        return json.dumps(state, default=str)
+            "SELECT payload FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s",
+            (tenant_key, comparison_id),
+        ).fetchone()
+    if row is None:
+        return None
+    payload = row[0]
+    if isinstance(payload, str):
+        payload = json.loads(payload)
+    if isinstance(payload, dict):
+        payload = dict(payload)
+        payload["reloaded"] = True
+        payload["receipt"] = {"persisted": True, "tool": "get_comparison"}
+        return payload
+    return None
 
-    content = getattr(messages[-1], "content", None)
+
-    if isinstance(content, str):
+async def _persist_receipt(
-        return content
+    ctx: RunContext[PlatformUserAuth],
-    if isinstance(content, list):
+    tenant_key: str,
-        parts: list[str] = []
+    tool_name: str,
-        for item in content:
+    comparison_id: str,
-            if isinstance(item, dict):
+    inputs: dict[str, Any],
-                text = item.get("text") or item.get("content")
+    result: dict[str, Any],
-                if text:
+) -> None:
-                    parts.append(str(text))
+    receipt_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{tenant_key}:{tool_name}:{comparison_id}:{hashlib.sha256(json.dumps(inputs, sort_keys=True, default=str).encode()).hexdigest()}"))
-            elif item:
+    payload = {
-                parts.append(str(item))
+        "receipt_id": receipt_id,
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+        "agent": QuoteJudgeStudioV1.name,
-    return str(content or messages[-1])
+        "version": QuoteJudgeStudioV1.version,
+        "tool": tool_name,
+        "comparison_id": comparison_id,
+        "task_id": getattr(ctx, "task_id", ""),
+        "ok": bool(result.get("ok")),
+        "input_hash": hashlib.sha256(json.dumps(inputs, sort_keys=True, default=str).encode()).hexdigest(),
+        "created_at": _now_iso(),
+    }
+    try:
+        with _connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO quote_execution_receipts (tenant_key, receipt_id, comparison_id, tool_name, ok, payload)
+                VALUES (%s, %s, %s, %s, %s, %s::jsonb)
+                ON CONFLICT (tenant_key, receipt_id) DO UPDATE SET payload = EXCLUDED.payload
+                """,
+                (tenant_key, receipt_id, comparison_id, tool_name, bool(result.get("ok")), json.dumps(payload, separators=(",", ":"))),
+            )
+    except Exception:
+        # Receipts are required in production, but a local no-DATABASE_URL unit
+        # test should still be able to exercise deterministic validation. The
+        # main persistence path raises if saving the comparison itself fails.
+        if _db_url():
+            raise