typed-nextflow

型付き Nextflow の記述（26.04+）

このスキルが必要な理由

事前学習データに含まれる Nextflow コードの大半は、レガシーな型なし DSL2 です。たとえば、tuple val(id), path(reads) の入力、publishDir、Channel.from、暗黙の it、set／tap、| パイプ、チャネル演算子としての splitCsv などです。Nextflow 25.10 / 26.04 では、厳格な構文と静的型付けのもとで、これらのパターンは非推奨または未サポートになっています。明示的に促さないと、モデルは古いスタイルに戻ってしまいます。プロジェクトが古い Nextflow を明示的に対象としている場合を除き、.nf ファイルを生成または編集するときは、必ずこのスキルを使用してください。

すばやい判断

• 新しいコードで、バージョン制約が指定されていない場合 → 型付きの厳格な構文（26.04 スタイル）を使用します。
• レガシーまたは部分的に型付けされたファイルを編集する場合 → まず、回答の先頭に型付きの同等表現を1～2行で示します（「最新化すると、次のようになります：tuple(id: String, …) と output {} を使い、publishDir は使いません」）。その後、ユーザーが書き換えを希望しない限り、ファイル内の既存スタイルで編集します。これにより、大規模な変更を強制せずに古い記述を明らかにできます。
• プロジェクトで Nextflow < 25.10 が固定されている場合 → レガシー DSL2 を使用します（ただし、非推奨のパターンである Channel.from、暗黙の it、set／tap、複数引数の mix は避けてください）。

注意 — 型付きコードでよくある後退

Nextflow の「型付き」コードを生成する場合でも、モデルは次の点を間違えがちです。提出前にそれぞれ確認してください。

1. プロセス入力の分解（26.04 の破壊的変更）。

   • 25.10: (id, fastq_1, fastq_2): Tuple<String, Path, Path>
   • 26.04: tuple(id: String, fastq_1: Path, fastq_2: Path) input: ブロック内で古い (name1, name2): Tuple<A, B> 構文を使うと、26.04 では拒否されます。Tuple<A, B> はチャネル要素型としては引き続き有効です（例：Channel<Tuple<String, Path>>）。ただし、プロセス入力の分解形式としては使えません。

2. groupBy はキーを返すクロージャを受け取りません。 チャネルがすでに (key, value) の2要素タプル（または (key, size, value) の3要素タプル）であることを前提とします。計算したキーでグループ化するには、先に map を使います。

   // wrong
   pairs.groupBy { p -> p[0] }
   // right
   pairs.map { id, file -> tuple(id, file) }.groupBy()
   // → emits (id, Bag<files>)
   

3. グループ化されたデータは、2つの並列チャネルではなく、1つのタプル入力として渡します。 (id, [files]) のチャネルは、2つの .map による派生チャネルに分割して位置引数で渡すのではなく、1つのプロセス入力に渡す必要があります。そうしないと同期が崩れます。プロセス内で分解してください。

   process MERGE_FILES {
       input:
       tuple(id: String, files: List<Path>)
       // ...
   }
   workflow {
       grouped = pairs.groupBy()      // Channel<Tuple<String, Bag<Path>>>
       MERGE_FILES(grouped)            // single channel argument
   }
   

4. プロセス内のセクションの順序は重要です。 必須の順序は、input: → stage:（任意）→ output: → topic:（任意）→ script:／exec: → stub:（任意）です。厳格な構文では、output: を script: の後に置くと解析エラーになります。

5. 出力は値であり、裸の識別子ではありません。 output: 内の各行は、名前なしの値（stdout()、file('x.txt')、record(...) など）または name: Type = value のいずれかです。レガシーなプロセスを移行する際に最もよくある後退は、入力／出力の修飾子を削除して、変数名だけを残してしまうことです。

   // legacy
   output:
   path results
   
   // wrong typed-style guess (model regression)
   output:
   results
   
   // correct typed output
   output:
   results: Path = file("${id}_results.csv")        // literal-ish name
   // or, if the script-level variable `results` is a Path/String:
   results: Path = file("${results}")
   // or, for a single unnamed output:
   file('results.csv')
   

   tuple(...) と record(...) の出力にも同じルールが適用されます。各出力行は、単独で記述した変数名ではなく、値を生成する式である必要があります。

