From d74c4ae9887a3dcbee307693f116ca4688e40aad Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 06:06:43 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 864 +++++++++---------------------------------------------- 1 file changed, 142 insertions(+), 722 deletions(-) diff --git a/agent.py b/agent.py index 32083a0..95e315f 100644 --- a/agent.py +++ b/agent.py @@ -1,769 +1,189 @@ -"""QuoteJudge Studio v1: deterministic quote comparison product agent. +"""quote-judge-studio-v1 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. +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 """ 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 pathlib import Path +from typing import Any -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel import a2a_pack as a2a from a2a_pack import ( A2AAgent, - AgentDatabase, - AgentDatabaseEnv, - AgentDatabaseMigrations, - AgentPlatformResources, - FileUpload, - PlatformUserAuth, + LLMProvisioning, + {{ auth_type }}, 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]] = [] +from a2a_pack.context import LLMCreds 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) +SYSTEM_PROMPT = """\ +You are a compact tool-calling agent. - @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 +Use the text_stats tool when the user asks about text, counts, summaries, +or anything where exact length/word numbers would help. Mention tool results +briefly instead of dumping raw JSON. +""" + +RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/" +DEEPAGENTS_RECURSION_LIMIT = 500 -class 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]): +class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]): 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" + 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 = PlatformUserAuth + auth_model = {{ auth_type }} + # 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=False, - notes="Deterministic quote scoring and managed Postgres persistence; no LLM required.", + caller_pays_llm=True, + notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", ) workspace_access = WorkspaceAccess.dynamic( - max_files=4, - allowed_modes=(WorkspaceMode.READ_ONLY,), + max_files=64, + allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), 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 = ("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", "file-upload") + await ctx.emit_progress("deepagent finished") + return _last_message_text(state) - @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( + def _build_deep_agent( 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 + *, + 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 - result = _score_quotes(comparison_id, quotes, weights) - result["tenant_scope"] = "user" - result["receipt"] = await _persist_receipt( + @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, - tenant, - "compare_quotes", - { - "comparison_id": comparison_id, - "quote_count": len(quotes), - "weights": weights.model_dump(mode="json"), - }, - result, + creds=creds, + backend=backend, + skills=skill_sources or None, + tools=[text_stats], + middleware=[log_model_call], + system_prompt=SYSTEM_PROMPT, ) - 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 _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 _valid_comparison_id(value: str) -> bool: - return bool(_COMPARISON_ID_RE.fullmatch(str(value or "").strip())) +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 _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 _last_message_text(state: dict[str, Any]) -> str: + messages = state.get("messages") or [] + if not messages: + return json.dumps(state, default=str) - -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() + 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])