a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,500 @@
-"""contract-clock-studio-v1 agent.
+"""ContractClock Studio full-stack A2A agent.
 
-Starter stack:
+Deterministic product backend for the ContractClock one-page startup. The agent
-  - DeepAgents for tool-calling orchestration
+extracts only explicit dated renewal obligations and notice windows, persists
-  - Caller-provided LLM credentials via ctx.llm
+contract timelines in user-scoped managed Postgres, and exposes the same typed
-  - A tiny model-call middleware hook you can replace with tracing,
+A2A tools to Agent API clients, MCP clients, and the packed React frontend.
-    routing, rate limits, or policy checks
 """
 from __future__ import annotations
 
-import json
+import asyncio
-from pathlib import Path
+import base64
-from typing import Any
+import binascii
+import hashlib
+import os
+import re
+from datetime import date, datetime, timedelta, timezone
+from typing import Annotated, Any
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    LLMProvisioning,
+    AgentDatabase,
-    {{ auth_type }},
+    AgentDatabaseEnv,
+    AgentDatabaseMigrations,
+    AgentPlatformResources,
+    FileUpload,
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
-    WorkspaceAccess,
+    State,
-    WorkspaceMode,
+    UploadedFile,
+)
+
+
+MAX_CONTRACT_TEXT_CHARS = 60_000
+MAX_UPLOAD_BYTES = 512_000
+ALLOWED_BROWSER_MEDIA_TYPES = (
+    "text/plain",
+    "text/markdown",
+    "application/octet-stream",
+)
+DATE_RE = re.compile(r"\b(20\d{2}-\d{2}-\d{2})\b")
+RENEWAL_RE = re.compile(r"\brenew(?:s|al|ed|ing)?\b", re.IGNORECASE)
+NOTICE_RE = re.compile(
+    r"(?:at\s+least\s+)?(?P<days>\d{1,3})\s+days?\s+(?:written\s+)?notice",
+    re.IGNORECASE,
+)
+AMBIGUOUS_DATE_RE = re.compile(
+    r"\b(sometime|spring|summer|fall|autumn|winter|soon|later|well\s+in\s+advance|next\s+\w+)\b",
+    re.IGNORECASE,
 )
-from a2a_pack.context import LLMCreds
 
 
 class ContractClockStudioV1Config(BaseModel):
     pass
 
 
-SYSTEM_PROMPT = """\
+class BrowserDocument(BaseModel):
-You are a compact tool-calling agent.
+    filename: Annotated[str, Field(min_length=1, max_length=160)]
-
+    media_type: Annotated[str, Field(min_length=1, max_length=120)]
-Use the text_stats tool when the user asks about text, counts, summaries,
+    data_base64: Annotated[str, Field(min_length=1, max_length=700_000)]
-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 ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]):
+class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, PlatformUserAuth]):
     name = "contract-clock-studio-v1"
-    description = "ContractClock one-page full-stack startup for extracting explicit contract renewal and notice deadlines, persisting user-scoped results, and reopening timelines."
+    description = (
-    version = "0.1.0"
+        "ContractClock extracts explicit renewal and notice deadlines from "
+        "contracts, saves user-scoped timelines, and reopens them in a packed "
+        "one-page studio."
+    )
+    version = "0.1.1"
 
     config_model = ContractClockStudioV1Config
-    auth_model = {{ auth_type }}
+    auth_model = PlatformUserAuth
 
-    # Hosted generated agents read the caller's saved LLM credential through
+    state = State.DURABLE
-    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
+    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120)
-    # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
+    platform_resources = AgentPlatformResources(
-    # directly.
+        databases=(
-    llm_provisioning = LLMProvisioning.PLATFORM
+            AgentDatabase(
+                name="contract-clock-studio-v1-data",
+                scope="user",
+                access_mode="read_write",
+                env=AgentDatabaseEnv(url="DATABASE_URL"),
+                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            ),
+        )
+    )
     pricing = Pricing(
         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="Deterministic local extraction; no LLM credential required.",
     )