6. meta をレコード内にネストしないでください。 record(meta: sample, logs: ...) を出力して「meta マップを保持する」というレガシーな nf-core の習慣は、レコードの目的に反します。meta フィールドをレコード内にフラット化してください。

   // anti-pattern (nf-core meta habit)
   record(meta: sample, logs: files('*_fastqc.zip'))
   // idiomatic
   record(id: sample.id, single_end: sample.single_end, logs: files('*_fastqc.zip'))
   

   後続の結合では、by: 'meta.id' ではなく by: 'id' を使用します。

新しいモードを有効にする

すべての型付きスクリプトには、次の両方が必要です。

nextflow.enable.types = true

// (strict syntax is on by default in 26.04; for 25.10 export NXF_SYNTAX_PARSER=v2)

params {} ブロックと output {} ブロックは、nextflow.enable.types がなくても機能します。型付きプロセスと型付きワークフロー（take:／emit: の注釈）には、このフラグが必要です。

基本要素

1. 型付きの params {} ブロック

トップレベルの params.foo = ... を、型付きブロックに置き換えます。

params {
    // Samplesheet path
    reads: Path = "${projectDir}/data/samples.csv"

    // Optional reference; required if no default given
    transcriptome: Path

    // Default-false boolean
    save_intermeds: Boolean

    outdir: Path = 'results'
}

ルール：

• params はエントリーワークフローと output {} ブロックでのみ使用し、サブワークフローやプロセスには明示的な入力として渡します。
• デフォルト値のないパラメーターは必須です（未設定の場合、実行は失敗します）。26.04 では、デフォルト値のないブール値は false になります。
• CLI の値は、宣言された型に変換されます。

2. レコードでタプルを置き換える

レコードは、名前付きフィールドを持つ複合型です。record(...) で構築し、record Name { ... } で型を定義します。

record Sample {
    id: String
    fastq_1: Path
    fastq_2: Path?     // `?` = nullable
}

def s = record(id: 'A', fastq_1: file('a_1.fq'), fastq_2: file('a_2.fq'))
s.id      // access by name, not index

レコードはダックタイピングされます。入力が宣言するすべてのフィールドを含んでいれば、その入力を満たします。追加のフィールドは無視されます。

2b. nf-core の meta マップからレコードへの移行

事前学習データで主流となっているパターンは、meta が Groovy マップである nf-core の (meta, files) タプルです。

// legacy nf-core
process FASTQC {
    tag "${meta.id}"
    input:
    tuple val(meta), path(reads)
    output:
    tuple val(meta), path('*_fastqc.zip'), emit: zip
    // ...
}
workflow {
    ch_reads = Channel
        .fromPath(params.samplesheet)
        .splitCsv(header: true)
        .map { row ->
            def meta = [id: row.sample, single_end: row.single_end.toBoolean()]
            def reads = meta.single_end ? [file(row.fastq_1)] : [file(row.fastq_1), file(row.fastq_2)]
            tuple(meta, reads)
        }
    FASTQC(ch_reads)
}

型付きの同等表現 — meta はネストせず、レコードにフラット化します。

nextflow.enable.types = true

record Sample {
    id: String
    single_end: Boolean
    strandedness: String?
    reads: List<Path>
}

process FASTQC {
    tag sample.id
    input:
    sample: Sample
    output:
    record(
        id: sample.id,
        single_end: sample.single_end,
        zip: files('*_fastqc.zip'),
        html: files('*_fastqc.html')
    )
    script:
    """
    fastqc --threads ${task.cpus} ${sample.reads.join(' ')}
    """
}

workflow {
    main:
    samples = channel.of(params.samplesheet)
        .flatMap { csv -> csv.splitCsv(header: true) }
        .map { row ->
            def single = row.single_end.toBoolean()
            record(
                id: row.sample,
                single_end: single,
                strandedness: row.strandedness ?: null,
                reads: single
                    ? [file(row.fastq_1, checkIfExists: true)]
                    : [file(row.fastq_1, checkIfExists: true), file(row.fastq_2, checkIfExists: true)]
            )
        }

    fastqc = FASTQC(samples)

    publish:
    fastqc = fastqc
}

output {
    fastqc: Channel<Record> {
        // per-sample subdir, by `id` directly (no `meta.id`)
        path { r -> "fastqc/${r.id}" }
        index { path 'fastqc.csv'; header true }
    }
}

