Execution graph

public function toArray(): array
{
    $sort = [];
    $previous = [];
    $sync = [];
    $toIndex = 0;
    foreach ($this->getSortAsc() as $job => $dependencies) {
        $matchCount = 0;
        foreach ($dependencies as $dependency) {
            if (in_array($dependency, $previous, true)) {
                ++$matchCount;
                
                break;
            }
        }
        if ($matchCount === 1) {
            $toIndex++;
            $previous = [];
        }
        $sort[$toIndex][] = $job;
        $previous[] = $job;
        if ($this->jobs->find($job) !== null) {
            $sync[$job] = $toIndex;
        }
    }
    
    return $this->getSortJobs($sort, $sync);
}

/**
 * Describes the component in charge of defining job execution order, where each node contains async jobs.
 */
interface GraphInterface extends StringMappedInterface
{
    /**
     * Determines if the graph has the given `$job`.
     */
    public function has(string $job): bool;

    /**
     * Retrieve dependencies for the given `$job`.
     */
    public function get(string $job): VectorInterface;

    /**
     * Determines if the given `$job` has the given `$dependencies`.
     */
    public function hasDependencies(string $job, string ...$dependencies): bool;

    /**
     * Return an instance with the specified `$name` and `$job` put.
     */
    public function withPut(string $name, JobInterface $job): self;

    /**
     * Returns the graph as an array of arrays, where each array is a node with async jobs.
     */
    public function toArray(): array;
}

$workflow = workflow(
    a: async(new A()),
    b: async(new B()),
    c: async(new C(), input: response('a')),
    d: async(new D(), input: response('a')),
    e: async(new E(), input: response('c'), data: response('d'))
);

$graph = $workflow->jobs()->graph()->toArray();
// Result:
// [
//   0 => ['a', 'b'],        // Level 0: No dependencies, run in parallel
//   1 => ['c', 'd'],        // Level 1: Depend on 'a', run in parallel
//   2 => ['e']              // Level 2: Depends on 'c' and 'd'
// ]

private function putAdded(JobInterface ...$job): void
{
    foreach ($job as $name => $item) {
        $this->jobDependencies = $item->dependencies();
        $name = strval($name);
        $this->addMap($name, $item);
        $this->jobs = $this->jobs->withPush($name);
        $this->handleArguments($name, $item);
        foreach ($item->runIf() as $runIf) {
            $this->handleRunIfReference($runIf);
            $this->handleRunIfVariable($name, $runIf);
        }
        foreach ($item->runIfNot() as $runIfNot) {
            $this->handleRunIfReference($runIfNot);
            $this->handleRunIfVariable($name, $runIfNot);
        }
        $this->storeReferences($name, $item);
        $this->assertDependencies($name);
        $this->graph = $this->graph->withPut($name, $item);
    }
}

workflow(
    fetch: async(new Fetch()),
    process: async(
        new Process(),
        data: response('fetch')  // Creates dependency: fetch → process
    )
)

workflow(
    validate: async(new Validate(), data: variable('input')),
    process: async(
        new Process(),
        data: variable('input')
    )->withDepends('validate')  // Explicit dependency: validate → process
)

workflow(
    validate: async(new Validate()),
    process: async(new Process())
        ->withRunIf(
            response('validate', 'isValid')  // Creates dependency: validate → process
        )
)

if ($job->isSync()) {
    if ($found === null) {
        $new->jobs = $new->jobs->withPush($name);
    }
} elseif ($found !== null) {
    $new->jobs = $new->jobs->withRemove($found);
}

workflow(
    a: async(new A()),
    b: async(new B()),
    c: sync(new C(), a: response('a')),  // Sync job
    d: async(new D(), c: response('c'))
)

// Graph:
// [
//   0 => ['a', 'b'],   // Parallel async
//   1 => ['c'],        // Sync job (blocks)
//   2 => ['d']         // Continues after sync
// ]

private function assertNotSelfDependency(string $job, VectorInterface $vector): void
{
    if (! $vector->contains($job)) {
        return;
    }
    
    throw new InvalidArgumentException(
        (string) message(
            'Cannot declare job **%job%** as a self-dependency',
            job: $job
        )
    );
}

// This will throw an error
workflow(
    a: async(
        new A(),
        data: response('b')
    ),
    b: async(
        new B(),
        data: response('a')  // Circular dependency!
    )
)

private function assertDependencies(string $job): void
{
    $dependencies = $this->jobDependencies->toArray();
    if (! $this->jobs->contains(...$dependencies)) {
        $missing = array_diff($dependencies, $this->jobs->toArray());
        
        throw new OutOfBoundsException(
            (string) message(
                'Job **%job%** has undeclared dependencies: `%dependencies%`',
                job: $job,
                dependencies: implode(', ', $missing),
            )
        );
    }
}

$workflow = workflow(
    // Level 0: No dependencies
    fetchUser: async(new FetchUser(), id: variable('userId')),
    fetchSettings: async(new FetchSettings(), id: variable('userId')),
    
    // Level 1: Depend on level 0
    validateUser: async(
        new ValidateUser(),
        user: response('fetchUser')
    ),
    applySettings: async(
        new ApplySettings(),
        user: response('fetchUser'),
        settings: response('fetchSettings')
    ),
    
    // Level 2: Depend on level 1
    processUser: async(
        new ProcessUser(),
        user: response('applySettings')
    )->withRunIf(
        response('validateUser', 'isValid')
    ),
    
    // Level 2: Parallel with processUser
    notifyError: async(
        new NotifyError(),
        errors: response('validateUser', 'errors')
    )->withRunIfNot(
        response('validateUser', 'isValid')
    ),
    
    // Level 3: Final step
    logResult: sync(
        new LogResult(),
        userId: variable('userId')
    )->withDepends('processUser')
);

$graph = $workflow->jobs()->graph()->toArray();
// [
//   0 => ['fetchUser', 'fetchSettings'],
//   1 => ['validateUser', 'applySettings'],
//   2 => ['processUser', 'notifyError'],
//   3 => ['logResult']
// ]