Guida allo sviluppo

Come si scrive una pipeline in un repo cliente: le cartelle, il catalogo, l'API pubblica del core, i comandi etl e il kit dell'agente. Per la visione di prodotto vedi la panoramica; per il percorso con Claude Code, dal bisogno al processo.

1. Cos'è in breve

Noeva ETL è un motore ETL code-first: le pipeline di integrazione dati (estrazione → trasformazione → caricamento) sono scritte in TypeScript, testate con vitest e documentate — non job grafici stile Talend, ma software vero.

Il motore è il core, il pacchetto @noeva/etl-core: nodi, connettori, transform, la CLI etl e questa guida. Le pipeline di un cliente vivono nel suo repo cliente, noeva-etl-<slug>, che dipende dal core e ne usa solo l'API pubblica. Un deploy esegue un repo cliente: le sue pipeline, con la versione del core che il repo dichiara.

Il servizio è stateless: non ha un proprio database, esegue una pipeline e restituisce un GraphRunReport (JSON). La CLI etl si avvia in tre modi:

• etl serve — il modo di un deploy: nello stesso processo espone i grafi via HTTP, per l'app desktop, e avvia i worker. Vedi il runbook di etl serve.
• etl worker — solo il daemon: resta in polling sulla tabella processes di Noeva, sempre attraverso noeva-server-api, ed esegue le pipeline dei job che Noeva crea (lanci dall'app, job pianificati). Mai insieme a un etl serve nello stesso ambiente, o i job vengono presi due volte. Vedi il runbook del worker.
• etl run <pipeline> — batch: una esecuzione, report su stdout, poi termina. Con --trace è una run tracciata, che salva anche la traccia dei record (La traccia di una run).

In tutti e tre i casi il nucleo che esegue la pipeline è lo stesso: cambia solo chi la fa partire e dove finisce il report.

2. Il repo cliente

Un repo cliente ha la stessa forma per ogni cliente:

etl.config.ts         # il catalogo: il cliente e le sue pipeline
src/
  pipelines/<nome>/   # una cartella per pipeline: schema, grafo, changelog, snapshot del piano, run, test
  sub-pipelines/<nome>/ # i pezzi di grafo riusati da più pipeline
  connectors/         # i connettori che conoscono il cliente, se servono
docs/                 # la documentazione del cliente, in solo markdown
  pipelines/          # una scheda per pipeline
  sub-pipelines/      # una scheda per sub-pipeline
  connectors/         # le schede dei connettori del cliente
  runbooks/           # i runbook delle sue pipeline
  architecture/adr/   # le sue decisioni, con numerazione propria
.env.example          # le istanze di connettore e la configurazione delle pipeline
AGENTS.md             # le istruzioni per l'agente, con il blocco del kit
.github/workflows/ci.yml  # la CI: il setup, poi `etl ci`
.github/workflows/image.yml  # l'immagine del deploy, a mano o su tag
ecosystem.config.cjs  # una riga: il deploy sotto PM2

Crearlo. etl init <slug> scrive nella cartella corrente il boilerplate del pacchetto, poi lancia git init con un primo commit e crea i branch dev, staging e main, con HEAD su dev. La cartella dev'essere vuota: altrimenti il comando si ferma con exit 2 senza scrivere niente, e lo stesso vale per uno slug che non è fatto di minuscole, cifre e trattini. Senza un'identità di git si ferma prima di scrivere, con exit 1. Non crea il repo su GitHub né il secret del PAT, che restano a mano. Il boilerplate è magro, perché ciò che deve evolvere arriva per dipendenza:

File	Cosa contiene
package.json	noeva-etl-<slug>, ESM, il core alla versione che genera il repo, il pnpm del core (fissato nel codice di etl init), gli alias pipeline:* e la toolchain nelle devDependencies
etl.config.ts	client e il catalogo vuoto
.env.example	le sole variabili obbligatorie del core (NOEVA_API_BASE_URL, NOEVA_API_KEY, NOEVA_WORKSPACE_ID)
.npmrc	il registry di @my-ginkgo su GitHub Packages, senza token
tsconfig.json, eslint.config.js, vitest.config.ts	una riga sul preset
.gitignore	.env, .runs/, .state/, logs/ di PM2, il sito costruito, la sua cache e i file temporanei di VitePress, docs/catalogo.md
.prettierignore	il lockfile e gli snapshot del piano, che etl versions snapshot scrive con la sua formattazione
src/pipelines/, docs/pipelines/	vuote: le cartelle delle pipeline e le loro schede
AGENTS.md, CLAUDE.md	le istruzioni per l'agente, con il blocco gestito dal core; CLAUDE.md è solo @AGENTS.md
.claude/	file gestiti: il kit dell'agente in skills/, agents/ e hooks/ noeva-etl-*, e la voce dell'hook in settings.json (§10)
.github/workflows/ci.yml	file gestito: checkout con fetch-depth: 0, pnpm e node, install col PAT nel secret NOEVA_ETL_PACKAGES_TOKEN, poi pnpm exec etl ci
.github/workflows/image.yml	file gestito: l'immagine col Dockerfile del pacchetto, su ghcr.io/<owner>/noeva-etl-<slug> con tag = SHA, a mano o su tag; il PAT come BuildKit secret
ecosystem.config.cjs	file gestito: una riga su ecosystem() di @noeva/etl-core/pm2, che avvia etl serve sotto PM2

La toolchain sta nelle devDependencies del repo: vitest, eslint, typescript, prettier e @types/node, con i range che etl init legge dal package.json del core installato. Con pnpm le dipendenze del core non si vedono dal repo, e senza vitest e @types/node installati lì tsc --noEmit non trova i tipi dei test; con gli stessi range pnpm tiene una sola copia per il core e per il repo. Li riallinea etl upgrade a ogni bump, coi range del core nuovo.

La CI. ci.yml non usa workflow riusabili dell'org, così regge al transfer del repo, e gira sulle PR verso dev e main e sui push su main. Il secret NOEVA_ETL_PACKAGES_TOKEN del repo cliente è un PAT classic read:packages dell'utente invitato, che serve solo a installare il core. Nel repo del core lo stesso nome indica un PAT write:packages, che pubblica il pacchetto: non è lo stesso token.

I file gestiti. Il blocco di AGENTS.md, fra <!-- noeva-etl:inizio … --> e <!-- noeva-etl:fine -->, lo scrive etl agent sync; il resto del file è del repo. Dice all'agente di lanciare etl upgrade --check prima di ogni nuovo sviluppo, di leggere il changelog del core prima di ogni bump, e porta le regole su doc nello stesso commit, --realign e i tre rollback. I file gestiti che etl init scrive sono i due YAML (ci.yml e image.yml), ecosystem.config.cjs, il kit dell'agente in .claude/{skills,agents,hooks}/noeva-etl-* con la voce dell'hook in .claude/settings.json (§10), e il blocco di AGENTS.md; tutto il resto è del repo. Un file gestito che il repo vuole tenersi si dichiara in etl.config.ts con unmanaged: [".github/workflows/ci.yml"] (per il kit, la cartella di una skill, il file di un subagent, AGENTS.md per il blocco o .claude/settings.json per la voce dell'hook), e da lì l'aggiornamento è del repo. image.yml costruisce l'immagine del deploy col Dockerfile del pacchetto (Deploy § Immagine del repo cliente), e ecosystem.config.cjs è la riga che avvia etl serve sotto PM2 (Aggiornare un deploy PM2).