nf-core の meta から移行する際の規約：

• meta + [type: 'long'] によるマップのマージ → s + record(type: 'long') によるレコードのマージ。
• meta.id でのアクセス → s.id で直接アクセス。tag "${meta.id}" → tag sample.id。
• meta.single_end ? [r1] : [r1, r2] の形はそのままですが、読み取りデータはレコード上の型付き List<Path> フィールドになります。
• プロセスごとの publishDir { "results/fastqc/${meta.id}" } による公開 → output { fastqc { path { r -> "fastqc/${r.id}" } } }。
• nf-core の task.ext.args / task.ext.prefix は、型付きプロセスでも引き続き使えます。非推奨にはなっていません。書き換えが必要なのは meta マップの規約だけです。

3. 型付きプロセス

nextflow.enable.types = true

process FASTQC {
    tag sample.id
    conda 'bioconda::fastqc=0.12.1'

    input:
    sample: Sample                  // by record type
    // or destructured:
    // record(id: String, fastq_1: Path, fastq_2: Path)
    // or tuple destructure:
    // tuple(id: String, fastq: Path)
    // or scalar:
    // index: Path

    output:
    record(
        id: sample.id,
        fastqc: file("fastqc_${sample.id}_logs")
    )

    script:
    """
    fastqc.sh ${sample.id} ${sample.fastq_1} ${sample.fastq_2}
    """
}

ポイント：

• 各入力は name: Type（または分割代入された record(...) / tuple(...)）です。
• 型が修飾子に置き換わります。path → Path、val → 実際の型、path '*' のコレクション → Set<Path>（順序が重要な場合は List<Path>）。
• Path 入力と Path コレクションは自動的にステージングされます。コレクションのデフォルトのステージパターンは '*' です。
• null 許容入力：input: Path?（そうでない場合、null ではタスクが失敗します）。
• 出力は、file()、files()、stdout()、env('VAR')、record(...)、tuple(...) で構築する通常の値です。名前のない出力を 1 つだけ指定することもできます。
• file('x', optional: true) は、ファイルが見つからない場合に失敗せず null を返します。
• each 入力修飾子は廃止されました — 呼び出し箇所で combine 演算子を使ってください。

4. stage: セクション（残っている修飾子を置き換えます）

以前は入力修飾子に付随していたカスタムステージングは、stage: に記述します。

process grep {
    input:
    id: String
    fasta: Path

    stage:
    stageAs fasta, "${id}.fa"      // note: (value, pattern) order in 26.04
    env 'SAMPLE_ID', id
    stdin id

    script:
    "cat ${id}.fa | grep '>'"
}

26.04 で stageAs のメソッドシグネチャが (value, pattern) に変更されました — 先に value を指定します。

5. 型付きワークフロー

nextflow.enable.types = true

workflow RNASEQ {
    take:
    samples: Channel<Sample>
    transcriptome: Path

    main:
    index   = INDEX(transcriptome)
    fastqc  = FASTQC(samples)
    quant   = QUANT(samples, index)
    joined  = fastqc.join(quant, by: 'id')

    emit:
    samples: Channel<AlignedSample> = joined
}

record AlignedSample {
    id: String
    fastqc: Path
    quant: Path
}

• ストリーミングには Channel<T>、単一のデータフロー値には Value<T>（または省略形の T）を使います。
• Emit アノテーションは省略可能ですが、ドキュメントおよび健全性チェックとして推奨されます。
• 型付きワークフローでは、Channel. の大文字表記、暗黙の it、set / tap、| / &、.out は使用できません。通常の代入とメソッド呼び出しを使ってください。

6. ワークフロー出力（publishDir を置き換えます）

公開する内容をトップレベルで一度定義します。プロセスから publishDir を削除してください。

workflow {
    main:
    samples_ch = channel.of(params.reads)
        .flatMap { csv -> csv.splitCsv(header: true) }
        .map { row -> record(id: row.id, fastq_1: file(row.fastq_1), fastq_2: file(row.fastq_2)) }

    aligned_ch = RNASEQ(samples_ch, params.transcriptome)
    report     = MULTIQC(aligned_ch.flatMap { s -> [s.fastqc, s.quant] }.collect(), params.multiqc)

    publish:
    samples         = aligned_ch
    multiqc_report  = report
}

