a2a-source-edit: write agent.py

agent.py

@@ -1,594 +1,189 @@
-"""QuoteJudge Studio full-stack A2A agent.
+"""quote-judge-studio-v1 agent.
 
-Deterministic product backend for comparing vendor quotes with weighted price,
+Starter stack:
-delivery, and warranty scoring. The packed React frontend calls these same
+  - DeepAgents for tool-calling orchestration
-public tools, and external clients retain typed FileUpload support.
+  - 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
-import os
+from pathlib import Path
-import re
+from typing import Any
-import uuid
-from datetime import datetime, timezone
-from decimal import Decimal, ROUND_HALF_UP
-from typing import Annotated, Any
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    AgentDatabase,
+    LLMProvisioning,
-    AgentDatabaseEnv,
+    {{ auth_type }},
-    AgentDatabaseMigrations,
-    AgentPlatformResources,
-    FileUpload,
-    PlatformUserAuth,
     Pricing,
-    Resources,
     RunContext,
-    State,
-    UploadedFile,
     WorkspaceAccess,
     WorkspaceMode,
 )
-
+from a2a_pack.context import LLMCreds
-MAX_BROWSER_UPLOAD_BYTES = 256_000
-MAX_BROWSER_UPLOADS = 4
-ALLOWED_UPLOAD_MEDIA_TYPES = {
-    "application/json",
-    "text/csv",
-    "text/plain",
-    "application/vnd.ms-excel",
-}
-
-_MEMORY_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {}
-_MEMORY_RECEIPTS: list[dict[str, Any]] = []
 
 
 class QuoteJudgeStudioV1Config(BaseModel):
-    """No user-configurable settings; all state is tenant-scoped by platform auth."""
+    pass
 
 
-class QuoteJudgeDurableState(BaseModel):
+SYSTEM_PROMPT = """\
-    """Marker state model: durable product state lives in managed Postgres."""
+You are a compact tool-calling agent.
 
-    database: str = "quote-judge-studio-v1-data"
+Use the text_stats tool when the user asks about text, counts, summaries,
+or anything where exact length/word numbers would help. Mention tool results
+briefly instead of dumping raw JSON.
+"""
+
+RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/"
+DEEPAGENTS_RECURSION_LIMIT = 500
 
 
-class QuoteInput(BaseModel):
+class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
-    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=100_000_000)
-    delivery_days: int = Field(..., ge=0, le=3650)
-    warranty_months: int = Field(..., ge=0, le=600)
-    currency: str = Field(default="BRL", min_length=3, max_length=8)
-
-    @field_validator("vendor", "currency")
-    @classmethod
-    def _clean_text(cls, value: str) -> str:
-        return " ".join(str(value).strip().split())
-
-
-class WeightInput(BaseModel):
-    price: float = Field(..., ge=0, le=100)
-    delivery: float = Field(..., ge=0, le=100)
-    warranty: float = Field(..., ge=0, le=100)
-
-
-class BrowserQuoteDocument(BaseModel):
-    filename: str = Field(..., min_length=1, max_length=160)
-    media_type: str = Field(..., min_length=3, max_length=120)
-    data_base64: str = Field(..., min_length=1)
-
-    @field_validator("filename")
-    @classmethod
-    def _safe_filename(cls, value: str) -> str:
-        cleaned = str(value).replace("\\", "/").split("/")[-1].strip()
-        if not cleaned or cleaned in {".", ".."}:
-            raise ValueError("filename must be a simple file name")
-        return cleaned
-
-
-class ValidationIssue(BaseModel):
-    code: str
-    message: str
-    field: str | None = None
-    action: str | None = None
-
-
-class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
     name = "quote-judge-studio-v1"
-    description = (
+    description = "QuoteJudge Studio compares vendor quotes with weighted price, delivery, and warranty scoring, persists results, supports uploads, and exposes MCP-compatible A2A tools."
-        "Compare at least two vendor quotes with price, delivery, and warranty "
+    version = "0.1.0"
-        "weights; persist results; reopen prior comparisons; accept browser "
-        "base64 uploads and typed FileUpload inputs; and persist execution receipts."
-    )
-    version = "0.1.3"
 
     config_model = QuoteJudgeStudioV1Config
