Files
contract-clock-studio-v1/agent.py

732 lines
25 KiB
Python

"""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))