| 1094 | run_batch_inputs_step(step, context); |
| 1095 | } else if (type == "convert_audio") { |
| 1096 | run_convert_audio_step(step, options, context); |
| 1097 | } else if (type == "model") { |
| 1098 | run_model_step_impl(registry, step, options, context); |
| 1099 | } else if (type == "format_subtitles") { |
| 1100 | run_format_subtitles_step(step, context); |
| 1101 | } else if (type == "chunked_model") { |
| 1102 | run_chunked_model_step(registry, step, options, context); |
| 1103 | } else if (type == "mix_audio") { |
| 1104 | run_mix_audio_step(step, context); |
| 1105 | } else { |
| 1106 | throw std::runtime_error("unsupported workflow step type: " + type); |
| 1107 | } |
| 1108 | } |
| 1109 | |
| 1110 | void run_workflow_step( |
| 1111 | const engine::runtime::ModelRegistry & registry, |
| 1112 | const engine::io::json::Value & step, |
| 1113 | const WorkflowRunOptions & options, |
| 1114 | WorkflowContext & context) { |
| 1115 | if (!workflow_has_field(step, "foreach")) { |
| 1116 | run_workflow_step_once(registry, step, options, context); |
| 1117 | return; |
| 1118 | } |
| 1119 | |
| 1120 | const std::string id = workflow_string(step, "id"); |
| 1121 | const std::string type = workflow_string(step, "type"); |
| 1122 | if (type == "batch_inputs") { |
| 1123 | throw std::runtime_error("batch_inputs cannot use foreach: " + id); |
| 1124 | } |
| 1125 | if (type == "model") { |
| 1126 | run_model_step_foreach(registry, step, options, context); |
| 1127 | return; |
| 1128 | } |
| 1129 | const std::string source_id = parse_foreach_source(step); |
| 1130 | if (source_id == id) { |
| 1131 | throw std::runtime_error("workflow foreach cannot reference the current step: " + id); |
| 1132 | } |
| 1133 | const auto source_it = context.batches.find(source_id); |
| 1134 | if (source_it == context.batches.end()) { |
| 1135 | throw std::runtime_error("workflow foreach source is not a batch: " + source_id); |
| 1136 | } |
| 1137 | |
| 1138 | const auto base_values = context.values; |
| 1139 | std::vector<std::unordered_map<std::string, std::string>> output_items; |
| 1140 | output_items.reserve(source_it->second.size()); |
| 1141 | for (const auto & item : source_it->second) { |
| 1142 | context.values = base_values; |
| 1143 | for (const auto & [key, value] : item) { |
| 1144 | context.values["item." + key] = value; |
| 1145 | } |
| 1146 | run_workflow_step_once(registry, step, options, context); |
| 1147 | |
| 1148 | auto output_item = item; |
| 1149 | const std::string prefix = id + "."; |
| 1150 | for (const auto & [key, value] : context.values) { |
| 1151 | if (key.rfind(prefix, 0) == 0) { |
| 1152 | output_item[key.substr(prefix.size())] = value; |
no test coverage detected