a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
777
agent.py
777
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))
|
||||
|
||||
Reference in New Issue
Block a user