Example of run-until Nextflow

main.nf

workflow {
    def statham = new groovy.json.JsonSlurper()

    updates = Channel.topic('model_updates')
    limit = updates.until { path -> statham.parseText(path.text).score >= 100 }
    
    Setup()
    Model(updates, limit)
    
    Model.out.last().view { model -> "Last model: ${model.text}" }
}

process Model {
    publishDir 'results', mode: 'copy', saveAs: { "out.iteration_${task.index}.json" }

    input:
        path("start.json")
        val(limiter)

    output:
        path("updated.json"), topic: model_updates

    script:
    """
    #!/usr/bin/env python3
    import json
    import random

    # Read in updated.json
    with open('start.json') as f:
        data = json.load(f)

    # Add a new step object
    new_step = { 
        "step": data['step'] + 1, 
        "score": data['score'] + random.randint(1, 10)
    }

    # Pretty print the updated data to updated.json
    with open('updated.json', 'w') as f:
        json.dump(new_step, f)
    """
}

process Setup {
    output: path("start.json"), topic: model_updates
    script:
    """
    echo '{ "step": 0, "score": 0 }' > start.json
    """
}

A slightly different idea (same outcome) for the workflow block below. I think this is a slightly cleaner syntax, but it just depends on stylistic preference:

workflow {
    def statham = new groovy.json.JsonSlurper()

    Channel.topic('model_updates')
        .tap { updates }
        .map { path -> statham.parseText(path.text) }
        .until { obj -> obj?.score > 100 }
        .tap { limit }
        .last()
        .view { model -> "Final model: $model" }
    
    Setup()
    Model(updates, limit)
}