Il catalogo. etl.config.ts esporta defineEtlConfig({ client, pipelines }): client è lo slug del cliente, pipelines le pipeline importate come valori. Ogni comando che lavora sulle pipeline (list, plan, run, state, versions, env, catalog, docs, ci, serve, worker) legge questo file: la CLI lo cerca risalendo dalla cartella corrente, e --config <file> lo indica a mano. Il TypeScript gira dai sorgenti con tsx, quindi il package.json del repo dichiara "type": "module". Un config assente o che non si importa, un export di forma sbagliata, una voce che non è una RunnablePipeline, un nome ripetuto (anche fra pipeline e sub-pipeline) o un client che non è uno slug fermano il comando con exit 2 e un errore che nomina la causa; un catalogo vuoto è permesso, con un avviso. Grafi, dispatch del worker e annuncio del catalogo a noeva-server vedono le stesse pipeline: non c'è un registro a parte.

// etl.config.ts
import { defineEtlConfig } from "@noeva/etl-core";
import { testPosta } from "@noeva/etl-core/pipelines";
import { fattureDaily } from "./src/pipelines/fatture-daily/run.js";

export default defineEtlConfig({
  client: "acme",
  pipelines: [fattureDaily, testPosta],
});

Le pipeline pronte arrivano dal core: si dichiarano nel catalogo come le altre, importate da @noeva/etl-core/pipelines. Hanno versione, scheda e scopi nel core, quindi il repo cliente non le versiona e non le documenta: etl list le marca con pronta, e l'indice del catalogo del cliente le linka nella sezione «Pipeline pronte» di questa guida.

Il nome di una pipeline combacia con processes.function e con scheduled_jobs.process_function: il worker risolve per nome, senza traduzione.

3. L'API pubblica

Un repo cliente importa solo dal pacchetto e dai suoi subpath, mai da percorsi interni: è l'unico confine, e i tipi del core arrivano con il pacchetto, nelle sue .d.ts.

Import	Cosa contiene
@noeva/etl-core	il DSL graph(), i tipi dei nodi, le transform, gli errori, z di Zod, defineEtlConfig
@noeva/etl-core/connectors/<nome>	un connettore: excel, file, http, imap, jira, mail, mysql, noeva, oauth, odoo, postgres, process, sendgrid, smtp, sqlserver, zoho
@noeva/etl-core/connector-kit	gli strumenti per scrivere un connettore o leggere l'ambiente di una pipeline
@noeva/etl-core/testing	i doppi in memoria dei sistemi esterni (IMAP, Noeva, SendGrid), per i test
@noeva/etl-core/pipelines	le pipeline pronte
@noeva/etl-core/contract	i tipi del contratto fra etl serve e l'app desktop
@noeva/etl-core/tsconfig, /eslint, /vitest	i preset della toolchain
@noeva/etl-core/pm2, /deploy/Dockerfile	le confezioni del deploy: la configurazione PM2 e il Dockerfile dell'immagine

