Connettore: Noeva (file dei workspace)

Legge e scrive i file dei workspace Noeva passando da noeva-server-api, mai da Supabase diretto. Codice in src/connectors/noeva/.

È una famiglia a sé, non uno storage del connettore File: le operazioni che servono (risolvere un file per nome o per somiglianza, nuova versione contro sovrascrittura, marcatori nella metadata) non stanno nel contratto path-based list/get/put. I byte però sono gli stessi: per decodificarli si compone un codec del connettore File a valle.

Funzione	Tipo	Cosa fa
noevaFileSource	Source<NoevaFileContent>	Un file, risolto per id, nome esatto o ricerca fuzzy
noevaFolderSource	Source<NoevaFileContent>	Tutti i file non ancora elaborati di una cartella
noevaFileSink	Sink<Uint8Array>	Un file dal nome fisso, da uno stream di byte
noevaFilesSink	Sink<NoevaOutputFile>	Un file per record, ciascuno col suo nome e la sua metadata
noevaUploadAction	Action<I, O> (nodo action)	Carica un file per record, con elaborazione AI facoltativa, e passa avanti il fileId
noevaElaborazioneLookup	LookupResolver	Per fileId: elaborazione AI pronto / in-corso / assente, con attesa per run facoltativa
noevaContenutoLookup	LookupResolver	Per fileId: il testo estratto dall'elaborazione AI
noevaAiExtract	AiConfig<I, O> (nodo ai)	Estrazione AI di campi strutturati dai documenti di un record, validata con Zod

Il client (NoevaHttpClient) espone anche converse, per chiamare un modello tramite l'inference di Noeva: lo usa noevaAiExtract. E le operazioni sulle cartelle e sui nomi dei file che servono a una pipeline per scrivere in un albero annidato: vedi Cartelle e nomi dei file.

Configurazione

Il client (NoevaHttpClient) si costruisce da una NoevaClientConfig:

Variabile	Obbligatoria	Note
NOEVA_API_BASE_URL	sì	es. https://api.noeva.ai; lo slash finale viene tolto; niente credenziali (utente:password@): il fetch di Node rifiuta l'URL
NOEVA_API_KEY	sì	API key service-role del workspace, mai una chiave personale
NOEVA_WORKSPACE_ID	sì	il workspace; con più workspace si usa NOEVA_WORKSPACE_IDS
NOEVA_WORKSPACE_IDS	no	lista separata da virgole; ha la precedenza su NOEVA_WORKSPACE_ID

La chiave deve essere service-role perché la ricerca file dell'API, con una chiave personale, vede solo i file di quell'utente. Il client la manda come Authorization: Bearer … e il workspace come header x-noeva-workspace.

Workspace ammessi. Le pipeline non costruiscono il client a mano: passano dal WorkspaceRegistry, che contiene solo i workspace configurati e fa fail-fast su qualunque altro. Così una pipeline non tocca mai un workspace non previsto. L'API key è una sola per tutti i workspace, cambia solo l'id.

const registry = loadWorkspaceRegistry(env);
const client = registry.client(workspaceId); // ConfigError se non configurato

loadWorkspaceRegistry(env) è buildWorkspaceRegistry(loadNoevaWorkspaces(env)): i workspace che il worker serve, senza leggere la sua configurazione di polling.

loadNoevaWorkspaces(env) legge NOEVA_WORKSPACE_IDS, o il singolo NOEVA_WORKSPACE_ID, con la stessa base URL e la stessa chiave per tutti; è ciò che usa anche il worker. Le variabili sono dichiarate in MANIFEST_NOEVA: ogni config porta le variabili da cui arriva, col workspace che punta alla lista quando viene da lì, e un ConfigError elenca tutte le mancanti. primoWorkspace(env) è il primo servito, e istanzaNoevaPerIlPiano(env) è lo stesso per il piano, che ricade sull'esempio da solo.

loadNoevaClientConfig("NOEVA_") legge un solo workspace, ed è comodo nei test di integrazione.

Source — noevaFileSource

