public static RunRequest createInternalWorkflowRunRequest(
      WorkflowSummary workflowSummary,
      StepRuntimeSummary runtimeSummary,
      List<Tag> tags,
      Map<String, ParamDefinition> runParams,
      String dedupKey) {

    UpstreamInitiator initiator =
        UpstreamInitiator.withType(Initiator.Type.valueOf(runtimeSummary.getType().name()));
    UpstreamInitiator.Info parent = new UpstreamInitiator.Info();
    parent.setWorkflowId(workflowSummary.getWorkflowId());
    parent.setInstanceId(workflowSummary.getWorkflowInstanceId());
    parent.setRunId(workflowSummary.getWorkflowRunId());
    parent.setStepId(runtimeSummary.getStepId());
    parent.setStepAttemptId(runtimeSummary.getStepAttemptId());

    List<UpstreamInitiator.Info> ancestors = new ArrayList<>();
    if (workflowSummary.getInitiator() instanceof UpstreamInitiator) {
      ancestors.addAll(((UpstreamInitiator) workflowSummary.getInitiator()).getAncestors());
    }
    ancestors.add(parent);
    initiator.setAncestors(ancestors);
    // enforce depth limit, e.g. 10, it can also prevent workflow-subworkflow cycle
    Checks.checkTrue(
        initiator.getDepth() < Constants.WORKFLOW_DEPTH_LIMIT,
        "Workflow initiator [%s] is not less than the depth limit: %s",
        initiator,
        Constants.WORKFLOW_DEPTH_LIMIT);

    return RunRequest.builder()
        .initiator(initiator)
        .requestId(IdHelper.createUuid(dedupKey))
        .requestTime(System.currentTimeMillis())
        .currentPolicy(workflowSummary.getRunPolicy()) // default and might be updated
        .runtimeTags(tags)
        .correlationId(workflowSummary.getCorrelationId())
        .instanceStepConcurrency(workflowSummary.getInstanceStepConcurrency()) // pass it down
        .runParams(runParams)
        .restartConfig(
            copyRestartConfigWithClonedPath(
                ObjectHelper.valueOrDefault(
                    runtimeSummary.getRestartConfig(),
                    workflowSummary.getRestartConfig()))) // to be updated
        .build();
  }
@Test
  public void testErrorForTooDeepWorkflow() {
    WorkflowSummary summary = new WorkflowSummary();
    summary.setWorkflowId("test-workflow");
    summary.setWorkflowInstanceId(123);
    summary.setWorkflowRunId(2);
    SubworkflowInitiator initiator = new SubworkflowInitiator();
    UpstreamInitiator.Info info = new UpstreamInitiator.Info();
    info.setWorkflowId("foo");
    initiator.setAncestors(Arrays.asList(info, info, info, info, info, info, info, info, info));
    summary.setInitiator(initiator);

    StepRuntimeSummary runtimeSummary =
        StepRuntimeSummary.builder()
            .type(StepType.SUBWORKFLOW)
            .stepId("bar")
            .stepAttemptId(3)
            .build();

    AssertHelper.assertThrows(
        "Workflow depth is not less than the depth limit",
        IllegalArgumentException.class,
        "is not less than the depth limit: 10",
        () ->
            StepHelper.createInternalWorkflowRunRequest(summary, runtimeSummary, null, null, null));
  }