feat(contract-toolkit): express Automation Engine activity nodes in the manifest DAG [FND-4754]

contract-toolkit/AGENTS.md

@@ -56,6 +56,7 @@ scripts/
 |-- check-ruff-stable.py  # ruff changes no generated .py (default settings)
 `-- test-sdk-import.py
 examples/
+|-- ae-activities/         # AE activity nodes (FetchEntitiesNode, MergeNode)
 |-- connection-ref/        # ConnectionRefInput typed output
 |-- fanin/                 # DependencyCondition and fan-in
 |-- openapi/               # OpenAPI loader with extra object-store credential

contract-toolkit/CLAUDE.md

@@ -71,6 +71,8 @@ demonstrates distinct feature surface, verified by `tests/*.pkl`.
 - `connection-ref`: `ConnectionRefInput` widget, `pipeline.publish = null`.
 - `publish-controls`: publish toggles, `includeInputFields`, `errorHandling`, `zeroOutConfig`.
 - `fanin`: multi-parent fan-in via `dependsOn`, explicit `DependencyCondition`.
+- `ae-activities`: `FetchEntitiesNode`, `MergeNode` and generic `ActivityNode` in
+  `extraNodes`; AE activity nodes mixed with a `DAGNode`, no `extract` step.
 - `artifact-schemas`: `artifactSchemas` data hand-off declarations — parquet + NDJSON, nested paths, arrays of structs (`[]` element step), an input artifact.
 
 ## Editing Rules

contract-toolkit/README.md

@@ -112,6 +112,7 @@ The `examples/` directory contains executable contracts that teach stable toolki
 - [`examples/connection-ref/`](examples/connection-ref/) — `ConnectionRefInput` widget, `pipeline.publish = null`.
 - [`examples/publish-controls/`](examples/publish-controls/) — publish toggles, `includeInputFields`, `errorHandling`, `zeroOutConfig`.
 - [`examples/fanin/`](examples/fanin/) — multi-parent fan-in via `dependsOn`, explicit `DependencyCondition`.
+- [`examples/ae-activities/`](examples/ae-activities/) — Automation Engine activity nodes in `extraNodes`: `FetchEntitiesNode` → `MergeNode` branches fanning in to an app workflow, plus a generic `ActivityNode`.
 - [`examples/agent-e2e/`](examples/agent-e2e/) — agent/SDR e2e codegen: `_e2e_credential.py` emits both `<Name>CredentialBody` (direct) and `<Name>AgentCredentialBody` (lightweight), plus an `extraction-method` ConditionalInput whose `overrideEnum` widens the substitutions `Literal` to `["direct", "agent"]`.
 - [`examples/scheduled/`](examples/scheduled/) — cron background job via `schedules`; renders `triggers.schedules` into `manifest.json` (multiple schedules, non-UTC timezone, a `PAUSED` one). See [Schedules](docs/reference.md#schedules-background-jobs).
 - [`examples/streaming/`](examples/streaming/) — streaming dispatch via `streaming.enabled`: AE starts one short run per Kafka micro-batch and hands it the batch (the generated input model reads it from `batch_key`), instead of a per-batch run that reads the Iceberg events table. Two streaming triggers and `streamingWorkflowType` on the entrypoint. Its own app because a streaming entrypoint cannot also carry schedules or batch triggers — one extract node cannot serve both shells. See [Streaming Dispatch](docs/reference.md#streaming-dispatch-on-event-triggers).
@@ -402,6 +403,24 @@ To replace the generated node, define `extraNodes["notifications"]`.
 
 `extraNodes` is still available as an escape hatch for nodes that fall outside the canonical pipeline (e.g. custom fan-in steps).
 
+**AE activity nodes.** A `DAGNode` always starts a child workflow (`execute_workflow`). To call an Automation Engine activity directly, put a `FetchEntitiesNode`, a `MergeNode`, or the generic `ActivityNode` in `extraNodes`. These render `node_type: "activity"` with the activity's own `inputs`, and allow `errorHandling.heartbeatTimeoutSeconds`. Their `errorHandling` defaults to `null`, so AE's per-activity timeouts apply. See [AE Activity Nodes](docs/reference.md#ae-activity-nodes-fetchentitiesnode-mergenode-activitynode) and [`examples/ae-activities/`](examples/ae-activities/).
+
+```pkl
+extraNodes {
+  ["fetch_assets"] = new FetchEntitiesNode {
+    displayName = "Fetch Assets"
+    sqlInputFile = "artifacts/queries/assets.sql"
+    engine = "duckdb"
+  }
+  ["merge"] = new MergeNode {
+    displayName = "Aggregate Assets"
+    inputPrefixes { "$.fetch_assets.outputs.output_path" }
+    sql = "SELECT * FROM input_0"
+    dependsOn { "fetch_assets" }
+  }
+}
+```
+
 ## Pools Map
 
 The `pools` map declares named worker pools. It is **fully optional and explicit-overrides-only** — an empty map (the default) emits nothing. Declare named pools to configure per-pool KEDA, resources, or env.

contract-toolkit/docs/reference.md

@@ -3387,6 +3387,77 @@ non-standard node ids. Use `waitForAssetPublish = false` when the Lineage app
 node already depends on asset publish. Use `dependsOn = new Listing { ... }`
 when replacing the default fan-in entirely.
 
+### AE Activity Nodes (`FetchEntitiesNode`, `MergeNode`, `ActivityNode`)
+
+A `DAGNode` always renders `activity_name = "execute_workflow"` and starts a child
+workflow. To call an Automation Engine activity directly, such as `fetch_entities` or
+`merge`, add an `AEActivityNode` to `extraNodes` instead. `extraNodes` is typed
+`Mapping<String, DAGNode|AEActivityNode>`, so both kinds mix in one DAG.
+
+```pkl
+abstract class AEActivityNode {
+  displayName: String
+  appName: String = "automation-engine"     // the app that registers the activity
+  dependsOn: Listing<String>?               // same as DAGNode.dependsOn
+  dependsOnCondition: DependencyCondition?  // same as DAGNode.dependsOnCondition
+  errorHandling: ErrorHandlingConfig?       // default null → AE's per-activity defaults
+}
+
+class FetchEntitiesNode extends AEActivityNode {   // activity_name = "fetch_entities"
+  sqlInput: String?                    // exactly one of sqlInput / sqlInputFile
+  sqlInputFile: String?
+  engine: ("daft"|"duckdb")?           // AE default: "daft"
+  extraInput: Mapping<String, Any>     // other FetchEntitiesInput fields (sql_config, incremental)
+}
+
+class MergeNode extends AEActivityNode {           // activity_name = "merge"
+  inputPrefixes: Listing<String>(!isEmpty)  // input_0, input_1, … in the SQL
+  sql: String?                         // exactly one of sql / sqlFile
+  sqlFile: String?
+  declaredTier: ("s"|"m"|"l")?
+  extraInputs: Mapping<String, Any>    // other MergeInput fields
+}
+
+class ActivityNode extends AEActivityNode {        // any other AE activity
+  activityName: String
+  inputs: Mapping<String, Any>         // rendered verbatim; not checked by the toolkit
+}
+```
+
+Rendered shape (`FetchEntitiesNode { displayName = "Fetch Assets"; sqlInputFile = ""; engine = "duckdb" }`):
+
+```json
+{
+  "node_type": "activity",
+  "activity_name": "fetch_entities",
+  "activity_display_name": "Fetch Assets",
+  "app_name": "automation-engine",
+  "inputs": { "input": { "sql_input_file": "", "engine": "duckdb" } }
+}
+```
+
+What changes compared with a `DAGNode`:
+
+- `node_type: "activity"` is rendered. `DAGNode` output does not change and still omits it.
+- `inputs` is the activity's own shape. There is no `workflow_type`, `task_queue` or `args`
+  envelope, and the toolkit stamps no `app_name` arg (CNCT-93 applies to `args` only).
+- `errorHandling` defaults to `null`, not the 1-day `DAGNode` default, so AE applies the
+  activity's registered timeouts. `heartbeatTimeoutSeconds` is allowed. It must still be
+  less than `startToCloseTimeoutSeconds`.
+- `depends_on` renders exactly as on a `DAGNode`.
+
+Caveats:
+
+- The SQL-source checks treat an empty string as set. An app that fills `sql_input_file` or
+  `sql` in at submission time can render `""`. AE rejects an empty value at run time, so
+  something must fill it.
+- `extraInput` / `extraInputs` cannot set a typed field. Use the typed property.
+- `ActivityNode.inputs` is not validated by the toolkit. AE validates it on submission.
+- Downstream nodes read a fetch or merge result from `$.<node>.outputs.output_path`.
+
+See `examples/ae-activities/` for three fetch → merge branches fanning in to an app
+workflow.
+
 ### Custom Pipeline Example
 
 Insert QI and enrichment nodes between extract and publish:

contract-toolkit/examples/ae-activities/app.pkl

@@ -0,0 +1,84 @@
+/// AE activity nodes — a DAG that calls Automation Engine activities directly.
+///
+/// `FetchEntitiesNode` and `MergeNode` render `node_type: "activity"` with the
+/// activity's own `inputs` instead of the `execute_workflow` envelope a
+/// `DAGNode` renders. Here three fetch → merge branches fan in to an app
+/// workflow (`post_fetch`), which a generic `ActivityNode` (`upload`) follows.
+/// There is no `extract` step: the fetch nodes read the metadata lakehouse.
+///
+/// Each fetch reads its query from an uploaded `.sql` file (`sqlInputFile`, an
+/// object-store key); each merge runs inline `sql` over `input_0`.
+/// `ActivityNode` covers any AE activity without a typed node.
+amends "../../src/App.pkl"
+
+name = "ae-activities"
+displayName = "AE Activities Example"
+icon = "https://assets.atlan.com/assets/fullscreen-exit.svg"
+hasCredentialConfig = false
+notifications = false
+
+pipeline {
+  extract = null
+  publish = null
+}
+
+uiConfig = new UIConfig {
+  tasks {
+    ["Configuration"] {
+      inputs {
+        ["qn-prefix"] = new TextInput {
+          title = "Qualified name prefix"
+          helpText = "Export assets under this prefix."
+          placeholderText = "default"
+        }
+      }
+    }
+  }
+}
+
+local fetchErrorHandling = new ErrorHandlingConfig {
+  startToCloseTimeoutSeconds = 3600
+  heartbeatTimeoutSeconds = 60
+}
+
+extraNodes {
+  for (kind, label in Map("asset", "Assets", "glossary", "Glossaries", "product", "Products")) {
+    ["fetch_\(kind)"] = new FetchEntitiesNode {
+      displayName = "Fetch \(label)"
+      sqlInputFile = "artifacts/apps/ae-activities/queries/\(kind).sql"
+      engine = "duckdb"
+      errorHandling = fetchErrorHandling
+    }
+    ["merge_\(kind)"] = new MergeNode {
+      displayName = "Aggregate \(label)"
+      inputPrefixes { "$.fetch_\(kind).outputs.output_path" }
+      sql = "SELECT * FROM input_0"
+      errorHandling = fetchErrorHandling
+      dependsOn { "fetch_\(kind)" }
+    }
+  }
+  ["post_fetch"] = new DAGNode {
+    displayName = "Enrich and Export"
+    appName = name
+    workflowType = "ae-activities:post-fetch"
+    args {
+      ["asset_path"] = "$.merge_asset.outputs.output_path"
+      ["glossary_path"] = "$.merge_glossary.outputs.output_path"
+      ["product_path"] = "$.merge_product.outputs.output_path"
+      ["metadata"] = new Mapping {
+        ["qn-prefix"] = "{{qn-prefix}}"
+      }
+    }
+    errorHandling { startToCloseTimeoutSeconds = 43200 }
+    dependsOn { "merge_asset"; "merge_glossary"; "merge_product" }
+  }
+  // Generic form, for an AE activity with no typed node.
+  ["upload"] = new ActivityNode {
+    displayName = "Upload Export"
+    activityName = "object_store_upload"
+    inputs {
+      ["source_path"] = "$.post_fetch.outputs.export_path"
+    }
+    dependsOn { "post_fetch" }
+  }
+}

contract-toolkit/examples/ae-activities/app.yaml

@@ -0,0 +1,5 @@
+# AUTO-GENERATED from contract/app.pkl — DO NOT EDIT MANUALLY.
+# To regenerate: pkl eval -m . contract/app.pkl
+app_name: ae-activities
+app_image: ${APP_IMAGE}
+app_port: 8000

contract-toolkit/examples/ae-activities/app/generated/_e2e_base.py

@@ -0,0 +1,10 @@
+# Generated from contract/app.pkl via contract-toolkit. DO NOT EDIT.
+# Regenerate with: pkl eval -m . contract/app.pkl
+from application_sdk.testing.e2e import BaseE2ETest
+
+
+class AeActivitiesGeneratedE2EBase(BaseE2ETest):
+    connector_short_name = "ae-activities"
+    argo_package_name = "@atlan/ae-activities"
+    argo_template_name = "atlan-ae-activities"
+    app_service_url = "http://ae-activities.ae-activities-app.svc.cluster.local"

contract-toolkit/examples/ae-activities/app/generated/_e2e_substitutions.py

@@ -0,0 +1,10 @@
+# Generated from contract/app.pkl via contract-toolkit. DO NOT EDIT.
+# Regenerate with: pkl eval -m . contract/app.pkl
+from __future__ import annotations
+
+from application_sdk.testing.e2e.substitutions import MustacheSubstitutions
+from pydantic import Field
+
+
+class AeActivitiesMustacheSubstitutions(MustacheSubstitutions):
+    qn_prefix: str = Field(default="", alias="{{qn-prefix}}")

contract-toolkit/examples/ae-activities/app/generated/_input.py

@@ -0,0 +1,10 @@
+# AUTO-GENERATED from contract/app.pkl — DO NOT EDIT MANUALLY.
+# To regenerate: pkl eval -m . contract/app.pkl
+from __future__ import annotations
+
+from application_sdk.templates.contracts import ExtractionInput
+
+
+class AppInputContract(ExtractionInput):
+    qn_prefix: str = ""
+    """Export assets under this prefix."""

contract-toolkit/examples/ae-activities/app/generated/ae-activities.json

@@ -0,0 +1,31 @@
+{
+  "id": "ae-activities",
+  "name": "AE Activities Example",
+  "logo": "https://assets.atlan.com/assets/fullscreen-exit.svg",
+  "config": {
+    "properties": {
+      "qn-prefix": {
+        "type": "string",
+        "required": false,
+        "ui": {
+          "widget": "input",
+          "label": "Qualified name prefix",
+          "hidden": false,
+          "help": "Export assets under this prefix.",
+          "placeholder": "default",
+          "grid": 8
+        }
+      }
+    },
+    "steps": [
+      {
+        "title": "Configuration",
+        "description": "",
+        "id": "configuration",
+        "properties": [
+          "qn-prefix"
+        ]
+      }
+    ]
+  }
+}