diff --git a/agent.py b/agent.py index 462ee44..e554bbd 100644 --- a/agent.py +++ b/agent.py @@ -1,679 +1,189 @@ -"""QuoteJudge full-stack A2A agent. +"""quote-judge-studio-v1 agent. -Deterministic quote comparison, per-user persistence in managed Postgres, -browser-safe upload parsing, and production execution receipt storage. +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 csv -import hashlib -import io import json -import os -import re -import uuid -from contextlib import contextmanager -from datetime import datetime, timezone -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, - PlatformUserAuth, + LLMProvisioning, + {{ auth_type }}, Pricing, - Resources, RunContext, - State, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.workspace import FileUpload, UploadedFile - -try: # psycopg is installed in deployment and sandbox via requirements.txt. - import psycopg - from psycopg.rows import dict_row -except Exception: # pragma: no cover - import guard for card-only tooling - psycopg = None - dict_row = None - - -MAX_QUOTES = 20 -MAX_VENDOR_LENGTH = 120 -MAX_UPLOADS = 5 -MAX_UPLOAD_BYTES = 128_000 -MAX_UPLOAD_BASE64_CHARS = ((MAX_UPLOAD_BYTES + 2) // 3) * 4 -ALLOWED_UPLOAD_MEDIA_TYPES = {"text/csv", "text/plain", "application/json"} -_DATABASE_NAME = "quote-judge-studio-v1-data" -_LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {} -_LOCAL_RECEIPTS: list[dict[str, Any]] = [] +from a2a_pack.context import LLMCreds class QuoteJudgeStudioV1Config(BaseModel): - """No caller-configured settings are required.""" + pass -class QuoteInput(BaseModel): - vendor: str = Field(..., min_length=1, max_length=MAX_VENDOR_LENGTH) - quantity: int = Field(..., gt=0, le=10_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=600) +SYSTEM_PROMPT = """\ +You are a compact tool-calling agent. - @field_validator("vendor") - @classmethod - def clean_vendor(cls, value: str) -> str: - cleaned = " ".join(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=100) - delivery: float = Field(..., ge=0, le=100) - warranty: float = Field(..., ge=0, le=100) - - - -class BrowserQuoteDocument(BaseModel): - filename: str = Field(..., min_length=1, max_length=160) - media_type: str = Field(..., min_length=1, max_length=100) - data_base64: str = Field(..., min_length=1, max_length=MAX_UPLOAD_BASE64_CHARS) - - @field_validator("filename") - @classmethod - def safe_filename(cls, value: str) -> str: - cleaned = value.strip().replace("\\", "/").split("/")[-1] - if not cleaned or cleaned in {".", ".."}: - raise ValueError("filename is required") - return cleaned[:160] - - -class QuoteJudgeState(BaseModel): - """Durable state is stored in managed Postgres; this schema advertises that fact.""" - - persistence: str = "managed_postgres" - - -class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]): +class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]): name = "quote-judge-studio-v1" - description = ( - "QuoteJudge compares vendor quotes with weighted price, delivery, and " - "warranty scoring, persists per-user comparisons, and serves a one-page product UI." - ) - version = "0.1.5" + 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 = PlatformUserAuth - state = State.DURABLE - state_model = QuoteJudgeState - resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300) - platform_resources = AgentPlatformResources( - databases=( - AgentDatabase( - name=_DATABASE_NAME, - scope="user", - access_mode="read_write", - env=AgentDatabaseEnv(url="DATABASE_URL"), - migrations=AgentDatabaseMigrations(path="db/migrations"), - ), - ) - ) + 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 local quote scoring; no LLM credential is required.", + caller_pays_llm=True, + notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", ) workspace_access = WorkspaceAccess.dynamic( - max_files=MAX_UPLOADS, - 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_UPLOADS * MAX_UPLOAD_BYTES, ) - tools_used = ("postgres", "mcp") + tools_used = ("deepagents", "langchain") - @a2a.tool( - description=( - "Compare at least two vendor quotes using typed price, delivery, and warranty " - "weights; persist the recommendation and a production receipt." - ), - timeout_seconds=30, - idempotent=False, - cost_class="cheap", - ) - async def compare_quotes( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: str, - quotes: list[QuoteInput], - weights: QuoteWeights, - ) -> dict[str, Any]: - tenant = _tenant_key(ctx) - cleaned_id = _clean_comparison_id(comparison_id) - started = _now_iso() - input_payload = { - "comparison_id": cleaned_id, - "quotes": [quote.model_dump() for quote in quotes], - "weights": weights.model_dump(), - } - - if weights.price + weights.delivery + weights.warranty <= 0: - result = { - "ok": False, - "code": "positive_weight_required", - "message": "Set at least one weight above zero before comparing.", - "comparison_id": cleaned_id, - } - await _persist_receipt( - tenant_key=tenant, - comparison_id=cleaned_id, - skill_name="compare_quotes", - input_payload=input_payload, - result_payload=result, - status="validation_error", - started_at=started, + @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." ) - return result - - if len(quotes) < 2: - result = { - "ok": False, - "code": "at_least_two_quotes_required", - "message": "Add at least two vendor quotes before comparing.", - "comparison_id": cleaned_id, - } - await _persist_receipt( - tenant_key=tenant, - comparison_id=cleaned_id, - skill_name="compare_quotes", - input_payload=input_payload, - result_payload=result, - status="validation_error", - started_at=started, - ) - return result - if len(quotes) > MAX_QUOTES: - result = { - "ok": False, - "code": "too_many_quotes", - "message": f"Compare at most {MAX_QUOTES} quotes at a time.", - "comparison_id": cleaned_id, - } - await _persist_receipt( - tenant_key=tenant, - comparison_id=cleaned_id, - skill_name="compare_quotes", - input_payload=input_payload, - result_payload=result, - status="validation_error", - started_at=started, - ) - return result - - result = _score_quotes(cleaned_id, quotes, weights) - await _save_comparison(tenant, result) - await _persist_receipt( - tenant_key=tenant, - comparison_id=cleaned_id, - skill_name="compare_quotes", - input_payload=input_payload, - result_payload=result, - status="ok", - started_at=started, + graph = self._build_deep_agent(ctx=ctx, creds=creds) + state = await graph.ainvoke( + {"messages": [{"role": "user", "content": prompt}]}, + config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, ) - return result + await ctx.emit_progress("deepagent finished") + return _last_message_text(state) - @a2a.tool( - description="Reopen a saved QuoteJudge comparison for the authenticated user.", - timeout_seconds=15, - idempotent=False, - cost_class="cheap", - ) - async def get_comparison( + def _build_deep_agent( self, - ctx: RunContext[PlatformUserAuth], - comparison_id: str, - ) -> dict[str, Any]: - tenant = _tenant_key(ctx) - cleaned_id = _clean_comparison_id(comparison_id) - started = _now_iso() - record = await _load_comparison(tenant, cleaned_id) - if record is None: - result = { - "ok": False, - "code": "comparison_not_found", - "message": "No saved comparison exists for this id and user.", - "comparison_id": cleaned_id, - } - else: - result = dict(record) - result["ok"] = True - await _persist_receipt( - tenant_key=tenant, - comparison_id=cleaned_id, - skill_name="get_comparison", - input_payload={"comparison_id": cleaned_id}, - result_payload=result, - status="ok" if result.get("ok") else "not_found", - started_at=started, - ) - return result + *, + 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 - @a2a.tool( - description=( - "Browser upload bridge: parse bounded base64 JSON/CSV/text quote files, " - "compare extracted quotes, and persist the result." - ), - timeout_seconds=30, - idempotent=False, - cost_class="cheap", - ) - async def compare_uploaded_quotes_browser( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: str, - documents: list[BrowserQuoteDocument], - weights: QuoteWeights, - ) -> dict[str, Any]: - parsed = _parse_browser_documents(documents) - if not parsed["ok"]: - return parsed - return await self.compare_quotes( + @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, - comparison_id=comparison_id, - quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]], - weights=weights, - ) - - @a2a.tool( - description=( - "External-client upload path: parse an uploaded CSV, JSON, or plain-text quote " - "file and compare extracted quotes." - ), - timeout_seconds=30, - idempotent=False, - cost_class="cheap", - ) - async def compare_uploaded_quote_file( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: str, - document: Annotated[ - UploadedFile, - FileUpload( - accept=("text/csv", "text/plain", "application/json"), - max_bytes=MAX_UPLOAD_BYTES, - description="CSV, JSON, or text file containing vendor quote rows.", - ), - ], - weights: QuoteWeights, - ) -> dict[str, Any]: - if document.size_bytes > MAX_UPLOAD_BYTES: - return { - "ok": False, - "code": "upload_too_large", - "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller.", - "comparison_id": _clean_comparison_id(comparison_id), - } - if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: - return { - "ok": False, - "code": "unsupported_media_type", - "message": "Upload must be CSV, JSON, or plain text.", - "comparison_id": _clean_comparison_id(comparison_id), - } - try: - reader = getattr(ctx.workspace, "read_bytes") - raw = reader(document.path) - except Exception: - return { - "ok": False, - "code": "upload_read_failed", - "message": "Could not read the uploaded file from the granted workspace.", - "comparison_id": _clean_comparison_id(comparison_id), - } - parsed = _parse_document_bytes(document.filename, document.media_type, raw) - if not parsed["ok"]: - return parsed - return await self.compare_quotes( - ctx, - comparison_id=comparison_id, - quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]], - weights=weights, + 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 - ident = auth.user_id if auth.user_id is not None else (auth.sub or auth.email) - if ident is None: - raise ValueError("authenticated user identity is required") - return f"user:{ident}" +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 _clean_comparison_id(value: str) -> str: - cleaned = str(value or "").strip() - if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,79}", cleaned): - raise ValueError("comparison_id must be 1-80 safe characters") - return cleaned +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 _score_sort_key(row: dict[str, Any]) -> tuple[float, float, int, int, str]: - return ( - -float(row["scores"]["weighted_total"]), - float(row["subtotal"]), - int(row["delivery_days"]), - -int(row["warranty_months"]), - str(row["vendor"]).lower(), - ) +def _last_message_text(state: dict[str, Any]) -> str: + messages = state.get("messages") or [] + if not messages: + return json.dumps(state, default=str) - -def _subtotal_key(row: dict[str, Any]) -> float: - return float(row["subtotal"]) - - -def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: - quote_rows = [quote.model_dump() for quote in quotes] - for row in quote_rows: - row["subtotal"] = round(row["quantity"] * row["unit_price"], 2) - - totals = [row["subtotal"] for row in quote_rows] - deliveries = [row["delivery_days"] for row in quote_rows] - warranties = [row["warranty_months"] for row in quote_rows] - total_weight = weights.price + weights.delivery + weights.warranty - normalized_weights = { - "price": weights.price / total_weight, - "delivery": weights.delivery / total_weight, - "warranty": weights.warranty / total_weight, - } - - scored: list[dict[str, Any]] = [] - for row in quote_rows: - price_score = _lower_is_better(row["subtotal"], totals) - delivery_score = _lower_is_better(row["delivery_days"], deliveries) - warranty_score = _higher_is_better(row["warranty_months"], warranties) - weighted_score = ( - price_score * normalized_weights["price"] - + delivery_score * normalized_weights["delivery"] - + warranty_score * normalized_weights["warranty"] - ) - scored.append( - { - **row, - "scores": { - "price": round(price_score, 4), - "delivery": round(delivery_score, 4), - "warranty": round(warranty_score, 4), - "weighted_total": round(weighted_score, 4), - }, - } - ) - - scored.sort(key=_score_sort_key) - winner = scored[0] - baseline = min(scored, key=_subtotal_key) - rationale_bits = [ - f"{winner['vendor']} has the highest weighted score ({winner['scores']['weighted_total']:.2f}).", - f"Its normalized total is {winner['subtotal']:.2f} for {winner['quantity']} units.", - ] - if winner["delivery_days"] == min(deliveries): - rationale_bits.append("It also has the fastest delivery among the submitted quotes.") - if winner["warranty_months"] == max(warranties): - rationale_bits.append("It offers the strongest warranty among the submitted quotes.") - if winner["vendor"] == baseline["vendor"]: - rationale_bits.append("It is the lowest-price option under the selected quantity.") - - return { - "ok": True, - "comparison_id": comparison_id, - "recommendation": { - "vendor": winner["vendor"], - "score": winner["scores"]["weighted_total"], - "normalized_total": winner["subtotal"], - "rationale": " ".join(rationale_bits), - }, - "weights": { - "price": weights.price, - "delivery": weights.delivery, - "warranty": weights.warranty, - "normalized": {key: round(value, 4) for key, value in normalized_weights.items()}, - }, - "quotes": scored, - "created_at": _now_iso(), - } - - -def _lower_is_better(value: float, values: list[float]) -> float: - min_value = min(values) - max_value = max(values) - if max_value == min_value: - return 1.0 - return (max_value - value) / (max_value - min_value) - - -def _higher_is_better(value: float, values: list[float]) -> float: - min_value = min(values) - max_value = max(values) - if max_value == min_value: - return 1.0 - return (value - min_value) / (max_value - min_value) - - -def _parse_browser_documents(documents: list[BrowserQuoteDocument]) -> dict[str, Any]: - if not documents: - return {"ok": False, "code": "no_uploads", "message": "Upload at least one quote file."} - if len(documents) > MAX_UPLOADS: - return {"ok": False, "code": "too_many_uploads", "message": f"Upload at most {MAX_UPLOADS} files."} - quotes: list[dict[str, Any]] = [] - for document in documents: - if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: - return {"ok": False, "code": "unsupported_media_type", "message": "Upload must be CSV, JSON, or plain text."} - try: - raw = base64.b64decode(document.data_base64, validate=True) - except Exception: - return {"ok": False, "code": "invalid_base64", "message": "Uploaded file data is not valid base64."} - if len(raw) > MAX_UPLOAD_BYTES: - return {"ok": False, "code": "upload_too_large", "message": f"Each upload must be {MAX_UPLOAD_BYTES} bytes or smaller."} - parsed = _parse_document_bytes(document.filename, document.media_type, raw) - if not parsed["ok"]: - return parsed - quotes.extend(parsed["quotes"]) - return {"ok": True, "quotes": quotes[:MAX_QUOTES]} - - -def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]: - if len(raw) > MAX_UPLOAD_BYTES: - return {"ok": False, "code": "upload_too_large", "message": f"Upload must be {MAX_UPLOAD_BYTES} bytes or smaller."} - text = raw.decode("utf-8-sig", errors="replace") - try: - if media_type == "application/json" or filename.lower().endswith(".json"): - data = json.loads(text) - rows = data.get("quotes", data) if isinstance(data, dict) else data - if not isinstance(rows, list): - raise ValueError("JSON must be a list or contain a quotes list") - quotes = [QuoteInput.model_validate(row).model_dump() for row in rows] - elif media_type == "text/csv" or filename.lower().endswith(".csv"): - reader = csv.DictReader(io.StringIO(text)) - quotes = [QuoteInput.model_validate(_coerce_quote_row(row)).model_dump() for row in reader] - else: - quotes = _parse_plain_text_quotes(text) - except Exception as exc: - return {"ok": False, "code": "quote_parse_failed", "message": f"Could not parse quote upload: {exc}"} - if not quotes: - return {"ok": False, "code": "no_quotes_found", "message": "No quote rows were found in the upload."} - return {"ok": True, "quotes": quotes} - - -def _coerce_quote_row(row: dict[str, Any]) -> dict[str, Any]: - aliases = { - "vendor": ["vendor", "supplier", "name"], - "quantity": ["quantity", "qty"], - "unit_price": ["unit_price", "price", "unit price", "unitPrice"], - "delivery_days": ["delivery_days", "delivery", "delivery days", "lead_time", "lead time"], - "warranty_months": ["warranty_months", "warranty", "warranty months"], - } - normalized = {str(k).strip().lower(): v for k, v in row.items()} - out: dict[str, Any] = {} - for target, names in aliases.items(): - for name in names: - if name.lower() in normalized and normalized[name.lower()] not in (None, ""): - out[target] = normalized[name.lower()] - break - return out - - -def _parse_plain_text_quotes(text: str) -> list[dict[str, Any]]: - quotes: list[dict[str, Any]] = [] - for line in text.splitlines(): - if not line.strip(): - continue - parts = [part.strip() for part in re.split(r"[,;|]", line) if part.strip()] - if len(parts) >= 5: - quotes.append( - QuoteInput.model_validate( - { - "vendor": parts[0], - "quantity": parts[1], - "unit_price": parts[2], - "delivery_days": parts[3], - "warranty_months": parts[4], - } - ).model_dump() - ) - return quotes - - -@contextmanager -def _db_conn() -> Any: - url = os.environ.get("DATABASE_URL") - if not url or psycopg is None: - yield None - return - conn = psycopg.connect(url, row_factory=dict_row) - try: - yield conn - conn.commit() - finally: - conn.close() - - -async def _save_comparison(tenant_key: str, payload: dict[str, Any]) -> None: - key = (tenant_key, payload["comparison_id"]) - with _db_conn() as conn: - if conn is None: - _LOCAL_COMPARISONS[key] = dict(payload) - return - with conn.cursor() as cur: - cur.execute( - """ - INSERT INTO quote_comparisons - (tenant_key, comparison_id, recommended_vendor, recommendation_vendor, - score, weighted_score, normalized_total, payload, updated_at) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW()) - ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET - recommended_vendor = EXCLUDED.recommended_vendor, - recommendation_vendor = EXCLUDED.recommendation_vendor, - score = EXCLUDED.score, - weighted_score = EXCLUDED.weighted_score, - normalized_total = EXCLUDED.normalized_total, - payload = EXCLUDED.payload, - updated_at = NOW() - """, - ( - tenant_key, - payload["comparison_id"], - payload["recommendation"]["vendor"], - payload["recommendation"]["vendor"], - payload["recommendation"]["score"], - payload["recommendation"]["score"], - payload["recommendation"]["normalized_total"], - json.dumps(payload), - ), - ) - - -async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None: - key = (tenant_key, comparison_id) - with _db_conn() as conn: - if conn is None: - return _LOCAL_COMPARISONS.get(key) - with conn.cursor() as cur: - cur.execute( - """ - SELECT payload - FROM quote_comparisons - WHERE tenant_key = %s AND comparison_id = %s - """, - (tenant_key, comparison_id), - ) - row = cur.fetchone() - if not row: - return None - payload = row["payload"] - return payload if isinstance(payload, dict) else json.loads(payload) - - -async def _persist_receipt( - *, - tenant_key: str, - comparison_id: str, - skill_name: str, - input_payload: dict[str, Any], - result_payload: dict[str, Any], - status: str, - started_at: str, -) -> dict[str, Any]: - receipt = { - "receipt_id": str(uuid.uuid4()), - "agent": QuoteJudgeStudioV1.name, - "agent_version": QuoteJudgeStudioV1.version, - "skill": skill_name, - "comparison_id": comparison_id, - "status": status, - "started_at": started_at, - "completed_at": _now_iso(), - "input_hash": _hash_json(input_payload), - "result_hash": _hash_json(result_payload), - } - with _db_conn() as conn: - if conn is None: - _LOCAL_RECEIPTS.append({"tenant_key": tenant_key, **receipt}) - return receipt - with conn.cursor() as cur: - cur.execute( - """ - INSERT INTO quote_execution_receipts - (tenant_key, comparison_id, receipt_id, skill_name, tool_name, status, ok, - input_hash, result_hash, payload, created_at) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) - """, - ( - tenant_key, - comparison_id, - receipt["receipt_id"], - skill_name, - skill_name, - status, - status == "ok", - receipt["input_hash"], - receipt["result_hash"], - json.dumps(receipt), - ), - ) - return receipt - - -def _hash_json(payload: dict[str, Any]) -> str: - encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") - return hashlib.sha256(encoded).hexdigest() - - -def _now_iso() -> str: - return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + 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])