Risolve un solo file e ne emette metadati e byte (in streaming). Va indicata una fra tre opzioni:

• fileId — diretto, nessuna ricerca.
• fileName — match esatto sul nome. Nessun match è un ExtractError.
• fuzzyQuery — best-match per somiglianza (fuse.js, lato client, perché l'API cerca solo per sottostringa). fuzzyThreshold va da 0 (solo match perfetto) a 1 (accetta tutto), default 0.4. Nessun candidato sopra soglia, o due candidati a pari punteggio, è un ExtractError: il connettore non tira a indovinare.

noevaFileSource(client, { fileName: "test-etl.csv" });

Source — noevaFolderSource

Emette tutti i file non ancora elaborati di una cartella, uno per record. I byte di ogni file si scaricano solo quando vengono consumati: la cartella non finisce in memoria (salvo con skipVersion, vedi sotto).

noevaFolderSource(client, {
  folderId,
  includeSubfolders: false, // default
  skipIfMetadataKey: "elaborato_da_etl",
  maxFiles: 50, // default
  skipVersion: (bytes, meta) => false, // opzionale
});

• skipIfMetadataKey — un file la cui metadata contiene già questa chiave è considerato elaborato e viene saltato. È il modo di rendere incrementale una pipeline che lavora su una cartella: la run scrive il marcatore quando carica l'output (vedi noevaFilesSink), la run successiva lo trova e salta il file.
• maxFiles — tetto per run (default 50, DEFAULT_MAX_FILES): protegge la prima run su una cartella con anni di storico. I file oltre il tetto passano alla run successiva.
• includeSubfolders — di default solo la cartella indicata.
• skipVersion — per le pipeline che ricaricano il proprio output come nuova versione dello stesso file. Dice se una versione va saltata: la sorgente parte dalla corrente e risale verso la v1, ed emette la prima non scartata, con il numero in NoevaFileContent.version. Serve a rielaborare un file a cui è stato tolto il marcatore senza rileggere il risultato di una run precedente. Costa il download completo di ogni versione esaminata, quindi i byte non sono più pigri. Se nessuna versione è buona solleva ExtractError.

⚠️ Versioni e API. GET /api/files/:fileId/presigned-url?version=N accetta solo le versioni storiche (da 1 a currentVersion - 1, file <base>_vN). La corrente si scarica senza version: chiederla per numero risponde «Versione N del file non trovata nello storage».

Costo: il marcatore sta nella metadata, che l'elenco della cartella non restituisce. Serve quindi una GET /api/files/:fileId per ogni file della cartella, anche quelli già elaborati. A pochi file al giorno è irrilevante; su una cartella molto grande andrà ripensato.

⚠️ Il nome della chiave del marcatore è un identificatore stabile: cambiarlo fa risultare non elaborato tutto lo storico della cartella.

Sink — noevaFileSink

Concatena lo stream di byte in un solo file dal nome fisso (fileName), per esempio l'output di un codec.

noevaFileSink(client, {
  fileName: "report.csv",
  mode: "new-version",
  mimeType: "text/csv", // default application/octet-stream
  folderId, // opzionale
});

Sink — noevaFilesSink

Carica un file per record: ogni NoevaOutputFile porta fileName, bytes, mimeType? e metadata?. È il sink per le pipeline che riscrivono N file d'ingresso.

noevaFilesSink(client, { mode: "new-version", folderId });

In più rispetto a noevaFileSink:

• loadOne: il sink è isolabile per-record. Dentro uno scope try-catch il fallimento dell'upload di un file finisce sul ramo catch con il file in mano, e gli altri passano.
• La metadata viene fusa, non sostituita. L'API, a ogni upload, rimpiazza l'intera metadata del file: il sink rilegge quella esistente e la rimanda insieme alle chiavi nuove. Senza, ogni nuova versione cancellerebbe il marcatore scritto dalla run precedente.
• Salta solo se non cambierebbe nulla. In new-version l'upload si salta quando il contenuto coincide con la versione corrente e le chiavi di metadata da scrivere ci sono già (a prescindere dal valore). Un output identico all'ingresso riceve così comunque il marcatore, al prezzo di una versione con gli stessi byte, una volta sola.
• versioniEquivalenti (opzionale, per record): hash SHA-256 di contenuti che, se sono la versione corrente, rendono inutile l'upload. Serve a chi carica più versioni in sequenza (ingresso, poi output) e deve poter ripetere la sequenza dopo un fallimento a metà senza accumulare versioni: l'ingresso dichiara l'output come equivalente.

Comportamenti comuni ai due sink

• mode — overwrite sostituisce il file; new-version ne crea una versione nuova.
• Idempotenza di new-version: prima di caricare, il sink scarica l'ultima versione e ne confronta lo SHA-256 col contenuto nuovo; se coincidono salta l'upload (written: 0). L'API non espone un checksum, quindi l'hash si calcola lato client. noevaFilesSink aggiunge le due condizioni descritte sopra (chiavi di metadata e versioniEquivalenti).
• folderId — dopo l'upload il file viene assegnato alla cartella con una chiamata dedicata (addFileToFolder), perché l'upload da solo non lo fa. Su un file già nella cartella è un no-op.
• File grandi: se l'API sceglie l'upload TUS (file oltre la soglia del presigned upload) il sink fallisce con un LoadError: oggi è supportato solo l'upload presigned.
• Ogni errore di rete o di API diventa ExtractError (source) o LoadError (sink), con uno stage che identifica il connettore. Un errore della source aborta la run (una source non ha dead-letter e sta a monte di ogni guard); un errore di un sink dentro uno scope try-catch finisce sul ramo catch.

Elaborazione AI: upload, stato, testo

Le API sono quelle che usava il plugin Thunderbird, lette sul codice di noeva-server-api il 2026-09-21.

Action — noevaUploadAction

.action("mail", "archivia", noevaUploadAction(client, {
  seGiaPresente: "riusa",               // obbligatoria: riusa · errore · nuova-versione · sovrascrivi
  aiReady: { skipTags: true },          // facoltativa: chiede l'elaborazione AI, con le parti da saltare
  folderId, categoria: "email",         // facoltative, dopo l'upload
  file: (mail) => ({ fileName: `${mail.id}.eml`, bytes: mail.eml, mimeType: "message/rfc822" }),
  merge: (mail, { fileId, riusato }) => ({ ...mail, fileId }),
}))

• seGiaPresente non ha default, perché la scelta dipende dalla pipeline e dal cliente:
  • riusa: il 409 «file già esistente» di upload/prepare restituisce il file esistente (details.existingFileId), senza ricaricare, se il suo upload era completato (metadata.upload_status). Un file rimasto a metà (byte mai arrivati) si ricarica in overwrite. È ciò che rende l'upload idempotente al retry.
  • errore: il record fallisce con un LoadError.
  • nuova-versione e sovrascrivi: i flag version / overwrite del server. ⚠️ Una nuova versione azzera i dati derivati, testo estratto compreso, e l'elaborazione riparte.
• aiReady (skip*, textExtractionType, textExtractionPresetId, prompt aggiuntivi per tag e categorie) è validato strict, come sul server: una chiave sconosciuta è un errore prima della chiamata, non un 400. complete crea il processo doc_ai_ready.
• Un file riusato non rilancia l'elaborazione AI. Cartella e categoria si riassegnano comunque: il server le tratta in modo idempotente, e un retry dopo un fallimento proprio lì deve poterle completare.
• Gli altri errori del server passano, e li traduce il nodo action.

Resolver — noevaElaborazioneLookup e noevaContenutoLookup

• noevaElaborazioneLookup: pronto se lo stato AI del file è ai-ready o ai-ready-plus (testo ed embedding); in-corso se un processo doc_ai_ready è pending o running; altrimenti assente. «Assente» copre fallito, annullato e mai partito: il server ritrova un processo solo finché è in corso, e non espone la storia per file. ⚠️ La ricerca del processo filtra anche per utente: va interrogata con le stesse credenziali che hanno caricato il file.
• Attesa (attesa: { massimaMs, intervalloMs }, facoltativa): il resolver ricontrolla ogni intervalloMs i file in-corso finché diventano pronti. Il budget massimaMs è per run: la scadenza si fissa alla prima chiamata della run, e dopo chi arriva non aspetta più. Un file assente o non trovato interrompe subito l'attesa, perché non diventerebbe pronto da solo. Un errore del server durante l'attesa (non al primo controllo) non fa fallire il lookup: si logga e si restituisce l'ultimo stato letto. La pausa si interrompe se la run viene annullata. Senza attesa il resolver risponde con lo stato del momento.
• noevaContenutoLookup: il testo estratto. Un 404 (testo non ancora disponibile) è «non trovato»: con onNotFound il nodo decide se è un errore.
• Una chiave alla volta: le API non hanno una forma a batch, e i volumi (qualche allegato per record) non la giustificano. Un file che non esiste più è «non trovato».
• Il testo non finisce mai nel descriptor né nei log.

Nodo ai — noevaAiExtract

Estrae campi strutturati dai documenti di un record: una chiamata converse per documento, risposta validata con uno schema Zod, risultati combinati con regole dichiarate per campo. Restituisce la configurazione completa del nodo ai, che si esegue come un lookup ma nel DAG compare come nodo AI:

const esito = z.object({
  esito: z.enum(["AVANTI", "REVIEW"]).describe("AVANTI solo se tutto è in ordine"),
  numero_ordine: z.string().nullable().describe("Numero ordine NOLIT o W"),
  motivo: z.string().nullable(),
});

.ai("testi", "verifica-pagamento", noevaAiExtract(client, {
  modelId: "eu.anthropic.claude-sonnet-4-6",
  prompt: "…istruzioni della pipeline…",
  schema: esito,
  temperature: 0.2,                        // facoltative
  cache: false,                            // default true; vedi «Chiavi e cache»
  maxTokens: 1024,
  documenti: (r) => r.allegati,            // DocumentoAi: { id, nome?, testo }
  contesto: (r) => `EMAIL:\n${r.corpo}`,   // facoltativo, aggiunto a ogni documento
  regole: {                                // una per campo, obbligatoria nel tipo
    esito: regole.priorita(["REVIEW", "AVANTI"]),
    numero_ordine: regole.concordi(),
    motivo: regole.unisci(),
  },
  merge: (estratto, r, { documenti }) => ({ ...r, estratto }),
}), { deadLetter })

• Prompt dallo schema. Il prompt di sistema elenca i campi con tipo, valori ammessi, .describe() e se ammettono null; in fondo, le istruzioni della pipeline. Lo stesso schema valida la risposta. promptDiSistema(schema, istruzioni) e leggiRisposta(testo, schema, documento) sono esportati, per provare schema e prompt di una pipeline senza chiamare il modello.

• Risposta non valida = errore del record. Niente JSON, JSON rotto, campo mancante o del tipo sbagliato: LookupError, quindi dead-letter o ramo catch. Mai un {}.

• Regole di aggregazione (regole, da src/connectors/noeva):

  Regola	Con più documenti
  concordi()	lo stesso valore ovunque; in disaccordo è un errore
  priorita([...])	vince il primo valore dell'ordine che compare
  unisci(sep?)	le stringhe distinte e non vuote, unite (" | ")
  primo()	il primo valore non nullo

  Una regola può essere anche una funzione (valori, campo) => valore. Le regole valgono anche con un solo documento; senza documenti il merge riceve estratto: null e nessuna chiamata parte.

• Errori. L'errore di un documento (risposta non valida, NoevaApiError dell'inference) è l'errore del suo record, col messaggio che nomina documento e campi.