-    workspace_access = WorkspaceAccess.dynamic(
+    tools_used = ("postgres", "mcp", "packed-frontend")
-        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=(
-        creds = ctx.llm
+            "Analyze pasted contract text, extract explicit renewal and notice "
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+            "deadlines, atomically persist the contract timeline and execution receipt."
-        if not creds.api_key:
+        ),
-            return (
+        timeout_seconds=30,
-                "LLM key required. Add an LLM credential in Settings > LLM "
+        idempotent=True,
-                "credentials before running this agent; for local --invoke "
+        cost_class="cheap",
-                "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,
-        *,
+        ctx: RunContext[PlatformUserAuth],
-        ctx: RunContext[{{ auth_type }}],
+        contract_id: Annotated[str, Field(min_length=1, max_length=120)],
-        creds: LLMCreds,
+        text: Annotated[str, Field(min_length=1, max_length=MAX_CONTRACT_TEXT_CHARS)],
-    ) -> Any:
+        title: Annotated[str, Field(min_length=1, max_length=200)] = "Untitled contract",
-        # Lazy imports keep `a2a card` usable before local dependencies are
+    ) -> dict[str, Any]:
-        # installed. `a2a deploy` installs requirements.txt during the build.
+        tenant = _tenant_key(ctx)
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        if tenant is None:
-        from langchain.agents.middleware import wrap_model_call
+            return _auth_failure()
-        from langchain_core.tools import tool
+        analysis = _analyze_text(contract_id=contract_id, title=title, text=text)
+        if not analysis["ok"]:
+            return analysis
+        try:
+            await _save_analysis(
+                tenant_key=tenant,
+                contract_id=contract_id,
+                title=title,
+                text=text,
+                analysis=analysis,
+                skill_name="analyze_contract",
+                task_id=getattr(ctx, "task_id", ""),
+            )
+        except DatabaseUnavailable as exc:
+            return _db_failure(exc)
+        return analysis
 