Due regole valgono per ogni pipeline:

• z arriva dalla radice del pacchetto, non da zod: così i contratti usano l'istanza e la versione del core.
• Un driver passa dal suo connettore: una pipeline non importa pg, mssql o mysql2 direttamente.

I preset della toolchain rendono di una riga i tre file di configurazione del repo:

// tsconfig.json
{ "extends": "@noeva/etl-core/tsconfig" }

// eslint.config.js
export { default } from "@noeva/etl-core/eslint";

// vitest.config.ts
export { default } from "@noeva/etl-core/vitest";

./tsconfig ha moduleResolution: bundler, che legge i subpath della mappa exports, e noEmit; ./eslint mette no-explicit-any e no-console a error; ./vitest cerca i test in src/ e tests/, senza globali e senza soglia di coverage.

4. Le primitive

Tutto ruota attorno a quattro concetti base, mappati ai componenti Talend:

Primitiva	Cosa fa
Source<T>	estrae record da un sistema esterno come AsyncIterable<T> (streaming, mai tutto in memoria)
Transform<I,O>	trasforma uno stream I in uno stream O; pura, senza side-effect esterni
Sink<T>	carica i record su un sistema di destinazione; deve essere idempotente (upsert/MERGE)
GraphPipeline	il processo: un grafo di nodi costruito col DSL graph(name)…build()

Il grafo è l'unico modo di costruire un processo. Anche una catena semplice è un grafo con un nodo per stadio:

graph("orders-daily")
  .source("orders", source)
  .transform("orders", "raw-contract", validate("raw-contract", rawSchema), {
    deadLetter: rejects,
  })
  .transform("raw-contract", "to-fact", mapTransform("to-fact", toFact))
  .transform("to-fact", "fact-contract", validate("fact-contract", factSchema))
  .sink("fact-contract", sink)
  .build();

Gli undici tipi di nodo. A questi si aggiunge il nodo pipeline, .pipeline(from, id, sub, { params?, deps?, fromPort?, deadLetter?, purpose? }), che istanzia una sub-pipeline e alla costruzione diventa i suoi nodi:

Nodo	Metodo	Cosa fa
source	.source(id, source)	punto d'ingresso di uno stream
transform	.transform(from, id, t, opts)	trasformazione pura; accetta deadLetter
lookup	.lookup(from, id, config, opts)	lettura in sola lettura in mezzo al grafo, a batch, da un DB o un'API REST; accetta deadLetter e recordReport
ai	.ai(from, id, config, opts)	domanda a un modello per ogni record, con la risposta che arricchisce il record (es. noevaAiExtract); si esegue come un lookup; accetta deadLetter e recordReport
action	.action(from, id, action, opts)	scrittura per record in mezzo al grafo, con un record in uscita su cui il grafo prosegue; accetta deadLetter e recordReport
route	.route(from, id, { routes }, opts)	smistamento esclusivo: ogni record esce sulla porta della prima condizione vera, o su default; ogni porta va collegata; accetta deadLetter
join	.join(id, { build, probe, … })	fan-in a chiave: join inner/left, lato build in memoria e probe in streaming; accetta deadLetter
union	.union(id, { from, order })	fan-in di N flussi con la stessa forma, record invariati: interleave (default) nell'ordine di arrivo, concat nell'ordine di from
try-catch	.tryCatch(from, id, opts)	apre uno scope di gestione errori sui nodi a valle, con una porta catch
sink	.sink(from, sink)	scrittura verso l'esterno; l'id del nodo è sink.name
task	.task(id, task, opts)	operazione senza record (truncate, scambio di tabelle), avviata dagli edge di controllo after

Il cheatsheet dei nodi, con una scheda di dettaglio per ciascuno (opzioni, errori, esempi), è in Tipi di nodo.

Il fan-out invece è libero: basta agganciare più nodi allo stesso from, e il runner duplica lo stream. La forma del grafo è validata a runtime su .build(): cicli, id sconosciuti o più di un ingresso su un nodo diverso da join e union esplodono come GraphValidationError alla costruzione. Esempio di riferimento completo: orders-enriched-daily.

Oltre agli edge di dati ci sono gli edge di controllo: l'opzione after di .task() e .source() dice «parti solo dopo che questi nodi sono riusciti». Partono da un task o da un sink e arrivano a un task o a una source. Dettagli nella scheda task.

Ogni nodo ha uno scopo, purpose, scritto per chi legge il grafo nell'app: che cosa fa il nodo e perché serve, in una-tre frasi di business. Lo verificano etl catalog check ed etl docs build.