• Isolamento. Va configurato con deadLetter o sotto un try-catch: così un documento non valido ferma solo il suo record. Senza, il primo record in errore fa fallire la run.

• Chiavi e cache. La chiave del nodo è la richiesta serializzata (id, nome, testo, contesto): la stessa richiesta non parte due volte nella stessa finestra. Con documenti grandi e molti record, l'opzione cache: false.

• Prompt e testi non finiscono mai nel descriptor né negli errori, che nominano documento e campi.

Errori del client

Un errore HTTP è un NoevaApiError, con status e details; il 409 della rinomina è la sua sottoclasse NoevaFileNameConflictError (vedi Cartelle e nomi dei file). È un EtlError, quindi dentro un nodo action, lookup o ai resta isolabile (dead-letter o ramo catch). I dettagli sono normalizzati dalle chiavi che il server usa a seconda del gestore: details nei controller dei file, payload nel gestore globale. Il messaggio resta quello del server, salvo per la sottoclasse della rinomina.

Cartelle e nomi dei file

Operazioni del client, senza un nodo proprio: le chiama il codice di una pipeline che deve preparare dove scrivere. Poggiano su rotte che l'API ha già (/api/folders/* e PATCH /api/files/:fileId/rename).

const cartella = await client.ensureFolderPath([
  "CE Report Mensile",
  "2026",
  "07",
]);
await client.moveFile({
  fileId,
  fromFolderId: cartellaDiArrivo,
  toFolderId: cartella,
});
await client.renameFile(fileId, "2026-07.pdf");

