"""ContractClock Studio v1 full-stack A2A agent. Deterministic, no-LLM contract deadline extraction for explicit dated renewal clauses. State is partitioned by the authenticated A2A Cloud user and persisted in managed Postgres with a same-transaction execution receipt row. """ from __future__ import annotations import base64 import binascii import hashlib import json import os import re import uuid from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from typing import Annotated, Any, Literal from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, AgentDatabase, AgentDatabaseEnv, AgentDatabaseMigrations, AgentPlatformResources, PlatformUserAuth, Pricing, Resources, RunContext, State, UploadedFile, FileUpload, WorkspaceAccess, WorkspaceMode, ) MAX_CONTRACT_TEXT_CHARS = 80_000 MAX_BROWSER_UPLOAD_BYTES = 512_000 ALLOWED_UPLOAD_MEDIA_TYPES = ("text/plain", "application/octet-stream") DB_STATEMENT_TIMEOUT_MS = 5_000 DB_LOCK_TIMEOUT_MS = 3_000 DB_IDLE_TRANSACTION_TIMEOUT_MS = 5_000 ContractId = Annotated[ str, Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}$"), ] ContractTitle = Annotated[str, Field(min_length=1, max_length=200)] ContractText = Annotated[str, Field(min_length=1, max_length=MAX_CONTRACT_TEXT_CHARS)] class ContractClockStudioV1Config(BaseModel): pass class ContractClockState(BaseModel): persistence: Literal["managed_postgres"] = "managed_postgres" class DeadlineOut(BaseModel): kind: Literal["renewal", "notice"] date: str source_text: str | None = None rationale: str | None = None class ContractResult(BaseModel): ok: bool contract_id: str title: str | None = None deadlines: list[DeadlineOut] = Field(default_factory=list) receipt_id: str | None = None saved_at: str | None = None code: str | None = None message: str | None = None 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_BROWSER_UPLOAD_BYTES * 2) @dataclass(frozen=True) class ExtractedDeadline: kind: Literal["renewal", "notice"] deadline_date: date source_text: str rationale: str def public(self) -> DeadlineOut: return DeadlineOut( kind=self.kind, date=self.deadline_date.isoformat(), source_text=self.source_text, rationale=self.rationale, ) class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, PlatformUserAuth]): name = "contract-clock-studio-v1" description = ( "ContractClock Studio extracts explicit renewal and notice deadlines " "from contracts, persists user-scoped timelines, and serves a packed " "one-page React product at /app." ) version = "0.1.8" config_model = ContractClockStudioV1Config auth_model = PlatformUserAuth # Deterministic local logic: no LLM credential is read or required. pricing = Pricing( price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic deadline extraction; no LLM usage.", ) resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) workspace_access = WorkspaceAccess.dynamic( max_files=1, allowed_modes=(WorkspaceMode.READ_ONLY,), require_reason=False, max_total_size_bytes=MAX_BROWSER_UPLOAD_BYTES, ) state = State.DURABLE state_model = ContractClockState tools_used = ("managed-postgres", "mcp", "packed-frontend") platform_resources = AgentPlatformResources( databases=( AgentDatabase( name="contract-clock-studio-v1-data", scope="user", access_mode="read_write", env=AgentDatabaseEnv(url="DATABASE_URL"), migrations=AgentDatabaseMigrations(path="db/migrations"), ), ) ) @a2a.tool( description=( "Analyze pasted contract text, extract explicit renewal/notice " "deadlines, atomically persist the contract timeline and receipt." ), timeout_seconds=30, idempotent=False, cost_class="deterministic-db-write", ) async def analyze_contract( self, ctx: RunContext[PlatformUserAuth], contract_id: ContractId, title: ContractTitle, text: ContractText, ) -> ContractResult: tenant = _tenant_key(ctx) if tenant is None: return _auth_required(contract_id) result = _analyze_contract_text(contract_id=contract_id, title=title, text=text) if not result.ok: await _persist_receipt_only(ctx, tenant, "analyze_contract", {"contract_id": contract_id}, result) return result return await _persist_contract_result( ctx=ctx, tenant_key=tenant, tool_name="analyze_contract", contract_id=contract_id, title=title, text=text, result=result, ) @a2a.tool( description="Reload a previously saved user-scoped ContractClock deadline timeline.", timeout_seconds=20, idempotent=True, cost_class="deterministic-db-read", ) async def get_contract_deadlines( self, ctx: RunContext[PlatformUserAuth], contract_id: ContractId, ) -> ContractResult: tenant = _tenant_key(ctx) if tenant is None: return _auth_required(contract_id) return await _load_contract_result(ctx, tenant, contract_id) @a2a.tool( description=( "Browser upload bridge: accept one bounded base64 text document, " "then analyze and persist its contract deadlines." ), timeout_seconds=30, idempotent=False, cost_class="deterministic-db-write", ) async def analyze_contract_upload_base64( self, ctx: RunContext[PlatformUserAuth], contract_id: ContractId, title: ContractTitle, document: BrowserDocument, ) -> ContractResult: decoded = _decode_browser_document(document) if isinstance(decoded, ContractResult): return decoded.model_copy(update={"contract_id": contract_id, "title": title}) return await self.analyze_contract(ctx, contract_id=contract_id, title=title, text=decoded) @a2a.tool( description=( "External client upload bridge: accept a typed A2A FileUpload text " "document, then analyze and persist its contract deadlines." ), timeout_seconds=30, idempotent=False, cost_class="deterministic-db-write", ) async def analyze_contract_file( self, ctx: RunContext[PlatformUserAuth], contract_id: ContractId, title: ContractTitle, document: Annotated[ UploadedFile, FileUpload( accept=ALLOWED_UPLOAD_MEDIA_TYPES, max_bytes=MAX_BROWSER_UPLOAD_BYTES, description="Plain-text contract document up to 512 KB.", ), ], ) -> ContractResult: text_or_error = _read_uploaded_text(ctx, contract_id, title, document) if isinstance(text_or_error, ContractResult): return text_or_error return await self.analyze_contract(ctx, contract_id=contract_id, title=title, text=text_or_error) def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str | None: auth = getattr(ctx, "auth", None) if auth is None: return None user_id = getattr(auth, "user_id", None) if user_id is not None: return f"user:{user_id}" subject = (getattr(auth, "sub", None) or "").strip() return f"user:{subject}" if subject else None def _auth_required(contract_id: str) -> ContractResult: return ContractResult( ok=False, code="auth_required", message="A signed-in A2A Cloud user session is required to access ContractClock records.", contract_id=contract_id, ) def _analyze_contract_text(contract_id: str, title: str, text: str) -> ContractResult: normalized = _normalize_text(text) if not _contains_explicit_iso_date(normalized): if _mentions_deadline_domain(normalized): return ContractResult( ok=False, code="ambiguous_date", message="ContractClock only extracts explicit ISO calendar dates; ambiguous dates were not guessed.", contract_id=contract_id, title=title, ) return ContractResult( ok=False, code="no_explicit_date", message="No explicit ISO calendar date was found in a renewal or notice clause.", contract_id=contract_id, title=title, ) deadlines = _extract_deadlines(normalized) if not deadlines: return ContractResult( ok=False, code="ambiguous_date", message="Explicit dates were present, but none were clause-bound to a renewal obligation.", contract_id=contract_id, title=title, ) return ContractResult( ok=True, contract_id=contract_id, title=title, deadlines=[item.public() for item in deadlines], ) def _deadline_sort_key(deadline: ExtractedDeadline) -> tuple[int, str]: return (0 if deadline.kind == "renewal" else 1, deadline.deadline_date.isoformat()) def _extract_deadlines(text: str) -> list[ExtractedDeadline]: deadlines: list[ExtractedDeadline] = [] seen: set[tuple[str, date]] = set() for clause in _deadline_clauses(text): lower = clause.lower() if "renew" not in lower: continue renewal_offset = lower.find("renew") dates_after_renewal = [ parsed for offset, parsed in _explicit_iso_dates_with_positions(clause) if offset > renewal_offset ] if not dates_after_renewal: continue renewal_date = dates_after_renewal[0] if ("renewal", renewal_date) not in seen: deadlines.append( ExtractedDeadline( kind="renewal", deadline_date=renewal_date, source_text=clause, rationale="Explicit ISO date appears inside the renewal clause.", ) ) seen.add(("renewal", renewal_date)) notice_days = _notice_days(clause) if notice_days is not None: notice_date = renewal_date - timedelta(days=notice_days) if ("notice", notice_date) not in seen: deadlines.append( ExtractedDeadline( kind="notice", deadline_date=notice_date, source_text=clause, rationale=f"Notice deadline is {notice_days} days before the explicit renewal date.", ) ) seen.add(("notice", notice_date)) # Public contract wants renewal before the derived notice date for a clause. deadlines.sort(key=_deadline_sort_key) return deadlines def _deadline_clauses(text: str) -> list[str]: parts = re.split(r"(?<=[.;])\s+|\n+", text) return [part.strip() for part in parts if part.strip()] def _normalize_text(text: str) -> str: normalized_lines = [ re.sub(r"[^\S\n]+", " ", line).strip() for line in text.splitlines() ] return "\n".join(line for line in normalized_lines if line).strip() def _contains_explicit_iso_date(text: str) -> bool: return bool(_explicit_iso_dates(text)) def _mentions_deadline_domain(text: str) -> bool: lower = text.lower() return any(word in lower for word in ("renew", "notice", "deadline", "due", "spring", "advance")) def _explicit_iso_dates(text: str) -> list[date]: return [parsed for _, parsed in _explicit_iso_dates_with_positions(text)] def _explicit_iso_dates_with_positions(text: str) -> list[tuple[int, date]]: out: list[tuple[int, date]] = [] for match in re.finditer(r"\b(20\d{2})-(0[1-9]|1[0-2])-([0-2]\d|3[01])\b", text): try: out.append((match.start(), date.fromisoformat(match.group(0)))) except ValueError: continue return out def _notice_days(text: str) -> int | None: lower = text.lower() if "notice" not in lower: return None match = re.search(r"(?:at\s+least\s+)?(\d{1,3})\s+days?\s+(?:written\s+)?notice", lower) if not match: match = re.search(r"notice\s+(?:of\s+)?(?:at\s+least\s+)?(\d{1,3})\s+days?", lower) if not match: return None days = int(match.group(1)) return days if 1 <= days <= 366 else None def _decode_browser_document(document: BrowserDocument) -> str | ContractResult: filename = document.filename.replace("\\", "/").split("/")[-1].strip() if not filename: return ContractResult(ok=False, code="invalid_upload", message="Filename is required.", contract_id="") if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: return ContractResult( ok=False, code="unsupported_media_type", message="Only plain-text uploads are supported by this ContractClock bridge.", contract_id="", ) try: raw = base64.b64decode(document.data_base64, validate=True) except (binascii.Error, ValueError): return ContractResult(ok=False, code="invalid_base64", message="Upload data_base64 is not valid base64.", contract_id="") if len(raw) > MAX_BROWSER_UPLOAD_BYTES: return ContractResult( ok=False, code="upload_too_large", message=f"Upload exceeds {MAX_BROWSER_UPLOAD_BYTES} bytes.", contract_id="", ) try: text = raw.decode("utf-8") except UnicodeDecodeError: return ContractResult(ok=False, code="invalid_text", message="Uploaded file must be UTF-8 text.", contract_id="") if len(text) > MAX_CONTRACT_TEXT_CHARS: return ContractResult(ok=False, code="text_too_large", message="Contract text exceeds the supported size.", contract_id="") return text def _read_uploaded_text( ctx: RunContext[PlatformUserAuth], contract_id: str, title: str, document: UploadedFile, ) -> str | ContractResult: if document.size_bytes > MAX_BROWSER_UPLOAD_BYTES: return ContractResult( ok=False, code="upload_too_large", message=f"Upload exceeds {MAX_BROWSER_UPLOAD_BYTES} bytes.", contract_id=contract_id, title=title, ) if document.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES: return ContractResult( ok=False, code="unsupported_media_type", message="Only plain-text uploads are supported.", contract_id=contract_id, title=title, ) try: raw = ctx.workspace.read_bytes(document.path) # type: ignore[attr-defined] except Exception: return ContractResult( ok=False, code="upload_read_failed", message="The uploaded file could not be read from the granted workspace.", contract_id=contract_id, title=title, ) if len(raw) > MAX_BROWSER_UPLOAD_BYTES: return ContractResult( ok=False, code="upload_too_large", message=f"Upload exceeds {MAX_BROWSER_UPLOAD_BYTES} bytes.", contract_id=contract_id, title=title, ) try: return raw.decode("utf-8") except UnicodeDecodeError: return ContractResult( ok=False, code="invalid_text", message="Uploaded file must be UTF-8 text.", contract_id=contract_id, title=title, ) async def _persist_contract_result( *, ctx: RunContext[PlatformUserAuth], tenant_key: str, tool_name: str, contract_id: str, title: str, text: str, result: ContractResult, ) -> ContractResult: if not os.environ.get("DATABASE_URL"): return result.model_copy( update={ "ok": False, "code": "setup_required", "message": "Managed Postgres is not configured; DATABASE_URL is missing.", } ) try: import psycopg from psycopg.rows import dict_row receipt_id = _receipt_id() now = _now_iso() input_payload = {"contract_id": contract_id, "text_sha256": _sha256(text)} result_payload = result.model_dump(mode="json") with psycopg.connect( os.environ["DATABASE_URL"], connect_timeout=3, options=_database_options(), row_factory=dict_row, ) as conn: with conn.transaction(): with conn.cursor() as cur: cur.execute( """ INSERT INTO contracts (tenant_key, contract_id, title, source_text_hash, updated_at) VALUES (%s, %s, %s, %s, NOW()) ON CONFLICT (tenant_key, contract_id) DO UPDATE SET title = EXCLUDED.title, source_text_hash = EXCLUDED.source_text_hash, updated_at = NOW() RETURNING id """, (tenant_key, contract_id, title, _sha256(text)), ) row = cur.fetchone() contract_pk = row["id"] cur.execute("DELETE FROM contract_deadlines WHERE contract_pk = %s", (contract_pk,)) for deadline in result.deadlines: cur.execute( """ INSERT INTO contract_deadlines (contract_pk, kind, deadline_date, source_text, rationale) VALUES (%s, %s, %s, %s, %s) """, ( contract_pk, deadline.kind, deadline.date, deadline.source_text, deadline.rationale, ), ) cur.execute( """ INSERT INTO execution_receipts (receipt_id, tenant_key, contract_id, tool_name, task_id, input_hash, result_hash, status, payload) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb) """, ( receipt_id, tenant_key, contract_id, tool_name, getattr(ctx, "task_id", ""), _hash_json(input_payload), _hash_json(result_payload), "ok", json.dumps( {"input": input_payload, "result": _receipt_result_summary(result)}, sort_keys=True, ), ), ) return result.model_copy(update={"receipt_id": receipt_id, "saved_at": now}) except Exception as exc: # noqa: BLE001 return result.model_copy( update={ "ok": False, "code": "database_error", "message": f"ContractClock could not persist this result ({type(exc).__name__}).", } ) async def _persist_receipt_only( ctx: RunContext[PlatformUserAuth], tenant_key: str, tool_name: str, input_payload: dict[str, Any], result: ContractResult, ) -> None: if not os.environ.get("DATABASE_URL"): return try: import psycopg with psycopg.connect( os.environ["DATABASE_URL"], connect_timeout=3, options=_database_options(), ) as conn: with conn.transaction(): with conn.cursor() as cur: cur.execute( """ INSERT INTO execution_receipts (receipt_id, tenant_key, contract_id, tool_name, task_id, input_hash, result_hash, status, payload) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb) """, ( _receipt_id(), tenant_key, input_payload.get("contract_id"), tool_name, getattr(ctx, "task_id", ""), _hash_json(input_payload), _hash_json(result.model_dump(mode="json")), "error", json.dumps( {"input": input_payload, "result": _receipt_result_summary(result)}, sort_keys=True, ), ), ) except Exception: return async def _load_contract_result( ctx: RunContext[PlatformUserAuth], tenant_key: str, contract_id: str, ) -> ContractResult: if not os.environ.get("DATABASE_URL"): return ContractResult( ok=False, code="setup_required", message="Managed Postgres is not configured; DATABASE_URL is missing.", contract_id=contract_id, ) try: import psycopg from psycopg.rows import dict_row with psycopg.connect( os.environ["DATABASE_URL"], connect_timeout=3, options=_database_options(), row_factory=dict_row, ) as conn: with conn.cursor() as cur: cur.execute( """ SELECT id, title, updated_at FROM contracts WHERE tenant_key = %s AND contract_id = %s """, (tenant_key, contract_id), ) contract = cur.fetchone() if not contract: return ContractResult( ok=False, code="not_found", message="No saved ContractClock timeline exists for this contract_id.", contract_id=contract_id, ) cur.execute( """ SELECT kind, deadline_date, source_text, rationale FROM contract_deadlines WHERE contract_pk = %s ORDER BY CASE kind WHEN 'renewal' THEN 0 WHEN 'notice' THEN 1 ELSE 2 END, deadline_date ASC """, (contract["id"],), ) deadlines = [ DeadlineOut( kind=row["kind"], date=row["deadline_date"].isoformat(), source_text=row["source_text"], rationale=row["rationale"], ) for row in cur.fetchall() ] updated_at = contract["updated_at"] saved_at = updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at) return ContractResult( ok=True, contract_id=contract_id, title=contract["title"], deadlines=deadlines, saved_at=saved_at, ) except Exception as exc: # noqa: BLE001 return ContractResult( ok=False, code="database_error", message=f"ContractClock could not reload this timeline ({type(exc).__name__}).", contract_id=contract_id, ) def _database_options() -> str: return ( f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} " f"-c lock_timeout={DB_LOCK_TIMEOUT_MS} " f"-c idle_in_transaction_session_timeout={DB_IDLE_TRANSACTION_TIMEOUT_MS}" ) def _receipt_result_summary(result: ContractResult) -> dict[str, Any]: return { "ok": result.ok, "code": result.code, "contract_id": result.contract_id, "deadline_count": len(result.deadlines), "deadlines": [ {"kind": deadline.kind, "date": deadline.date} for deadline in result.deadlines ], } def _receipt_id() -> str: return f"ccr_{uuid.uuid4().hex}" def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _sha256(value: str | bytes) -> str: raw = value.encode("utf-8") if isinstance(value, str) else value return hashlib.sha256(raw).hexdigest() def _hash_json(value: Any) -> str: return _sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), default=str))