def derive_from_bob_output(wildcards):
    with checkpoints.bob.get().output[0].open() as f:
        return f.read().strip()


def derive_from_alice_output(wildcards):
    with checkpoints.alice.get().output[0].open() as f:
        return f.read().strip()


rule all:
    input:
        derive_from_bob_output


rule intermediate:
    input:
        "alice.txt",
        "bob.txt",


rule needs_alice:
    input:
        derive_from_alice_output
    output:
        touch("second.txt")
    shell:
        "test {input} = 'first.txt' || exit 1"


rule first:
    output:
        touch("first.txt")


checkpoint alice:
    output:
        "alice.txt"
    shell:
        "echo first.txt > {output}"


checkpoint bob:
    output:
        "bob.txt"
    shell:
        "echo second.txt > {output}"

        def handle_incomplete_checkpoint(exception):
            """If checkpoint is incomplete, target it such that it is completed
            before this rule gets executed."""
            return exception.targetfile

    def get(self, **wildcards):
        missing = self.rule.wildcard_names.difference(wildcards.keys())
        if missing:
            raise WorkflowError(
                "Missing wildcard values for {}".format(", ".join(missing))
            )

        output, _ = self.rule.expand_output(wildcards)
        if self.checkpoints.created_output:
            missing_output = set(output) - set(self.checkpoints.created_output)
            if not missing_output:
                return CheckpointJob(self.rule, output)
            else:
                logger.debug(
                    f"Missing checkpoint output for {self.rule.name} "
                    f"(wildcards: {wildcards}): {','.join(missing_output)} of {','.join(output)}"
                )

        raise IncompleteCheckpointException(self.rule, checkpoint_target(output[0]))

    async def update_checkpoint_dependencies(self, jobs=None):
        """Update dependencies of checkpoints."""

        async def is_output_present(job):
            return (self.finished(job) or not self.needrun(job)) and all(
                await asyncio.gather(*(out.exists() for out in job.output))
            )

        job_queue = defaultdict(set)
        if jobs is None:
            jobs = [
                job
                for job in self.jobs
                if job.is_checkpoint and await is_output_present(job)
            ]
        else:
            # called with finished jobs, no need to check if output is present
            jobs = [job for job in jobs if job.is_checkpoint]
        for job in jobs:
            for depending in self.depending[job]:
                job_queue[depending].add(job)
            self.workflow.checkpoints.created_output.update(job.output)

        updated = len(job_queue) > 0
        if updated:
            logger.info("Updating checkpoint dependencies.")

        i = 1
        while job_queue:
            logger.debug(f"Checkpoint dependency update round {i}")
            for noneedrun_checkpoint_deps in job_queue.values():
                for checkpoint in noneedrun_checkpoint_deps:
                    self.workflow.checkpoints.created_output.update(checkpoint.output)

            candidate_job_queue = defaultdict(set)
            for job in job_queue.keys():
                prior_checkpoint_targets = {
                    infile
                    for infile in job.input
                    if is_flagged(infile, "checkpoint_target")
                }
                updated_job = await job.updated()

                await self.replace_job(job, updated_job, recursive=False)

                posterior_checkpoint_targets = {
                    infile
                    for infile in updated_job.input
                    if is_flagged(infile, "checkpoint_target")
                    and infile not in prior_checkpoint_targets
                }
                posterior_checkpoint_deps = {
                    dep
                    for dep, files in self.dependencies[updated_job].items()
                    if dep.is_checkpoint
                    and not files.isdisjoint(posterior_checkpoint_targets)
                }

                if posterior_checkpoint_deps:
                    # there is at least one dep that is a new potentially
                    # unfinished checkpoint target
                    candidate_job_queue[updated_job].update(posterior_checkpoint_deps)

            job_queue = defaultdict(set)
            if candidate_job_queue:
                await self.update_needrun()
                for job, posterior_checkpoint_deps in candidate_job_queue.items():
                    for checkpoint in posterior_checkpoint_deps:
                        if not self.needrun(checkpoint):
                            job_queue[job].add(checkpoint)
            i += 1

        if updated:
            self.set_until_jobs()
            self.delete_omitfrom_jobs()
            await self.postprocess_after_update()

        return updated