Metodo	Cosa fa
getFolderTree()	L'albero del workspace, piatto: id, name, parentFolderId, path
listFolderFiles(folderId, { page, limit })	I file di una cartella, paginati: vedi sotto
createFolder(name, parentFolderId)	Crea una cartella (null: alla radice) e ne restituisce l'id
ensureFolderPath(segmenti)	Riusa i segmenti che esistono, crea i mancanti, restituisce l'id della foglia
addFileToFolder(folderId, fileId)	Mette un file in una cartella; idempotente
removeFileFromFolder(folderId, fileId)	Toglie un file da una cartella (il file resta nel workspace); idempotente
moveFile({ fileId, fromFolderId, toFolderId })	Aggiunge alla destinazione, poi toglie dall'origine
renameFile(fileId, newName)	Rinomina; un nome già usato è un NoevaFileNameConflictError

• listFolderFiles non porta né metadata né versione, solo id, name, folderId, createdAt (quando il file è stato caricato nel workspace) e addedAt (quando è entrato in quella cartella). Per una cartella usata come coda l'ordine giusto è addedAt: un file spostato lì dentro ci è arrivato dopo il suo caricamento. Con includeSubfolders elenca anche le sottocartelle; il totale per la paginazione arriva da meta.total.
• ensureFolderPath è idempotente. Legge l'albero una volta e crea solo i segmenti che mancano, ognuno sotto il suo genitore: su un percorso completo non scrive niente. Il percorso è un array di segmenti, non una stringa con /, perché un nome di cartella può contenere una barra; gli spazi ai bordi si tolgono, come fa il server. Se un'altra run crea lo stesso segmento fra la lettura e la scrittura, il server risponde 409 e la cartella si riusa rileggendo l'albero.
• Cartelle omonime alla radice. Il vincolo di unicità di folders (workspace_id, parent_folder_id, name) tratta i NULL come distinti: alla radice il server accetta due cartelle con lo stesso nome, e non risponde 409. Due cartelle omonime allo stesso livello sono un NoevaFolderPathError che nomina il percorso e gli id, invece di una scelta a caso che cambierebbe da una run all'altra. Va sistemato a mano nel workspace. Lo stesso errore copre un percorso vuoto o con un segmento vuoto, prima di qualunque chiamata.
• Spostamento in due passi. moveFile aggiunge alla destinazione e poi toglie dall'origine, con le rotte del file singolo. Un errore a metà lascia il file in due cartelle, non in nessuna, e rifare la chiamata completa lo spostamento: i due passi sono idempotenti. Verso la stessa cartella non toglie niente. Lo spostamento atomico dell'API (POST /api/folders/bulk-membership, operation: "move") è pensato per la selezione multipla della UI e rifiuta origine e destinazione uguali; qui non serve. Gli argomenti hanno un nome perché sono tre id dello stesso tipo.
• Nomi unici per workspace. Su Noeva il nome di un file è unico nel workspace, non nella cartella. NoevaFileNameConflictError (un NoevaApiError con status: 409) porta conflictingName (senza spazi ai bordi, come lo confronta il server) e, se il server lo dà, conflictingFileId. Il messaggio è costruito dal client, non preso dal server, e li nomina entrambi. Rinominare un file col nome che ha già non è un conflitto.

