Emit job definition, run phases and dependency graph artifacts from dbt Cloud tasks

## Summary

Make `dbt.cloud.TriggerRun` and `dbt.cloud.CheckStatus` expose the full lifecycle of an external dbt Cloud run as artifacts: a pre-execution snapshot of the job definition, a run-phase timeline (queued / running / heartbeat / finished) with per-step status, a link to the run and its logs, and the same model dependency graph, test results and timing artifacts as `DbtCLI`, built from the `manifest.json` / `run_results.json` artifacts the task already downloads. Enrich the table assets already emitted with run metadata.

## Motivation

A dbt Cloud run is an external job: Kestra currently shows streamed step logs and two storage URIs. Users need to see, without leaving Kestra, what was sent (job steps, environment, git branch), whether the job is healthy while it runs (heartbeat, queue time, current step), what came back (models and tests by status, the DAG), and to jump to dbt Cloud only when they need the raw logs. This is the reference case for the "external job phases" model shared with Airbyte, Fivetran and Databricks tasks.

## Context

### Artifact API as implemented in Core today

1. An "artifact" is a Vue 3 micro-frontend shipped inside the plugin jar, built from a `ui/` folder with `@kestra-io/artifact-sdk` (Vite + Module Federation) and packaged by Gradle (`buildUI` -> `src/main/resources/plugin-ui/`).
2. Core discovers `plugin-ui/manifest.json` in each plugin jar (`io.kestra.core.plugins.PluginScanner`, `UI_MANIFEST_PATH`) and serves it through `POST /api/v1/plugins/pluginUiManifest` (`PluginController`), modelled by `io.kestra.core.models.ui.PluginUiManifest` / `PluginUiModule` (`uiModule`, `staticInfo`, `styles`, `distribution` OSS|EE).
3. The manifest is keyed by task type FQCN -> list of modules. Supported slots (SDK `SLOT_NAMES`): `topology-details`, `topology-task-drawer`, `topology-task-modal`. There is no declarative "table / chart / graph" artifact type: the component renders whatever it wants with `@kestra-io/design-system` (`KsTopologyDetails`, `KsEditor`, ...).
4. The host (`ui/src/remoteComponents/useFederatedModule.ts`, `LowCodeEditor.vue`, `executions/Topology.vue`) injects props: `taskType`, `task` (merged flow-source task), `progress`, `execution`, `namespace`, `flowId`, `tenant`, `source`, `fetchOutputs({taskRunId})`, `fetchMetrics({page,size,sort,taskRunId})`.
5. `staticInfo` for `topology-details` carries `height`, `heightWithExecution` and an optional `customAction {label, taskProp, lang}`.
6. Rule from kestra#18884: any value displayed by an artifact must be a task output read through `fetchOutputs`, never a Pebble expression rendered on the fly.
7. There is no backend-side artifact interface yet (kestra-ee#9994): a plugin cannot expose a custom endpoint. That issue explicitly names "a button that fetches dbt Cloud logs through a plugin endpoint" and "backend parsing of the dbt manifest" as the target use cases, so the graph and on-demand logs depend on it.
8. Reference implementation: `plugin-gcp/ui/vite.config.ts` (`exposes` map) and `plugin-gcp/ui/src/components/BigqueryQueryTopologyDetails.vue`.
9. Assets API: `io.kestra.core.models.assets.{Asset, Custom, External, AssetIdentifier, AssetsInOut, AssetsDeclaration}`; a task emits with `runContext.assets().emit(new AssetEmit(inputs, outputs))`; `AssetsDeclaration` exposes `assets.enableAuto`, `inputs`, `outputs`, `assetFailureBehavior`. Table type is `io.kestra.plugin.ee.assets.Table`; OSS falls back to `NoopAssetService`.
10. The original epic (data table, data summaries, dependency graph, topology nodes) is https://github.com/kestra-io/kestra/issues/12696 (closed); this issue extends it.

Related:
- https://github.com/kestra-io/kestra/issues/12696
- https://github.com/kestra-io/kestra-ee/issues/9994
- https://github.com/kestra-io/kestra/issues/18884

## Current State

Files (`plugin-dbt/src/main/java/io/kestra/plugin/dbt/cloud/`):
- `TriggerRun.java` — `POST /api/v2/accounts/{accountId}/jobs/{jobId}/run/` with optional `steps_override` and other overrides; `wait` (default true) delegates polling to `CheckStatus`; `Output { runId, runResults (URI), manifest (URI) }`.
- `CheckStatus.java` — polls `GET /api/v2/accounts/{accountId}/runs/{runId}/?include_related=["trigger","job","run_steps","environment"(,"debug_logs")]`, streams step logs (l.567), fetches final debug logs best-effort (l.264), downloads `run_results.json` and `manifest.json` from `/runs/{runId}/artifacts/{path}` (l.600-611), calls `ResultParser.parseManifestWithAssets` / `parseRunResult`, dedupes lineage per run with KV key `dbt-cloud-last-run.*` (l.117). `Output { runId, lineageEmitted, assets[], runResults (URI), manifest (URI) }`. `parseRunResults` property (default true).
- `AbstractDbtCloud.java` — base URL handling (legacy `cloud.getdbt.com`), auth, retry/backoff on 429/502/503/504.
- `models/Run.java` — full run mapping: `status`, `statusMessage`, `gitBranch`, `gitSha`, `dbtVersion`, `createdAt`, `dequeuedAt`, `startedAt`, `finishedAt`, `lastHeartbeatAt`, `queuedDuration`, `runDuration`, `artifactsSaved`, `deferringRunId`, `trigger`, `job`, `environment`, `runSteps[]`.
- `models/Step.java` — `name`, `status`, `statusColor`, `startedAt`, `finishedAt`, `duration`, `logs`, `debugLogs`, `truncatedDebugLogs`, `logLocation`.
- `models/Job.java`, `JobSettings.java`, `JobTriggers.java`, `Environment.java`, `ManifestArtifact.java`.
- Shared with CLI: `../ResultParser.java`, `../models/Manifest.java`, `../models/RunResult.java`.
- No `ui/` module in this repo yet.

Existing outputs: `runId`, `runResults`, `manifest`, `assets[]`, `lineageEmitted`. Existing assets: Table assets from the manifest (shared `ResultParser`), emitted once per dbt Cloud run.

## Proposed Artifacts

| artifact | type | phase | data source |
|---|---|---|---|
| `job-definition` | table + link | pre (setup) | `GET /api/v2/accounts/{accountId}/jobs/{jobId}/` before triggering: `name`, `execute_steps[]` (or `stepsOverride`), `environment {name, dbt_version, type}`, `triggers {github_webhook, schedule}`, `generate_docs`, `run_generate_sources`, `deferring_job_definition_id`, `project_id`; persisted as `jobDefinition.json` (URI) + inline `jobDefinitionSummary`; link `{baseUrl}/deploy/{accountId}/projects/{projectId}/jobs/{jobId}` |
| `run-phases` | table / timeline | during + post (job health) | `Run`: `createdAt`, `dequeuedAt`, `startedAt`, `finishedAt`, `lastHeartbeatAt`, `queuedDuration`, `runDuration`, `status`, `statusMessage`, `gitBranch`, `gitSha`, `dbtVersion`, `deferringRunId`, `artifactsSaved`; inline output `run {…}`; heartbeat age exposed as `heartbeatAgeSeconds` at each poll (WARN log when > 5 min) |
| `run-steps` | table | during + post (sub-node parsing) | `runSteps[*] {index, name, status, startedAt, finishedAt, duration}`; today streamed as logs only; stored as ION `steps` (URI) and, like `parseRunResult` does for models, one dynamic task run per step |
| `run-link` / `run-logs` | link (+ on-demand logs modal) | post | `{baseUrl}/deploy/{accountId}/projects/{projectId}/runs/{runId}` as output `runUrl`; `debug_logs` fetched on demand through the future backend endpoint (not stored by default: size) |
| `model-graph` | graph coloured by status | post (result parsing) | same `graph.json` builder as `DbtCLI` from the downloaded `manifest.json` + `run_results.json`; output `graph` (URI) + `graphSummary` |
| `test-results` | table | post | same as CLI: `tests` (URI) + `testSummary` |
| `model-timing` | chart | post | same as CLI: `modelRuns` (URI) |
| `run-summary` | table | post (teardown verification) | `runSummary {invocationId, dbtVersion, adapter, elapsedTime, models{}, tests{}, rowsAffected}` plus `artifactsSaved`, `isComplete`, `notificationsSent` to confirm the remote side finished cleanly |
| `changed-since-last-run` | table | post | previous manifest = artifact of `deferringRunId` when set (`/runs/{deferringRunId}/artifacts/manifest.json`), else the last `manifest.json` output stored for this task (KV key `dbt-cloud-last-run.*` already tracks the last processed run id); `manifestDiff` (URI) + `manifestDiffSummary` |

## Proposed Assets

- Keep Table assets from the manifest (shared `ResultParser.emitAssets`), still emitted once per dbt Cloud run.
- Enrich metadata with `lastRunStatus`, `lastRunAt`, `executionTime`, `rowsAffected`, `materialized`, `tags`, `dbtCloudRunId`, `dbtCloudJobId`, `environment`, `gitSha` so the catalog answers "which job / commit built this table".
- Optionally emit one `External` asset for the dbt Cloud job itself (`id: dbt-cloud.job.{accountId}.{jobId}`, metadata = job definition summary) as the lineage "producer" of the tables; behind `assets.enableAuto`.

## Implementation Notes

Build now (no dependency on the Core interface):
1. `TriggerRun`: fetch the job definition before triggering (one extra GET, reuse `AbstractDbtCloud` HTTP + retry), persist it with `runContext.storage().putFile()`, add outputs `jobDefinition` (URI), `jobDefinitionSummary`, `jobUrl`. Fail soft (WARN) if the GET fails: triggering must not depend on it.
2. `CheckStatus`: map the polled `Run` into an inline `run` output and the `steps` ION; compute `heartbeatAgeSeconds`; emit one dynamic task run per `runSteps[*]` (same mechanism as `ResultParser.parseRunResult`). Add `runUrl`.
3. Reuse the `ArtifactBuilder` introduced for `DbtCLI` (graph / tests / modelRuns / runSummary / manifestDiff) on the already downloaded artifacts; `TriggerRun` copies the new outputs from `CheckStatus.Output` as it already does for `runResults` / `manifest` (l.461).
4. Previous-manifest source: `deferringRunId` artifacts when present, else the previous run's stored manifest URI recorded next to the existing `dbt-cloud-last-run.*` KV entry.
5. Metrics: `Timer` `queued.duration`, `run.duration`; `Counter` `steps.success/error`, `models.*`, `tests.*`.
6. Use `io.kestra.core.serializers.JacksonMapper` for the new models (`JobDefinition` = existing `Job` + `JobSettings` + `JobTriggers`, extend if fields are missing).

Build after kestra-ee#9994 lands:
7. Share the `ui/` module with the CLI tasks: `DbtRunTopologyDetails.vue` gains a dbt Cloud header (job name, environment, branch, phases) when `run` is present; `DbtRunDrawer.vue` adds tabs Job / Steps / Logs. Exposed for `cloud.TriggerRun` and `cloud.CheckStatus`.
8. "Show logs" button calls the plugin's backend endpoint which proxies `GET /runs/{runId}/?include_related=["debug_logs"]` using the task's stored credentials server-side; the token never reaches the browser.

Security / redaction:
- Never output `token`, account-level secrets, or environment variables returned by the jobs / environments API (`custom_environment_variables`, `credentials`); the job-definition snapshot is an allow-list of fields, not the raw response.
- Step `logs` / `debugLogs` may contain connection details echoed by adapters: they are streamed to task logs today and stay there; the `steps` ION stores status/timing only.
- `runUrl` / `jobUrl` are plain links built from `baseUrl`, `accountId`, `projectId`, `jobId`, `runId`; no query params.
- All payloads in internal storage; the KV store keeps only the dedupe marker it already has.

## YAML Example

```yaml
id: dbt_cloud_artifacts
namespace: company.team

tasks:
- id: dbt_cloud
type: io.kestra.plugin.dbt.cloud.TriggerRun
accountId: "{{ secret('DBT_CLOUD_ACCOUNT_ID') }}"
token: "{{ secret('DBT_CLOUD_API_TOKEN') }}"
baseUrl: https://cloud.getdbt.com
jobId: "12345"
wait: true
pollFrequency: PT10S
maxDuration: PT1H
parseRunResults: true
assets:
enableAuto: true

- id: check
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.dbt_cloud.testSummary.fail > 0 or outputs.dbt_cloud.run.status != 'SUCCESS' }}"
then:
- id: notify
type: io.kestra.plugin.core.log.Log
level: ERROR
message: |
dbt Cloud run {{ outputs.dbt_cloud.runId }} ({{ outputs.dbt_cloud.run.gitBranch }}@{{ outputs.dbt_cloud.run.gitSha }})
queued {{ outputs.dbt_cloud.run.queuedDuration }}, ran {{ outputs.dbt_cloud.run.runDuration }}
failed tests: {{ outputs.dbt_cloud.testSummary.fail }} -> {{ outputs.dbt_cloud.runUrl }}

- id: report
type: io.kestra.plugin.core.log.Log
message: |
job: {{ outputs.dbt_cloud.jobDefinitionSummary.name }} / {{ outputs.dbt_cloud.jobDefinitionSummary.environment }}
steps: {{ outputs.dbt_cloud.steps }}
graph: {{ outputs.dbt_cloud.graph }}
```

## Acceptance Criteria

- [ ] `TriggerRun` persists a job-definition snapshot before triggering and exposes `jobDefinition` (URI), `jobDefinitionSummary`, `jobUrl`; a failing GET only logs a warning
- [ ] `CheckStatus` / `TriggerRun` expose inline `run` (phases, branch, sha, dbt version, heartbeat), `runUrl`, `steps` (URI) and one dynamic task run per dbt Cloud step with correct CREATED/RUNNING/terminal history
- [ ] `graph`, `tests`, `modelRuns`, `manifestDiff`, `runSummary`, `graphSummary`, `testSummary`, `manifestDiffSummary` are produced from the downloaded artifacts, identical shape to the `DbtCLI` outputs (shared builder, shared fixtures)
- [ ] `changed-since-last-run` uses `deferringRunId` artifacts when present, else the previous processed run's manifest; empty and non-failing otherwise
- [ ] Missing `manifest.json` (source freshness runs) or asynchronous artifact upload still never fails a successful run; artifacts simply absent with `runContext.logger().warn`
- [ ] Lineage still emitted exactly once per run (existing dedupe test), asset metadata enriched with run / job / environment / git fields
- [ ] Snapshot and outputs never contain the API token, environment variables or credentials (test with a mocked `/jobs/{id}/` response containing `custom_environment_variables`)
- [ ] New properties use `Property`; logging via `runContext.logger()`; JSON via `io.kestra.core.serializers.JacksonMapper`; ION via `FileSerde`
- [ ] `MockTriggerRunTest` / `CheckStatusTest` extended with `run_steps`, job definition and `deferringRunId` fixtures
- [ ] Follow-up issue filed for the shared `ui/` module (Job / Steps / Logs tabs, on-demand debug logs via plugin backend endpoint) once https://github.com/kestra-io/kestra-ee/issues/9994 lands; displayed values are outputs only (https://github.com/kestra-io/kestra/issues/18884)
- [ ] `@Plugin(examples = ...)` gains the flow above; README updated

---
*[View as Artifact](https://claude.ai/code/artifact/4a0ba58a-5065-4afd-8c8a-d99f492a74d0)*