diff --git a/agent.py b/agent.py index 0695992..64bd8ec 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,561 @@ -"""contract-clock-studio-v1 agent. +"""ContractClock Studio: deterministic contract deadline extraction. -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 +This agent is intentionally no-LLM. It extracts only explicit ISO-dated +renewals and notice windows from contracts, persists user-scoped timelines in +managed Postgres, and never guesses ambiguous dates. """ from __future__ import annotations +import base64 +import binascii +import hashlib import json -from pathlib import Path -from typing import Any +import os +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 from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + PlatformUserAuth, Pricing, + Resources, RunContext, - WorkspaceAccess, - WorkspaceMode, + State, + 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\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): pass -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 = "contract-clock-studio-v1/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 +class Deadline(BaseModel): + kind: Literal["renewal", "notice", "obligation"] + date: str + summary: str + source_text: str -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" - 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" config_model = ContractClockStudioV1Config - 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 + # Deterministic local logic only: no LLM credential is required. 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="No LLM calls. Deterministic extraction and managed Postgres persistence only.", ) - 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}, + 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"), + ), ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + ) - def _build_deep_agent( + @a2a.tool( + description="Analyze pasted contract text and persist explicit renewal and notice deadlines.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def analyze_contract( 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 - - @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, - } + ctx: RunContext[PlatformUserAuth], + contract_id: str, + text: str, + title: str, + ) -> AnalyzeResult: + tenant = tenant_key(ctx) + try: + clean_id = validate_contract_id(contract_id) + 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, ) - @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)}" + payload = { + "contract_id": clean_id, + "title": clean_title, + "text_sha256": sha256_text(clean_text), + "deadlines": [deadline.model_dump(mode="json") for deadline in timeline.deadlines], + "warnings": timeline.warnings, + } + 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, ) - 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, + await ctx.emit_progress(f"Saved {len(timeline.deadlines)} deadlines for {clean_id}.") + return AnalyzeResult( + ok=True, + contract_id=clean_id, + title=clean_title, + deadlines=timeline.deadlines, + warnings=timeline.warnings, + receipt_id=receipt_id, + persisted=True, ) + @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: - 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}" + @a2a.tool( + description="Analyze a typed external FileUpload text contract and persist its timeline.", + timeout_seconds=30, + idempotent=True, + cost_class="deterministic", + ) + async def analyze_contract_file_upload( + 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]: - """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 tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + 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}" -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def validate_contract_id(contract_id: str) -> str: + value = (contract_id or "").strip() + 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 - 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 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( + """ + INSERT INTO contract_timelines (tenant_key, contract_id, title, payload, receipt_id, updated_at) + VALUES (%s, %s, %s, %s::jsonb, %s, NOW()) + ON CONFLICT (tenant_key, contract_id) + DO UPDATE SET title = EXCLUDED.title, payload = EXCLUDED.payload, + receipt_id = EXCLUDED.receipt_id, updated_at = NOW() + """, + (tenant, contract_id, title, json.dumps(payload), receipt_id), + ) + conn.commit() + + +def persist_receipt(tenant: str, receipt_id: str, skill_name: str, contract_id: str, payload: dict[str, Any], status: str) -> None: + receipt = { + "receipt_id": receipt_id, + "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() + + +def load_contract(tenant: str, contract_id: str) -> dict[str, Any] | None: + with db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT payload, receipt_id, updated_at + FROM contract_timelines + WHERE tenant_key = %s AND contract_id = %s + """, + (tenant, contract_id), + ) + row = cur.fetchone() + 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 + ]