diff --git a/agent.py b/agent.py index e554bbd..3ce5a67 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,565 @@ -"""quote-judge-studio-v1 agent. +"""QuoteJudge Studio full-stack A2A 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 +Deterministic, no-LLM quote comparison product: +- platform-authenticated user tenancy; +- managed Postgres persistence; +- typed A2A/MCP tools for comparison, reload, browser uploads, API uploads; +- production receipt records without secrets. """ 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 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, + FileUpload, + PlatformUserAuth, Pricing, + Resources, RunContext, + State, + UploadedFile, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds + +MAX_QUOTES = 25 +MAX_BROWSER_DOCUMENTS = 4 +MAX_BROWSER_DOCUMENT_BYTES = 64 * 1024 +ACCEPTED_UPLOAD_TYPES = ( + "application/json", + "text/csv", + "text/plain", +) +DATABASE_ENV = "DATABASE_URL" class QuoteJudgeStudioV1Config(BaseModel): - pass + """No user-editable config is required.""" -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +class QuoteInput(BaseModel): + vendor: Annotated[str, Field(min_length=1, max_length=160)] + unit_price: Annotated[float, Field(gt=0, le=1_000_000)] + quantity: Annotated[int, Field(gt=0, le=10_000_000)] + delivery_days: Annotated[int, Field(ge=0, le=3650)] + warranty_months: Annotated[int, Field(ge=0, le=600)] -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 = re.sub(r"\s+", " ", value).strip() + if not cleaned: + raise ValueError("vendor is required") + return cleaned -class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]): +class QuoteWeights(BaseModel): + price: Annotated[float, Field(ge=0, le=100)] = 50 + delivery: Annotated[float, Field(ge=0, le=100)] = 30 + warranty: Annotated[float, Field(ge=0, le=100)] = 20 + + +class BrowserDocument(BaseModel): + filename: Annotated[str, Field(min_length=1, max_length=180)] + media_type: Annotated[str, Field(min_length=1, max_length=100)] + data_base64: Annotated[str, Field(min_length=1, max_length=90_000)] + + +class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]): name = "quote-judge-studio-v1" - description = "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty scoring, persists recommendations, supports upload/paste workflows, and exposes typed A2A/MCP tools." + description = ( + "QuoteJudge compares vendor quotes with weighted price, delivery, " + "and warranty scoring, persists recommendations, supports upload/paste " + "workflows, and exposes typed A2A/MCP tools." + ) 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 is required.", ) + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) 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=512 * 1024, ) - 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}, + tools_used = ("postgres", "mcp", "packed-frontend") + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name="quote-judge-studio-v1-data", + scope="user", + access_mode="read_write", + env=AgentDatabaseEnv(url=DATABASE_ENV), + migrations=AgentDatabaseMigrations(path="db/migrations"), + ), ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + ) - def _build_deep_agent( + @a2a.tool( + description=( + "Compare at least two vendor quotes with price/delivery/warranty " + "weights, persist the result, and return a recommendation." + ), + timeout_seconds=60, + 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: Annotated[str, Field(min_length=1, max_length=120)], + quotes: Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)], + weights: QuoteWeights, + ) -> dict[str, Any]: + await ctx.emit_progress("Validating quote comparison inputs") + tenant = _tenant_key(ctx) + result = _build_comparison_result(comparison_id, quotes, weights) + if not result["ok"]: + await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "validation_error", {"code": result["code"]}, result) + return result - @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, - } + await ctx.emit_progress("Persisting comparison and production receipt") + await _save_comparison(tenant, comparison_id, result) + receipt = await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "ok", _safe_input_payload(comparison_id, quotes, weights), result) + result["receipt"] = receipt + return result + + @a2a.tool( + description="Reopen a previously persisted quote comparison by id.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def get_comparison( + self, + ctx: RunContext[PlatformUserAuth], + comparison_id: Annotated[str, Field(min_length=1, max_length=120)], + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + record = await _load_comparison(tenant, comparison_id) + if record is None: + result = { + "ok": False, + "code": "comparison_not_found", + "message": "No comparison was found for this id in your workspace.", + "comparison_id": comparison_id, + } + await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "not_found", {"comparison_id": comparison_id}, result) + return result + await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "ok", {"comparison_id": comparison_id}, record) + return record + + @a2a.tool( + description="List recently saved quote comparisons for the signed-in user.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def list_comparisons( + self, + ctx: RunContext[PlatformUserAuth], + limit: Annotated[int, Field(ge=1, le=50)] = 10, + ) -> dict[str, Any]: + rows = await _list_comparisons(_tenant_key(ctx), limit) + return {"ok": True, "comparisons": rows} + + @a2a.tool( + description=( + "Browser-safe upload bridge: accept bounded base64 JSON/CSV/text " + "quote files, parse them, compare, persist, and return the result." + ), + timeout_seconds=60, + idempotent=True, + cost_class="deterministic", + ) + async def compare_quotes_from_browser_upload( + self, + ctx: RunContext[PlatformUserAuth], + comparison_id: Annotated[str, Field(min_length=1, max_length=120)], + documents: Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_BROWSER_DOCUMENTS)], + weights: QuoteWeights, + ) -> dict[str, Any]: + parsed = _parse_browser_documents(documents) + if not parsed["ok"]: + await _persist_receipt(ctx, _tenant_key(ctx), comparison_id, "compare_quotes_from_browser_upload", "validation_error", {"comparison_id": comparison_id}, parsed) + return {**parsed, "comparison_id": comparison_id} + return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights) + + @a2a.tool( + description=( + "External API upload path: accept a typed FileUpload containing " + "JSON, CSV, or pasted text quote data and run the same comparison." + ), + timeout_seconds=60, + idempotent=True, + cost_class="deterministic", + grant_mode="read_only", + grant_allow_patterns=("**",), + ) + async def compare_quotes_from_upload( + self, + ctx: RunContext[PlatformUserAuth], + comparison_id: Annotated[str, Field(min_length=1, max_length=120)], + quote_file: Annotated[ + UploadedFile, + FileUpload( + accept=ACCEPTED_UPLOAD_TYPES, + max_bytes=MAX_BROWSER_DOCUMENT_BYTES, + description="JSON list/object, CSV, or text containing vendor quotes.", + ), + ], + weights: QuoteWeights, + ) -> dict[str, Any]: + if quote_file.size_bytes > MAX_BROWSER_DOCUMENT_BYTES: + return _validation_error( + "file_too_large", + f"Upload is {quote_file.size_bytes} bytes; maximum is {MAX_BROWSER_DOCUMENT_BYTES} bytes.", + comparison_id=comparison_id, ) - - @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)}" + if quote_file.media_type not in ACCEPTED_UPLOAD_TYPES: + return _validation_error( + "unsupported_media_type", + f"Unsupported media type {quote_file.media_type!r}. Use JSON, CSV, or plain text.", + comparison_id=comparison_id, ) - return await handler(request) + reader = getattr(ctx.workspace, "read_bytes", None) + if reader is None: + return _validation_error("workspace_read_unavailable", "The runtime did not expose uploaded file bytes.", comparison_id=comparison_id) + try: + raw = reader(quote_file.path) + except Exception: # noqa: BLE001 + return _validation_error("file_read_failed", "Could not read the uploaded file from the granted workspace.", comparison_id=comparison_id) + parsed = _parse_document_bytes(quote_file.filename, quote_file.media_type, raw) + if not parsed["ok"]: + return {**parsed, "comparison_id": comparison_id} + return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights) - 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, + +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 _safe_input_payload(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: + return { + "comparison_id": comparison_id, + "quote_count": len(quotes), + "vendors": [quote.vendor for quote in quotes], + "weights": weights.model_dump(mode="json"), + } + + +def _validation_error(code: str, message: str, *, comparison_id: str | None = None, details: list[dict[str, Any]] | None = None) -> dict[str, Any]: + payload: dict[str, Any] = { + "ok": False, + "code": code, + "message": message, + "validation_errors": details or [{"field": "quotes", "message": message}], + } + if comparison_id is not None: + payload["comparison_id"] = comparison_id + return payload + + +def _build_comparison_result(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: + if len(quotes) < 2: + return _validation_error( + "at_least_two_quotes_required", + "Provide at least two vendor quotes before requesting a recommendation.", + comparison_id=comparison_id, + details=[{"field": "quotes", "message": "At least two quotes are required."}], + ) + if len(quotes) > MAX_QUOTES: + return _validation_error( + "too_many_quotes", + f"Provide no more than {MAX_QUOTES} quotes.", + comparison_id=comparison_id, + ) + total_weight = weights.price + weights.delivery + weights.warranty + if total_weight <= 0: + return _validation_error( + "weights_must_sum_positive", + "At least one of price, delivery, or warranty weight must be greater than zero.", + comparison_id=comparison_id, + details=[{"field": "weights", "message": "Weights must sum to a positive number."}], ) + min_price = min(q.unit_price for q in quotes) + min_delivery = min(q.delivery_days for q in quotes) + max_warranty = max(q.warranty_months for q in quotes) + normalized_weights = { + "price": weights.price / total_weight, + "delivery": weights.delivery / total_weight, + "warranty": weights.warranty / total_weight, + } -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}" + rows: list[dict[str, Any]] = [] + for quote in quotes: + price_score = (min_price / quote.unit_price) * 100 if quote.unit_price else 0.0 + delivery_score = 100.0 if quote.delivery_days == 0 and min_delivery == 0 else ((min_delivery + 1) / (quote.delivery_days + 1)) * 100 + warranty_score = (quote.warranty_months / max_warranty) * 100 if max_warranty else 100.0 + weighted_score = ( + normalized_weights["price"] * price_score + + normalized_weights["delivery"] * delivery_score + + normalized_weights["warranty"] * warranty_score + ) + total_price = quote.unit_price * quote.quantity + rows.append( + { + "vendor": quote.vendor, + "unit_price": round(quote.unit_price, 4), + "quantity": quote.quantity, + "total_price": round(total_price, 2), + "delivery_days": quote.delivery_days, + "warranty_months": quote.warranty_months, + "scores": { + "price": round(price_score, 2), + "delivery": round(delivery_score, 2), + "warranty": round(warranty_score, 2), + "weighted_total": round(weighted_score, 2), + }, + } + ) + + rows.sort(key=lambda row: (-row["scores"]["weighted_total"], row["total_price"], row["delivery_days"], row["vendor"].lower())) + winner = rows[0] + recommendation = { + "vendor": winner["vendor"], + "score": winner["scores"]["weighted_total"], + "reason": ( + f"{winner['vendor']} has the strongest weighted score " + f"({winner['scores']['weighted_total']:.2f}) using price={weights.price:g}, " + f"delivery={weights.delivery:g}, warranty={weights.warranty:g}." + ), + } + return { + "ok": True, + "comparison_id": comparison_id, + "recommendation": recommendation, + "weights": { + "price": weights.price, + "delivery": weights.delivery, + "warranty": weights.warranty, + "normalized": {key: round(value, 4) for key, value in normalized_weights.items()}, + }, + "quote_count": len(quotes), + "comparison_table": rows, + "created_at": datetime.now(timezone.utc).isoformat(), + } -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 _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]: + quotes: list[QuoteInput] = [] + errors: list[dict[str, Any]] = [] + for index, doc in enumerate(documents): + if doc.media_type not in ACCEPTED_UPLOAD_TYPES: + errors.append({"field": f"documents[{index}].media_type", "message": "Use JSON, CSV, or plain text."}) + continue + try: + raw = base64.b64decode(doc.data_base64, validate=True) + except Exception: # noqa: BLE001 + errors.append({"field": f"documents[{index}].data_base64", "message": "Invalid base64 payload."}) + continue + if len(raw) > MAX_BROWSER_DOCUMENT_BYTES: + errors.append({"field": f"documents[{index}]", "message": f"File exceeds {MAX_BROWSER_DOCUMENT_BYTES} bytes."}) + continue + parsed = _parse_document_bytes(doc.filename, doc.media_type, raw) + if parsed["ok"]: + quotes.extend(parsed["quotes"]) + else: + errors.extend(parsed.get("validation_errors") or []) + if errors: + return _validation_error("upload_parse_failed", "One or more uploaded quote files could not be parsed.", details=errors) + return {"ok": True, "quotes": quotes[:MAX_QUOTES]} -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]: + if len(raw) > MAX_BROWSER_DOCUMENT_BYTES: + return _validation_error("file_too_large", f"{filename} exceeds the {MAX_BROWSER_DOCUMENT_BYTES} byte limit.") + text = raw.decode("utf-8", errors="replace").strip() + if not text: + return _validation_error("empty_upload", "The uploaded quote file was empty.") + try: + if media_type == "application/json" or filename.lower().endswith(".json"): + data = json.loads(text) + items = data.get("quotes") if isinstance(data, dict) else data + if not isinstance(items, list): + return _validation_error("invalid_json_quotes", "JSON must be a list of quotes or an object with a quotes list.") + return {"ok": True, "quotes": [QuoteInput.model_validate(item) for item in items]} + if media_type == "text/csv" or filename.lower().endswith(".csv"): + reader = csv.DictReader(io.StringIO(text)) + return {"ok": True, "quotes": [QuoteInput.model_validate(row) for row in reader]} + return {"ok": True, "quotes": _parse_loose_text_quotes(text)} + except Exception as exc: # noqa: BLE001 + return _validation_error("quote_parse_failed", f"Could not parse quote data: {type(exc).__name__}.") - 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 _parse_loose_text_quotes(text: str) -> list[QuoteInput]: + quotes: list[QuoteInput] = [] + for line in text.splitlines(): + parts = [part.strip() for part in re.split(r"[,|;]\s*", line) if part.strip()] + if len(parts) < 5 or parts[0].lower() in {"vendor", "supplier"}: + continue + 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("No quote rows found. Expected vendor, unit_price, quantity, delivery_days, warranty_months.") + return quotes + + +def _db_url() -> str: + return os.environ.get(DATABASE_ENV, "").strip() + + +def _connect(): + url = _db_url() + if not url: + raise RuntimeError("managed database is not configured") + import psycopg + + return psycopg.connect(url, autocommit=True) + + +async def _save_comparison(tenant_key: str, comparison_id: str, result: dict[str, Any]) -> None: + with _connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO quote_comparisons (tenant_key, comparison_id, recommendation_vendor, result) + VALUES (%s, %s, %s, %s) + ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET + recommendation_vendor = EXCLUDED.recommendation_vendor, + result = EXCLUDED.result, + updated_at = NOW() + """, + (tenant_key, comparison_id, result["recommendation"]["vendor"], json.dumps(result)), + ) + + +async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None: + with _connect() as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT result FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s", + (tenant_key, comparison_id), + ) + row = cur.fetchone() + if row is None: + return None + value = row[0] + if isinstance(value, str): + return json.loads(value) + return value + + +async def _list_comparisons(tenant_key: str, limit: int) -> list[dict[str, Any]]: + with _connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT comparison_id, recommendation_vendor, updated_at, result + FROM quote_comparisons + WHERE tenant_key = %s + ORDER BY updated_at DESC + LIMIT %s + """, + (tenant_key, limit), + ) + rows = cur.fetchall() + output: list[dict[str, Any]] = [] + for comparison_id, recommendation_vendor, updated_at, result in rows: + output.append( + { + "comparison_id": comparison_id, + "recommendation_vendor": recommendation_vendor, + "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at), + "quote_count": (result or {}).get("quote_count") if isinstance(result, dict) else None, + } + ) + return output + + +async def _persist_receipt( + ctx: RunContext[PlatformUserAuth], + tenant_key: str, + comparison_id: str, + tool_name: str, + status: str, + inputs: dict[str, Any], + result: dict[str, Any], +) -> dict[str, Any]: + payload = { + "agent": QuoteJudgeStudioV1.name, + "version": QuoteJudgeStudioV1.version, + "tool": tool_name, + "status": status, + "comparison_id": comparison_id, + "caller_scope": "platform_user", + "task_id": getattr(ctx, "task_id", ""), + "input_hash": _hash_json(inputs), + "result_hash": _hash_json(_receipt_safe_result(result)), + "created_at": datetime.now(timezone.utc).isoformat(), + } + payload["receipt_id"] = "qjr_" + _hash_json(payload)[:24] + try: + with _connect() as 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 NOTHING + """, + (payload["receipt_id"], tenant_key, comparison_id, tool_name, status, payload["input_hash"], json.dumps(payload)), + ) + except Exception as exc: # noqa: BLE001 + await ctx.emit_error("Receipt persistence failed; comparison result was still computed.", code="receipt_persist_failed") + return {"persisted": False, "code": "receipt_persist_failed", "message": type(exc).__name__} + return {"persisted": True, "receipt_id": payload["receipt_id"], "input_hash": payload["input_hash"]} + + +def _receipt_safe_result(result: dict[str, Any]) -> dict[str, Any]: + return { + "ok": result.get("ok"), + "comparison_id": result.get("comparison_id"), + "recommendation": result.get("recommendation"), + "code": result.get("code"), + } + + +def _hash_json(value: Any) -> str: + raw = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + return hashlib.sha256(raw).hexdigest()