Emit connection snapshot, per-stream sync stats and stream assets from Airbyte Sync

## Summary

Make `airbyte.connections.Sync` / `CheckStatus` (self-hosted API) and `airbyte.cloud.jobs.Sync` (Airbyte Cloud public API) expose the lifecycle of the external sync job as artifacts: a pre-run snapshot of the connection (source, destination, streams and sync modes), a per-stream records/bytes table, an attempt timeline with failure reasons, a job summary and a link to the job logs. Auto-emit one table asset per synced stream in the destination, with source -> destination lineage.

## Motivation

The plugin already collects everything Airbyte knows about a job (attempts, per-stream stats, failure summaries, logs) but exposes it only as scattered metrics and streamed log lines; outputs are `jobId` and `alreadyRunning`. Users cannot answer "which streams moved how many records", "why did attempt 1 fail", "did the connection definition change" without opening Airbyte. Streams are also the natural table assets for ingestion lineage and nothing is emitted today.

## 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, so on-demand job logs or large stream tables served from storage 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-airbyte/src/main/java/io/kestra/plugin/airbyte/`):
- `connections/Sync.java` — `POST /api/v1/connections/sync/` (l.164); `wait` (default true) delegates to `CheckStatus`; `Output { jobId, alreadyRunning }`; `failOnActiveSync`.
- `connections/CheckStatus.java` — polls `POST /api/v1/jobs/get/` (l.140); streams attempt logs (l.220); emits metrics `attempts.count`, `records.committed`, `records.emitted`, `bytes.emitted`, `state.emitted` tagged by `stream` (l.189-208); logs `failureSummary`; `Output { finalJobStatus }`.
- `connections/AbstractAirbyteConnection.java` — `url`, `username`/`password`, `token`, `applicationCredentials` (`/api/v1/applications/token`), `httpTimeout`.
- `models/JobInfo {job, attempts[]}`, `Job {id, configType, configId, createdAt, updatedAt, status, resetConfig}`, `AttemptInfo {attempt, logs, logType}`, `Attempt {id, status, createdAt, updatedAt, endedAt, bytesSynced, recordsSynced, totalStats, streamStats[], failureSummary}`, `AttemptStreamStats {streamName, stats}`, `AttemptStats {recordsEmitted, bytesEmitted, stateMessagesEmitted, recordsCommitted}`, `AttemptFailureSummary` / `AttemptFailureReason {origin, type, ...}`, `Log`, `StreamDescriptor`.
- `cloud/jobs/Sync.java`, `cloud/jobs/Reset.java`, `cloud/jobs/AbstractTrigger.java` — Airbyte Cloud public API; `Output { job, jobId, jobType, status, duration, bytesSynced, rowsSynced }`; metrics `bytes_synced`, `rows_synced`, `duration`.
- `cloud/AbstractAirbyteCloud.java` — token / client credentials.
- No asset emission anywhere in this repo. No `ui/` module.

## Proposed Artifacts

| artifact | type | phase | data source |
|---|---|---|---|
| `connection-snapshot` | table (+ streams sub-table) | pre (setup) | self-hosted: `POST /api/v1/web_backend/connections/get {connectionId}` (single call returning `name`, `status`, `schedule`, `namespaceDefinition`, `namespaceFormat`, `prefix`, `source {name, sourceName}`, `destination {name, destinationName}`, `syncCatalog.streams[*] {stream.name, stream.namespace, config.selected, config.syncMode, config.destinationSyncMode, config.cursorField, config.primaryKey}`); cloud: `GET /v1/connections/{connectionId}` + `GET /v1/sources/{id}` + `GET /v1/destinations/{id}`; persisted as `connection.json` (URI) + inline `connection {name, sourceType, destinationType, streamCount, selectedStreamCount, schedule}` |
| `job-summary` | table | post (teardown verification) | `Job {id, configType, status, createdAt, updatedAt}` + totals over the last attempt (`recordsCommitted`, `recordsEmitted`, `bytesEmitted`), `attemptCount`, `durationMs`; inline output `job` |
| `stream-stats` | table (+ bar chart records per stream) | post (result parsing) | last successful attempt `streamStats[*]`: `streamName`, `namespace` (from snapshot), `syncMode`, `recordsEmitted`, `recordsCommitted`, `bytesEmitted`, `stateMessagesEmitted`, `recordsDropped = emitted - committed`; ION `streams` (URI) + inline `streamSummary {streams, recordsCommitted, bytesEmitted}` |
| `attempt-timeline` | chart (gantt) / table | during + post (job health) | `attempts[*].attempt {id, status, createdAt, endedAt, bytesSynced, recordsSynced, failureSummary.failures[*] {failureOrigin, failureType, externalMessage, timestamp}}`; ION `attempts` (URI); one dynamic task run per attempt (CREATED/RUNNING/terminal) so retries show as sub-bars |
| `job-link` / `job-logs` | link (+ on-demand logs modal) | post | self-hosted: `{url}/workspaces/{workspaceId}/connections/{connectionId}/job-history#{jobId}::0` (workspaceId from the snapshot); cloud: `https://cloud.airbyte.com/workspaces/{workspaceId}/connections/{connectionId}/job-history`; logs already streamed to task logs; full `attempts[*].logs.logLines` on demand through the future backend endpoint |
| `connection-drift` | table | pre | diff of `connection.json` against the previous execution's snapshot (streams added/removed, sync mode changed, source/destination version changed); opt-in `compareWithPreviousExecution`; inline `connectionDriftSummary` |

