From 3468f58f1259751401a11b9b662ae079826160ad Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 06:08:41 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 687 +++++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 543 insertions(+), 144 deletions(-) diff --git a/agent.py b/agent.py index 95e315f..58eed19 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,588 @@ -"""quote-judge-studio-v1 agent. - -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 -""" +"""QuoteJudge Studio v1: deterministic quote comparison with durable per-user state.""" from __future__ import annotations +import base64 +import csv +import hashlib +import io import json -from pathlib import Path -from typing import Any +import os +import re +from datetime import datetime, timezone +from decimal import Decimal, ROUND_HALF_UP +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, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + PlatformUserAuth, Pricing, + Resources, RunContext, + State, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds +from a2a_pack.workspace import FileUpload, UploadedFile + +try: # psycopg is installed in the deployed image via requirements.txt. + import psycopg + from psycopg.rows import dict_row + from psycopg.types.json import Jsonb +except Exception: # pragma: no cover - lets `a2a card` run before deps install. + psycopg = None + dict_row = None + Jsonb = None + + +COMPARISON_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,79}$") +MAX_QUOTES = 10 +MAX_BROWSER_DOCUMENTS = 4 +MAX_UPLOAD_BYTES = 64 * 1024 +ALLOWED_UPLOAD_MEDIA_TYPES = { + "application/json", + "text/json", + "text/csv", + "application/csv", + "application/vnd.ms-excel", + "text/plain", +} + +# Local/dev fallback only. Hosted production uses DATABASE_URL and managed Postgres. +_LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {} +_LOCAL_RECEIPTS: list[dict[str, Any]] = [] class QuoteJudgeStudioV1Config(BaseModel): - pass + """Runtime config is intentionally empty; the app uses platform auth + DB.""" -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +class QuoteInput(BaseModel): + vendor: str = Field(..., min_length=1, max_length=120) + quantity: int = Field(..., ge=1, le=1_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=240) -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 + @field_validator("vendor") + @classmethod + def clean_vendor(cls, value: str) -> str: + cleaned = " ".join(str(value or "").split()) + if not cleaned: + raise ValueError("vendor is required") + return cleaned -class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]): +class QuoteWeights(BaseModel): + price: float = Field(..., ge=0, le=1000) + delivery: float = Field(..., ge=0, le=1000) + warranty: float = Field(..., ge=0, le=1000) + + @field_validator("warranty") + @classmethod + def at_least_one_weight(cls, value: float, info: Any) -> float: + values = dict(info.data) + total = float(values.get("price") or 0) + float(values.get("delivery") or 0) + float(value or 0) + if total <= 0: + raise ValueError("at least one weight must be greater than zero") + return value + + +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=90_000) + + @field_validator("filename") + @classmethod + def safe_filename(cls, value: str) -> str: + cleaned = str(value or "").strip().replace("\\", "/") + if not cleaned or "/" in cleaned or cleaned in {".", ".."}: + raise ValueError("filename must be a simple file name") + return cleaned + + @field_validator("media_type") + @classmethod + def known_media_type(cls, value: str) -> str: + cleaned = str(value or "").split(";", 1)[0].strip().lower() + if cleaned not in ALLOWED_UPLOAD_MEDIA_TYPES: + raise ValueError("unsupported media type") + 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 per-user comparisons, supports browser uploads, and serves a polished packed product UI." + description = ( + "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty " + "scoring, persists per-user comparisons, supports browser uploads, and serves " + "a polished packed 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 + state = State.DURABLE pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + caller_pays_llm=False, + notes="Deterministic quote scoring; no LLM credential or provider key is required.", ) + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600) + tools_used = ("postgres", "mcp", "packed-react-frontend") workspace_access = WorkspaceAccess.dynamic( - max_files=64, + max_files=8, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, + max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_BROWSER_DOCUMENTS, ) - tools_used = ("deepagents", "langchain") - - @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful") - async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: - creds = ctx.llm - await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") - if not creds.api_key: - return ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_KEY." - ) - graph = self._build_deep_agent(ctx=ctx, creds=creds) - state = await graph.ainvoke( - {"messages": [{"role": "user", "content": prompt}]}, - config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name="quote-judge-studio-v1-data", + scope="user", + access_mode="read_write", + env=AgentDatabaseEnv(url="DATABASE_URL"), + migrations=AgentDatabaseMigrations(path="db/migrations"), + ), ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + ) - def _build_deep_agent( + @a2a.tool( + description="Compare two to ten vendor quotes with weighted price, delivery, and warranty scoring; persist the result and a production receipt.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def compare_quotes( self, - *, - ctx: RunContext[{{ auth_type }}], - creds: LLMCreds, - ) -> Any: - # Lazy imports keep `a2a card` usable before local dependencies are - # installed. `a2a deploy` installs requirements.txt during the build. - from a2a_pack.deepagents import create_a2a_deep_agent - from langchain.agents.middleware import wrap_model_call - from langchain_core.tools import tool + ctx: RunContext[PlatformUserAuth], + comparison_id: str, + quotes: list[QuoteInput], + weights: QuoteWeights, + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + validation = _validate_comparison_request(comparison_id, quotes, weights) + if validation is not None: + return validation - @tool - 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, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + result = _score_quotes(comparison_id, quotes, weights) + receipt = _build_receipt( + tenant_key=tenant, + skill_name="compare_quotes", + comparison_id=comparison_id, + inputs={ + "comparison_id": comparison_id, + "quotes": [quote.model_dump(mode="json") for quote in quotes], + "weights": weights.model_dump(mode="json"), + }, + result=result, ) + persisted = _persist_comparison(tenant, comparison_id, result, receipt) + result["receipt"] = { + "receipt_id": receipt["receipt_id"], + "persisted": persisted["ok"], + "created_at": receipt["created_at"], + } + if persisted.get("warning"): + result.setdefault("warnings", []).append(persisted["warning"]) + await ctx.emit_progress(f"saved comparison {comparison_id}") + return result + + @a2a.tool( + description="Reopen a saved quote comparison for the authenticated user.", + timeout_seconds=20, + 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 {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id} + record = _load_comparison(tenant, comparison_id) + if record is None: + return {"ok": False, "code": "comparison_not_found", "comparison_id": comparison_id} + result = dict(record["result"]) + result["comparison_id"] = comparison_id + result["ok"] = True + result["reloaded"] = True + result["saved_at"] = record.get("updated_at") + if record.get("latest_receipt_id"): + result["latest_receipt_id"] = record["latest_receipt_id"] + return result + + @a2a.tool( + description="Browser-safe upload bridge: accept bounded base64 JSON/CSV/text quote documents, parse them, compare, persist, and receipt the result.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def compare_quotes_from_browser_upload( + self, + ctx: RunContext[PlatformUserAuth], + comparison_id: str, + documents: list[BrowserDocument], + weights: QuoteWeights, + ) -> dict[str, Any]: + parsed = _decode_browser_documents(documents) + if not parsed["ok"]: + return {"ok": False, "comparison_id": comparison_id, **parsed} + quotes = parsed["quotes"] + result = await self.compare_quotes(ctx, comparison_id, quotes, weights) + result["source"] = "browser_upload" + result["parsed_documents"] = parsed["documents"] + return result + + @a2a.tool( + description="External Agent API upload path: accept a typed FileUpload JSON/CSV/text document, parse quotes, compare, persist, and receipt the result.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def compare_quotes_from_file( + self, + ctx: RunContext[PlatformUserAuth], + comparison_id: str, + document: Annotated[ + UploadedFile, + FileUpload( + accept=sorted(ALLOWED_UPLOAD_MEDIA_TYPES), + max_bytes=MAX_UPLOAD_BYTES, + description="JSON or CSV quote file with vendor, quantity, unit_price, delivery_days, warranty_months.", + ), + ], + weights: QuoteWeights, + ) -> dict[str, Any]: + if document.size_bytes > MAX_UPLOAD_BYTES: + return {"ok": False, "code": "file_too_large", "comparison_id": comparison_id} + media_type = (document.media_type or "").split(";", 1)[0].strip().lower() + if media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: + return {"ok": False, "code": "unsupported_media_type", "comparison_id": comparison_id} + try: + reader = getattr(ctx.workspace, "read_bytes", None) + if reader is None: + return {"ok": False, "code": "workspace_read_unavailable", "comparison_id": comparison_id} + data = reader(document.path) + except Exception: + return {"ok": False, "code": "file_read_failed", "comparison_id": comparison_id} + parsed = _parse_quote_document(data, filename=document.filename, media_type=media_type) + if not parsed["ok"]: + return {"ok": False, "comparison_id": comparison_id, **parsed} + result = await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights) + result["source"] = "file_upload" + result["parsed_documents"] = [{"filename": document.filename, "quotes": len(parsed["quotes"])}] + return result -def _runtime_skills_root(ctx: RunContext[Any]) -> str: - workspace = getattr(ctx, "_workspace", None) - prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) - 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 _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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """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 _valid_comparison_id(value: str) -> bool: + return bool(COMPARISON_ID_RE.fullmatch(str(value or "").strip())) -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def _validate_comparison_request( + comparison_id: str, + quotes: list[QuoteInput], + weights: QuoteWeights, +) -> dict[str, Any] | None: + if not _valid_comparison_id(comparison_id): + return {"ok": False, "code": "invalid_comparison_id", "comparison_id": comparison_id} + if len(quotes) < 2: + return {"ok": False, "code": "at_least_two_quotes_required", "comparison_id": comparison_id} + if len(quotes) > MAX_QUOTES: + return {"ok": False, "code": "too_many_quotes", "comparison_id": comparison_id} + vendor_names = [quote.vendor.casefold() for quote in quotes] + if len(set(vendor_names)) != len(vendor_names): + return {"ok": False, "code": "duplicate_vendor", "comparison_id": comparison_id} + if weights.price + weights.delivery + weights.warranty <= 0: + return {"ok": False, "code": "weights_sum_to_zero", "comparison_id": comparison_id} + return None - content = getattr(messages[-1], "content", None) - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, dict): - text = item.get("text") or item.get("content") - if text: - parts.append(str(text)) - elif item: - parts.append(str(item)) - return "\n".join(parts) if parts else json.dumps(content, default=str) - return str(content or messages[-1]) + +def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: + total_weight = Decimal(str(weights.price + weights.delivery + weights.warranty)) + prices = [_money(q.unit_price) for q in quotes] + totals = [_money(q.unit_price * q.quantity) for q in quotes] + deliveries = [Decimal(q.delivery_days) for q in quotes] + warranties = [Decimal(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 * Decimal(str(weights.price)) + + delivery_score * Decimal(str(weights.delivery)) + + warranty_score * Decimal(str(weights.warranty)) + ) / total_weight + rows.append( + { + "vendor": quote.vendor, + "quantity": quote.quantity, + "unit_price": _float2(prices[index]), + "total_price": _float2(totals[index]), + "delivery_days": quote.delivery_days, + "warranty_months": quote.warranty_months, + "score": _float2(weighted_score), + "score_breakdown": { + "price": _float2(price_score), + "delivery": _float2(delivery_score), + "warranty": _float2(warranty_score), + }, + } + ) + rows.sort(key=lambda item: (-item["score"], item["total_price"], item["delivery_days"], -item["warranty_months"], item["vendor"])) + for rank, row in enumerate(rows, start=1): + row["rank"] = rank + recommendation = rows[0] + return { + "ok": True, + "comparison_id": comparison_id, + "recommendation": { + "vendor": recommendation["vendor"], + "score": recommendation["score"], + "total_price": recommendation["total_price"], + "delivery_days": recommendation["delivery_days"], + "warranty_months": recommendation["warranty_months"], + "rationale": ( + f"{recommendation['vendor']} has the best weighted score using " + f"price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}." + ), + }, + "rankings": rows, + "weights": weights.model_dump(mode="json"), + "quote_count": len(quotes), + } + + +def _money(value: float | int | Decimal) -> Decimal: + return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + + +def _float2(value: Decimal) -> float: + return float(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) + + +def _lower_is_better(value: Decimal, values: list[Decimal]) -> Decimal: + low, high = min(values), max(values) + if low == high: + return Decimal("100") + return ((high - value) / (high - low) * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + + +def _higher_is_better(value: Decimal, values: list[Decimal]) -> Decimal: + low, high = min(values), max(values) + if low == high: + return Decimal("100") + return ((value - low) / (high - low) * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + + +def _decode_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]: + if not documents: + return {"ok": False, "code": "no_documents_provided"} + if len(documents) > MAX_BROWSER_DOCUMENTS: + return {"ok": False, "code": "too_many_documents"} + all_quotes: list[QuoteInput] = [] + summaries: list[dict[str, Any]] = [] + for doc in documents: + try: + data = base64.b64decode(doc.data_base64.encode("ascii"), validate=True) + except Exception: + return {"ok": False, "code": "invalid_base64", "filename": doc.filename} + if len(data) > MAX_UPLOAD_BYTES: + return {"ok": False, "code": "file_too_large", "filename": doc.filename} + parsed = _parse_quote_document(data, filename=doc.filename, media_type=doc.media_type) + if not parsed["ok"]: + return {"ok": False, "filename": doc.filename, **parsed} + all_quotes.extend(parsed["quotes"]) + summaries.append({"filename": doc.filename, "quotes": len(parsed["quotes"])}) + return {"ok": True, "quotes": all_quotes, "documents": summaries} + + +def _parse_quote_document(data: bytes, *, filename: str, media_type: str) -> dict[str, Any]: + if len(data) > MAX_UPLOAD_BYTES: + return {"ok": False, "code": "file_too_large"} + text = data.decode("utf-8-sig", errors="replace").strip() + if not text: + return {"ok": False, "code": "empty_document"} + try: + if media_type in {"application/json", "text/json"} or filename.lower().endswith(".json"): + raw = json.loads(text) + items = raw.get("quotes", raw) if isinstance(raw, dict) else raw + if not isinstance(items, list): + return {"ok": False, "code": "quotes_array_required"} + return {"ok": True, "quotes": [QuoteInput.model_validate(item) for item in items]} + rows = list(csv.DictReader(io.StringIO(text))) + if rows and rows[0]: + return {"ok": True, "quotes": [_quote_from_row(row) for row in rows]} + return {"ok": False, "code": "unsupported_document_format"} + except Exception: + return {"ok": False, "code": "document_parse_failed"} + + +def _quote_from_row(row: dict[str, Any]) -> QuoteInput: + normalized = {str(key).strip().lower(): value for key, value in row.items()} + return QuoteInput( + vendor=str(normalized.get("vendor") or normalized.get("supplier") or ""), + quantity=int(float(normalized.get("quantity") or normalized.get("qty") or 0)), + unit_price=float(normalized.get("unit_price") or normalized.get("price") or 0), + delivery_days=int(float(normalized.get("delivery_days") or normalized.get("delivery") or 0)), + warranty_months=int(float(normalized.get("warranty_months") or normalized.get("warranty") or 0)), + ) + + +def _database_url() -> str | None: + return os.environ.get("DATABASE_URL") + + +def _connect() -> Any: + if psycopg is None: + raise RuntimeError("psycopg is not installed") + url = _database_url() + if not url: + raise RuntimeError("DATABASE_URL is not configured") + return psycopg.connect(url, row_factory=dict_row) + + +def _persist_comparison( + tenant_key: str, + comparison_id: str, + result: dict[str, Any], + receipt: dict[str, Any], +) -> dict[str, Any]: + if not _database_url() or psycopg is None: + _LOCAL_COMPARISONS[(tenant_key, comparison_id)] = { + "result": json.loads(json.dumps(result)), + "updated_at": receipt["created_at"], + "latest_receipt_id": receipt["receipt_id"], + } + _LOCAL_RECEIPTS.append(receipt) + return {"ok": True, "warning": "local_ephemeral_persistence"} + try: + with _connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO quote_judge_comparisons + (tenant_key, comparison_id, input_json, result_json, recommendation_vendor, latest_receipt_id, updated_at) + VALUES (%s, %s, %s, %s, %s, %s, NOW()) + ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET + input_json = EXCLUDED.input_json, + result_json = EXCLUDED.result_json, + recommendation_vendor = EXCLUDED.recommendation_vendor, + latest_receipt_id = EXCLUDED.latest_receipt_id, + updated_at = NOW() + """, + ( + tenant_key, + comparison_id, + Jsonb({"weights": result["weights"], "quote_count": result["quote_count"]}), + Jsonb(result), + result["recommendation"]["vendor"], + receipt["receipt_id"], + ), + ) + cur.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_key, + comparison_id, + receipt["skill_name"], + receipt["skill_name"], + receipt["status"], + receipt["input_hash"], + receipt["result_hash"], + Jsonb(receipt), + Jsonb(receipt), + ), + ) + conn.commit() + return {"ok": True} + except Exception: + return {"ok": False, "warning": "persistence_unavailable"} + + +def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None: + if not _database_url() or psycopg is None: + return _LOCAL_COMPARISONS.get((tenant_key, comparison_id)) + try: + with _connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT result_json, updated_at, latest_receipt_id + FROM quote_judge_comparisons + WHERE tenant_key = %s AND comparison_id = %s + """, + (tenant_key, comparison_id), + ) + row = cur.fetchone() + if not row: + return None + updated_at = row["updated_at"] + return { + "result": row["result_json"], + "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at), + "latest_receipt_id": row.get("latest_receipt_id"), + } + except Exception: + return None + + +def _build_receipt( + *, + tenant_key: str, + skill_name: str, + comparison_id: str, + inputs: dict[str, Any], + result: dict[str, Any], +) -> dict[str, Any]: + now = datetime.now(timezone.utc).isoformat() + tenant_hash = _hash_json({"tenant_key": tenant_key}) + input_hash = _hash_json(inputs) + result_hash = _hash_json(result) + receipt_id = "qjr_" + _hash_json({ + "tenant_hash": tenant_hash, + "skill_name": skill_name, + "comparison_id": comparison_id, + "input_hash": input_hash, + "result_hash": result_hash, + })[:24] + return { + "schema": "quotejudge.production_receipt.v1", + "receipt_id": receipt_id, + "agent_name": "quote-judge-studio-v1", + "agent_version": "0.1.0", + "skill_name": skill_name, + "comparison_id": comparison_id, + "status": "ok" if result.get("ok") else "error", + "tenant_hash": tenant_hash, + "input_hash": input_hash, + "result_hash": result_hash, + "created_at": now, + } + + +def _hash_json(value: Any) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + return hashlib.sha256(encoded).hexdigest()