Il RunContext che ogni nodo riceve porta env, parametri di lancio (ctx.params), logger, metriche, il segnale d'arresto e lo store dello stato di pipeline. Gli errori di dominio sono EtlError (ValidationError, ExtractError, LoadError, ConfigError, GraphValidationError, JoinError, LookupError, ActionError, TaskError, AssertionError, GraphRunFailure): un errore che non lo è conta come bug.

5. Come gira una pipeline

Il grafo è pull-based: la domanda parte dai sink e risale verso le source, quindi è il ritmo di scrittura a regolare quello di lettura (backpressure gratis). I record scorrono in streaming, mai tutti in memoria — tranne il lato build di un join e le transform che trattengono lo stream, come sort, che hanno un tetto obbligatorio.

Quando un record fallisce ci sono tre percorsi, dal più locale al più drastico:

• Dead-letter per nodo — { deadLetter: sink } su un transform, lookup, ai, action, route o join. Il record che solleva EtlError esce come envelope DeadLetter su quel sink e lo stream continua. Adatto alle trasformazioni stateless (map/mapColumns/flatMap/explode/unpivot/filter/validate, parse, hashRow, surrogateKey, mask). Sul .pipeline() della root la deadLetter si distribuisce a ogni nodo isolabile dell'istanza, istanze annidate comprese (deadLetter sull'istanza).
• Nodo try-catch — protegge tutti i nodi raggiungibili dalla sua porta out, sink compresi. Di default è per-record: ogni record che fallisce va sul ramo catch, che è fatto di nodi veri e può loggare, rimappare o instradare altrove. In circuit-breaker al primo errore la sub-stream si ferma.
• Fail-fast — il default, senza dead-letter né guard: l'errore aborta la run. È l'unica scelta sensata per le trasformazioni stateful (es. groupBy, sort, serialize): un'aggregazione spezzata per record darebbe un risultato falso. Vale per tutto il grafo: al primo errore fatale il runner annulla anche i rami indipendenti e aspetta che ogni sink e ogni task chiuda prima di riportare la run fallita. Un task che fallisce è sempre fatale.

Un errore che non è un EtlError (un TypeError, un bug) aborta sempre la run, anche dentro un guard: un bug non va scaricato in un sink di scarto, va visto. Qualunque sia la scelta, nessun record sparisce in silenzio: ogni scarto è contato nel report.

Ogni run() produce un GraphRunReport:

• stato — success, partial (arrivata in fondo con scarti o errori catturati) o failed;
• totali — extracted (somma delle source), written (somma dei sink: su un fan-out lo stesso record scritto due volte conta due), rejected, rejectedReasons;
• per nodo — nodes[] ed edges[] con la topologia reale e i contatori di ogni nodo, caught sui guard, tempi di attraversamento;
• stato di pipeline — state, una voce per source incrementale: da dove ha letto, cosa propone, se il watermark è stato committato;
• metriche ed errore — mai segreti o payload.

È il "prodotto" del servizio: chi invoca la pipeline decide se archiviarlo, alertare, ecc. Schema completo in Il report di run.

6. Cosa c'è già pronto

Connettori, uno per subpath di @noeva/etl-core/connectors/:

• PostgreSQL (postgres) — source con cursor streaming, sink con upsert ON CONFLICT o in modalità SCD2 (dimensioni storicizzate), task SQL postgresSqlTask (statement in una transazione, per il nodo task).
• SQL Server (sqlserver) — source streaming, sink con MERGE o in modalità SCD2, task SQL mssqlSqlTask.
• MySQL e MariaDB (mysql) — source streaming, sink con ON DUPLICATE KEY UPDATE, lookup mysqlLookup, task SQL mysqlSqlTask (senza SCD2). Date e BIGINT arrivano come stringhe.
• Excel (excel) — workbook .xlsx/.xlsm che si leggono e si modificano senza perdere ciò che non si è scritto: excelSource (uno o più fogli, a flusso), excelSink (append, upsert per chiave, replace; fogli nuovi, formati, formule, modelli) ed excelTask (fogli e colonne), più la transform excelValidate. Solo storage filesystem per scrivere. Vedi la scheda e l'ADR-0110.
• File (file) — architettura a due assi componibili:
  • storage: dove vivono i file (locale, S3, SFTP, FTP/FTPS);
  • codec: come si (de)serializzano — CSV (anche grezzo, csvRawCodec), JSON/NDJSON, XML, XLSX, Parquet. Storage e codec si combinano liberamente (es. CSV su S3, Parquet su SFTP).
