From b098a66c67b36774c4bd5f4db5ccd11fd1168c7d Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 15:50:50 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 872 ++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 733 insertions(+), 139 deletions(-) diff --git a/agent.py b/agent.py index f62d0cc..bd678cb 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,783 @@ -"""type-data-csv-excel-7rh4pz-057e66 agent. +"""Data workflow builder 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 +Creates reusable extraction/transformation workflow specs from uploaded CSV, +Excel, JSON, JSON-LD, or other structured data and a caller-provided JSON +Schema. The core workflow builder is deterministic so smoke/MCP calls do not +require an LLM credential. The generated workflow is saved as an artifact and, +when the managed database is available, recorded with an execution receipt for +platform acceptance evidence. """ from __future__ import annotations +import base64 +import csv +import hashlib +import io import json +import os +import re +import uuid +from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Annotated, Any, Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field, ValidationError, field_validator import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + FileUpload, + PlatformUserAuth, Pricing, + Resources, RunContext, + UploadedFile, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds + +try: # Optional at card-load time; required only when DATABASE_URL exists. + import psycopg +except Exception: # pragma: no cover - sandbox card load before optional deps + psycopg = None # type: ignore[assignment] + + +MAX_UPLOAD_BYTES = 2 * 1024 * 1024 +MAX_SCHEMA_BYTES = 64 * 1024 +MAX_SAMPLE_ROWS = 25 +WORKFLOW_OUTPUT_PREFIX = "outputs/data-workflows/" +DB_NAME = "type-data-csv-excel-7rh4pz-057e66-data" class TypeDataCsvExcel7rh4pz057e66Config(BaseModel): - pass + max_upload_bytes: int = MAX_UPLOAD_BYTES -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +class BrowserDataUpload(BaseModel): + filename: str = Field(..., min_length=1, max_length=160) + media_type: str = Field(..., min_length=1, max_length=120) + data_base64: str = Field(..., min_length=1, max_length=MAX_UPLOAD_BYTES * 2) -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 = "type-data-csv-excel-7rh4pz-057e66/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 + @field_validator("filename") + @classmethod + def safe_filename(cls, value: str) -> str: + clean = value.strip().replace("\\", "/").split("/")[-1] + if not clean or clean in {".", ".."}: + raise ValueError("filename must be a safe basename") + if not re.fullmatch(r"[A-Za-z0-9._ ()@+-]{1,160}", clean): + raise ValueError("filename contains unsupported characters") + return clean -class TypeDataCsvExcel7rh4pz057e66(A2AAgent[TypeDataCsvExcel7rh4pz057e66Config, {{ auth_type }}]): +class WorkflowResult(BaseModel): + status: Literal["ok", "validation_error", "storage_warning"] + workflow_id: str + workflow_name: str + source_format: str + fields_detected: list[str] + mapped_fields: list[dict[str, Any]] + unmapped_required_fields: list[str] + workflow_path: str + artifact_uri: str | None = None + receipt_id: str + persisted: bool + warnings: list[str] = Field(default_factory=list) + + +class TypeDataCsvExcel7rh4pz057e66(A2AAgent[TypeDataCsvExcel7rh4pz057e66Config, PlatformUserAuth]): name = "type-data-csv-excel-7rh4pz-057e66" - description = "Creates reusable agent-powered extraction and transformation workflows from uploaded CSV, Excel, JSON, JSON-LD, and other tabular or structured data using a caller-provided JSON Schema." + description = ( + "Builds reusable extraction and transformation workflows from CSV, " + "Excel, JSON, JSON-LD, and other structured data into a requested JSON Schema." + ) version = "0.1.0" config_model = TypeDataCsvExcel7rh4pz057e66Config - 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 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 workflow generation; no LLM credential is required.", ) + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600) workspace_access = WorkspaceAccess.dynamic( - max_files=64, + max_files=32, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, + max_total_size_bytes=10 * 1024 * 1024, ) - 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}, + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name=DB_NAME, + 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) + ) + tools_used = ("jsonschema", "csv", "openpyxl") - def _build_deep_agent( + @a2a.tool( + description=( + "Create and save a reusable extraction/transformation workflow from " + "a browser-uploaded base64 data file and a target JSON Schema." + ), + timeout_seconds=120, + cost_class="standard", + grant_mode="read_write_overlay", + grant_allow_patterns=("outputs/data-workflows/**",), + grant_outputs_prefix=WORKFLOW_OUTPUT_PREFIX, + grant_write_prefixes=(WORKFLOW_OUTPUT_PREFIX,), + ) + async def create_workflow_from_browser_upload( + self, + ctx: RunContext[PlatformUserAuth], + source_file: BrowserDataUpload, + output_schema: dict[str, Any], + workflow_name: str = "data-transform-workflow", + ) -> WorkflowResult: + """Browser-safe bounded base64 upload bridge for the packed frontend.""" + try: + data = _decode_browser_upload(source_file, max_bytes=self.config.max_upload_bytes) + except ValueError as exc: + return WorkflowResult( + status="validation_error", + workflow_id="", + workflow_name=workflow_name, + source_format="unknown", + fields_detected=[], + mapped_fields=[], + unmapped_required_fields=[], + workflow_path="", + receipt_id="", + persisted=False, + warnings=[str(exc)], + ) + return await self._create_workflow( + ctx=ctx, + filename=source_file.filename, + media_type=source_file.media_type, + data=data, + output_schema=output_schema, + workflow_name=workflow_name, + upload_mode="browser_base64", + ) + + @a2a.tool( + description=( + "Create and save a reusable extraction/transformation workflow from " + "an Agent API FileUpload and a target JSON Schema." + ), + timeout_seconds=120, + cost_class="standard", + grant_mode="read_write_overlay", + grant_allow_patterns=("outputs/data-workflows/**",), + grant_outputs_prefix=WORKFLOW_OUTPUT_PREFIX, + grant_write_prefixes=(WORKFLOW_OUTPUT_PREFIX,), + ) + async def create_workflow_from_file_upload( + self, + ctx: RunContext[PlatformUserAuth], + source_file: Annotated[ + UploadedFile, + FileUpload( + accept=( + "text/csv", + "application/csv", + "application/json", + "application/ld+json", + "application/vnd.ms-excel", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "text/plain", + ), + max_bytes=MAX_UPLOAD_BYTES, + description="CSV, Excel, JSON, JSON-LD, or text data file", + ), + ], + output_schema: dict[str, Any], + workflow_name: str = "data-transform-workflow", + ) -> WorkflowResult: + """Typed FileUpload path for external Agent API/MCP-compatible clients.""" + if source_file.size_bytes and source_file.size_bytes > self.config.max_upload_bytes: + return WorkflowResult( + status="validation_error", + workflow_id="", + workflow_name=workflow_name, + source_format="unknown", + fields_detected=[], + mapped_fields=[], + unmapped_required_fields=[], + workflow_path="", + receipt_id="", + persisted=False, + warnings=[f"file exceeds {self.config.max_upload_bytes} byte limit"], + ) + data = _read_uploaded_workspace_file(ctx, source_file.path) + if len(data) > self.config.max_upload_bytes: + return WorkflowResult( + status="validation_error", + workflow_id="", + workflow_name=workflow_name, + source_format="unknown", + fields_detected=[], + mapped_fields=[], + unmapped_required_fields=[], + workflow_path="", + receipt_id="", + persisted=False, + warnings=[f"file exceeds {self.config.max_upload_bytes} byte limit"], + ) + return await self._create_workflow( + ctx=ctx, + filename=source_file.filename, + media_type=source_file.media_type, + data=data, + output_schema=output_schema, + workflow_name=workflow_name, + upload_mode="agent_file_upload", + ) + + @a2a.tool( + description="List recent saved workflow receipts for the signed-in user.", + timeout_seconds=30, + cost_class="cheap", + idempotent=True, + ) + async def list_workflow_receipts( + self, + ctx: RunContext[PlatformUserAuth], + limit: int = 10, + ) -> dict[str, Any]: + safe_limit = max(1, min(int(limit), 50)) + rows = _list_receipts(_tenant_key(ctx), safe_limit) + return {"status": "ok", "receipts": rows, "persisted": bool(os.environ.get("DATABASE_URL"))} + + async def _create_workflow( 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 - - @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, - } + ctx: RunContext[PlatformUserAuth], + filename: str, + media_type: str, + data: bytes, + output_schema: dict[str, Any], + workflow_name: str, + upload_mode: str, + ) -> WorkflowResult: + tenant = _tenant_key(ctx) + warnings: list[str] = [] + schema_error = _validate_json_schema(output_schema) + if schema_error: + return WorkflowResult( + status="validation_error", + workflow_id="", + workflow_name=workflow_name, + source_format="unknown", + fields_detected=[], + mapped_fields=[], + unmapped_required_fields=[], + workflow_path="", + receipt_id="", + persisted=False, + warnings=[schema_error], + ) + if len(json.dumps(output_schema, default=str).encode("utf-8")) > MAX_SCHEMA_BYTES: + return WorkflowResult( + status="validation_error", + workflow_id="", + workflow_name=workflow_name, + source_format="unknown", + fields_detected=[], + mapped_fields=[], + unmapped_required_fields=[], + workflow_path="", + receipt_id="", + persisted=False, + warnings=[f"output_schema exceeds {MAX_SCHEMA_BYTES} bytes"], ) - @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)}" - ) - return await handler(request) + profile = _profile_source_data(filename, media_type, data) + warnings.extend(profile.get("warnings", [])) + workflow_id = _stable_workflow_id(filename, data, output_schema, workflow_name) + receipt_id = f"rcpt_{uuid.uuid4().hex}" + workflow_path = f"{WORKFLOW_OUTPUT_PREFIX}{workflow_id}.workflow.json" + workflow_spec = _build_workflow_spec( + workflow_id=workflow_id, + workflow_name=workflow_name, + filename=filename, + media_type=media_type, + upload_mode=upload_mode, + profile=profile, + output_schema=output_schema, + receipt_id=receipt_id, + ) - 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, + artifact_uri: str | None = None + ref = await ctx.write_artifact( + f"{workflow_id}.workflow.json", + json.dumps(workflow_spec, indent=2, sort_keys=True).encode("utf-8"), + "application/json", + ) + await ctx.emit_artifact(ref) + artifact_uri = ref.uri + + await _write_workspace_output(ctx, workflow_path, workflow_spec, warnings) + + persisted = _persist_workflow_and_receipt( + tenant_key=tenant, + workflow_id=workflow_id, + receipt_id=receipt_id, + workflow_name=workflow_name, + workflow_path=workflow_path, + source_format=str(profile["format"]), + source_filename=filename, + spec=workflow_spec, + result_summary={ + "mapped_count": len(workflow_spec["mapping"]["fields"]), + "unmapped_required_fields": workflow_spec["mapping"]["unmapped_required_fields"], + "artifact_uri": artifact_uri, + }, + warnings=warnings, + ) + if not persisted: + warnings.append("DATABASE_URL is not configured; workflow artifact was created but database receipt was not persisted") + + return WorkflowResult( + status="ok" if persisted else "storage_warning", + workflow_id=workflow_id, + workflow_name=workflow_name, + source_format=str(profile["format"]), + fields_detected=list(profile.get("fields", [])), + mapped_fields=list(workflow_spec["mapping"]["fields"]), + unmapped_required_fields=list(workflow_spec["mapping"]["unmapped_required_fields"]), + workflow_path=workflow_path, + artifact_uri=artifact_uri, + receipt_id=receipt_id, + persisted=persisted, + warnings=warnings, ) -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}" +def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + auth = ctx.auth + stable_id = auth.user_id if auth.user_id is not None else auth.sub + if stable_id is None or str(stable_id).strip() == "": + raise PermissionError("stable platform identity required") + return f"user:{stable_id}" -def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """Copy packaged DeepAgents skills into the invocation workspace. +def _decode_browser_upload(source_file: BrowserDataUpload, *, max_bytes: int) -> bytes: + if source_file.media_type not in _accepted_media_types(): + raise ValueError(f"unsupported media_type {source_file.media_type!r}") + try: + data = base64.b64decode(source_file.data_base64, validate=True) + except Exception as exc: # noqa: BLE001 + raise ValueError("data_base64 must be valid base64") from exc + if not data: + raise ValueError("uploaded file is empty") + if len(data) > max_bytes: + raise ValueError(f"uploaded file exceeds {max_bytes} byte limit") + return data - 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(): + +def _accepted_media_types() -> set[str]: + return { + "text/csv", + "application/csv", + "application/json", + "application/ld+json", + "application/vnd.ms-excel", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "text/plain", + "application/octet-stream", + } + + +def _read_uploaded_workspace_file(ctx: RunContext[PlatformUserAuth], path: str) -> bytes: + workspace = ctx.workspace + reader = getattr(workspace, "read_bytes", None) + if callable(reader): + return bytes(reader(path)) + raise RuntimeError("workspace file reader is unavailable for uploaded file") + + +async def _write_workspace_output(ctx: RunContext[PlatformUserAuth], path: str, payload: dict[str, Any], warnings: list[str]) -> None: + data = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8") + workspace = ctx.workspace + writer = getattr(workspace, "write_bytes", None) + if callable(writer): + writer(path, data) + return + try: + grant = await workspace.request_access( + files=(path,), + mode=WorkspaceMode.READ_WRITE_OVERLAY, + reason="Save generated data workflow spec", + purpose="workflow output", + ) + view = await workspace.open_view( + purpose="workflow output", + hints=(path,), + max_files=1, + mode=WorkspaceMode.READ_WRITE_OVERLAY, + reason="Save generated data workflow spec", + ) + await view.write(path, data) + _ = grant + except Exception as exc: # noqa: BLE001 + warnings.append(f"workspace write warning: {type(exc).__name__}: {str(exc)[:160]}") + + +def _validate_json_schema(schema: dict[str, Any]) -> str | None: + if not isinstance(schema, dict): + return "output_schema must be a JSON object" + if schema.get("type") != "object": + return "output_schema must describe a JSON object with type='object'" + props = schema.get("properties") + if not isinstance(props, dict) or not props: + return "output_schema.properties must be a non-empty object" + required = schema.get("required", []) + if required is not None and not isinstance(required, list): + return "output_schema.required must be a list when provided" + try: + json.dumps(schema) + except Exception as exc: # noqa: BLE001 + return f"output_schema must be JSON serializable: {exc}" + return None + + +def _profile_source_data(filename: str, media_type: str, data: bytes) -> dict[str, Any]: + ext = Path(filename.lower()).suffix + text = data.decode("utf-8-sig", errors="replace") + warnings: list[str] = [] + if media_type in {"application/json", "application/ld+json"} or ext in {".json", ".jsonld"}: + try: + parsed = json.loads(text) + rows, fields = _json_rows_and_fields(parsed) + return { + "format": "jsonld" if media_type == "application/ld+json" or ext == ".jsonld" else "json", + "fields": fields, + "sample_rows": rows[:MAX_SAMPLE_ROWS], + "row_count_sampled": min(len(rows), MAX_SAMPLE_ROWS), + "warnings": warnings, + } + except json.JSONDecodeError as exc: + warnings.append(f"JSON parse warning: {exc.msg}") + if ext in {".xlsx", ".xlsm", ".xltx", ".xltm"} or "spreadsheetml" in media_type: + try: + return _profile_excel(data, warnings) + except Exception as exc: # noqa: BLE001 + warnings.append(f"Excel parse warning: {type(exc).__name__}: {str(exc)[:120]}; falling back to delimited text") + delimiter = _sniff_delimiter(text) + try: + reader = csv.DictReader(io.StringIO(text), delimiter=delimiter) + fields = [str(f).strip() for f in (reader.fieldnames or []) if str(f).strip()] + rows = [] + for idx, row in enumerate(reader): + if idx >= MAX_SAMPLE_ROWS: + break + rows.append({str(k): v for k, v in row.items() if k is not None}) + if fields: + return { + "format": "csv" if delimiter == "," else "delimited_text", + "fields": fields, + "sample_rows": rows, + "row_count_sampled": len(rows), + "delimiter": delimiter, + "warnings": warnings, + } + except Exception as exc: # noqa: BLE001 + warnings.append(f"CSV parse warning: {type(exc).__name__}: {str(exc)[:120]}") + lines = [line.strip() for line in text.splitlines() if line.strip()][:MAX_SAMPLE_ROWS] + return { + "format": "text", + "fields": [], + "sample_rows": [{"line": line} for line in lines], + "row_count_sampled": len(lines), + "warnings": warnings, + } + + +def _profile_excel(data: bytes, warnings: list[str]) -> dict[str, Any]: + from openpyxl import load_workbook + + wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True) + ws = wb[wb.sheetnames[0]] + rows_iter = ws.iter_rows(values_only=True) + header = next(rows_iter, None) + fields = [str(cell).strip() for cell in (header or []) if cell is not None and str(cell).strip()] + rows: list[dict[str, Any]] = [] + for idx, row in enumerate(rows_iter): + if idx >= MAX_SAMPLE_ROWS: + break + rows.append({fields[i]: row[i] for i in range(min(len(fields), len(row)))}) + if not fields: + warnings.append("first Excel sheet has no header row") + return { + "format": "excel", + "fields": fields, + "sample_rows": rows, + "row_count_sampled": len(rows), + "sheet": ws.title, + "warnings": warnings, + } + + +def _json_rows_and_fields(parsed: Any) -> tuple[list[dict[str, Any]], list[str]]: + if isinstance(parsed, list): + rows = [item for item in parsed if isinstance(item, dict)] + elif isinstance(parsed, dict): + if "@graph" in parsed and isinstance(parsed["@graph"], list): + rows = [item for item in parsed["@graph"] if isinstance(item, dict)] + else: + list_values = [value for value in parsed.values() if isinstance(value, list)] + rows = [item for item in (list_values[0] if list_values else [parsed]) if isinstance(item, dict)] + else: + rows = [] + fields: list[str] = [] + for row in rows[:MAX_SAMPLE_ROWS]: + for key in row: + if key not in fields: + fields.append(str(key)) + return rows, fields + + +def _sniff_delimiter(text: str) -> str: + sample = text[:4096] + try: + return csv.Sniffer().sniff(sample, delimiters=",\t;|").delimiter + except Exception: # noqa: BLE001 + return "," + + +def _build_workflow_spec( + *, + workflow_id: str, + workflow_name: str, + filename: str, + media_type: str, + upload_mode: str, + profile: dict[str, Any], + output_schema: dict[str, Any], + receipt_id: str, +) -> dict[str, Any]: + source_fields = list(profile.get("fields", [])) + properties = output_schema.get("properties", {}) if isinstance(output_schema.get("properties"), dict) else {} + required = [str(item) for item in output_schema.get("required", [])] + mappings: list[dict[str, Any]] = [] + unmapped_required: list[str] = [] + for target, prop_schema in properties.items(): + source = _best_field_match(str(target), source_fields) + target_type = prop_schema.get("type") if isinstance(prop_schema, dict) else None + rule = { + "target_field": str(target), + "source_field": source, + "required": str(target) in required, + "transformations": _transformations_for_type(target_type, prop_schema if isinstance(prop_schema, dict) else {}), + "confidence": 1.0 if source == target else 0.82 if source else 0.0, + } + mappings.append(rule) + if not source and str(target) in required: + unmapped_required.append(str(target)) + return { + "workflow_version": "1.0", + "workflow_id": workflow_id, + "workflow_name": workflow_name.strip()[:120] or "data-transform-workflow", + "created_at": datetime.now(timezone.utc).isoformat(), + "source": { + "filename": filename, + "media_type": media_type, + "upload_mode": upload_mode, + "format": profile.get("format"), + "fields": source_fields, + "sample_rows": _json_safe(profile.get("sample_rows", [])), + "sample_limit": MAX_SAMPLE_ROWS, + }, + "target_schema": output_schema, + "mapping": { + "strategy": "normalized_field_name_match_with_type_coercion", + "fields": mappings, + "unmapped_required_fields": unmapped_required, + }, + "validation": { + "required_fields": required, + "additional_properties": output_schema.get("additionalProperties", True), + "gates": [ + "parse source with declared source.format parser", + "apply per-field mapping and transformations", + "validate each output object against target_schema", + "send rows with unmapped required fields to review queue", + ], + }, + "receipt": {"receipt_id": receipt_id, "kind": "workflow_created"}, + } + + +def _best_field_match(target: str, source_fields: list[str]) -> str | None: + if target in source_fields: + return target + normalized_target = _normalize_name(target) + normalized = {_normalize_name(field): field for field in source_fields} + if normalized_target in normalized: + return normalized[normalized_target] + for norm, original in normalized.items(): + if normalized_target and (normalized_target in norm or norm in normalized_target): + return original + return None + + +def _normalize_name(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "", value.lower()) + + +def _transformations_for_type(target_type: Any, schema: dict[str, Any]) -> list[str]: + t = target_type[0] if isinstance(target_type, list) and target_type else target_type + steps = ["trim strings", "treat blank values as null"] + if t in {"integer", "number"}: + steps.append("parse numeric value") + elif t == "boolean": + steps.append("parse boolean synonyms true/false/yes/no/1/0") + elif t == "array": + steps.append("wrap scalar values or split delimited lists when configured") + elif t == "object": + steps.append("parse nested JSON object when source value is text") + else: + steps.append("coerce to string when non-null") + if "enum" in schema: + steps.append("validate against enum values") + if schema.get("format") == "date" or schema.get("format") == "date-time": + steps.append("normalize date/date-time to ISO 8601; flag ambiguous dates") + return steps + + +def _stable_workflow_id(filename: str, data: bytes, schema: dict[str, Any], workflow_name: str) -> str: + digest = hashlib.sha256() + digest.update(filename.encode("utf-8")) + digest.update(data[:65536]) + digest.update(json.dumps(schema, sort_keys=True, default=str).encode("utf-8")) + digest.update(workflow_name.encode("utf-8")) + return f"wf_{digest.hexdigest()[:16]}" + + +def _json_safe(value: Any) -> Any: + try: + json.dumps(value) + return value + except TypeError: + return json.loads(json.dumps(value, default=str)) + + +def _db_connect() -> Any | None: + url = os.environ.get("DATABASE_URL") + if not url or psycopg is None: + return None + return psycopg.connect( + url, + options="-c statement_timeout=5000 -c lock_timeout=2000 -c idle_in_transaction_session_timeout=10000", + ) + + +def _persist_workflow_and_receipt( + *, + tenant_key: str, + workflow_id: str, + receipt_id: str, + workflow_name: str, + workflow_path: str, + source_format: str, + source_filename: str, + spec: dict[str, Any], + result_summary: dict[str, Any], + warnings: list[str], +) -> bool: + conn = _db_connect() + if conn is None: + return False + try: + with conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO data_workflows + (tenant_key, workflow_id, workflow_name, workflow_path, source_format, source_filename, spec) + VALUES (%s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (tenant_key, workflow_id) DO UPDATE SET + workflow_name = EXCLUDED.workflow_name, + workflow_path = EXCLUDED.workflow_path, + source_format = EXCLUDED.source_format, + source_filename = EXCLUDED.source_filename, + spec = EXCLUDED.spec, + updated_at = NOW() + """, + ( + tenant_key, + workflow_id, + workflow_name, + workflow_path, + source_format, + source_filename, + json.dumps(spec), + ), + ) + cur.execute( + """ + INSERT INTO execution_receipts + (tenant_key, receipt_id, workflow_id, skill_name, status, input_hash, result_summary, warnings) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + tenant_key, + receipt_id, + workflow_id, + "create_workflow", + "ok", + hashlib.sha256(json.dumps(spec.get("source", {}), sort_keys=True, default=str).encode("utf-8")).hexdigest(), + json.dumps(result_summary), + json.dumps(warnings), + ), + ) + return True + except Exception: # noqa: BLE001 + return False + finally: + conn.close() + + +def _list_receipts(tenant_key: str, limit: int) -> list[dict[str, Any]]: + conn = _db_connect() + if conn is None: 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 _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) - - 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]) + try: + with conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT receipt_id, workflow_id, skill_name, status, result_summary, created_at + FROM execution_receipts + WHERE tenant_key = %s + ORDER BY created_at DESC + LIMIT %s + """, + (tenant_key, limit), + ) + rows = [] + for receipt_id, workflow_id, skill_name, status, result_summary, created_at in cur.fetchall(): + rows.append( + { + "receipt_id": receipt_id, + "workflow_id": workflow_id, + "skill_name": skill_name, + "status": status, + "result_summary": result_summary, + "created_at": created_at.isoformat() if hasattr(created_at, "isoformat") else str(created_at), + } + ) + return rows + finally: + conn.close()