From d931b598f4639f03f8457694ae3710c0716018ad Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 10:12:18 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 636 ++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 491 insertions(+), 145 deletions(-) diff --git a/agent.py b/agent.py index 3f44b83..b82a2f5 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,535 @@ -"""invoice-guard-studio-v1 agent. +"""InvoiceGuard Studio v1. -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 +A deterministic full-stack A2A product for reviewing pasted or uploaded +invoices. The agent does not call an LLM: it validates invoice records, +detects duplicate invoice numbers and stated-total mismatches, persists +user-scoped review cases/decisions/receipts in managed Postgres, and exposes +small typed tools that are also served through MCP. """ from __future__ import annotations +import base64 +import csv +import hashlib +import io import json -from pathlib import Path -from typing import Any +import os +import re +from contextlib import contextmanager +from datetime import UTC, datetime +from decimal import Decimal, InvalidOperation, ROUND_HALF_UP +from typing import Annotated, Any -from pydantic import BaseModel +from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + FileUpload, + PlatformUserAuth, Pricing, + Resources, RunContext, - WorkspaceAccess, - WorkspaceMode, + State, + UploadedFile, ) -from a2a_pack.context import LLMCreds + +MAX_INVOICES = 50 +MAX_UPLOADS = 5 +MAX_UPLOAD_BYTES = 256 * 1024 +ACCEPTED_MEDIA_TYPES = {"application/json", "text/json", "text/csv", "text/plain"} +DB_CONNECT_OPTIONS = "-c statement_timeout=5000 -c lock_timeout=3000 -c idle_in_transaction_session_timeout=10000" class InvoiceGuardStudioV1Config(BaseModel): - pass + """No operator configuration is required for this deterministic product.""" -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. - -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 = "invoice-guard-studio-v1/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 +class BrowserDocument(BaseModel): + filename: str = Field(min_length=1, max_length=180) + media_type: str = Field(min_length=1, max_length=120) + data_base64: str = Field(min_length=1, max_length=((MAX_UPLOAD_BYTES * 4) // 3) + 16) -class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, {{ auth_type }}]): +class DecisionInput(BaseModel): + decision_id: str = Field(min_length=1, max_length=120) + invoice_number: str | None = Field(default=None, max_length=120) + decision: str = Field(min_length=1, max_length=40) + note: str = Field(default="", max_length=1000) + + +class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth]): name = "invoice-guard-studio-v1" - description = "One-page InvoiceGuard app for deterministic invoice duplicate and total mismatch review with user-scoped persistence." + description = ( + "One-page InvoiceGuard app for deterministic duplicate invoice number " + "and stated-total mismatch review with user-scoped Postgres persistence." + ) version = "0.1.0" config_model = InvoiceGuardStudioV1Config - auth_model = {{ auth_type }} + auth_model = PlatformUserAuth - # Hosted generated agents read the caller's saved LLM credential through - # ctx.llm. The platform may proxy that credential through LiteLLM, but agent - # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY - # directly. - llm_provisioning = LLMProvisioning.PLATFORM + state = State.DURABLE + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) 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 invoice validation; no LLM credentials are required.", ) - workspace_access = WorkspaceAccess.dynamic( - max_files=64, - allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=False, - ) - 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 = ("managed-postgres", "mcp") + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name="invoice-guard-studio-v1-data", + 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) + ) - def _build_deep_agent( + @a2a.tool( + description="Review pasted invoice objects, detect duplicate invoice numbers and stated-total mismatches, persist a user-scoped case, and record a production receipt.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def review_invoices( 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], + case_id: str, + invoices: list[dict[str, Any]], + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + result = _review_payload(case_id, invoices) + if not result["ok"]: + await ctx.emit_error(result["message"], code=result["code"]) + return result + persisted = _persist_case_with_receipt(tenant, result, source="json") + await ctx.emit_progress( + f"reviewed {result['invoice_count']} invoice(s): {result['duplicate_count']} duplicate group(s), {result['total_mismatch_count']} total mismatch(es)" + ) + return {**result, "receipt": persisted["receipt"], "saved_at": persisted["saved_at"]} - @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( + @a2a.tool( + description="Review invoices uploaded by browser JSON/base64 bridge, then persist the same review case and receipt as pasted invoices.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def review_invoice_uploads( + self, + ctx: RunContext[PlatformUserAuth], + case_id: str, + documents: list[BrowserDocument], + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + parsed = _parse_browser_documents(documents) + if not parsed["ok"]: + await ctx.emit_error(parsed["message"], code=parsed["code"]) + return parsed + result = _review_payload(case_id, parsed["invoices"]) + if not result["ok"]: + await ctx.emit_error(result["message"], code=result["code"]) + return result + persisted = _persist_case_with_receipt(tenant, result, source="browser_upload") + return {**result, "receipt": persisted["receipt"], "saved_at": persisted["saved_at"]} + + @a2a.tool( + description="Review a typed Agent API file upload containing JSON, CSV, or plain-text invoice rows.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def review_invoice_file( + self, + ctx: RunContext[PlatformUserAuth], + case_id: str, + document: Annotated[ + UploadedFile, + FileUpload( + accept=sorted(ACCEPTED_MEDIA_TYPES), + max_bytes=MAX_UPLOAD_BYTES, + description="JSON array/object, CSV, or simple key-value text invoice data.", + ), + ], + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + if document.size_bytes > MAX_UPLOAD_BYTES: + return _validation_error("upload_too_large", f"Upload exceeds {MAX_UPLOAD_BYTES} bytes.") + if document.media_type not in ACCEPTED_MEDIA_TYPES: + return _validation_error("unsupported_media_type", "Upload must be JSON, CSV, or plain text.") + data = ctx.workspace.read_bytes(document.path) + parsed = _parse_upload_bytes(document.filename, document.media_type, data) + if not parsed["ok"]: + await ctx.emit_error(parsed["message"], code=parsed["code"]) + return parsed + result = _review_payload(case_id, parsed["invoices"]) + if not result["ok"]: + await ctx.emit_error(result["message"], code=result["code"]) + return result + persisted = _persist_case_with_receipt(tenant, result, source="file_upload") + return {**result, "receipt": persisted["receipt"], "saved_at": persisted["saved_at"]} + + @a2a.tool( + description="Reopen a previously saved user-scoped invoice review case, including decisions and receipt metadata.", + timeout_seconds=15, + idempotent=True, + cost_class="deterministic", + ) + async def get_invoice_case( + self, + ctx: RunContext[PlatformUserAuth], + case_id: str, + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + record = _load_case(tenant, case_id) + if record is None: + return _validation_error("case_not_found", "No invoice review case exists for this signed-in user and case_id.", case_id=case_id) + return record + + @a2a.tool( + description="Record or update a user decision for a saved invoice review case.", + timeout_seconds=15, + idempotent=True, + cost_class="deterministic", + ) + async def record_decision( + self, + ctx: RunContext[PlatformUserAuth], + case_id: str, + decision: DecisionInput, + ) -> dict[str, Any]: + tenant = _tenant_key(ctx) + updated = _persist_decision(tenant, case_id, decision.model_dump()) + if updated is None: + return _validation_error("case_not_found", "Save or review the invoice case before recording decisions.", case_id=case_id) + return updated + + +def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + stable_id = ctx.auth.user_id if ctx.auth.user_id is not None else ctx.auth.sub + if not stable_id: + raise PermissionError("stable platform identity required") + return f"user:{stable_id}" + + +def _clean_case_id(case_id: str) -> str: + cleaned = str(case_id or "").strip() + if not cleaned or len(cleaned) > 120 or not re.fullmatch(r"[A-Za-z0-9_.:-]+", cleaned): + raise ValueError("case_id must be 1-120 chars using letters, numbers, dot, colon, underscore, or dash") + return cleaned + + +def _money(value: Any) -> Decimal | None: + if value is None or value == "": + return None + try: + return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + except (InvalidOperation, ValueError): + return None + + +def _money_json(value: Decimal) -> float: + return float(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) + + +def _review_payload(case_id: str, raw_invoices: list[dict[str, Any]]) -> dict[str, Any]: + try: + clean_case_id = _clean_case_id(case_id) + except ValueError as exc: + return _validation_error("invalid_case_id", str(exc), case_id=str(case_id or "")) + if not isinstance(raw_invoices, list) or not raw_invoices: + return _validation_error("no_invoices", "Provide at least one invoice.", case_id=clean_case_id) + if len(raw_invoices) > MAX_INVOICES: + return _validation_error("too_many_invoices", f"At most {MAX_INVOICES} invoices are allowed per review.", case_id=clean_case_id) + + invoices: list[dict[str, Any]] = [] + for index, item in enumerate(raw_invoices): + if not isinstance(item, dict): + return _validation_error("invalid_invoice", f"Invoice at index {index} must be an object.", case_id=clean_case_id) + missing = [name for name in ("invoice_number", "vendor", "subtotal", "tax", "total") if item.get(name) in (None, "")] + if missing: + return _validation_error( + "incomplete_invoice", + f"Invoice at index {index} is missing required field(s): {', '.join(missing)}.", + case_id=clean_case_id, + ) + subtotal = _money(item.get("subtotal")) + tax = _money(item.get("tax")) + total = _money(item.get("total")) + if subtotal is None or tax is None or total is None: + return _validation_error("invalid_amount", f"Invoice at index {index} has a non-numeric amount.", case_id=clean_case_id) + invoices.append( + { + "index": index, + "invoice_number": str(item["invoice_number"]).strip(), + "vendor": str(item["vendor"]).strip(), + "subtotal": _money_json(subtotal), + "tax": _money_json(tax), + "total": _money_json(total), + "expected_total": _money_json(subtotal + tax), + } + ) + + groups: dict[str, list[int]] = {} + mismatches: list[dict[str, Any]] = [] + for inv in invoices: + key = inv["invoice_number"].casefold() + groups.setdefault(key, []).append(inv["index"]) + if Decimal(str(inv["total"])).quantize(Decimal("0.01")) != Decimal(str(inv["expected_total"])).quantize(Decimal("0.01")): + mismatches.append( { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, + "index": inv["index"], + "invoice_number": inv["invoice_number"], + "vendor": inv["vendor"], + "subtotal": inv["subtotal"], + "tax": inv["tax"], + "stated_total": inv["total"], + "expected_total": inv["expected_total"], + "difference": _money_json(Decimal(str(inv["total"])) - Decimal(str(inv["expected_total"]))), } ) + duplicates = [ + {"invoice_number": invoices[indexes[0]]["invoice_number"], "indexes": indexes, "count": len(indexes)} + for indexes in groups.values() + if len(indexes) > 1 + ] + return { + "ok": True, + "case_id": clean_case_id, + "invoice_count": len(invoices), + "duplicate_count": len(duplicates), + "total_mismatch_count": len(mismatches), + "duplicates": duplicates, + "total_mismatches": mismatches, + "invoices": invoices, + "decisions": [], + } - @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)}" + +def _validation_error(code: str, message: str, *, case_id: str | None = None) -> dict[str, Any]: + payload: dict[str, Any] = {"ok": False, "code": code, "message": message} + if case_id is not None: + payload["case_id"] = case_id + return payload + + +def _safe_filename(name: str) -> bool: + clean = str(name or "").strip() + return bool(clean and len(clean) <= 180 and "/" not in clean and "\\" not in clean and "\x00" not in clean) + + +def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]: + if not documents: + return _validation_error("no_uploads", "Attach at least one invoice document.") + if len(documents) > MAX_UPLOADS: + return _validation_error("too_many_uploads", f"At most {MAX_UPLOADS} uploads are allowed.") + invoices: list[dict[str, Any]] = [] + for document in documents: + if not _safe_filename(document.filename): + return _validation_error("invalid_filename", "Upload filename is missing or unsafe.") + if document.media_type not in ACCEPTED_MEDIA_TYPES: + return _validation_error("unsupported_media_type", "Upload must be JSON, CSV, or plain text.") + try: + data = base64.b64decode(document.data_base64, validate=True) + except Exception: + return _validation_error("invalid_base64", "Upload data_base64 is not valid base64.") + if len(data) > MAX_UPLOAD_BYTES: + return _validation_error("upload_too_large", f"Each upload must be at most {MAX_UPLOAD_BYTES} bytes.") + parsed = _parse_upload_bytes(document.filename, document.media_type, data) + if not parsed["ok"]: + return parsed + invoices.extend(parsed["invoices"]) + return {"ok": True, "invoices": invoices} + + +def _parse_upload_bytes(filename: str, media_type: str, data: bytes) -> dict[str, Any]: + try: + text = data.decode("utf-8-sig") + except UnicodeDecodeError: + return _validation_error("invalid_text_encoding", "Upload must be UTF-8 text.") + if media_type in {"application/json", "text/json"} or filename.lower().endswith(".json"): + try: + payload = json.loads(text) + except json.JSONDecodeError: + return _validation_error("invalid_json", "Upload JSON could not be parsed.") + if isinstance(payload, dict) and isinstance(payload.get("invoices"), list): + return {"ok": True, "invoices": payload["invoices"]} + if isinstance(payload, list): + return {"ok": True, "invoices": payload} + return _validation_error("invalid_json_shape", "JSON upload must be an invoice array or {invoices: [...]} object.") + if media_type == "text/csv" or filename.lower().endswith(".csv"): + rows = list(csv.DictReader(io.StringIO(text))) + return {"ok": True, "invoices": rows} + # Plain text bridge: accept one JSON object per line for simple browser demos. + invoices: list[dict[str, Any]] = [] + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + item = json.loads(line) + except json.JSONDecodeError: + return _validation_error("invalid_text_invoice", "Plain text uploads must contain one JSON invoice object per line.") + if not isinstance(item, dict): + return _validation_error("invalid_text_invoice", "Plain text upload lines must be JSON objects.") + invoices.append(item) + return {"ok": True, "invoices": invoices} + + +@contextmanager +def _db_connection(): + database_url = os.environ.get("DATABASE_URL") + if not database_url: + raise RuntimeError("DATABASE_URL is not configured for the managed Postgres resource") + import psycopg + + with psycopg.connect(database_url, options=DB_CONNECT_OPTIONS) as conn: + yield conn + + +def _now_iso() -> str: + return datetime.now(UTC).isoformat() + + +def _receipt_for(tenant: str, case: dict[str, Any], source: str, saved_at: str) -> dict[str, Any]: + public_case = { + "case_id": case["case_id"], + "duplicate_count": case["duplicate_count"], + "total_mismatch_count": case["total_mismatch_count"], + "invoice_count": case["invoice_count"], + "source": source, + } + digest = hashlib.sha256(json.dumps(public_case, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + return { + "receipt_id": f"invoice_guard:{case['case_id']}:{digest[:16]}", + "kind": "production_execution_receipt", + "agent": "invoice-guard-studio-v1", + "agent_version": "0.1.0", + "skill": "review_invoices", + "tenant_hash": hashlib.sha256(tenant.encode("utf-8")).hexdigest()[:16], + "input_hash": digest, + "created_at": saved_at, + } + + +def _persist_case_with_receipt(tenant: str, case: dict[str, Any], source: str) -> dict[str, Any]: + saved_at = _now_iso() + receipt = _receipt_for(tenant, case, source, saved_at) + case_payload = {**case, "receipt": receipt, "saved_at": saved_at, "updated_at": saved_at} + with _db_connection() as conn: + with conn.transaction(): + conn.execute( + """ + INSERT INTO invoice_cases ( + tenant_key, case_id, invoice_count, duplicate_count, total_mismatch_count, + invoices, duplicates, total_mismatches, decisions, receipt, created_at, updated_at + ) VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, NOW(), NOW()) + ON CONFLICT (tenant_key, case_id) DO UPDATE SET + invoice_count = EXCLUDED.invoice_count, + duplicate_count = EXCLUDED.duplicate_count, + total_mismatch_count = EXCLUDED.total_mismatch_count, + invoices = EXCLUDED.invoices, + duplicates = EXCLUDED.duplicates, + total_mismatches = EXCLUDED.total_mismatches, + receipt = EXCLUDED.receipt, + updated_at = NOW() + """, + ( + tenant, + case["case_id"], + case["invoice_count"], + case["duplicate_count"], + case["total_mismatch_count"], + json.dumps(case["invoices"]), + json.dumps(case["duplicates"]), + json.dumps(case["total_mismatches"]), + json.dumps(case.get("decisions", [])), + json.dumps(receipt), + ), ) - 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, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, - ) + conn.execute( + """ + INSERT INTO invoice_execution_receipts (tenant_key, receipt_id, case_id, skill, input_hash, receipt, created_at) + VALUES (%s, %s, %s, %s, %s, %s::jsonb, NOW()) + ON CONFLICT (tenant_key, receipt_id) DO UPDATE SET receipt = EXCLUDED.receipt, created_at = NOW() + """, + (tenant, receipt["receipt_id"], case["case_id"], "review_invoices", receipt["input_hash"], json.dumps(receipt)), + ) + return {"receipt": receipt, "saved_at": saved_at, "case": case_payload} -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 _row_to_case(row: Any) -> dict[str, Any]: + payload = { + "ok": True, + "case_id": row["case_id"], + "invoice_count": row["invoice_count"], + "duplicate_count": row["duplicate_count"], + "total_mismatch_count": row["total_mismatch_count"], + "invoices": row["invoices"], + "duplicates": row["duplicates"], + "total_mismatches": row["total_mismatches"], + "decisions": row["decisions"], + "receipt": row["receipt"], + "saved_at": row["created_at"].isoformat() if hasattr(row["created_at"], "isoformat") else str(row["created_at"]), + "updated_at": row["updated_at"].isoformat() if hasattr(row["updated_at"], "isoformat") else str(row["updated_at"]), + } + return payload -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 _load_case(tenant: str, case_id: str) -> dict[str, Any] | None: + clean_case_id = _clean_case_id(case_id) + with _db_connection() as conn: + conn.row_factory = _dict_row_factory() + row = conn.execute( + """ + SELECT case_id, invoice_count, duplicate_count, total_mismatch_count, + invoices, duplicates, total_mismatches, decisions, receipt, created_at, updated_at + FROM invoice_cases + WHERE tenant_key = %s AND case_id = %s + """, + (tenant, clean_case_id), + ).fetchone() + return _row_to_case(row) if row else 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 _persist_decision(tenant: str, case_id: str, decision: dict[str, Any]) -> dict[str, Any] | None: + clean_case_id = _clean_case_id(case_id) + decided_at = _now_iso() + decision_payload = {**decision, "decided_at": decided_at} + with _db_connection() as conn: + conn.row_factory = _dict_row_factory() + with conn.transaction(): + row = conn.execute( + """ + SELECT decisions FROM invoice_cases + WHERE tenant_key = %s AND case_id = %s + FOR UPDATE + """, + (tenant, clean_case_id), + ).fetchone() + if row is None: + return None + decisions = [item for item in (row["decisions"] or []) if item.get("decision_id") != decision_payload["decision_id"]] + decisions.append(decision_payload) + conn.execute( + """ + UPDATE invoice_cases + SET decisions = %s::jsonb, updated_at = NOW() + WHERE tenant_key = %s AND case_id = %s + """, + (json.dumps(decisions), tenant, clean_case_id), + ) + return _load_case(tenant, clean_case_id) - 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 _dict_row_factory(): + from psycopg.rows import dict_row + + return dict_row