quote-judge-studio-v1/agent.py

"""QuoteJudge Studio v1: deterministic quote comparison product agent.

This agent intentionally does not call an LLM. It exposes a compact typed A2A
surface, a packed React app, managed Postgres persistence, MCP-compatible tools,
and bounded upload paths for both browser JSON and external FileUpload clients.
"""
from __future__ import annotations

import base64
import binascii
import csv
import hashlib
import io
import json
import os
import re
from datetime import UTC, datetime
from typing import Annotated, Any

from pydantic import BaseModel, Field, field_validator

import a2a_pack as a2a
from a2a_pack import (
    A2AAgent,
    AgentDatabase,
    AgentDatabaseEnv,
    AgentDatabaseMigrations,
    AgentPlatformResources,
    FileUpload,
    PlatformUserAuth,
    Pricing,
    RunContext,
    UploadedFile,
    WorkspaceAccess,
    WorkspaceMode,
)

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


class QuoteInput(BaseModel):
    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)

    @field_validator("vendor")
    @classmethod
    def _clean_vendor(cls, value: str) -> str:
        cleaned = " ".join(str(value).strip().split())
        if not cleaned:
            raise ValueError("vendor is required")
        return cleaned


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 bounded uploads, and "
        "exposes typed MCP-compatible A2A tools."
    )
    version = "0.1.1"

    config_model = QuoteJudgeStudioV1Config
    auth_model = PlatformUserAuth

    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic quote scoring and managed Postgres persistence; no LLM required.",
    )
    workspace_access = WorkspaceAccess.dynamic(
        max_files=4,
        allowed_modes=(WorkspaceMode.READ_ONLY,),
        require_reason=False,
        max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_UPLOAD_FILES,
    )
    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="Compare at least two vendor quotes with price/delivery/warranty weights and persist a production receipt.",
        timeout_seconds=30,
        cost_class="deterministic",
    )
    async def compare_quotes(
        self,
        ctx: RunContext[PlatformUserAuth],
        comparison_id: ComparisonId,
        quotes: QuoteList,
        weights: QuoteWeights,
    ) -> dict[str, Any]:
        tenant = _tenant_key(ctx)
        validation = _validate_compare_inputs(comparison_id, quotes, weights)
        if validation is not None:
            await _persist_receipt(ctx, tenant, "compare_quotes", {"comparison_id": comparison_id}, validation)
            return validation

        result = _score_quotes(comparison_id, quotes, weights)
        result["tenant_scope"] = "user"
        result["receipt"] = await _persist_receipt(
            ctx,
            tenant,
            "compare_quotes",
            {
                "comparison_id": comparison_id,
                "quote_count": len(quotes),
                "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(
        key=lambda row: (
            -row["scores"]["weighted_total"],
            row["total_price"],
            row["delivery_days"],
            -row["warranty_months"],
            row["vendor"].casefold(),
        )
    )
    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 _lower_is_better(value: float, min_value: float, max_value: float) -> float:
    if max_value == min_value:
        return 100.0
    return max(0.0, min(100.0, 100.0 * (max_value - value) / (max_value - min_value)))


def _higher_is_better(value: float, min_value: float, max_value: float) -> float:
    if max_value == min_value:
        return 100.0
    return max(0.0, min(100.0, 100.0 * (value - min_value) / (max_value - min_value)))


def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
    all_quotes: list[QuoteInput] = []
    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(
                """
                INSERT INTO quote_judge_comparisons
                    (tenant_key, comparison_id, recommendation_vendor, result_json, updated_at)
                VALUES (%s, %s, %s, %s, NOW())
                ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
                    recommendation_vendor = EXCLUDED.recommendation_vendor,
                    result_json = EXCLUDED.result_json,
                    updated_at = NOW()
                """,
                (
                    tenant,
                    result["comparison_id"],
                    result["recommendation"]["vendor"],
                    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.",
        }


async def _load_comparison(tenant: str, comparison_id: str) -> dict[str, Any] | None:
    key = (tenant, comparison_id)
    db_url = os.environ.get("DATABASE_URL")
    if not db_url:
        return _LOCAL_COMPARISONS.get(key)
    try:
        import psycopg

        with psycopg.connect(db_url) as conn:
            _ensure_schema(conn)
            row = conn.execute(
                """
                SELECT result_json FROM quote_judge_comparisons
                WHERE tenant_key = %s AND comparison_id = %s
                ORDER BY updated_at DESC LIMIT 1
                """,
                (tenant, comparison_id),
            ).fetchone()
            if row and row[0]:
                return dict(row[0])
            row = conn.execute(
                """
                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,
            receipt_id TEXT,
            tenant_key TEXT NOT NULL,
            comparison_id TEXT,
            skill_name TEXT,
            tool_name TEXT,
            status TEXT,
            input_hash TEXT,
            result_hash TEXT,
            receipt_json JSONB,
            payload JSONB,
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
        )
        """
    )
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS receipt_id TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS skill_name TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS tool_name TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS status TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS input_hash TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS result_hash TEXT")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS receipt_json JSONB")
    conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS payload JSONB")
    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,
        "tool_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
                    (receipt_id, tenant_key, comparison_id, skill_name, tool_name,
                     status, input_hash, result_hash, receipt_json, payload)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    receipt["receipt_id"],
                    tenant,
                    comparison_id or None,
                    skill_name,
                    skill_name,
                    receipt["status"],
                    receipt["input_hash"],
                    receipt["result_hash"],
                    Jsonb(receipt),
                    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()