Cloud API limits to document: the public API job object has `bytesSynced`, `rowsSynced`, `duration` but no per-attempt or per-stream stats; `stream-stats` and `attempt-timeline` are therefore self-hosted-only until the public API exposes them (`GET /v1/jobs/{jobId}` returns aggregate only).

## Proposed Assets

- One `Custom` asset of type `io.kestra.plugin.ee.assets.Table` per selected stream, representing the destination table: `id: {destinationName}.{namespace}.{prefix}{streamName}` where `namespace` is resolved from `namespaceDefinition` (`source` -> stream namespace, `destination` -> destination default, `customformat` -> rendered `namespaceFormat`), `displayName: {streamName}`, metadata `{system: destinationName (e.g. Postgres, Snowflake, BigQuery), connectionId, streamName, syncMode, destinationSyncMode, cursorField, primaryKey, lastJobId, lastJobStatus, recordsCommitted, bytesEmitted, lastSyncedAt}`.
- One `External` asset per source stream as lineage input (`id: airbyte.source.{sourceName}.{namespace}.{streamName}`), so the catalog shows source -> destination table.
- Emitted from `connections.CheckStatus` after the job ends (stats known) and from `cloud.jobs.Sync` (connection-level metadata only); gated by `assets.enableAuto` / `assetFailureBehavior`; skip silently on OSS as the JDBC plugin does.

## Implementation Notes

Build now (no dependency on the Core interface):
1. New models `WebBackendConnection`, `SyncCatalog`, `AirbyteStream`, `StreamConfig` (self-hosted) and the cloud equivalents, mapped with `io.kestra.core.serializers.JacksonMapper` (`ofJson(false)` to survive API additions).
2. `connections.Sync`: fetch the snapshot before `POST /connections/sync/`, persist with `runContext.storage().putFile()`, add outputs `connection` (inline summary), `connectionSnapshot` (URI), `jobUrl`. Snapshot failure is a WARN, never blocks the sync.
3. `connections.CheckStatus`: build `job`, `streams` (URI), `streamSummary`, `attempts` (URI) from the final `JobInfo`; emit one dynamic task run per attempt; keep existing metrics; `Sync.Output` copies all of it when `wait: true`.
4. `cloud.jobs.Sync` / `Reset`: snapshot via the three public-API GETs, `job` summary from the existing `Job` response, `jobUrl`.
5. `connection-drift`: opt-in `Property compareWithPreviousExecution`; previous snapshot = `connectionSnapshot` output URI of the last execution of the same flow/task (skip silently when unavailable).
6. Assets: `StreamAssetEmitter` helper building inputs/outputs from snapshot + stats, called at the end of `CheckStatus` and `cloud.jobs.Sync`.
7. Metrics: add `Timer` `duration` (self-hosted), `Counter` `records.dropped` per stream, `Counter` `attempts.failed`.

Build after kestra-ee#9994 lands:
8. `ui/` module: `AirbyteSyncTopologyDetails.vue` (`topology-details`: source -> destination, streams count, records/bytes, attempts, status) and `AirbyteSyncDrawer.vue` (`topology-task-drawer`: Streams table + bar chart, Attempts timeline, Connection snapshot, Logs button). Exposed for `connections.Sync`, `connections.CheckStatus`, `cloud.jobs.Sync`, `cloud.jobs.Reset`.
9. Logs button proxies `attempts[*].logs` through the plugin backend endpoint with server-side credentials.