Descriptor (UI del DAG)

Tutti espongono baseUrl e workspace — mai l'API key, che il client non espone nemmeno con un getter — più i puntamenti propri. Con un client costruito dal manifest, baseUrl e workspace portano la loro variabile e la loro origine, e subito dopo c'è la riga credenziale con il solo stato della chiave: chiave API · impostata, o chiave API · manca NOEVA_API_KEY. Una variabile si attribuisce a una riga solo se ne ha prodotto il valore: un client su un altro workspace, costruito dalla stessa config, mostra il suo id senza variabile.

baseUrl si mostra senza le credenziali che l'URL avesse (utente:password@). Gli errori di rete del client, del worker e dello store di stato su API le riportano come ***@: il fetch di Node rifiuta un URL con le credenziali e lo cita per intero, e senza maschera la password finirebbe nei log del worker, nell'error_message del processo e nel report della run. Gli helper sono in src/connectors/shared/credenziali-url.ts.

kind	Da	Dettagli in più
noeva-file	noevaFileSource	fileId / fileName / fuzzyQuery
noeva-file	noevaFileSink	fileName, mode, folderId
noeva-folder	noevaFolderSource	folderId, includeSubfolders, skipIfMetadataKey, maxFiles, skipVersion (se attivo)
noeva-files	noevaFilesSink	mode, folderId
noeva-upload	noevaUploadAction	se già presente, elaborazione AI (se chiesta), folderId, categoria
noeva-elaborazione	noevaElaborazioneLookup	pronto (la definizione usata), attesa (se configurata)
noeva-contenuto	noevaContenutoLookup	—
noeva-ai-extract	noevaAiExtract	modello, campi, temperature e maxTokens se impostati; mai prompt né testi