• Noeva (noeva) — file dei workspace Noeva via noeva-server-api: lettura di un file o di una cartella intera, scrittura di uno o più file in overwrite o new-version, upload con elaborazione AI (noevaUploadAction), lookup su stato e testo estratto ed estrazione AI di campi strutturati (noevaAiExtract, nodo ai).
• IMAP (imap) — caselle di posta: imapSource legge i messaggi di una cartella (allegati ed EML grezzo), le action imapFlagAction e imapSpostaAction marcano e spostano, con la cartella come coda.
• SMTP (smtp) — notifiche email: smtpAction e smtpSink inviano una mail costruita da una funzione del record, con Message-ID deterministico contro i doppioni.
• SendGrid (sendgrid) — mail via API REST v3: sendgridAction e sendgridSink, con la stessa mail di SMTP (MailInUscita), sandbox per le prove, retry di 429 e 5xx; mailAction e mailSink (mail) scelgono fra SMTP e SendGrid da configurazione.
• HTTP (http) — API REST per i nodi lookup e action: httpLookup legge in sola lettura, una richiesta per batch o una GET per chiave, httpAction scrive e prosegue con l'esito; la risposta è validata da uno schema Zod e le credenziali vengono dal registry HTTP_INSTANCES. Il loro trasporto, richiestaHttp, è condiviso con i connettori REST di dominio, come Jira.
• Jira Cloud (jira) — jiraIssues legge le issue di una JQL, pagina dopo pagina, con i siti nel registry JIRA_INSTANCES; i campi si dichiarano per id o per nome e diventano colonne normalizzate, con gli ADF convertiti in testo. jiraIssueLookup legge le stesse issue per chiave, e jiraChangelogLookup, jiraCommentsLookup e jiraWorklogsLookup danno a ogni issue la sua lista completa. jiraCreateIssue, jiraUpdateIssue, jiraTransition, jiraUpsertIssue (per ID esterno), jiraAddComment e jiraAddWorklog (con una chiave di deduplica obbligatoria) e jiraLinkIssues scrivono come nodi action, con l'esito come dato.
• Zoho CRM (zoho) — zohoCrmRecords legge i record di un modulo con i soli campi dichiarati, che sono anche la regola di privacy, in ordine di id: con la Records API fino a 100.000 record, con la Query API (COQL) a chiavi crescenti di id oltre. Le organizzazioni stanno nel registry ZOHO_INSTANCES, il token lo rinnova un'identità OAuth2 e il trasporto è quello condiviso di http.
• Odoo (odoo) — odooRecords legge i record di un modello con i soli campi dichiarati, a pagine in ordine di id, e odooConteggi i search_count da confrontare con le righe caricate. Solo metodi di lettura, al più una chiamata al secondo per database e mai due in parallelo (Acceptable Use Policy di Odoo); XML-RPC oggi, JSON-2 con Odoo 19. I database stanno nel registry ODOO_INSTANCES.
• Processo esterno (process) — processoEsternoAction lancia un programma per record, senza shell, in una cartella temporanea sempre cancellata: i file del record entrano, i file di uscita/ e un riepilogo JSON validato con Zod escono. Ambiente da allowlist e timeout obbligatorio.
• OAuth2 (oauth) — non un connettore ma l'identità che IMAP e SMTP usano dove la password non basta più (Gmail, Workspace, Microsoft 365), e con cui Zoho CRM autentica le sue richieste: tiene il refresh token e ne scambia uno nuovo quando l'access token scade, con un registry proprio OAUTH_INSTANCES.

Le schede, con configurazione e variabili d'ambiente di ognuno, sono in Connettori.