Security / redaction:
- The snapshot is an allow-list; `source.connectionConfiguration` / `destination.connectionConfiguration` (hosts, users, passwords, keys) are never persisted. Only `sourceName` / `destinationName` (connector type), `name`, ids and catalog structure are kept.
- Never output `token`, `password`, `clientSecret`, or the application access token.
- `failureSummary.externalMessage` / `internalMessage` may echo connector configuration; keep `externalMessage` only, truncated to 2 KiB, and drop `stacktrace`.
- Job logs stay in task logs (already masked by the run context) and are not persisted as artifacts by default.
- All payloads in internal storage; nothing in KV.

## YAML Example

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

tasks:
- id: sync
type: io.kestra.plugin.airbyte.connections.Sync
url: http://airbyte.internal:8000
username: "{{ secret('AIRBYTE_USER') }}"
password: "{{ secret('AIRBYTE_PASSWORD') }}"
connectionId: e3b1c6a2-7f0e-4c2d-9a1b-2f3e4d5c6b7a
wait: true
pollFrequency: PT15S
maxDuration: PT2H
compareWithPreviousExecution: true
assets:
enableAuto: true

- id: check
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.sync.job.status != 'succeeded' or outputs.sync.streamSummary.recordsCommitted == 0 }}"
then:
- id: alert
type: io.kestra.plugin.core.log.Log
level: ERROR
message: |
Airbyte job {{ outputs.sync.jobId }} on {{ outputs.sync.connection.name }} ended {{ outputs.sync.job.status }}
attempts: {{ outputs.sync.job.attemptCount }}, records committed: {{ outputs.sync.streamSummary.recordsCommitted }}
{{ outputs.sync.jobUrl }}

- id: report
type: io.kestra.plugin.core.log.Log
message: |
{{ outputs.sync.connection.sourceType }} -> {{ outputs.sync.connection.destinationType }}
streams: {{ outputs.sync.streamSummary.streams }} ({{ outputs.sync.streamSummary.bytesEmitted }} bytes)
drift: {{ outputs.sync.connectionDriftSummary }}
per-stream table: {{ outputs.sync.streams }}
```

## Acceptance Criteria

- [ ] `connections.Sync` persists an allow-listed connection snapshot before triggering and exposes `connection`, `connectionSnapshot` (URI), `jobUrl`; snapshot failure only warns
- [ ] `connections.CheckStatus` / `Sync(wait: true)` expose inline `job` (status, timestamps, attemptCount, totals, durationMs), `streams` (URI), `streamSummary`, `attempts` (URI), and one dynamic task run per attempt with correct state history
- [ ] `cloud.jobs.Sync` / `Reset` expose `connection`, `connectionSnapshot`, `job`, `jobUrl`; documented that per-stream / per-attempt stats are self-hosted only
- [ ] `compareWithPreviousExecution: true` produces `connectionDriftSummary` (added / removed / changed streams, changed sync modes); empty and non-failing when no previous snapshot exists
- [ ] One `io.kestra.plugin.ee.assets.Table` asset per selected stream with resolved destination namespace (`source` / `destination` / `customformat`) and source `External` inputs as lineage; gated by `assets.enableAuto`; no-op on OSS
- [ ] Snapshot and outputs never contain `connectionConfiguration`, tokens or passwords (test with a mocked `web_backend/connections/get` response containing credentials)
- [ ] Failure reasons keep `externalMessage` only, truncated; no stack traces in outputs
- [ ] Existing metrics unchanged; new `duration`, `records.dropped`, `attempts.failed` added and declared in `@Plugin(metrics = ...)`
- [ ] New properties use `Property`; logging via `runContext.logger()`; JSON via `io.kestra.core.serializers.JacksonMapper`; ION via `FileSerde`
- [ ] Unit tests with recorded API fixtures (self-hosted `jobs/get` with 2 attempts and 3 streams; cloud `jobs/{id}`); integration test against the existing Airbyte test container if present
- [ ] Follow-up issue filed for the `ui/` module (streams table + chart, attempts timeline, logs button 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/0d995593-b756-4d87-af9f-3860868f74be)*