clientDetails e noevaFileSinkDescriptor sono esportati: lo stub di getSpec di una pipeline li usa per mostrare nel DAG lo stesso nodo della run, senza costruire un client (se il baseUrl o il workspace non sono noti, il dettaglio si omette).

Contratto generale: node-descriptor.md.

Chi lo usa

Fra gli esempi di questa guida:

• cross-workspace-csv-merge — noevaFileSink su due workspace dello stesso registry. In lettura usa una source propria per workspace (workspace-source.ts) che chiama direttamente il NoevaHttpClient, perché un file assente vale 0 righe invece di un errore; le due source confluiscono in un union.

Test

• Unit colocati in src/connectors/noeva/*.test.ts: client (envelope delle risposte; cartelle e rinomina contro un fetch finto che tiene lo stato delle cartelle, client-folders.test.ts), config, ricerca fuzzy, source e sink (hash-compare, fusione della metadata, folderId), descriptor con test anti-segreto, registry dei workspace.
• WorkspaceNoevaFinto (src/testing/noeva-finto.ts, esportato da @noeva/etl-core/testing): un workspace Noeva in memoria con la forma del client (cartelle annidate, file con versioni e metadata, nome unico nel workspace, guasti a comando anche a metà spostamento), per i test di pipeline che leggono e scrivono cartelle. Con versiona() si prepara la versione lasciata da una run precedente, con sovrascrivi() il «Sovrascrivi» dell'operatore (stessa versione, byte nuovi, is_overwrite: true, che il finto scrive solo qui e non negli upload della pipeline), con fotografia() si confronta lo stato di due workspace senza id né date.
• bumpVersion, esportato dal connettore, modifica il JSON di un file prima di ricaricarlo: è l'helper di un test di integrazione manuale contro un workspace reale.