output {
    samples: Channel<AlignedSample> {
        path { s ->
            s.fastqc >> "fastqc/${s.id}"
            s.quant  >> "quant/${s.id}"
        }
        index {
            path 'samples.csv'
            header true
        }
    }

    multiqc_report: Path {
        path 'multiqc_report.html'
    }
}

設定側：

workflow.output.mode = 'copy'
outputDir = 'results'      // or use -output-dir on CLI

26.04 では manifest.defaultBranch は非推奨です — 削除してください。

厳密な構文のチートシート（禁止事項／必須事項）

禁止／非推奨	代わりに使用
import groovy.json.JsonSlurper	new groovy.json.JsonSlurper()（完全修飾名）
スクリプト内の class Foo { ... }	enum、record、または lib/ に配置
宣言と混在するトップレベル文	文を workflow {} 内に配置
for / while ループ	.each {}、.collect {}、.find {} など
switch	if/else if/else
スプレッド構文 *list	列挙するか分割代入する
${PWD} による暗黙的な環境変数参照	env('PWD')
include での addParams / params	明示的な入力として渡す
def Map x = [:] のような Groovy 型付き変数	def x: Map = [:]（Nextflow 型）または型指定なしの def x = [:]
${...} を含むスラッシー文字列、または複数行のスラッシー文字列	ダブルクォートまたはトリプルクォート
(Map) x によるソフトキャスト	x as Map
クォートなしの env FOO	env 'FOO'
process.shell セクション	script:
Channel.of(...)	channel.of(...)
暗黙の it	{ v -> ... }
トップレベルの workflow.onComplete { ... }	エントリーワークフロー内の onComplete: セクションに配置
プロセスの when: セクション	呼び出し箇所でフィルター処理
エントリー以外のワークフロー内の params.foo	明示的な入力として受け取る

nextflow.enable.typesでのオペレーター移行マップ

コアオペレーターを優先し、レガシーなものを書き換えてください。

レガシー	置き換え
set { ch }	ch = …
tap { ch }	次のオペレーターの前後で代入する
branch { ... }	各分岐に1つずつfilter／mapを使用する（レコードを使うと簡単です）
multiMap { ... }	各分岐に1つずつmapを使用するか、すべてのフィールドを持つ単一のレコードを使用する
groupTuple	groupBy — 入力はすでに(key, value)の2要素タプルである必要があります。キー用クロージャーはありません。派生キーでグループ化するには、.map { x -> tuple(key(x), x) }.groupBy()を使用します。(key, Bag<values>)を出力します。上記のコールアウト#2を参照してください。
cross	レコードに対してjoin(other, by: 'id')を使用する
combine(by: 0)	join(other, by: 'id')
combine(other)	combine(other)（引き続きコアです）。定数フィールドを追加するには、ch.combine(field: 'auto', ref: x)を使用します
concat	mix（1回の呼び出しにつき引数は1つです：a.mix(b).mix(c)）
mix(b, c)（複数引数）	a.mix(b).mix(c)
merge	join（決定的）
flatten	flatMap { … -> [...] }
transpose	flatMap { k, vs -> vs.collect { v -> tuple(k, v) } }
オペレーターとしてのsplitCsv／splitFasta／splitFastq／splitJson／splitText	flatMap { f -> f.splitCsv(...) }（Path標準ライブラリメソッド）
toList	collect（現在のcollectはtoListと同等です。フラット化されず、空の場合は空リストになります）
toSortedList	.collect().map { xs -> xs.toSorted() }
count／max／min／sum	.collect().map { xs -> xs.size() / xs.max() / ... }
first／last／take	収集した値に対するリストメソッド（オペレーター版は非決定的です）
buffer／collate	チャンクインデックスをキーとするgroupBy、またはList::collate
distinct	unique
randomSample	明示的に実装してください。使用しないでください（非決定的です）
dump	view(tag: '…')
collectFile	ワークフロー出力インデックスファイル、またはexec:プロセス
ifEmpty(err)	.collect().subscribe { xs -> if (xs.isEmpty()) error '...' }
ValueでのifEmpty(default)	.map { v -> v ?: default }（型付きプロセスは、任意出力が欠落している場合にnullを出力します）