Trasformazioni riusabili, dalla radice del pacchetto: groupBy (aggregazione stateful generica init/reduce/finalize), dedup (duplicati per chiave), mapColumns (select, rename, cast e colonne calcolate, con lineage di colonna) e castValue (i suoi cast stretti, usabili da soli), explode (un campo array → un record per elemento), unpivot e pivot (dal formato largo al lungo e ritorno, con colonne dichiarate), sort (ordinamento stabile su più chiavi), chunk (record in lotti di N per un sink o un'API con limite di payload), parse (byte di un file → righe, con un FileCodec) e serialize (record → un file per gruppo, stateful), hashRow e surrogateKey (hash di riga e chiave surrogata deterministici), mask (mascheramento e pseudonimizzazione delle colonne con dati personali), profile (profilo delle colonne nel report del nodo, a memoria costante), assert (controlli sul dataset a fine stream). Il catalogo completo è in transform.

Pipeline d'esempio, con la loro scheda nella sezione «Esempi» di questa guida:

• orders-enriched-daily — dataset demo in memoria, esempio di riferimento per una pipeline a grafo: due sorgenti, join a chiave, fan-out su due sink, dead-letter isolato su due nodi indipendenti, try/catch e un'istanza della sub-pipeline riepilogo-ordini.
• movimenti-per-conto — dataset demo in memoria, esempio di riferimento per le transform di file e di ordinamento in mezzo al grafo: parse, sort e serialize, da estratti conto CSV a un CSV per conto.
• crm-leads-sync — copia fra due istanze Postgres dello stesso registry, con un try/catch per record.
• cross-workspace-csv-merge — merge di un CSV fra due workspace Noeva, con un union e la scrittura in new-version su entrambi.

Pipeline pronte, da dichiarare nel catalogo: test-posta, la prova dei connettori IMAP e SMTP di un'installazione.

7. I comandi

pnpm exec etl init <slug>              # in una cartella vuota: il boilerplate di un repo cliente
pnpm exec etl ci                       # il gate completo del repo, lo stesso della CI
pnpm exec etl agent sync               # riscrive il kit dell'agente dal core installato (§10)
pnpm exec etl upgrade --check          # versione installata e ultima, changelog e deriva dei file gestiti
pnpm exec etl upgrade                  # il bump del core e il sync dei file gestiti
pnpm exec etl upgrade --sync           # solo il sync dei file gestiti col core installato
pnpm etl list                          # le pipeline del catalogo; le pronte marcate con «pronta»
pnpm etl plan                          # la topologia di ogni pipeline (JSON, ultima riga)
pnpm etl run <pipeline>                # esegue una pipeline
pnpm etl run <pipeline> --trace        # run tracciata: il percorso dei primi record
pnpm etl runs list                     # lo storico locale delle run
pnpm etl state show <pipeline>         # i watermark delle source incrementali
pnpm etl env check                     # le variabili d'ambiente mancanti, con la provenienza
pnpm etl catalog check                 # la conformità del catalogo
pnpm etl versions snapshot <pipeline>  # lo snapshot del piano della versione corrente
pnpm etl versions check                # il gate delle versioni
pnpm etl docs build                    # il gate della documentazione, poi il sito del cliente
pnpm etl docs dev                      # lo stesso sito in sviluppo
pnpm etl serve                         # grafi via HTTP + worker
pnpm etl worker                        # solo il daemon su processes

etl ci è il gate del repo, lo stesso in locale e nella CI. Esegue, in quest'ordine, tsc --noEmit, ESLint (lint), Prettier in --check (format), vitest con --passWithNoTests (test), etl versions check, etl catalog check ed etl docs build. Su stderr ogni fase si annuncia con ▶ <fase>, e la prima che fallisce ferma il gate: etl ci: la fase «<fase>» è fallita, con l'exit code della fase. Gli strumenti sono quelli installati nel repo, o in mancanza quelli fra le dipendenze del core; i comandi etl girano in processo sul catalogo già caricato. Un catalogo vuoto passa, con l'avviso del caricamento, e un repo appena creato senza test pure; senza origin/dev, prima del primo push, il gate delle versioni dà l'avviso «nessuna base» e controlla solo gli snapshot (vedi Versioni di pipeline). Quali fasi girano arriva con il bump del core, mentre ci.yml resta fermo. Prima delle fasi etl ci guarda l'ultima versione del core sul registry: un core non aggiornato è un avviso (Avviso: il core installato è …) e mai un errore, così un commit vecchio non diventa rosso perché è uscita una release; anche un registry che non risponde è solo un avviso.

etl upgrade --check è in sola lettura, e il blocco di AGENTS.md lo chiede prima di ogni nuovo sviluppo. Su stdout dà la versione del core installata e l'ultima sul registry, le sezioni del CHANGELOG.md del core fra le due, lette dal tarball della versione nuova senza installarla, con le major ([MAJOR]) e le voci «Cambia il piano» ([CAMBIA IL PIANO]) in evidenza in testa, e la deriva dei file gestiti: modificato (diverso dal template del core installato), mancante o in più (un file sotto noeva-etl-* che il kit non porta), con saltato per quelli in unmanaged. Per AGENTS.md e .claude/settings.json conta solo la parte gestita, il blocco e la voce dell'hook. Chi ha installato una versione rilasciata segue il dist-tag latest; chi ha una prerelease next segue la più recente fra latest e next. Il registry è npmjs, o quello di npm_config_registry. Esce con 0, o con 1 se il registry non risponde.

etl upgrade fa il bump. Si ferma con exit 1 su un file gestito modificato o in più, che l'aggiornamento sovrascriverebbe, e lo nomina; procede con --force, e le modifiche si perdono. Un file in unmanaged non lo ferma e non si riscrive. Poi stampa il changelog, porta @noeva/etl-core all'ultima nel package.json, col prefisso che il range aveva (^, ~ o un pin esatto), e la toolchain delle devDependencies ai range del core nuovo, lancia pnpm install e infine la CLI del core appena installato (pnpm exec etl upgrade --sync --force), che risincronizza i file gestiti col suo template: i due YAML, ecosystem.config.cjs e il kit come etl agent sync. Se il changelog dice «Cambia il piano» suggerisce pnpm exec etl versions snapshot --realign --ticket <chiave>. Con il core già all'ultima versione risincronizza soltanto i file gestiti; etl upgrade --sync fa lo stesso senza andare sul registry, per esempio dopo un bump fatto a mano con pnpm, e si ferma come l'upgrade su un file toccato a mano, salvo --force. Package.json, lockfile e file gestiti si committano insieme, in una PR del bump. Un core dichiarato con file:, link: o workspace: si aggiorna a mano.

etl env check verifica l'ambiente sull'intero catalogo: su stdout una riga per variabile obbligatoria mancante, <VARIABILE>\t<provenienze>, dove la provenienza è core, pipeline <nome> o istanza <connettore> <id> (<pipeline che la usano>); su stderr un avviso per ogni chiave del .env che nessuno dichiara (solo il nome, mai il valore) e il riepilogo. Esce con 1 se manca un'obbligatoria, 0 altrimenti, anche con sconosciute. Le variabili del core sono in Variabili d'ambiente del core; quelle di una pipeline le dà il suo environment(env), scritto con ambienteDellaPipeline di ./connector-kit dagli stessi manifest della run (vedi test-posta). Senza, il check ricade sul piano, che non mostra tutte le variabili di un'istanza (l'utente di un DB, l'identità OAuth2), e quelle variabili, se stanno nel .env, compaiono fra le sconosciute. Legge l'ambiente del processo, con il .env alla radice del repo che la CLI carica da sola prima di ogni comando (le variabili già nel processo vincono). Resta fuori dalla CI: serve l'ambiente vero.

etl catalog check è la conformità del catalogo: gli stessi controlli in ogni repo, che arrivano aggiornati con il bump del core. Su stdout una riga per violazione, <regola>\t<pipeline>: <dettaglio>; su stderr il riepilogo e gli avvisi del caricamento. Esce con 0 se il catalogo è conforme (anche vuoto, con l'avviso), 1 con almeno una violazione, 2 per un errore d'uso o di configurazione. Le regole:

Regola	Cosa verifica
changelog	numeri da 1 senza buchi, date AAAA-MM-GG in ordine, ticket, voci non vuote
piano-senza-credenziali	getSpec() senza .env non lancia, il piano ha il nome della pipeline, almeno una source e un sink, si serializza in JSON e i nodi stanno nella proiezione
scopo-dei-nodi	ogni nodo e ogni istanza ha uno scopo; esenti il collettore dei dead-letter e i nodi di confine di un'istanza, che genera il core
piano-senza-segreti	con l'ambiente di .env.example e un valore riconoscibile al posto di ogni segreto il piano si disegna dalla configurazione risolta e non contiene segreti; poi lo stesso con le credenziali alternative
cartella-orfana	ogni run.ts sotto src/pipelines/ esporta una pipeline del catalogo: una pipeline tolta da etl.config.ts non resta senza controlli

L'ambiente di prova di piano-senza-segreti è il .env.example alla radice del config, più le variabili segrete che le pipeline dichiarano con environment(env). È segreta una variabile il cui nome contiene PASSWORD, SECRET, TOKEN, API_KEY, ACCESS_KEY o PRIVATE_KEY. Le credenziali commentate (# IMAP_ACCESS_TOKEN=) sono alternative: il secondo passaggio le attiva e toglie quella attiva con lo stesso prefisso (IMAP_PASSWORD). Per questo .env.example deve essere completo: una pipeline che con quelle variabili si disegna dall'esempio (un valore con la parola «esempio» nel piano) è una violazione, perché il controllo sui segreti guarderebbe solo i segnaposto.

etl versions tiene le versioni di pipeline: ogni cambio di piano è una versione nuova, con la sua voce nel changelog e lo snapshot del piano. Il gate, il rollback di una pipeline sola e il riallineamento al core sono in Versioni di pipeline e nel runbook del rollback.

etl docs build costruisce la documentazione del cliente, con configurazione, tema e componenti dei grafi dal pacchetto: in un repo cliente docs/ è solo markdown. Il sito va in docs/.vitepress/dist (--outDir per un'altra cartella), col nome del cliente nel titolo, e comprende tutto docs/, ADR compresi, più una pagina indice del catalogo, docs/catalogo.md, che il build genera e che non si versiona. etl docs dev serve lo stesso sito in sviluppo (--port <n>). Il sito costruito lo serve etl serve su /guida/, accanto alla guida del core su /guida/core/ (runbook del server). Prima di costruire il comando controlla la documentazione, e con un problema si ferma: una riga per problema su stdout, <regola>\t<dove>: <dettaglio>, ed exit 1.

Regola	Cosa verifica
scheda-mancante	ogni pipeline del catalogo ha una pagina con <PipelineDag name="…" />, ogni sub-pipeline che una di loro istanzia una con <SubPipelineDag>
scopo-mancante	ogni nodo e ogni istanza ha uno scopo, come scopo-dei-nodi di etl catalog check
link-rotto	ogni link porta a una pagina pubblicata o a un file di docs/; un link fuori da docs/ è rotto, perché non si pubblica
core-non-risolto	un link core:nodes/lookup.md porta a una pagina della guida del core installata
catalogo-del-repo	docs/catalogo.md è quella generata: una scritta a mano non si sovrascrive

Le pipeline pronte sono esenti: la loro scheda sta in questa guida, e nell'indice del catalogo sono una riga col link qui. Un link core: diventa /guida/core/<pagina>, la guida del core servita dallo stesso deploy; si valida contro le pagine della guida nel pacchetto installato, quindi se un bump rinomina una pagina il build del cliente fallisce. Il piano si disegna senza l'ambiente del processo, come lo snapshot.

8. Aggiungere le cose più comuni

Una nuova pipeline

Una cartella src/pipelines/<nome>/ con: schema.ts (contratti Zod input/output), pipeline.ts (il grafo con graph()...build(), sorgenti e sink iniettati, non scritti dentro), changelog.ts (le versioni di pipeline, una nuova parte dalla v1), run.ts (il collegamento dei connettori concreti e la lettura dell'ambiente), pipeline.test.ts (end-to-end con arraySource/arraySink: record validi caricati, invalidi al dead-letter, metriche corrette). Se legge variabili d'ambiente, il suo run.ts dichiara anche environment(env) con ambienteDellaPipeline, dagli stessi manifest e istanze della run, così etl env check le vede tutte. Le sue variabili obbligatorie vanno in .env.example, che è anche l'ambiente di prova di etl catalog check. Poi si aggiunge al catalogo in etl.config.ts, se ne scrive lo snapshot del piano con pnpm etl versions snapshot <nome> (versions/v1.plan.json) e si crea la scheda docs/pipelines/<nome>.md (vedi Le schede di una pipeline).

Una nuova sub-pipeline

Un pezzo di grafo riusato da più pipeline, o più volte nella stessa, è una sub-pipeline: src/sub-pipelines/<nome>/ con schema.ts (contratti d'ingresso e d'uscita, e lo z.object dei parametri se ne ha), sub-pipeline.ts (subPipeline({ … }), con params per i dati che ogni istanza fissa e deps per i connettori che riceve dalla root) e sub-pipeline.test.ts (con runSubPipeline, recordIds acceso e spento). Un parametro compare in chiaro nel piano: le credenziali arrivano solo dentro una dipendenza. Lo scopo dei nodi del corpo si scrive nella definizione, quello dell'istanza al punto di chiamata (Scopo). Non si dichiara nel catalogo: arriva insieme alla prima root che la istanzia con .pipeline(). Scheda in docs/sub-pipelines/<nome>.md (vedi Le schede di una sub-pipeline).

Un connettore del cliente

Un sistema che conosce solo il cliente (il suo gestionale, la sua API) è un connettore del repo cliente, in src/connectors/: implementa Source<T>, Sink<T> o l'interfaccia di un lookup o di un'action, con gli strumenti di @noeva/etl-core/connector-kit, e ha la sua scheda in docs/connectors/. Un connettore che serve a più clienti va nel core.

Una nuova trasformazione

Funzione pura, senza side-effect verso sistemi esterni (quelli stanno solo nei nodi che li dichiarano), così è testabile in isolamento. Prima di scriverne una, guarda le transform pronte. Se è stateful (come groupBy), va usata in fail-fast: niente dead-letter sul suo nodo (vedi §5).

9. Cosa NON fare (vincoli non negoziabili)

• Niente import dai percorsi interni del core — solo il pacchetto e i suoi subpath (§3).
• Niente any — dati esterni entrano come unknown, si restringono con Zod.
• Niente IO dentro una Transform — per una lettura c'è il nodo lookup, per una scrittura il nodo action, per un'operazione senza record il nodo task.
• Niente scarti silenziosi — ogni record invalido ha un percorso esplicito: dead-letter o fail-fast, mai un semplice skip.
• Sink non idempotenti — un retry/riesecuzione non deve mai duplicare dati (upsert su chiave naturale, MERGE, o write-then-swap, vedi il runbook).
• Niente segreti nel codice, nei log o nel report — solo variabili d'ambiente, documentate in .env.example.
• Codice senza test o senza scheda aggiornata non è "finito" — codice e documentazione cambiano nello stesso commit, e etl docs build lo verifica.

10. Il kit dell'agente

Il repo cliente porta il kit con cui Claude Code, o un altro agente, lavora secondo le regole del core: la skill etl-graph (come si costruisce, si cabla e si configura un grafo sull'API pubblica), le skill di scaffolding e di review (/new-pipeline, /new-connector, /new-transform, /etl-review, /doc-sync) e i subagent che materializzano e rivedono una pipeline, più l'hook che blocca commit e push su dev, staging e main. I file del kit sono del core e arrivano con la sua versione: etl init li scrive alla nascita del repo, e pnpm exec etl agent sync li riscrive dal core installato in .claude/skills/noeva-etl-*, .claude/agents/noeva-etl-* e .claude/hooks/noeva-etl-*, insieme alla voce dell'hook in .claude/settings.json e al blocco gestito di AGENTS.md. Toglie anche le voci noeva-etl-* che il kit non ha più, e salta quelle in unmanaged. I file si committano: li vedono anche gli agenti che non sono Claude, e un bump del core mostra il diff del kit. Il resto di .claude/ e di AGENTS.md è del cliente. Il kit rimanda a questa guida invece di ricopiarla. Il catalogo completo è in Skill e subagent.