-    auth_model = PlatformUserAuth
+    auth_model = {{ auth_type }}
 
-    state = State.DURABLE
+    # Hosted generated agents read the caller's saved LLM credential through
-    state_model = QuoteJudgeDurableState
+    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
-    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
+    # 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=False,
+        caller_pays_llm=True,
-        notes="Deterministic quote scoring; no LLM credential required.",
+        notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
     )
     workspace_access = WorkspaceAccess.dynamic(
-        max_files=8,
+        max_files=64,
         allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
         require_reason=False,
-        max_total_size_bytes=2_000_000,
     )
-    tools_used = ("postgres", "mcp", "file-upload")
+    tools_used = ("deepagents", "langchain")
-    platform_resources = AgentPlatformResources(
+
-        databases=(
+    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
-            AgentDatabase(
+    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
-                name="quote-judge-studio-v1-data",
+        creds = ctx.llm
-                provider="neon",
+        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
-                engine="postgres",
+        if not creds.api_key:
-                scope="user",
+            return (
-                branch="main",
+                "LLM key required. Add an LLM credential in Settings > LLM "
-                access_mode="read_write",
+                "credentials before running this agent; for local --invoke "
-                env=AgentDatabaseEnv(url="DATABASE_URL"),
+                "runs set AGENT_LLM_KEY."
-                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            )
-                scale_to_zero=True,
+        graph = self._build_deep_agent(ctx=ctx, creds=creds)
-            ),
+        state = await graph.ainvoke(
+            {"messages": [{"role": "user", "content": prompt}]},
+            config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
         )
-    )
+        await ctx.emit_progress("deepagent finished")
+        return _last_message_text(state)
 
-    @a2a.tool(
+    def _build_deep_agent(
-        description="Compare vendor quotes with price/delivery/warranty weights and persist the result plus a production receipt.",
-        timeout_seconds=60,
-        idempotent=True,
-        cost_class="deterministic",
-    )
-    async def compare_quotes(
         self,
-        ctx: RunContext[PlatformUserAuth],
+        *,
-        comparison_id: str,
+        ctx: RunContext[{{ auth_type }}],
-        quotes: list[QuoteInput],
+        creds: LLMCreds,
-        weights: WeightInput,
+    ) -> Any:
-    ) -> dict[str, Any]:
+        # Lazy imports keep `a2a card` usable before local dependencies are
-        tenant = _tenant_key(ctx)
+        # installed. `a2a deploy` installs requirements.txt during the build.
-        issues = _validate_request(comparison_id, quotes, weights)
+        from a2a_pack.deepagents import create_a2a_deep_agent
-        if issues:
+        from langchain.agents.middleware import wrap_model_call
-            result = _validation_result(comparison_id, issues)
+        from langchain_core.tools import tool
-            await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", result, {"comparison_id": comparison_id, "quotes": [q.model_dump() for q in quotes], "weights": weights.model_dump()})
-            return result
 
-        result = _score_quotes(comparison_id, quotes, weights)
+        @tool
-        result["receipt"] = await _persist_receipt(
+        def text_stats(text: str) -> str:
+            """Return exact word, character, and line counts for text."""
+            words = [part for part in text.split() if part.strip()]
+            return json.dumps(
+                {
+                    "characters": len(text),
+                    "words": len(words),
+                    "lines": len(text.splitlines()) or 1,
+                }
+            )
+
+        @wrap_model_call
+        async def log_model_call(request: Any, handler: Any) -> Any:
+            messages = request.state.get("messages", [])
+            print(
+                "[middleware] model_call "
+                f"model={creds.model} source={creds.source} messages={len(messages)}"
+            )
+            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,
-            tenant,
+            creds=creds,
-            comparison_id,
+            backend=backend,
-            "compare_quotes",
+            skills=skill_sources or None,
-            result,
+            tools=[text_stats],
-            {"comparison_id": comparison_id, "quotes": [q.model_dump() for q in quotes], "weights": weights.model_dump()},
+            middleware=[log_model_call],
+            system_prompt=SYSTEM_PROMPT,
         )
-        await _save_comparison(tenant, comparison_id, result)
-        await ctx.emit_progress(f"saved comparison {comparison_id}: {result['recommendation']['vendor']}")
-        return result
-
-    @a2a.tool(
-        description="Reopen a persisted QuoteJudge comparison by id for the authenticated user.",
-        timeout_seconds=30,
-        idempotent=True,
-        cost_class="deterministic",
-    )
-    async def get_comparison(
-        self,
-        ctx: RunContext[PlatformUserAuth],
-        comparison_id: str,
-    ) -> dict[str, Any]:
-        tenant = _tenant_key(ctx)
-        if not _valid_comparison_id(comparison_id):
-            return _validation_result(
-                comparison_id,
-                [ValidationIssue(code="invalid_comparison_id", message="comparison_id must contain only letters, numbers, dot, underscore, or hyphen.", field="comparison_id", action="Use a stable id such as studio-quote-v1.")],
-            )
-        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 id.",
-                "action": "Run compare_quotes or upload/paste at least two quotes first.",
-            }
-        result = dict(result)
-        result["receipt"] = await _persist_receipt(
-            ctx,
-            tenant,
-            comparison_id,
-            "get_comparison",
-            result,
-            {"comparison_id": comparison_id},
-        )
-        return result
-
-    @a2a.tool(
-        description="Browser-safe upload bridge: parse bounded base64 JSON/CSV/text quote files, compare, persist, and return a receipt.",
-        timeout_seconds=60,
-        idempotent=True,
-        cost_class="deterministic",
-    )
-    async def compare_uploaded_quotes_browser(
-        self,
-        ctx: RunContext[PlatformUserAuth],
-        comparison_id: str,
-        documents: list[BrowserQuoteDocument],
-        weights: WeightInput,
-    ) -> dict[str, Any]:
-        parsed = _parse_browser_documents(documents)
-        if not parsed["ok"]:
-            return _validation_result(comparison_id, parsed["issues"])
-        return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
-
-    @a2a.tool(
-        description="Compatibility FileUpload entrypoint for external clients; parses an uploaded quote file and compares/persists results.",
-        timeout_seconds=60,
-        idempotent=True,
-        cost_class="deterministic",
-    )
-    async def compare_uploaded_quote_file(
-        self,
-        ctx: RunContext[PlatformUserAuth],
-        comparison_id: str,
-        quote_file: Annotated[
-            UploadedFile,
-            FileUpload(
-                accept=("application/json", "text/csv", "text/plain"),
-                max_bytes=MAX_BROWSER_UPLOAD_BYTES,
-                description="JSON array/object, CSV, or line-oriented quote text with vendor/unit_price/quantity/delivery_days/warranty_months.",
-            ),
-        ],
-        weights: WeightInput,
-    ) -> dict[str, Any]:
-        issues = _validate_uploaded_file_meta(quote_file.filename, quote_file.media_type, quote_file.size_bytes)
-        if issues:
-            return _validation_result(comparison_id, issues)
-        try:
-            data = ctx.workspace.read_bytes(quote_file.path)  # concrete runtime clients expose this helper
-        except Exception:  # noqa: BLE001
-            return _validation_result(
-                comparison_id,
-                [ValidationIssue(code="upload_read_failed", message="The uploaded file could not be read from the invocation workspace.", field="quote_file", action=f"Re-upload the file and ensure it is below {MAX_BROWSER_UPLOAD_BYTES} bytes.")],
-            )
-        parsed = _parse_document_bytes(quote_file.filename, quote_file.media_type, data)
-        if not parsed["ok"]:
-            return _validation_result(comparison_id, parsed["issues"])
-        return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
 
 
-def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
+def _runtime_skills_root(ctx: RunContext[Any]) -> str:
-    auth = ctx.auth
+    workspace = getattr(ctx, "_workspace", None)
-    stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
+    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
-    return f"user:{stable}"
+    if not prefixes:
+        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+        prefixes = (outputs_prefix or "outputs/",)
+    prefix = str(prefixes[0]).strip("/")
+    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
 
 
-def _valid_comparison_id(value: str) -> bool:
+def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
-    return bool(re.fullmatch(r"[A-Za-z0-9._-]{1,96}", str(value or "")))
+    """Copy packaged DeepAgents skills into the invocation workspace.
+
+    DeepAgents loads skills from its backend, while source-controlled
+    ``skills/`` folders live in the image. This bridge lets generated agents
+    ship reusable SKILL.md bundles without giving up durable A2A workspace
+    files.
+    """
+    root = Path(__file__).parent / "skills"
+    if not root.exists():
+        return []
+    runtime_skills_root = _runtime_skills_root(ctx)
+    uploads: list[tuple[str, bytes]] = []
+    for path in root.rglob("*"):
+        if path.is_file():
+            rel = path.relative_to(root).as_posix()
+            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+    if uploads:
+        backend.upload_files(uploads)
+        return [runtime_skills_root]
+    return []
 
 
-def _validate_request(comparison_id: str, quotes: list[QuoteInput], weights: WeightInput) -> list[ValidationIssue]:
+def _last_message_text(state: dict[str, Any]) -> str:
-    issues: list[ValidationIssue] = []
+    messages = state.get("messages") or []
-    if not _valid_comparison_id(comparison_id):
+    if not messages:
-        issues.append(ValidationIssue(code="invalid_comparison_id", message="comparison_id must contain only letters, numbers, dot, underscore, or hyphen.", field="comparison_id", action="Use a stable id such as studio-quote-v1."))
+        return json.dumps(state, default=str)
-    if len(quotes) < 2:
-        issues.append(ValidationIssue(code="at_least_two_quotes_required", message="QuoteJudge needs at least two vendor quotes to make a comparison.", field="quotes", action="Add at least one more vendor quote."))
-    vendors = [q.vendor.casefold() for q in quotes]
-    if len(vendors) != len(set(vendors)):
-        issues.append(ValidationIssue(code="duplicate_vendor", message="Each quote vendor must be unique within a comparison.", field="quotes.vendor", action="Rename or merge duplicate vendor rows."))
-    total_weight = weights.price + weights.delivery + weights.warranty
-    if total_weight <= 0:
-        issues.append(ValidationIssue(code="positive_weight_required", message="At least one scoring weight must be greater than zero.", field="weights", action="Set price, delivery, or warranty to a positive weight."))
-    return issues
 