型を使う場合の厳格化されたセマンティクス：

• filterにはクロージャーが必要です（正規表現リテラルだけでは使えません）。
• joinにはby:が必要です（タプルではint、レコードではstring）。レコードの結合ではfailOnDuplicate／failOnMismatchは使えません。remainder:を使って確認してください。
• flatMapはマップ／タプルを自動的にフラット化しません。クロージャーはコレクションを返す必要があります。
• mapはnullを削除しません。.filter { v -> v != null }を連結してください。
• mixの引数は1回の呼び出しにつき1つです。

典型的な骨組み

nextflow.enable.types = true

params {
    samplesheet: Path
    reference:   Path
    outdir:      Path = 'results'
}

record Sample {
    id: String
    fastq_1: Path
    fastq_2: Path?
}

include { ALIGN } from './modules/align'

workflow {
    main:
    samples = channel.of(params.samplesheet)
        .flatMap { csv -> csv.splitCsv(header: true) }
        .map { row ->
            record(
                id: row.id,
                fastq_1: file(row.fastq_1),
                fastq_2: row.fastq_2 ? file(row.fastq_2) : null
            )
        }

    aligned = ALIGN(samples, params.reference)

    publish:
    aligned = aligned
}

output {
    aligned: Channel<Record> {
        path { s -> "aligned/${s.id}" }
        index { path 'aligned.csv'; header true }
    }
}

よくある間違い（コードを出力する前にセルフチェックしてください）

• 型付きのinput:／output:／take:／emit:を使っているのに、nextflow.enable.types = trueを設定し忘れる。
• tuple val(id), path(reads)と書く。これはレガシーです。record(id: String, reads: Path)またはtuple(id: String, reads: Path)を使用してください。
• Set<Path>／List<Path>を指定せずに、pathコレクションを使用する。
• チャネルに対してsplitCsvを直接呼び出す。flatMap { f -> f.splitCsv(...) }でラップしてください。
• Channel.from(...)（大文字で、非推奨のファクトリー）を使用する。代わりにchannel.of(...)またはchannel.fromList(...)を使用してください。
• クロージャー内で暗黙のitを使う。常に{ v -> ... }のように宣言してください。
• MY_WF.out.fooを使う。まず呼び出し結果を代入してください：out = MY_WF(); out.foo。
• 複数引数のmix(a, b, c)を使う。.mix(a).mix(b)のように連結してください。
• 結合時にレコードなら名前を維持できるのに、タプルを出力する。branch／multiMapによる複雑な処理を避けるため、レコードに切り替えてください。
• プロセス内でoutput {}ブロックと併せてpublishDirを残す。
• stageAs '*', valueを使う。26.04では引数の順序が逆になり、stageAs value, '*'になりました。

参照ドキュメント

• 26.04 への移行：https://nextflow.io/docs/latest/migrations/26-04.html
• 静的型付けへの移行：https://nextflow.io/docs/latest/tutorials/static-types.html
• 静的型付けでのオペレーター：https://nextflow.io/docs/latest/tutorials/static-types-operators.html
• ワークフローの出力：https://nextflow.io/docs/latest/tutorials/workflow-outputs.html
• 型付きプロセス：https://nextflow.io/docs/latest/process-typed.html
• 型付きワークフロー：https://nextflow.io/docs/latest/workflow-typed.html
• 厳密な構文：https://nextflow.io/docs/latest/strict-syntax.html

テスト結果（サブエージェントによる RED/GREEN、2026-05-12）

シナリオ	結果
CSV サンプルシートからの一般的な FastQC プロセス	初回で問題なし
nf-core の (meta, reads) タプル → 型付き	初回は出力レコード内に meta をネストしていた。注意事項 #6 を追加 → 修正済み
レガシーの部分的に型付けされたファイル（このリポジトリの throughput/benchmark.nf）を編集し、新しいプロセスを追加	初回は 25.10 の分割代入 (name): Tuple<...> と識別子のみの出力を使用していた。注意事項 #1 と #5 を追加 → 再テストで修正済み
(id, file) ペアを ID ごとにグループ化し、ID ごとに一度プロセスを呼び出す	初回は groupBy { closure } を使用し、2 つの並列チャネルに分割していた。注意事項 #2 と #3 を追加 → 修正済み