diff --git a/agent.py b/agent.py index 12f2873..b3935bd 100644 --- a/agent.py +++ b/agent.py @@ -1,583 +1,189 @@ -"""ContractClock Studio: deterministic contract deadline extraction. +"""contract-clock-studio-v1 agent. -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. +Starter stack: + - DeepAgents for tool-calling orchestration + - Caller-provided LLM credentials via ctx.llm + - A tiny model-call middleware hook you can replace with tracing, + routing, rate limits, or policy checks """ from __future__ import annotations -import base64 -import binascii -import hashlib import json -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 pathlib import Path +from typing import Any -from pydantic import BaseModel, Field +from pydantic import BaseModel import a2a_pack as a2a from a2a_pack import ( A2AAgent, - AgentDatabase, - AgentDatabaseEnv, - AgentDatabaseMigrations, - AgentPlatformResources, - FileUpload, - PlatformUserAuth, + LLMProvisioning, + {{ auth_type }}, Pricing, - Resources, RunContext, - State, - UploadedFile, WorkspaceAccess, WorkspaceMode, ) - -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 -VERSION = "0.1.1" - -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) -DEADLINE_KIND_ORDER = {"renewal": 0, "notice": 1, "obligation": 2} +from a2a_pack.context import LLMCreds class ContractClockStudioV1Config(BaseModel): pass -class ContractClockState(BaseModel): - """Durable-state marker; records live in managed Postgres tables.""" +SYSTEM_PROMPT = """\ +You are a compact tool-calling agent. - durable_store: str = "managed_postgres" +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 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]): +class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]): name = "contract-clock-studio-v1" - 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 = VERSION + description = "ContractClock one-page full-stack startup for extracting explicit contract renewal and notice deadlines, persisting user-scoped results, and reopening timelines." + version = "0.1.0" config_model = ContractClockStudioV1Config - auth_model = PlatformUserAuth + auth_model = {{ auth_type }} - # Deterministic local logic only: no LLM credential is read or required. + # Hosted generated agents read the caller's saved LLM credential through + # ctx.llm. The platform may proxy that credential through LiteLLM, but agent + # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY + # directly. + llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=False, - notes="No LLM calls. Deterministic extraction and managed Postgres persistence only.", + caller_pays_llm=True, + notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", ) - resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) - state = State.DURABLE - state_model = ContractClockState - tools_used = ("postgres", "mcp") workspace_access = WorkspaceAccess.dynamic( - max_files=4, - allowed_modes=(WorkspaceMode.READ_ONLY,), + max_files=64, + allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, - max_total_size_bytes=MAX_UPLOAD_BYTES, ) - 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"), - ), + 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}, ) - ) + await ctx.emit_progress("deepagent finished") + return _last_message_text(state) - @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( + def _build_deep_agent( self, - ctx: RunContext[PlatformUserAuth], - contract_id: str, - text: str, - title: str, - ) -> AnalyzeResult: - try: - tenant = tenant_key(ctx) - 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, + *, + 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, + } ) - 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_with_receipt( - tenant, - clean_id, - clean_title, - payload, - receipt_id, - "analyze_contract", - "ok", - ) - except Exception: - 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 + async def log_model_call(request: Any, handler: Any) -> Any: + messages = request.state.get("messages", []) + print( + "[middleware] model_call " + f"model={creds.model} source={creds.source} messages={len(messages)}" ) + return await handler(request) - 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, + 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) - @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.") - await ctx.workspace.request_access( - files=[document.path], - mode=WorkspaceMode.READ_ONLY, - reason="Read the caller-uploaded contract text for deadline extraction.", - purpose="contract upload analysis", - ) - read_bytes = getattr(ctx.workspace, "read_bytes", None) - if read_bytes is None: - raise RuntimeError("workspace backend does not expose direct file reads") - raw = read_bytes(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: - try: - tenant = tenant_key(ctx) - 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: - try: - tenant = tenant_key(ctx) - safe_limit = max(1, min(int(limit), 50)) - rows = list_contract_rows(tenant, safe_limit) - except ValidationFailure as exc: - return ListContractsResult(ok=False, code=exc.code, message=exc.message) - 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 _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 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 _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 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 +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_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: - date_token = parsed.isoformat() - date_index = text.find(date_token) - local_context = text[max(0, date_index - 160): date_index + 160] if date_index >= 0 else excerpt - if RENEWAL_RE.search(excerpt) or RENEWAL_RE.search(local_context): - renewal_dates.append(parsed) - deadlines.append(Deadline(kind="renewal", date=date_token, summary="Automatic renewal date explicitly stated in contract.", source_text=excerpt)) - else: - deadlines.append(Deadline(kind="obligation", date=date_token, 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.") - - deduped: dict[tuple[str, str], Deadline] = {} - for deadline in sorted(deadlines, key=deadline_sort_key): - deduped.setdefault((deadline.kind, deadline.date), deadline) - return ExtractedTimeline(deadlines=list(deduped.values()), warnings=warnings) - - -def deadline_sort_key(deadline: Deadline) -> tuple[int, str, str]: - return (DEADLINE_KIND_ORDER[deadline.kind], deadline.date, deadline.summary) - - -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: - yield conn - - -def persist_contract_with_receipt( - tenant: str, - contract_id: str, - title: str, - payload: dict[str, Any], - receipt_id: str, - skill_name: str, - 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 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), - ) - 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 - ] + 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])