From 3fe5efbffee0363a07c61a3695adfe5c3fcf9dba Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 08:10:33 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 777 ++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 633 insertions(+), 144 deletions(-) diff --git a/agent.py b/agent.py index 7f9f977..e3e044c 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,678 @@ -"""contract-clock-studio-v1 agent. +"""ContractClock Studio v1 full-stack A2A agent. -Starter stack: - - DeepAgents for tool-calling orchestration - - Caller-provided LLM credentials via ctx.llm - - A tiny model-call middleware hook you can replace with tracing, - routing, rate limits, or policy checks +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 -from pathlib import Path -from typing import Any +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 +from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + PlatformUserAuth, Pricing, + Resources, RunContext, - WorkspaceAccess, - WorkspaceMode, + State, + UploadedFile, + FileUpload, ) -from a2a_pack.context import LLMCreds + + +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 + +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 -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. - -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 DeadlineOut(BaseModel): + kind: Literal["renewal", "notice"] + date: str + source_text: str | None = None + rationale: str | None = None -class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]): +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: upload or paste contracts, extract explicit renewal and notice deadlines, persist user-scoped timelines, and reopen them in a packed React app." - version = "0.1.0" + 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.5" config_model = ContractClockStudioV1Config - auth_model = {{ auth_type }} + auth_model = PlatformUserAuth - # 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 + # Deterministic local logic: no LLM credential is read or required. pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + caller_pays_llm=False, + notes="Deterministic deadline extraction; no LLM usage.", ) - workspace_access = WorkspaceAccess.dynamic( - 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") - async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: - creds = ctx.llm - await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") - if not creds.api_key: - return ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_KEY." - ) - graph = self._build_deep_agent(ctx=ctx, creds=creds) - state = await graph.ainvoke( - {"messages": [{"role": "user", "content": prompt}]}, - config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) + state = State.DURABLE + 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"), + ), ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + ) - def _build_deep_agent( + @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[{{ auth_type }}], - creds: LLMCreds, - ) -> Any: - # Lazy imports keep `a2a card` usable before local dependencies are - # installed. `a2a deploy` installs requirements.txt during the build. - from a2a_pack.deepagents import create_a2a_deep_agent - from langchain.agents.middleware import wrap_model_call - from langchain_core.tools import tool + 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, + ) - @tool - def text_stats(text: str) -> str: - """Return exact word, character, and line counts for text.""" - words = [part for part in text.split() if part.strip()] - return json.dumps( - { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, - } + @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 getattr(auth, "email", 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, + ) - @wrap_model_call - async def log_model_call(request: Any, handler: Any) -> Any: - messages = request.state.get("messages", []) - print( - "[middleware] model_call " - f"model={creds.model} source={creds.source} messages={len(messages)}" + 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 _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 + dates = _explicit_iso_dates(clause) + if not dates: + continue + renewal_date = dates[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.", + ) ) - return await handler(request) + 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)) + deadlines.sort(key=lambda item: (item.deadline_date, item.kind != "notice")) + # Public contract wants renewal before derived notice for the same clause. + deadlines.sort(key=lambda item: 0 if item.kind == "renewal" else 1) + return deadlines - backend = ctx.workspace_backend() - skill_sources = _seed_runtime_skills(backend, ctx) - # create_a2a_deep_agent resolves provider:model strings with - # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing, - # provider-specific extra body, and runtime model overrides. - return create_a2a_deep_agent( - ctx, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + +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: + return re.sub(r"\s+", " ", text).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]: + out: list[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(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, ) -def _runtime_skills_root(ctx: RunContext[Any]) -> str: - workspace = getattr(ctx, "_workspace", None) - prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) - if not prefixes: - outputs_prefix = getattr(workspace, "outputs_prefix", None) - prefixes = (outputs_prefix or "outputs/",) - prefix = str(prefixes[0]).strip("/") - return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" +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, "title": title, "text_sha256": _sha256(text)} + result_payload = result.model_dump(mode="json") + with psycopg.connect( + os.environ["DATABASE_URL"], + connect_timeout=3, + options=f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} -c lock_timeout={DB_LOCK_TIMEOUT_MS}", + row_factory=dict_row, + ) as conn: + with conn.transaction(): + with conn.cursor() as cur: + cur.execute("SET LOCAL statement_timeout = %s", (DB_STATEMENT_TIMEOUT_MS,)) + cur.execute("SET LOCAL lock_timeout = %s", (DB_LOCK_TIMEOUT_MS,)) + 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": result_payload}, 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__}).", + } + ) -def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """Copy packaged DeepAgents skills into the invocation workspace. +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 - DeepAgents loads skills from its backend, while source-controlled - ``skills/`` folders live in the image. This bridge lets generated agents - ship reusable SKILL.md bundles without giving up durable A2A workspace - files. - """ - root = Path(__file__).parent / "skills" - if not root.exists(): - return [] - runtime_skills_root = _runtime_skills_root(ctx) - uploads: list[tuple[str, bytes]] = [] - for path in root.rglob("*"): - if path.is_file(): - rel = path.relative_to(root).as_posix() - uploads.append((runtime_skills_root + rel, path.read_bytes())) - if uploads: - backend.upload_files(uploads) - return [runtime_skills_root] - return [] + with psycopg.connect( + os.environ["DATABASE_URL"], + connect_timeout=3, + options=f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} -c lock_timeout={DB_LOCK_TIMEOUT_MS}", + ) as conn: + with conn.transaction(): + with conn.cursor() as cur: + cur.execute("SET LOCAL statement_timeout = %s", (DB_STATEMENT_TIMEOUT_MS,)) + 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": result.model_dump(mode="json")}, sort_keys=True), + ), + ) + except Exception: + return -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +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 - content = getattr(messages[-1], "content", None) - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, dict): - text = item.get("text") or item.get("content") - if text: - parts.append(str(text)) - elif item: - parts.append(str(item)) - return "\n".join(parts) if parts else json.dumps(content, default=str) - return str(content or messages[-1]) + with psycopg.connect( + os.environ["DATABASE_URL"], + connect_timeout=3, + options=f"-c statement_timeout={DB_STATEMENT_TIMEOUT_MS} -c lock_timeout={DB_LOCK_TIMEOUT_MS}", + row_factory=dict_row, + ) as conn: + with conn.cursor() as cur: + cur.execute("SET statement_timeout = %s", (DB_STATEMENT_TIMEOUT_MS,)) + 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 _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))