bobseares-2/quote-judge-studio-v1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-18 04:43:38 +00:00
parent 8b6b5188a5
commit f333b49f75
1 changed files with 142 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files

774
agent.py
View File

@@ -1,679 +1,189 @@
"""QuoteJudge full-stack A2A agent. """quote-judge-studio-v1 agent.
Deterministic quote comparison, per-user persistence in managed Postgres, Starter stack:
browser-safe upload parsing, and production execution receipt storage. - 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 from __future__ import annotations
import base64
import csv
import hashlib
import io
import json import json
import os from pathlib import Path
import re from typing import Any
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Annotated, Any
from pydantic import BaseModel, Field, field_validator from pydantic import BaseModel
import a2a_pack as a2a import a2a_pack as a2a
from a2a_pack import ( from a2a_pack import (
A2AAgent, A2AAgent,
AgentDatabase, LLMProvisioning,
AgentDatabaseEnv, {{ auth_type }},
AgentDatabaseMigrations,
AgentPlatformResources,
PlatformUserAuth,
Pricing, Pricing,
Resources,
RunContext, RunContext,
State,
WorkspaceAccess, WorkspaceAccess,
WorkspaceMode, WorkspaceMode,
) )
from a2a_pack.workspace import FileUpload, UploadedFile from a2a_pack.context import LLMCreds
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]] = []
class QuoteJudgeStudioV1Config(BaseModel): class QuoteJudgeStudioV1Config(BaseModel):
"""No caller-configured settings are required.""" pass
class QuoteInput(BaseModel): SYSTEM_PROMPT = """\
vendor: str = Field(..., min_length=1, max_length=MAX_VENDOR_LENGTH) You are a compact tool-calling agent.
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)
@field_validator("vendor") Use the text_stats tool when the user asks about text, counts, summaries,
@classmethod or anything where exact length/word numbers would help. Mention tool results
def clean_vendor(cls, value: str) -> str: briefly instead of dumping raw JSON.
cleaned = " ".join(value.strip().split()) """
if not cleaned:
raise ValueError("vendor is required") RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/"
return cleaned DEEPAGENTS_RECURSION_LIMIT = 500
class QuoteWeights(BaseModel): class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
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]):
name = "quote-judge-studio-v1" name = "quote-judge-studio-v1"
description = ( description = "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty scoring, persists recommendations, supports upload/paste workflows, and exposes typed A2A/MCP tools."
"QuoteJudge compares vendor quotes with weighted price, delivery, and " version = "0.1.0"
"warranty scoring, persists per-user comparisons, and serves a one-page product UI."
)
version = "0.1.5"
config_model = QuoteJudgeStudioV1Config config_model = QuoteJudgeStudioV1Config
auth_model = PlatformUserAuth auth_model = {{ auth_type }}
state = State.DURABLE
state_model = QuoteJudgeState # Hosted generated agents read the caller's saved LLM credential through
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300) # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
platform_resources = AgentPlatformResources( # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
databases=( # directly.
AgentDatabase( llm_provisioning = LLMProvisioning.PLATFORM
name=_DATABASE_NAME,
scope="user",
access_mode="read_write",
env=AgentDatabaseEnv(url="DATABASE_URL"),
migrations=AgentDatabaseMigrations(path="db/migrations"),
),
)
)
pricing = Pricing( pricing = Pricing(
price_per_call_usd=0.0, price_per_call_usd=0.0,
caller_pays_llm=False, caller_pays_llm=True,
notes="Deterministic local quote scoring; no LLM credential is required.", notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
) )
workspace_access = WorkspaceAccess.dynamic( workspace_access = WorkspaceAccess.dynamic(
max_files=MAX_UPLOADS, max_files=64,
allowed_modes=(WorkspaceMode.READ_ONLY,), allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False, require_reason=False,
max_total_size_bytes=MAX_UPLOADS * MAX_UPLOAD_BYTES,
) )
tools_used = ("postgres", "mcp") tools_used = ("deepagents", "langchain")
@a2a.tool( @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
description=( async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
"Compare at least two vendor quotes using typed price, delivery, and warranty " creds = ctx.llm
"weights; persist the recommendation and a production receipt." await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
), if not creds.api_key:
timeout_seconds=30, return (
idempotent=False, "LLM key required. Add an LLM credential in Settings > LLM "
cost_class="cheap", "credentials before running this agent; for local --invoke "
) "runs set AGENT_LLM_KEY."
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,
) )
return result graph = self._build_deep_agent(ctx=ctx, creds=creds)
state = await graph.ainvoke(
if len(quotes) < 2: {"messages": [{"role": "user", "content": prompt}]},
result = { config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
"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,
) )
return result await ctx.emit_progress("deepagent finished")
return _last_message_text(state)
@a2a.tool( def _build_deep_agent(
description="Reopen a saved QuoteJudge comparison for the authenticated user.",
timeout_seconds=15,
idempotent=False,
cost_class="cheap",
)
async def get_comparison(
self, self,
ctx: RunContext[PlatformUserAuth], *,
comparison_id: str, ctx: RunContext[{{ auth_type }}],
) -> dict[str, Any]: creds: LLMCreds,
tenant = _tenant_key(ctx) ) -> Any:
cleaned_id = _clean_comparison_id(comparison_id) # Lazy imports keep `a2a card` usable before local dependencies are
started = _now_iso() # installed. `a2a deploy` installs requirements.txt during the build.
record = await _load_comparison(tenant, cleaned_id) from a2a_pack.deepagents import create_a2a_deep_agent
if record is None: from langchain.agents.middleware import wrap_model_call
result = { from langchain_core.tools import tool
"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
@a2a.tool( @tool
description=( def text_stats(text: str) -> str:
"Browser upload bridge: parse bounded base64 JSON/CSV/text quote files, " """Return exact word, character, and line counts for text."""
"compare extracted quotes, and persist the result." words = [part for part in text.split() if part.strip()]
), return json.dumps(
timeout_seconds=30, {
idempotent=False, "characters": len(text),
cost_class="cheap", "words": len(words),
) "lines": len(text.splitlines()) or 1,
async def compare_uploaded_quotes_browser( }
self, )
ctx: RunContext[PlatformUserAuth],
comparison_id: str, @wrap_model_call
documents: list[BrowserQuoteDocument], async def log_model_call(request: Any, handler: Any) -> Any:
weights: QuoteWeights, messages = request.state.get("messages", [])
) -> dict[str, Any]: print(
parsed = _parse_browser_documents(documents) "[middleware] model_call "
if not parsed["ok"]: f"model={creds.model} source={creds.source} messages={len(messages)}"
return parsed )
return await self.compare_quotes( 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, ctx,
comparison_id=comparison_id, creds=creds,
quotes=[QuoteInput.model_validate(item) for item in parsed["quotes"]], backend=backend,
weights=weights, skills=skill_sources or None,
) tools=[text_stats],
middleware=[log_model_call],
@a2a.tool( system_prompt=SYSTEM_PROMPT,
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,
) )
def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: def _runtime_skills_root(ctx: RunContext[Any]) -> str:
auth = ctx.auth workspace = getattr(ctx, "_workspace", None)
ident = auth.user_id if auth.user_id is not None else (auth.sub or auth.email) prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
if ident is None: if not prefixes:
raise ValueError("authenticated user identity is required") outputs_prefix = getattr(workspace, "outputs_prefix", None)
return f"user:{ident}" 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: def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
cleaned = str(value or "").strip() """Copy packaged DeepAgents skills into the invocation workspace.
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") DeepAgents loads skills from its backend, while source-controlled
return cleaned ``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]: def _last_message_text(state: dict[str, Any]) -> str:
return ( messages = state.get("messages") or []
-float(row["scores"]["weighted_total"]), if not messages:
float(row["subtotal"]), return json.dumps(state, default=str)
int(row["delivery_days"]),
-int(row["warranty_months"]),
str(row["vendor"]).lower(),
)
content = getattr(messages[-1], "content", None)
def _subtotal_key(row: dict[str, Any]) -> float: if isinstance(content, str):
return float(row["subtotal"]) return content
if isinstance(content, list):
parts: list[str] = []
def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: for item in content:
quote_rows = [quote.model_dump() for quote in quotes] if isinstance(item, dict):
for row in quote_rows: text = item.get("text") or item.get("content")
row["subtotal"] = round(row["quantity"] * row["unit_price"], 2) if text:
parts.append(str(text))
totals = [row["subtotal"] for row in quote_rows] elif item:
deliveries = [row["delivery_days"] for row in quote_rows] parts.append(str(item))
warranties = [row["warranty_months"] for row in quote_rows] return "\n".join(parts) if parts else json.dumps(content, default=str)
total_weight = weights.price + weights.delivery + weights.warranty return str(content or messages[-1])
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()