-        @tool
+    @a2a.tool(
-        def text_stats(text: str) -> str:
+        description="Reload a previously analyzed contract deadline timeline from user-scoped storage.",
-            """Return exact word, character, and line counts for text."""
+        timeout_seconds=20,
-            words = [part for part in text.split() if part.strip()]
+        idempotent=True,
-            return json.dumps(
+        cost_class="cheap",
-                {
+    )
-                    "characters": len(text),
+    async def get_contract_deadlines(
-                    "words": len(words),
+        self,
-                    "lines": len(text.splitlines()) or 1,
+        ctx: RunContext[PlatformUserAuth],
+        contract_id: Annotated[str, Field(min_length=1, max_length=120)],
+    ) -> dict[str, Any]:
+        tenant = _tenant_key(ctx)
+        if tenant is None:
+            return _auth_failure()
+        try:
+            record = await _load_contract(tenant_key=tenant, contract_id=contract_id)
+        except DatabaseUnavailable as exc:
+            return _db_failure(exc)
+        if record is None:
+            return {"ok": False, "code": "not_found", "contract_id": contract_id}
+        return {
+            "ok": True,
+            "contract_id": contract_id,
+            "title": record.get("title") or "Untitled contract",
+            "deadlines": record.get("deadlines") or [],
+            "receipt": record.get("receipt") or {},
+            "updated_at": record.get("updated_at"),
         }
-            )
 
-        @wrap_model_call
+    @a2a.tool(
-        async def log_model_call(request: Any, handler: Any) -> Any:
+        description=(
-            messages = request.state.get("messages", [])
+            "Analyze a text contract uploaded by an external Agent API client. "
-            print(
+            "The FileUpload schema is retained for typed multipart clients."
-                "[middleware] model_call "
+        ),
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+        timeout_seconds=30,
+        idempotent=True,
+        cost_class="cheap",
     )
-            return await handler(request)
+    async def analyze_contract_file(
+        self,
+        ctx: RunContext[PlatformUserAuth],
+        contract_id: Annotated[str, Field(min_length=1, max_length=120)],
+        document: Annotated[
+            UploadedFile,
+            FileUpload(
+                accept=ALLOWED_BROWSER_MEDIA_TYPES,
+                max_bytes=MAX_UPLOAD_BYTES,
+                description="UTF-8 text contract document",
+            ),
+        ],
+        title: Annotated[str, Field(min_length=1, max_length=200)] = "Uploaded contract",
+    ) -> dict[str, Any]:
+        try:
+            payload = _read_uploaded_file(ctx, document)
+        except UploadRejected as exc:
+            return exc.result
+        return await self.analyze_contract(ctx, contract_id=contract_id, text=payload, title=title)
 
-        backend = ctx.workspace_backend()
+    @a2a.tool(
-        skill_sources = _seed_runtime_skills(backend, ctx)
+        description=(
-        # create_a2a_deep_agent resolves provider:model strings with
+            "Browser-safe upload bridge accepting one bounded base64 JSON text "
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
+            "document from the packed frontend."
-        # provider-specific extra body, and runtime model overrides.
+        ),
-        return create_a2a_deep_agent(
+        timeout_seconds=30,
-            ctx,
+        idempotent=True,
-            creds=creds,
+        cost_class="cheap",
-            backend=backend,
+    )
-            skills=skill_sources or None,
+    async def analyze_contract_upload(
-            tools=[text_stats],
+        self,
-            middleware=[log_model_call],
+        ctx: RunContext[PlatformUserAuth],
-            system_prompt=SYSTEM_PROMPT,
+        contract_id: Annotated[str, Field(min_length=1, max_length=120)],
+        document: BrowserDocument,
+        title: Annotated[str, Field(min_length=1, max_length=200)] = "Uploaded contract",
+    ) -> dict[str, Any]:
+        try:
+            payload = _decode_browser_document(document)
+        except UploadRejected as exc:
+            return exc.result
+        return await self.analyze_contract(ctx, contract_id=contract_id, text=payload, title=title)
+
+
+class DatabaseUnavailable(RuntimeError):
+    def __init__(self, code: str, message: str) -> None:
+        self.code = code
+        super().__init__(message)
+
+
+class UploadRejected(ValueError):
+    def __init__(self, code: str, message: str) -> None:
+        self.result = {"ok": False, "code": code, "message": message}
+        super().__init__(message)
+
+
+def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str | None:
+    auth = getattr(ctx, "auth", None)
+    user_id = getattr(auth, "user_id", None)
+    sub = (getattr(auth, "sub", None) or "").strip()
+    email = (getattr(auth, "email", None) or "").strip().lower()
+    if user_id is not None:
+        return f"user:{user_id}"
+    if sub:
+        return f"user:{sub}"
+    if email:
+        return f"user:{email}"
+    return None
+
+
+def _auth_failure() -> dict[str, Any]:
+    return {
+        "ok": False,
+        "code": "auth_required",
+        "message": "A signed-in A2A Cloud user session is required to use ContractClock.",
+    }
+
+
+def _db_failure(exc: DatabaseUnavailable) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "code": exc.code,
+        "message": "ContractClock storage is not ready. Retry after the managed database is available.",
+    }
+
+
+def _analyze_text(*, contract_id: str, title: str, text: str) -> dict[str, Any]:
+    clean_text = (text or "").strip()
+    if not clean_text:
+        return {"ok": False, "code": "empty_contract", "contract_id": contract_id}
+    if len(clean_text) > MAX_CONTRACT_TEXT_CHARS:
+        return {"ok": False, "code": "text_too_large", "contract_id": contract_id}
+
+    explicit_dates = []
+    for match in DATE_RE.finditer(clean_text):
+        try:
+            explicit_dates.append(date.fromisoformat(match.group(1)))
+        except ValueError:
+            return {"ok": False, "code": "invalid_date", "contract_id": contract_id}
+
+    if not explicit_dates:
+        code = "ambiguous_date" if AMBIGUOUS_DATE_RE.search(clean_text) else "missing_explicit_date"
+        return {"ok": False, "code": code, "contract_id": contract_id}
+
+    deadlines: list[dict[str, str]] = []
+    if RENEWAL_RE.search(clean_text):
+        renewal_date = explicit_dates[0]
+        deadlines.append({"kind": "renewal", "date": renewal_date.isoformat()})
+        notice_match = NOTICE_RE.search(clean_text)
+        if notice_match:
+            notice_days = int(notice_match.group("days"))
+            if 1 <= notice_days <= 366:
+                deadlines.append(
+                    {"kind": "notice", "date": (renewal_date - timedelta(days=notice_days)).isoformat()}
+                )
+    else:
+        # Explicit dates outside a renewal clause are kept as dated obligations;
+        # relative invoice terms without a concrete anchor are intentionally not guessed.
+        for item in explicit_dates:
+            deadlines.append({"kind": "obligation", "date": item.isoformat()})
+
+    if not deadlines:
+        return {"ok": False, "code": "no_supported_deadline", "contract_id": contract_id}
+
+    return {
+        "ok": True,
+        "contract_id": contract_id,
+        "title": title,
+        "deadlines": deadlines,
+        "rationale": (
+            "Only ISO-formatted explicit dates were extracted. Notice deadlines "
+            "are calculated from explicit renewal dates and written-notice day counts."
+        ),
+    }
+
+
+def _decode_browser_document(document: BrowserDocument) -> str:
+    filename = document.filename.strip()
+    if "/" in filename or "\\" in filename or filename in {".", ".."}:
+        raise UploadRejected("invalid_filename", "Upload filename must be a simple file name.")
+    media_type = document.media_type.split(";", 1)[0].strip().lower()
+    if media_type not in ALLOWED_BROWSER_MEDIA_TYPES:
+        raise UploadRejected("unsupported_media_type", "Upload a UTF-8 text contract file.")
+    try:
+        raw = base64.b64decode(document.data_base64, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise UploadRejected("invalid_base64", "Upload data must be valid base64.") from exc
+    return _decode_uploaded_bytes(raw)
+
+
+def _read_uploaded_file(ctx: RunContext[PlatformUserAuth], document: UploadedFile) -> str:
+    media_type = document.media_type.split(";", 1)[0].strip().lower()
+    if media_type not in ALLOWED_BROWSER_MEDIA_TYPES:
+        raise UploadRejected("unsupported_media_type", "Upload a UTF-8 text contract file.")
+    if document.size_bytes > MAX_UPLOAD_BYTES:
+        raise UploadRejected("upload_too_large", "Contract uploads are limited to 512 KB.")
+    try:
+        raw = ctx.workspace.read_bytes(document.path)
+    except Exception as exc:  # noqa: BLE001
+        raise UploadRejected("upload_read_failed", "Could not read the uploaded contract file.") from exc
+    return _decode_uploaded_bytes(raw)
+
+
+def _decode_uploaded_bytes(raw: bytes) -> str:
+    if not raw:
+        raise UploadRejected("empty_upload", "Uploaded contract file is empty.")
+    if len(raw) > MAX_UPLOAD_BYTES:
+        raise UploadRejected("upload_too_large", "Contract uploads are limited to 512 KB.")
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise UploadRejected("invalid_text_encoding", "Uploaded contract must be UTF-8 text.") from exc
+    if not text.strip():
+        raise UploadRejected("empty_upload", "Uploaded contract file is empty.")
+    return text
+
+
+async def _save_analysis(
+    *,
+    tenant_key: str,
+    contract_id: str,
+    title: str,
+    text: str,
+    analysis: dict[str, Any],
+    skill_name: str,
+    task_id: str,
+) -> None:
+    await asyncio.to_thread(
+        _save_analysis_sync,
+        tenant_key=tenant_key,
+        contract_id=contract_id,
+        title=title,
+        text=text,
+        analysis=analysis,
+        skill_name=skill_name,
+        task_id=task_id,
     )
 
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+async def _load_contract(*, tenant_key: str, contract_id: str) -> dict[str, Any] | None:
-    workspace = getattr(ctx, "_workspace", None)
+    return await asyncio.to_thread(_load_contract_sync, tenant_key=tenant_key, contract_id=contract_id)
-    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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
+def _connect_db() -> Any:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    dsn = os.environ.get("DATABASE_URL", "").strip()
+    if not dsn:
+        raise DatabaseUnavailable("database_unavailable", "DATABASE_URL is not configured")
+    try:
+        import psycopg
+    except Exception as exc:  # noqa: BLE001
+        raise DatabaseUnavailable("database_driver_missing", "psycopg is not installed") from exc
+    try:
+        return psycopg.connect(
+            dsn,
+            connect_timeout=5,
+            options="-c statement_timeout=5000 -c idle_in_transaction_session_timeout=5000",
+        )
+    except Exception as exc:  # noqa: BLE001
+        raise DatabaseUnavailable("database_unavailable", "could not connect to managed Postgres") from exc
 
-    DeepAgents loads skills from its backend, while source-controlled
+
-    ``skills/`` folders live in the image. This bridge lets generated agents
+def _save_analysis_sync(
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+    *,
-    files.
+    tenant_key: str,
+    contract_id: str,
+    title: str,
+    text: str,
+    analysis: dict[str, Any],
+    skill_name: str,
+    task_id: str,
+) -> None:
+    try:
+        from psycopg.types.json import Jsonb
+    except Exception as exc:  # noqa: BLE001
+        raise DatabaseUnavailable("database_driver_missing", "psycopg JSON support is not installed") from exc
+
+    now = datetime.now(timezone.utc).isoformat()
+    input_hash = hashlib.sha256(
+        f"{tenant_key}\0{contract_id}\0{text}".encode("utf-8")
+    ).hexdigest()
+    receipt = {
+        "agent": ContractClockStudioV1.name,
+        "version": ContractClockStudioV1.version,
+        "skill": skill_name,
+        "task_id": task_id,
+        "input_hash": input_hash,
+        "status": "ok",
+        "created_at": now,
+    }
+    analysis_with_receipt = {**analysis, "receipt": receipt}
+
+    try:
+        with _connect_db() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
                     """
-    root = Path(__file__).parent / "skills"
+                    INSERT INTO contract_clock_contracts
-    if not root.exists():
+                        (tenant_key, contract_id, title, source_text_sha256, deadlines, analysis, updated_at)
-        return []
+                    VALUES (%s, %s, %s, %s, %s, %s, NOW())
-    runtime_skills_root = _runtime_skills_root(ctx)
+                    ON CONFLICT (tenant_key, contract_id) DO UPDATE SET
-    uploads: list[tuple[str, bytes]] = []
+                        title = EXCLUDED.title,
-    for path in root.rglob("*"):
+                        source_text_sha256 = EXCLUDED.source_text_sha256,
-        if path.is_file():
+                        deadlines = EXCLUDED.deadlines,
-            rel = path.relative_to(root).as_posix()
+                        analysis = EXCLUDED.analysis,
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+                        updated_at = NOW()
-    if uploads:
+                    """,
-        backend.upload_files(uploads)
+                    (
-        return [runtime_skills_root]
+                        tenant_key,
-    return []
+                        contract_id,
+                        title,
+                        hashlib.sha256(text.encode("utf-8")).hexdigest(),
+                        Jsonb(analysis["deadlines"]),
+                        Jsonb(analysis_with_receipt),
+                    ),
+                )
+                cur.execute(
+                    """
+                    INSERT INTO contract_clock_execution_receipts
+                        (tenant_key, contract_id, skill_name, task_id, input_hash, status, result_summary)
+                    VALUES (%s, %s, %s, %s, %s, %s, %s)
+                    """,
+                    (
+                        tenant_key,
+                        contract_id,
+                        skill_name,
+                        task_id,
+                        input_hash,
+                        "ok",
+                        Jsonb(
+                            {
+                                "deadline_count": len(analysis.get("deadlines") or []),
+                                "contract_id": contract_id,
+                            }
+                        ),
+                    ),
+                )
+    except DatabaseUnavailable:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        raise DatabaseUnavailable("database_write_failed", "could not persist contract analysis") from exc
 
 
-def _last_message_text(state: dict[str, Any]) -> str:
+def _load_contract_sync(*, tenant_key: str, contract_id: str) -> dict[str, Any] | None:
-    messages = state.get("messages") or []
+    try:
-    if not messages:
+        with _connect_db() as conn:
-        return json.dumps(state, default=str)
+            with conn.cursor() as cur:
-
+                cur.execute(
-    content = getattr(messages[-1], "content", None)
+                    """
-    if isinstance(content, str):
+                    SELECT title, deadlines, analysis, updated_at
-        return content
+                    FROM contract_clock_contracts
-    if isinstance(content, list):
+                    WHERE tenant_key = %s AND contract_id = %s
-        parts: list[str] = []
+                    """,
-        for item in content:
+                    (tenant_key, contract_id),
-            if isinstance(item, dict):
+                )
-                text = item.get("text") or item.get("content")
+                row = cur.fetchone()
-                if text:
+    except DatabaseUnavailable:
-                    parts.append(str(text))
+        raise
-            elif item:
+    except Exception as exc:  # noqa: BLE001
-                parts.append(str(item))
+        raise DatabaseUnavailable("database_read_failed", "could not load contract timeline") from exc
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+    if row is None:
-    return str(content or messages[-1])
+        return None
+    title, deadlines, analysis, updated_at = row
+    analysis = analysis or {}
+    return {
+        "title": title,
+        "deadlines": deadlines or [],
+        "receipt": analysis.get("receipt") or {},
+        "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
+    }