a2a-source-edit: write agent.py

This commit is contained in:
a2a-cloud
2026-07-18 07:11:47 +00:00
parent b75604ad6c
commit 6d3b4b7b6e

656
agent.py
View File

@@ -1,189 +1,561 @@
"""contract-clock-studio-v1 agent. """ContractClock Studio: deterministic contract deadline extraction.
Starter stack: This agent is intentionally no-LLM. It extracts only explicit ISO-dated
- DeepAgents for tool-calling orchestration renewals and notice windows from contracts, persists user-scoped timelines in
- Caller-provided LLM credentials via ctx.llm managed Postgres, and never guesses ambiguous dates.
- 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 binascii
import hashlib
import json import json
from pathlib import Path import os
from typing import Any import re
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import date, datetime, timezone, timedelta
from typing import Annotated, Any, Iterator, Literal
from pydantic import BaseModel from pydantic import BaseModel, Field
import a2a_pack as a2a import a2a_pack as a2a
from a2a_pack import ( from a2a_pack import (
A2AAgent, A2AAgent,
LLMProvisioning, AgentDatabase,
{{ auth_type }}, AgentDatabaseEnv,
AgentDatabaseMigrations,
AgentPlatformResources,
PlatformUserAuth,
Pricing, Pricing,
Resources,
RunContext, RunContext,
WorkspaceAccess, State,
WorkspaceMode, UploadedFile,
FileUpload,
) )
from a2a_pack.context import LLMCreds
MAX_TEXT_CHARS = 200_000
MAX_UPLOAD_BYTES = 256 * 1024
ALLOWED_MEDIA_TYPES = ("text/plain", "text/markdown", "application/octet-stream")
DATABASE_ENV = "DATABASE_URL"
DB_STATEMENT_TIMEOUT_MS = 5_000
DB_CONNECT_TIMEOUT_SECONDS = 3
ISO_DATE_RE = re.compile(r"\b(20\d{2}|19\d{2})-(0[1-9]|1[0-2])-([0-2]\d|3[01])\b")
NOTICE_RE = re.compile(
r"(?:at\s+least\s+)?(?P<days>\d{1,3})\s+days?\s+(?:prior\s+)?(?:written\s+)?notice",
re.IGNORECASE,
)
AMBIGUOUS_DATE_RE = re.compile(
r"\b(?:sometime|around|approximately|about|next\s+(?:spring|summer|fall|autumn|winter|month|year)|well\s+in\s+advance|tbd|to\s+be\s+determined)\b",
re.IGNORECASE,
)
RENEWAL_RE = re.compile(r"\brenew(?:s|al|ed|ing)?\b", re.IGNORECASE)
class ContractClockStudioV1Config(BaseModel): class ContractClockStudioV1Config(BaseModel):
pass pass
SYSTEM_PROMPT = """\ class Deadline(BaseModel):
You are a compact tool-calling agent. kind: Literal["renewal", "notice", "obligation"]
date: str
Use the text_stats tool when the user asks about text, counts, summaries, summary: str
or anything where exact length/word numbers would help. Mention tool results source_text: str
briefly instead of dumping raw JSON.
"""
RUNTIME_SKILLS_DIR = "contract-clock-studio-v1/.deepagents/skills/"
DEEPAGENTS_RECURSION_LIMIT = 500
class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]): 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 AnalyzeResult(BaseModel):
ok: bool
contract_id: str
title: str | None = None
deadlines: list[Deadline] = Field(default_factory=list)
code: str | None = None
message: str | None = None
warnings: list[str] = Field(default_factory=list)
receipt_id: str | None = None
persisted: bool = False
class ContractRecord(BaseModel):
ok: bool
contract_id: str
title: str | None = None
deadlines: list[Deadline] = Field(default_factory=list)
code: str | None = None
message: str | None = None
receipt_id: str | None = None
persisted_at: str | None = None
class ContractSummary(BaseModel):
contract_id: str
title: str
deadline_count: int
updated_at: str
class ListContractsResult(BaseModel):
ok: bool
contracts: list[ContractSummary] = Field(default_factory=list)
code: str | None = None
message: str | None = None
class ValidationFailure(Exception):
def __init__(self, code: str, message: str) -> None:
super().__init__(message)
self.code = code
self.message = message
@dataclass(frozen=True)
class ExtractedTimeline:
deadlines: list[Deadline]
warnings: list[str]
class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, PlatformUserAuth]):
name = "contract-clock-studio-v1" name = "contract-clock-studio-v1"
description = "Private full-stack ContractClock app for deterministic contract deadline extraction and user-scoped timeline persistence." description = (
"Private ContractClock app: paste or upload a contract, deterministically "
"extract explicit renewal and notice dates, persist the timeline in "
"user-scoped managed Postgres, and reopen it later."
)
version = "0.1.0" version = "0.1.0"
config_model = ContractClockStudioV1Config config_model = ContractClockStudioV1Config
auth_model = {{ auth_type }} auth_model = PlatformUserAuth
# Hosted generated agents read the caller's saved LLM credential through # Deterministic local logic only: no LLM credential is required.
# 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( pricing = Pricing(
price_per_call_usd=0.0, price_per_call_usd=0.0,
caller_pays_llm=True, caller_pays_llm=False,
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", notes="No LLM calls. Deterministic extraction and managed Postgres persistence only.",
)
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
state = State.DURABLE
tools_used = ("postgres", "mcp")
platform_resources = AgentPlatformResources(
databases=(
AgentDatabase(
name="contract-clock-studio-v1-data",
scope="user",
access_mode="read_write",
env=AgentDatabaseEnv(url=DATABASE_ENV),
migrations=AgentDatabaseMigrations(path="db/migrations"),
),
) )
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") @a2a.tool(
async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: description="Analyze pasted contract text and persist explicit renewal and notice deadlines.",
creds = ctx.llm timeout_seconds=30,
await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") idempotent=True,
if not creds.api_key: cost_class="deterministic",
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) async def analyze_contract(
state = await graph.ainvoke(
{"messages": [{"role": "user", "content": prompt}]},
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
)
await ctx.emit_progress("deepagent finished")
return _last_message_text(state)
def _build_deep_agent(
self, self,
*, ctx: RunContext[PlatformUserAuth],
ctx: RunContext[{{ auth_type }}], contract_id: str,
creds: LLMCreds, text: str,
) -> Any: title: str,
# Lazy imports keep `a2a card` usable before local dependencies are ) -> AnalyzeResult:
# installed. `a2a deploy` installs requirements.txt during the build. tenant = tenant_key(ctx)
from a2a_pack.deepagents import create_a2a_deep_agent try:
from langchain.agents.middleware import wrap_model_call clean_id = validate_contract_id(contract_id)
from langchain_core.tools import tool clean_title = validate_title(title)
clean_text = validate_text(text)
timeline = extract_timeline(clean_text)
except ValidationFailure as exc:
return AnalyzeResult(
ok=False,
contract_id=contract_id,
title=title,
code=exc.code,
message=exc.message,
)
@tool payload = {
def text_stats(text: str) -> str: "contract_id": clean_id,
"""Return exact word, character, and line counts for text.""" "title": clean_title,
words = [part for part in text.split() if part.strip()] "text_sha256": sha256_text(clean_text),
return json.dumps( "deadlines": [deadline.model_dump(mode="json") for deadline in timeline.deadlines],
{ "warnings": timeline.warnings,
"characters": len(text),
"words": len(words),
"lines": len(text.splitlines()) or 1,
} }
receipt_id = production_receipt_id(tenant, "analyze_contract", payload)
try:
persist_contract(tenant, clean_id, clean_title, payload, receipt_id)
persist_receipt(tenant, receipt_id, "analyze_contract", clean_id, payload, "ok")
except Exception:
# Fail closed: analysis without durable storage is not a success for this product.
return AnalyzeResult(
ok=False,
contract_id=clean_id,
title=clean_title,
deadlines=timeline.deadlines,
code="persistence_unavailable",
message="ContractClock could not save the timeline. Please retry after database setup is healthy.",
warnings=timeline.warnings,
receipt_id=receipt_id,
persisted=False,
) )
@wrap_model_call await ctx.emit_progress(f"Saved {len(timeline.deadlines)} deadlines for {clean_id}.")
async def log_model_call(request: Any, handler: Any) -> Any: return AnalyzeResult(
messages = request.state.get("messages", []) ok=True,
print( contract_id=clean_id,
"[middleware] model_call " title=clean_title,
f"model={creds.model} source={creds.source} messages={len(messages)}" deadlines=timeline.deadlines,
) warnings=timeline.warnings,
return await handler(request) receipt_id=receipt_id,
persisted=True,
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,
) )
@a2a.tool(
description="Analyze a bounded browser-uploaded base64 text contract and persist its timeline.",
timeout_seconds=30,
idempotent=True,
cost_class="deterministic",
)
async def analyze_contract_browser_upload(
self,
ctx: RunContext[PlatformUserAuth],
contract_id: str,
title: str,
document: BrowserDocument,
) -> AnalyzeResult:
try:
text = decode_browser_document(document)
except ValidationFailure as exc:
return AnalyzeResult(ok=False, contract_id=contract_id, title=title, code=exc.code, message=exc.message)
return await self.analyze_contract(ctx, contract_id=contract_id, title=title, text=text)
def _runtime_skills_root(ctx: RunContext[Any]) -> str: @a2a.tool(
workspace = getattr(ctx, "_workspace", None) description="Analyze a typed external FileUpload text contract and persist its timeline.",
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) timeout_seconds=30,
if not prefixes: idempotent=True,
outputs_prefix = getattr(workspace, "outputs_prefix", None) cost_class="deterministic",
prefixes = (outputs_prefix or "outputs/",) )
prefix = str(prefixes[0]).strip("/") async def analyze_contract_file_upload(
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" self,
ctx: RunContext[PlatformUserAuth],
contract_id: str,
title: str,
document: Annotated[
UploadedFile,
FileUpload(
accept=ALLOWED_MEDIA_TYPES,
max_bytes=MAX_UPLOAD_BYTES,
description="Plain-text contract file, max 256 KiB.",
),
],
) -> AnalyzeResult:
try:
if document.size_bytes > MAX_UPLOAD_BYTES:
raise ValidationFailure("file_too_large", "Upload is larger than 256 KiB.")
if document.media_type not in ALLOWED_MEDIA_TYPES:
raise ValidationFailure("unsupported_media_type", "Only plain text uploads are supported.")
grant = await ctx.workspace.request_access(
files=[document.path],
mode=a2a.WorkspaceMode.READ_ONLY,
reason="Read the caller-uploaded contract text for deadline extraction.",
purpose="contract upload analysis",
)
view = await ctx.workspace.open_view(
purpose="contract upload analysis",
hints=[document.path],
max_files=1,
mode=a2a.WorkspaceMode.READ_ONLY,
reason=f"Read uploaded file under grant {grant.grant_id}.",
)
raw = await view.read(document.path)
if len(raw) > MAX_UPLOAD_BYTES:
raise ValidationFailure("file_too_large", "Upload is larger than 256 KiB.")
text = raw.decode("utf-8")
except UnicodeDecodeError:
return AnalyzeResult(ok=False, contract_id=contract_id, title=title, code="invalid_encoding", message="Upload must be UTF-8 text.")
except ValidationFailure as exc:
return AnalyzeResult(ok=False, contract_id=contract_id, title=title, code=exc.code, message=exc.message)
except Exception:
return AnalyzeResult(ok=False, contract_id=contract_id, title=title, code="file_read_failed", message="Could not read the uploaded file from the workspace grant.")
return await self.analyze_contract(ctx, contract_id=contract_id, title=title, text=text)
@a2a.tool(
description="Reload a previously saved contract timeline for the signed-in user.",
timeout_seconds=15,
idempotent=True,
cost_class="deterministic",
)
async def get_contract_deadlines(
self,
ctx: RunContext[PlatformUserAuth],
contract_id: str,
) -> ContractRecord:
tenant = tenant_key(ctx)
try:
clean_id = validate_contract_id(contract_id)
record = load_contract(tenant, clean_id)
except ValidationFailure as exc:
return ContractRecord(ok=False, contract_id=contract_id, code=exc.code, message=exc.message)
except Exception:
return ContractRecord(ok=False, contract_id=contract_id, code="persistence_unavailable", message="ContractClock could not load timelines from storage.")
if record is None:
return ContractRecord(ok=False, contract_id=clean_id, code="not_found", message="No saved timeline exists for this contract and user.")
payload = record["payload"]
deadlines = [Deadline.model_validate(item) for item in payload.get("deadlines", [])]
return ContractRecord(
ok=True,
contract_id=clean_id,
title=payload.get("title"),
deadlines=deadlines,
receipt_id=record.get("receipt_id"),
persisted_at=record.get("updated_at"),
)
@a2a.tool(
description="List saved contract timelines for the signed-in user.",
timeout_seconds=15,
idempotent=True,
cost_class="deterministic",
)
async def list_contracts(self, ctx: RunContext[PlatformUserAuth], limit: int = 20) -> ListContractsResult:
tenant = tenant_key(ctx)
safe_limit = max(1, min(int(limit), 50))
try:
rows = list_contract_rows(tenant, safe_limit)
except Exception:
return ListContractsResult(ok=False, code="persistence_unavailable", message="ContractClock could not list saved timelines.")
return ListContractsResult(ok=True, contracts=[ContractSummary.model_validate(row) for row in rows])
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: def tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
"""Copy packaged DeepAgents skills into the invocation workspace. auth = ctx.auth
principal = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
if not principal:
raise ValidationFailure("auth_required", "A signed-in platform user is required.")
return f"user:{principal}"
DeepAgents loads skills from its backend, while source-controlled
``skills/`` folders live in the image. This bridge lets generated agents def validate_contract_id(contract_id: str) -> str:
ship reusable SKILL.md bundles without giving up durable A2A workspace value = (contract_id or "").strip()
files. if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", value):
raise ValidationFailure("invalid_contract_id", "contract_id must be 1-128 safe identifier characters.")
return value
def validate_title(title: str) -> str:
value = (title or "").strip()
if not value or len(value) > 180:
raise ValidationFailure("invalid_title", "title is required and must be at most 180 characters.")
return value
def validate_text(text: str) -> str:
value = text or ""
if not value.strip():
raise ValidationFailure("empty_contract", "Contract text is required.")
if len(value) > MAX_TEXT_CHARS:
raise ValidationFailure("text_too_large", "Contract text exceeds the 200,000 character limit.")
return value
def decode_browser_document(document: BrowserDocument) -> str:
name = document.filename.strip().replace("\\", "/").split("/")[-1]
if not name or name in {".", ".."}:
raise ValidationFailure("invalid_filename", "Upload filename is invalid.")
if document.media_type not in ALLOWED_MEDIA_TYPES:
raise ValidationFailure("unsupported_media_type", "Only plain text uploads are supported.")
try:
raw = base64.b64decode(document.data_base64, validate=True)
except (binascii.Error, ValueError):
raise ValidationFailure("invalid_base64", "Upload data_base64 must be valid base64.") from None
if len(raw) > MAX_UPLOAD_BYTES:
raise ValidationFailure("file_too_large", "Upload is larger than 256 KiB.")
try:
return raw.decode("utf-8")
except UnicodeDecodeError:
raise ValidationFailure("invalid_encoding", "Upload must be UTF-8 text.") from None
def extract_timeline(text: str) -> ExtractedTimeline:
if AMBIGUOUS_DATE_RE.search(text) and not ISO_DATE_RE.search(text):
raise ValidationFailure("ambiguous_date", "The contract uses ambiguous date language; no deadline was invented.")
dates: list[tuple[date, str]] = []
for match in ISO_DATE_RE.finditer(text):
token = match.group(0)
try:
parsed = date.fromisoformat(token)
except ValueError:
raise ValidationFailure("invalid_date", f"Invalid calendar date: {token}.") from None
dates.append((parsed, source_excerpt(text, match.start(), match.end())))
if not dates:
if RENEWAL_RE.search(text) or NOTICE_RE.search(text) or AMBIGUOUS_DATE_RE.search(text):
raise ValidationFailure("ambiguous_date", "No explicit ISO renewal date was found; ContractClock will not guess.")
raise ValidationFailure("no_dated_obligations", "No explicit ISO-dated obligations were found.")
deadlines: list[Deadline] = []
warnings: list[str] = []
renewal_dates = []
for parsed, excerpt in dates:
if RENEWAL_RE.search(excerpt) or RENEWAL_RE.search(text[max(0, text.find(parsed.isoformat()) - 160): text.find(parsed.isoformat()) + 160]):
renewal_dates.append(parsed)
deadlines.append(
Deadline(kind="renewal", date=parsed.isoformat(), summary="Automatic renewal date explicitly stated in contract.", source_text=excerpt)
)
else:
deadlines.append(
Deadline(kind="obligation", date=parsed.isoformat(), summary="Explicit dated obligation stated in contract.", source_text=excerpt)
)
notice_match = NOTICE_RE.search(text)
if notice_match:
days = int(notice_match.group("days"))
if not renewal_dates:
raise ValidationFailure("ambiguous_date", "Notice window found, but no explicit renewal date anchors the notice deadline.")
anchor = min(renewal_dates)
notice_date = anchor - timedelta(days=days)
deadlines.append(
Deadline(
kind="notice",
date=notice_date.isoformat(),
summary=f"Notice deadline calculated as {days} days before renewal date {anchor.isoformat()}.",
source_text=source_excerpt(text, notice_match.start(), notice_match.end()),
)
)
elif RENEWAL_RE.search(text):
warnings.append("No explicit notice-window duration was found for the renewal date.")
# Acceptance and UI prefer renewal before notice before other dated obligations.
order = {"renewal": 0, "notice": 1, "obligation": 2}
deduped: dict[tuple[str, str], Deadline] = {}
for deadline in sorted(deadlines, key=lambda item: (order[item.kind], item.date, item.summary)):
deduped.setdefault((deadline.kind, deadline.date), deadline)
return ExtractedTimeline(deadlines=list(deduped.values()), warnings=warnings)
def source_excerpt(text: str, start: int, end: int, radius: int = 140) -> str:
left = max(0, start - radius)
right = min(len(text), end + radius)
return re.sub(r"\s+", " ", text[left:right]).strip()
def sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def production_receipt_id(tenant: str, skill: str, payload: dict[str, Any]) -> str:
material = json.dumps({"tenant": tenant, "skill": skill, "payload": payload}, sort_keys=True, separators=(",", ":"))
return "ccr_" + hashlib.sha256(material.encode("utf-8")).hexdigest()[:24]
@contextmanager
def db_connection() -> Iterator[Any]:
url = os.environ.get(DATABASE_ENV)
if not url:
raise RuntimeError("DATABASE_URL is not configured")
import psycopg
with psycopg.connect(
url,
connect_timeout=DB_CONNECT_TIMEOUT_SECONDS,
options=f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} -c idle_in_transaction_session_timeout={DB_STATEMENT_TIMEOUT_MS}",
) as conn:
with conn.cursor() as cur:
cur.execute("SET LOCAL statement_timeout = %s", (DB_STATEMENT_TIMEOUT_MS,))
yield conn
def persist_contract(tenant: str, contract_id: str, title: str, payload: dict[str, Any], receipt_id: str) -> None:
with db_connection() as conn:
with conn.cursor() as cur:
cur.execute(
""" """
root = Path(__file__).parent / "skills" INSERT INTO contract_timelines (tenant_key, contract_id, title, payload, receipt_id, updated_at)
if not root.exists(): VALUES (%s, %s, %s, %s::jsonb, %s, NOW())
return [] ON CONFLICT (tenant_key, contract_id)
runtime_skills_root = _runtime_skills_root(ctx) DO UPDATE SET title = EXCLUDED.title, payload = EXCLUDED.payload,
uploads: list[tuple[str, bytes]] = [] receipt_id = EXCLUDED.receipt_id, updated_at = NOW()
for path in root.rglob("*"): """,
if path.is_file(): (tenant, contract_id, title, json.dumps(payload), receipt_id),
rel = path.relative_to(root).as_posix() )
uploads.append((runtime_skills_root + rel, path.read_bytes())) conn.commit()
if uploads:
backend.upload_files(uploads)
return [runtime_skills_root]
return []
def _last_message_text(state: dict[str, Any]) -> str: def persist_receipt(tenant: str, receipt_id: str, skill_name: str, contract_id: str, payload: dict[str, Any], status: str) -> None:
messages = state.get("messages") or [] receipt = {
if not messages: "receipt_id": receipt_id,
return json.dumps(state, default=str) "agent": ContractClockStudioV1.name,
"agent_version": ContractClockStudioV1.version,
"skill": skill_name,
"contract_id": contract_id,
"input_hash": sha256_text(json.dumps(payload, sort_keys=True)),
"status": status,
"created_at": datetime.now(timezone.utc).isoformat(),
}
with db_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO execution_receipts (tenant_key, receipt_id, skill_name, contract_id, status, payload)
VALUES (%s, %s, %s, %s, %s, %s::jsonb)
ON CONFLICT (tenant_key, receipt_id) DO NOTHING
""",
(tenant, receipt_id, skill_name, contract_id, status, json.dumps(receipt)),
)
conn.commit()
content = getattr(messages[-1], "content", None)
if isinstance(content, str): def load_contract(tenant: str, contract_id: str) -> dict[str, Any] | None:
return content with db_connection() as conn:
if isinstance(content, list): with conn.cursor() as cur:
parts: list[str] = [] cur.execute(
for item in content: """
if isinstance(item, dict): SELECT payload, receipt_id, updated_at
text = item.get("text") or item.get("content") FROM contract_timelines
if text: WHERE tenant_key = %s AND contract_id = %s
parts.append(str(text)) """,
elif item: (tenant, contract_id),
parts.append(str(item)) )
return "\n".join(parts) if parts else json.dumps(content, default=str) row = cur.fetchone()
return str(content or messages[-1]) if row is None:
return None
payload, receipt_id, updated_at = row
return {
"payload": payload if isinstance(payload, dict) else json.loads(payload),
"receipt_id": receipt_id,
"updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
}
def list_contract_rows(tenant: str, limit: int) -> list[dict[str, Any]]:
with db_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT contract_id, title, jsonb_array_length(payload->'deadlines') AS deadline_count, updated_at
FROM contract_timelines
WHERE tenant_key = %s
ORDER BY updated_at DESC
LIMIT %s
""",
(tenant, limit),
)
rows = cur.fetchall()
return [
{
"contract_id": contract_id,
"title": title,
"deadline_count": int(deadline_count or 0),
"updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
}
for contract_id, title, deadline_count, updated_at in rows
]