a2a-source-edit: write agent.py

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
+Starter stack:
-renewals and notice windows from contracts, persists user-scoped timelines in
+  - DeepAgents for tool-calling orchestration
-managed Postgres, and never guesses ambiguous dates.
+  - 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
+from pathlib import Path
-import re
+from typing import Any
-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, Field
+from pydantic import BaseModel
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    AgentDatabase,
+    LLMProvisioning,
-    AgentDatabaseEnv,
+    {{ auth_type }},
-    AgentDatabaseMigrations,
-    AgentPlatformResources,
-    FileUpload,
-    PlatformUserAuth,
     Pricing,
-    Resources,
     RunContext,
-    State,
-    UploadedFile,
     WorkspaceAccess,
     WorkspaceMode,
 )
-
+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
-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<days>\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}
 
 
 class ContractClockStudioV1Config(BaseModel):
     pass
 
 
-class ContractClockState(BaseModel):
+SYSTEM_PROMPT = """\
-    """Durable-state marker; records live in managed Postgres tables."""
+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):
+class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]):
-    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]):
     name = "contract-clock-studio-v1"
-    description = (
+    description = "ContractClock one-page full-stack startup for extracting explicit contract renewal and notice deadlines, persisting user-scoped results, and reopening timelines."
-        "Private ContractClock app: paste or upload a contract, deterministically "
+    version = "0.1.0"
-        "extract explicit renewal and notice dates, persist the timeline in "
-        "user-scoped managed Postgres, and reopen it later."
-    )
-    version = VERSION
 
     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,
+        caller_pays_llm=True,
-        notes="No LLM calls. Deterministic extraction and managed Postgres persistence only.",
+        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,
+        max_files=64,
-        allowed_modes=(WorkspaceMode.READ_ONLY,),
+        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
         require_reason=False,
-        max_total_size_bytes=MAX_UPLOAD_BYTES,
     )
-    platform_resources = AgentPlatformResources(
+    tools_used = ("deepagents", "langchain")
-        databases=(
+
-            AgentDatabase(
+    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
-                name="contract-clock-studio-v1-data",
+    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
-                scope="user",
+        creds = ctx.llm
-                access_mode="read_write",
+        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
-                env=AgentDatabaseEnv(url=DATABASE_ENV),
+        if not creds.api_key:
-                migrations=AgentDatabaseMigrations(path="db/migrations"),
+            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(
+    def _build_deep_agent(
-        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[PlatformUserAuth],
+        *,
-        contract_id: str,
+        ctx: RunContext[{{ auth_type }}],
-        text: str,
+        creds: LLMCreds,
-        title: str,
+    ) -> Any:
-    ) -> AnalyzeResult:
+        # Lazy imports keep `a2a card` usable before local dependencies are
-        try:
+        # installed. `a2a deploy` installs requirements.txt during the build.
-            tenant = tenant_key(ctx)
+        from a2a_pack.deepagents import create_a2a_deep_agent
-            clean_id = validate_contract_id(contract_id)
+        from langchain.agents.middleware import wrap_model_call
-            clean_title = validate_title(title)
+        from langchain_core.tools import tool
-            clean_text = validate_text(text)
+
-            timeline = extract_timeline(clean_text)
+        @tool
-        except ValidationFailure as exc:
+        def text_stats(text: str) -> str:
-            return AnalyzeResult(
+            """Return exact word, character, and line counts for text."""
-                ok=False,
+            words = [part for part in text.split() if part.strip()]
-                contract_id=contract_id,
+            return json.dumps(
-                title=title,
+                {
-                code=exc.code,
+                    "characters": len(text),
-                message=exc.message,
+                    "words": len(words),
+                    "lines": len(text.splitlines()) or 1,
+                }
             )
 
-        payload = {
+        @wrap_model_call
-            "contract_id": clean_id,
+        async def log_model_call(request: Any, handler: Any) -> Any:
-            "title": clean_title,
+            messages = request.state.get("messages", [])
-            "text_sha256": sha256_text(clean_text),
+            print(
-            "deadlines": [deadline.model_dump(mode="json") for deadline in timeline.deadlines],
+                "[middleware] model_call "
-            "warnings": timeline.warnings,
+                f"model={creds.model} source={creds.source} messages={len(messages)}"
-        }
-        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,
             )
+            return await handler(request)
 
-        await ctx.emit_progress(f"Saved {len(timeline.deadlines)} deadlines for {clean_id}.")
+        backend = ctx.workspace_backend()
-        return AnalyzeResult(
+        skill_sources = _seed_runtime_skills(backend, ctx)
-            ok=True,
+        # create_a2a_deep_agent resolves provider:model strings with
-            contract_id=clean_id,
+        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
-            title=clean_title,
+        # provider-specific extra body, and runtime model overrides.
-            deadlines=timeline.deadlines,
+        return create_a2a_deep_agent(
-            warnings=timeline.warnings,
+            ctx,
-            receipt_id=receipt_id,
+            creds=creds,
-            persisted=True,
+            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(
+def _runtime_skills_root(ctx: RunContext[Any]) -> str:
-        description="Analyze a typed external FileUpload text contract and persist its timeline.",
+    workspace = getattr(ctx, "_workspace", None)
-        timeout_seconds=30,
+    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
-        idempotent=True,
+    if not prefixes:
-        cost_class="deterministic",
+        outputs_prefix = getattr(workspace, "outputs_prefix", None)
-    )
+        prefixes = (outputs_prefix or "outputs/",)
-    async def analyze_contract_file_upload(
+    prefix = str(prefixes[0]).strip("/")
-        self,
+    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
-        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 tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
+def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
-    auth = ctx.auth
+    """Copy packaged DeepAgents skills into the invocation workspace.
-    principal = auth.user_id if auth.user_id is not None else (auth.sub or auth.email)
+
-    if not principal:
+    DeepAgents loads skills from its backend, while source-controlled
-        raise ValidationFailure("auth_required", "A signed-in platform user is required.")
+    ``skills/`` folders live in the image. This bridge lets generated agents
-    return f"user:{principal}"
+    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:
+def _last_message_text(state: dict[str, Any]) -> str:
-    value = (contract_id or "").strip()
+    messages = state.get("messages") or []
-    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", value):
+    if not messages:
-        raise ValidationFailure("invalid_contract_id", "contract_id must be 1-128 safe identifier characters.")
+        return json.dumps(state, default=str)
-    return value
 
-
+    content = getattr(messages[-1], "content", None)
-def validate_title(title: str) -> str:
+    if isinstance(content, str):
-    value = (title or "").strip()
+        return content
-    if not value or len(value) > 180:
+    if isinstance(content, list):
-        raise ValidationFailure("invalid_title", "title is required and must be at most 180 characters.")
+        parts: list[str] = []
-    return value
+        for item in content:
-
+            if isinstance(item, dict):
-
+                text = item.get("text") or item.get("content")
-def validate_text(text: str) -> str:
+                if text:
-    value = text or ""
+                    parts.append(str(text))
-    if not value.strip():
+            elif item:
-        raise ValidationFailure("empty_contract", "Contract text is required.")
+                parts.append(str(item))
-    if len(value) > MAX_TEXT_CHARS:
+        return "\n".join(parts) if parts else json.dumps(content, default=str)
-        raise ValidationFailure("text_too_large", "Contract text exceeds the 200,000 character limit.")
+    return str(content or messages[-1])
-    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
-    ]