-
+    content = getattr(messages[-1], "content", None)
-def _validation_result(comparison_id: str, issues: list[ValidationIssue]) -> dict[str, Any]:
+    if isinstance(content, str):
-    code = issues[0].code if issues else "validation_failed"
+        return content
-    return {
+    if isinstance(content, list):
-        "ok": False,
+        parts: list[str] = []
-        "code": code,
+        for item in content:
-        "comparison_id": comparison_id,
+            if isinstance(item, dict):
-        "message": issues[0].message if issues else "The request could not be validated.",
+                text = item.get("text") or item.get("content")
-        "issues": [issue.model_dump() for issue in issues],
+                if text:
-    }
+                    parts.append(str(text))
-
+            elif item:
-
+                parts.append(str(item))
-def _rank_key(item: dict[str, Any]) -> tuple[float, float, int, int, str]:
+        return "\n".join(parts) if parts else json.dumps(content, default=str)
-    return (-item["scores"]["weighted_total"], item["total_price"], item["delivery_days"], -item["warranty_months"], item["vendor"])
+    return str(content or messages[-1])
-
-
-def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: WeightInput) -> dict[str, Any]:
-    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,
-    }
-    totals = [q.unit_price * q.quantity for q in quotes]
-    deliveries = [q.delivery_days for q in quotes]
-    warranties = [q.warranty_months for q in quotes]
-
-    rows: list[dict[str, Any]] = []
-    for index, quote in enumerate(quotes):
-        price_score = _lower_is_better(totals[index], totals)
-        delivery_score = _lower_is_better(deliveries[index], deliveries)
-        warranty_score = _higher_is_better(warranties[index], warranties)
-        weighted_score = (
-            price_score * normalized_weights["price"]
-            + delivery_score * normalized_weights["delivery"]
-            + warranty_score * normalized_weights["warranty"]
-        )
-        rows.append(
-            {
-                "vendor": quote.vendor,
-                "unit_price": _money(quote.unit_price),
-                "quantity": quote.quantity,
-                "total_price": _money(quote.unit_price * quote.quantity),
-                "currency": quote.currency,
-                "delivery_days": quote.delivery_days,
-                "warranty_months": quote.warranty_months,
-                "scores": {
-                    "price": _round_score(price_score),
-                    "delivery": _round_score(delivery_score),
-                    "warranty": _round_score(warranty_score),
-                    "weighted_total": _round_score(weighted_score),
-                },
-            }
-        )
-    rows.sort(key=_rank_key)
-    winner = rows[0]
-    return {
-        "ok": True,
-        "comparison_id": comparison_id,
-        "recommendation": {
-            "vendor": winner["vendor"],
-            "weighted_score": winner["scores"]["weighted_total"],
-            "why": _recommendation_reason(winner),
-        },
-        "weights": {
-            "price": weights.price,
-            "delivery": weights.delivery,
-            "warranty": weights.warranty,
-            "normalized": {key: _round_score(value * 100) for key, value in normalized_weights.items()},
-        },
-        "quotes_ranked": rows,
-        "validation": {"ok": True, "issues": []},
-        "saved_at": datetime.now(timezone.utc).isoformat(),
-    }
-
-
-def _lower_is_better(value: float, values: list[float]) -> float:
-    low, high = min(values), max(values)
-    if high == low:
-        return 100.0
-    return 100.0 * (high - value) / (high - low)
-
-
-def _higher_is_better(value: float, values: list[float]) -> float:
-    low, high = min(values), max(values)
-    if high == low:
-        return 100.0
-    return 100.0 * (value - low) / (high - low)
-
-
-def _round_score(value: float) -> float:
-    return float(Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
-
-
-def _money(value: float) -> float:
-    return float(Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
-
-
-def _recommendation_reason(row: dict[str, Any]) -> str:
-    return (
-        f"{row['vendor']} has the highest weighted score ({row['scores']['weighted_total']}) "
-        f"after balancing total price {row['currency']} {row['total_price']}, "
-        f"{row['delivery_days']} delivery days, and {row['warranty_months']} months warranty."
-    )
-
-
-def _connect() -> Any | None:
-    url = os.environ.get("DATABASE_URL")
-    if not url:
-        return None
-    import psycopg
-    from psycopg.rows import dict_row
-
-    return psycopg.connect(url, row_factory=dict_row)
-
-
-async def _save_comparison(tenant: str, comparison_id: str, result: dict[str, Any]) -> None:
-    conn = _connect()
-    if conn is None:
-        _MEMORY_COMPARISONS[(tenant, comparison_id)] = json.loads(json.dumps(result))
-        return
-    with conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                """
-                INSERT INTO quote_comparisons (tenant_key, comparison_id, recommendation_vendor, result, payload, weighted_score, updated_at)
-                VALUES (%s, %s, %s, %s, %s, %s, NOW())
-                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
-                    recommendation_vendor = EXCLUDED.recommendation_vendor,
-                    result = EXCLUDED.result,
-                    payload = EXCLUDED.payload,
-                    weighted_score = EXCLUDED.weighted_score,
-                    updated_at = NOW()
-                """,
-                (
-                    tenant,
-                    comparison_id,
-                    result["recommendation"]["vendor"],
-                    json.dumps(result),
-                    json.dumps(result),
-                    result["recommendation"]["weighted_score"],
-                ),
-            )
-
-
-async def _load_comparison(tenant: str, comparison_id: str) -> dict[str, Any] | None:
-    conn = _connect()
-    if conn is None:
-        stored = _MEMORY_COMPARISONS.get((tenant, comparison_id))
-        return json.loads(json.dumps(stored)) if stored is not None else None
-    with conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT COALESCE(result, payload) AS result FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s",
-                (tenant, comparison_id),
-            )
-            row = cur.fetchone()
-            if not row:
-                return None
-            value = row["result"]
-            return value if isinstance(value, dict) else json.loads(value)
-
-
-async def _persist_receipt(
-    ctx: RunContext[PlatformUserAuth],
-    tenant: str,
-    comparison_id: str,
-    tool_name: str,
-    result: dict[str, Any],
-    inputs: dict[str, Any],
-) -> dict[str, Any]:
-    now = datetime.now(timezone.utc).isoformat()
-    input_hash = _sha256_json(inputs)
-    result_hash = _sha256_json({k: v for k, v in result.items() if k != "receipt"})
-    receipt = {
-        "receipt_id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"{tenant}:{comparison_id}:{tool_name}:{input_hash}:{result_hash}")),
-        "agent": QuoteJudgeStudioV1.name,
-        "agent_version": QuoteJudgeStudioV1.version,
-        "tool_name": tool_name,
-        "comparison_id": comparison_id,
-        "status": "ok" if result.get("ok") else "validation_error",
-        "input_hash": input_hash,
-        "result_hash": result_hash,
-        "created_at": now,
-        "production": True,
-    }
-    conn = _connect()
-    if conn is None:
-        _MEMORY_RECEIPTS.append({"tenant_key": tenant, "payload": receipt})
-    else:
-        with conn:
-            with conn.cursor() as cur:
-                cur.execute(
-                    """
-                    INSERT INTO quote_judge_receipts (receipt_id, tenant_key, comparison_id, tool_name, status, input_hash, payload)
-                    VALUES (%s, %s, %s, %s, %s, %s, %s)
-                    ON CONFLICT (receipt_id) DO UPDATE SET payload = EXCLUDED.payload
-                    """,
-                    (receipt["receipt_id"], tenant, comparison_id, tool_name, receipt["status"], input_hash, json.dumps(receipt)),
-                )
-    await ctx.emit_event(a2a.AgentEvent(kind="quote_judge_receipt_persisted", payload={"receipt_id": receipt["receipt_id"], "comparison_id": comparison_id, "tool_name": tool_name}))
-    return receipt
-
-
-def _sha256_json(value: Any) -> str:
-    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
-    return hashlib.sha256(payload).hexdigest()
-
-
-def _parse_browser_documents(documents: list[BrowserQuoteDocument]) -> dict[str, Any]:
-    issues: list[ValidationIssue] = []
-    if not documents:
-        issues.append(ValidationIssue(code="upload_required", message="Upload at least one JSON, CSV, or text file containing two or more quotes.", field="documents", action="Choose a quote file or paste quote rows instead."))
-    if len(documents) > MAX_BROWSER_UPLOADS:
-        issues.append(ValidationIssue(code="too_many_uploads", message=f"At most {MAX_BROWSER_UPLOADS} files can be uploaded at once.", field="documents", action="Upload fewer files or merge quote rows."))
-    quotes: list[QuoteInput] = []
-    for index, doc in enumerate(documents[:MAX_BROWSER_UPLOADS]):
-        if doc.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
-            issues.append(ValidationIssue(code="unsupported_media_type", message=f"{doc.filename} has unsupported media type {doc.media_type}.", field=f"documents[{index}].media_type", action="Use JSON, CSV, or plain text."))
-            continue
-        try:
-            data = base64.b64decode(doc.data_base64, validate=True)
-        except Exception:  # noqa: BLE001
-            issues.append(ValidationIssue(code="invalid_base64", message=f"{doc.filename} is not valid base64.", field=f"documents[{index}].data_base64", action="Re-select the file so the browser can encode it again."))
-            continue
-        if len(data) > MAX_BROWSER_UPLOAD_BYTES:
-            issues.append(ValidationIssue(code="upload_too_large", message=f"{doc.filename} exceeds {MAX_BROWSER_UPLOAD_BYTES} bytes.", field=f"documents[{index}]", action="Upload a smaller quote file."))
-            continue
-        parsed = _parse_document_bytes(doc.filename, doc.media_type, data)
-        if parsed["ok"]:
-            quotes.extend(parsed["quotes"])
-        else:
-            issues.extend(parsed["issues"])
-    if issues:
-        return {"ok": False, "issues": issues}
-    return {"ok": True, "quotes": quotes}
-
-
-def _validate_uploaded_file_meta(filename: str, media_type: str, size_bytes: int) -> list[ValidationIssue]:
-    issues: list[ValidationIssue] = []
-    if media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
-        issues.append(ValidationIssue(code="unsupported_media_type", message=f"{filename} has unsupported media type {media_type}.", field="quote_file.media_type", action="Upload JSON, CSV, or plain text."))
-    if size_bytes > MAX_BROWSER_UPLOAD_BYTES:
-        issues.append(ValidationIssue(code="upload_too_large", message=f"{filename} exceeds {MAX_BROWSER_UPLOAD_BYTES} bytes.", field="quote_file", action="Upload a smaller file."))
-    return issues
-
-
-def _parse_document_bytes(filename: str, media_type: str, data: bytes) -> dict[str, Any]:
-    try:
-        text = data.decode("utf-8-sig")
-    except UnicodeDecodeError:
-        return {"ok": False, "issues": [ValidationIssue(code="upload_not_utf8", message=f"{filename} must be UTF-8 text, CSV, or JSON.", field="quote_file", action="Export the quote file as UTF-8.")]}
-    try:
-        if media_type == "application/json" or filename.lower().endswith(".json"):
-            raw = json.loads(text)
-            rows = raw.get("quotes", raw) if isinstance(raw, dict) else raw
-            if not isinstance(rows, list):
-                raise ValueError("JSON must be an array or an object with a quotes array")
-            quotes = [QuoteInput.model_validate(row) for row in rows]
-        elif media_type in {"text/csv", "application/vnd.ms-excel"} or filename.lower().endswith(".csv"):
-            reader = csv.DictReader(io.StringIO(text))
-            quotes = [QuoteInput.model_validate(_normalize_quote_row(csv_row)) for csv_row in reader]
-        else:
-            quotes = _parse_plain_text_quotes(text)
-        return {"ok": True, "quotes": quotes}
-    except Exception as exc:  # noqa: BLE001
-        return {"ok": False, "issues": [ValidationIssue(code="quote_parse_failed", message=f"Could not parse {filename}: {exc}", field="quote_file", action="Use columns vendor, unit_price, quantity, delivery_days, warranty_months.")]}
-
-
-def _normalize_quote_row(row: dict[str, Any]) -> dict[str, Any]:
-    aliases = {
-        "vendor": ["vendor", "supplier", "name"],
-        "unit_price": ["unit_price", "unit price", "price", "unit_cost"],
-        "quantity": ["quantity", "qty", "units"],
-        "delivery_days": ["delivery_days", "delivery days", "lead_time", "lead time", "days"],
-        "warranty_months": ["warranty_months", "warranty months", "warranty", "months"],
-        "currency": ["currency"],
-    }
-    lower = {str(k).strip().lower(): v for k, v in row.items()}
-    out: dict[str, Any] = {}
-    for target, keys in aliases.items():
-        for key in keys:
-            if key in lower and lower[key] not in (None, ""):
-                out[target] = lower[key]
-                break
-    return out
-
-
-def _parse_plain_text_quotes(text: str) -> list[QuoteInput]:
-    quotes: list[QuoteInput] = []
-    for line in text.splitlines():
-        if not line.strip():
-            continue
-        parts = [part.strip() for part in re.split(r"[,;|]\s*", line) if part.strip()]
-        if len(parts) >= 5:
-            quotes.append(QuoteInput(vendor=parts[0], unit_price=float(parts[1]), quantity=int(float(parts[2])), delivery_days=int(float(parts[3])), warranty_months=int(float(parts[4]))))
-    if not quotes:
-        raise ValueError("plain text rows must be vendor, unit_price, quantity, delivery_days, warranty_months")
-    return quotes