diff --git a/agent.py b/agent.py index 7d813de..f05ad3f 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,747 @@ -"""quote-judge-studio-v1 agent. +"""QuoteJudge Studio v1: deterministic quote comparison product 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 +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 -from pathlib import Path -from typing import Any +import os +import re +import time +from datetime import UTC, datetime +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, RunContext, + UploadedFile, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds + +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 -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +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) -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).strip().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) + + +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 uploads, and exposes MCP-compatible A2A tools." + 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.0" config_model = QuoteJudgeStudioV1Config - auth_model = {{ auth_type }} + auth_model = PlatformUserAuth - # Hosted generated agents read the caller's saved LLM credential through - # ctx.llm. The platform may proxy that credential through LiteLLM, but agent - # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY - # directly. - llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + caller_pays_llm=False, + notes="Deterministic quote scoring and managed Postgres persistence; no LLM required.", ) workspace_access = WorkspaceAccess.dynamic( - max_files=64, - allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), + max_files=4, + allowed_modes=(WorkspaceMode.READ_ONLY,), require_reason=False, + max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_UPLOAD_FILES, ) - 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=DATABASE_NAME, + 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) + ) + tools_used = ("postgres", "mcp", "file-upload") - def _build_deep_agent( + @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[{{ 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: 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 - @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( + result = _score_quotes(comparison_id, quotes, weights) + result["tenant_scope"] = "user" + result["receipt"] = await _persist_receipt( ctx, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + 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 _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_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 - 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 _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, + tenant_key TEXT NOT NULL, + comparison_id TEXT, + skill_name TEXT NOT NULL, + receipt_json JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) + """ + ) + 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, + "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 + (tenant_key, comparison_id, skill_name, receipt_json) + VALUES (%s, %s, %s, %s) + """, + (tenant, comparison_id or None, skill_name, 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()