std::filesystem::path workflow_dir;

std::filesystem::path output_dir;

std::unordered_map<std::string, std::string> values;

std::unordered_map<std::string, std::vector<std::unordered_map<std::string, std::string>>> batches;

const auto * value = object.find(key);

if (value == nullptr || value->is_null()) {

throw std::runtime_error("workflow object missing required string field: " + key);

}

return value->as_string();

const engine::io::json::Value & object,

const std::string & key,

std::string fallback) {

const auto * value = object.find(key);

return value == nullptr || value->is_null() ? std::move(fallback) : value->as_string();

const engine::io::json::Value & object,

const std::string & key) {

const auto * value = object.find(key);

if (value == nullptr || value->is_null()) {

return std::nullopt;

}

return value->as_string();

const engine::io::json::Value & object,

const std::string & key) {

const auto * value = object.find(key);

if (value == nullptr || value->is_null()) {

return std::nullopt;

}

return value->as_number();

const engine::io::json::Value & object,

const std::string & key,

bool fallback) {

const auto * value = object.find(key);

if (value == nullptr || value->is_null()) {

return fallback;

}

if (value->is_bool()) {

return value->as_bool();

}

if (value->is_string()) {

const std::string & text = value->as_string();

if (text == "true") {

return true;

}

if (text == "false") {

return false;

}

}

throw std::runtime_error("workflow field must be a bool: " + key);

if (from.empty()) {

return text;

}

size_t pos = 0;

while ((pos = text.find(from, pos)) != std::string::npos) {

text.replace(pos, from.size(), to);

pos += to.size();

}

return text;

value = replace_all(value, "${out_dir}", context.output_dir.string());

value = replace_all(value, "${workflow_dir}", context.workflow_dir.string());

for (const auto & [key, item] : context.values) {

value = replace_all(value, "${" + key + "}", item);

}

return value;

if (value.empty()) {

throw std::runtime_error("workflow batch item id must not be empty");

}

for (char & ch : value) {

const unsigned char uch = static_cast<unsigned char>(ch);

if (std::isalnum(uch) == 0 && ch != '-' && ch != '_') {

ch = '_';

}

}

return value;

std::filesystem::path path(expand_value(value, context));

if (path.is_absolute()) {

return path;

}

if (value.find("${out_dir}") != std::string::npos ||

value.find("${workflow_dir}") != std::string::npos ||

value.find("${") != std::string::npos) {

throw std::runtime_error(

"workflow path inputs expanded from placeholders must be absolute: " + value +

" -> " + path.string());

}

return context.workflow_dir / path;

std::string ext = path.extension().string();

std::transform(ext.begin(), ext.end(), ext.begin(), [](unsigned char ch) {

return static_cast<char>(std::tolower(ch));

});

return ext;

std::ifstream input(path);

if (!input) {

throw std::runtime_error("failed to open workflow batch text file: " + path.string());

}

std::ostringstream raw;

raw << input.rdbuf();

return raw.str();

std::string normalized;

bool in_space = false;

for (unsigned char ch : text) {

if (std::isspace(ch) != 0) {

in_space = !normalized.empty();

continue;

}

if (in_space) {

normalized.push_back(' ');

in_space = false;

}

normalized.push_back(static_cast<char>(ch));

}

return normalized;

const auto ext = lowercase_extension(path);

if (ext == ".txt" || ext == ".md") {

return normalize_text_as_paragraph(read_text_file(path));

}

if (ext == ".json") {

const auto root = engine::io::json::parse_file(path);

if (root.is_string()) {

return normalize_text_as_paragraph(root.as_string());

}

if (root.is_object()) {

if (const auto * input = root.find("input"); input != nullptr && input->is_string()) {

return normalize_text_as_paragraph(input->as_string());

}

if (const auto * text = root.find("text"); text != nullptr && text->is_string()) {

return normalize_text_as_paragraph(text->as_string());

}

}

throw std::runtime_error("workflow batch JSON text file requires a string root, input, or text: " + path.string());

}

throw std::runtime_error("unsupported workflow batch text extension: " + ext);

const std::vector<std::unordered_map<std::string, std::string>> & items,

const std::string & step_id) {

std::unordered_set<std::string> ids;

for (const auto & item : items) {

const auto it = item.find("id");

if (it == item.end()) {

throw std::runtime_error("workflow batch item missing id: " + step_id);

}

if (!ids.insert(it->second).second) {

throw std::runtime_error("workflow batch item id collision in " + step_id + ": " + it->second);

}

}

const engine::io::json::Value & root,

const WorkflowRunOptions & options,

WorkflowContext & context) {

const auto * inputs = root.find("inputs");

if (inputs != nullptr && !inputs->is_null()) {

if (!inputs->is_object()) {

throw std::runtime_error("workflow inputs must be an object");

}

for (const auto & [key, value] : inputs->as_object()) {

if (!value.is_string()) {

throw std::runtime_error("workflow input default must be a string: " + key);

}

context.values[key] = value.as_string();

}

}

for (const auto & [key, value] : options.workflow_inputs) {

context.values[key] = value;

}

for (auto & [_, value] : context.values) {

value = expand_value(value, context);

}

const engine::io::json::Value & step,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

if (workflow_has_field(step, "foreach")) {

throw std::runtime_error("batch_inputs cannot use foreach: " + id);

}

const bool has_audio_dir = workflow_has_field(step, "audio_dir");

const bool has_text_dir = workflow_has_field(step, "text_dir");

if ((has_audio_dir ? 1 : 0) + (has_text_dir ? 1 : 0) != 1) {

throw std::runtime_error("batch_inputs requires exactly one of audio_dir or text_dir: " + id);

}

std::vector<std::filesystem::path> paths;

std::vector<std::unordered_map<std::string, std::string>> items;

if (has_audio_dir) {

const auto dir = resolve_workflow_path(workflow_string(step, "audio_dir"), context);

if (!std::filesystem::is_directory(dir)) {

throw std::runtime_error("batch_inputs audio_dir must be an existing directory: " + dir.string());

}

for (const auto & entry : std::filesystem::directory_iterator(dir)) {

if (entry.is_regular_file() && is_wav_path(entry.path())) {

paths.push_back(entry.path());

}

}

std::sort(paths.begin(), paths.end());

if (paths.empty()) {

throw std::runtime_error("batch_inputs audio_dir contains no .wav files: " + dir.string());

}

items.reserve(paths.size());

for (const auto & path : paths) {

items.push_back({

{"id", safe_workflow_id(path.stem().string())},

{"audio_path", path.string()},

});

}

} else {

const auto dir = resolve_workflow_path(workflow_string(step, "text_dir"), context);

if (!std::filesystem::is_directory(dir)) {

throw std::runtime_error("batch_inputs text_dir must be an existing directory: " + dir.string());

}

for (const auto & entry : std::filesystem::directory_iterator(dir)) {

if (entry.is_regular_file() && is_text_batch_path(entry.path())) {

paths.push_back(entry.path());

}

}

std::sort(paths.begin(), paths.end());

if (paths.empty()) {

throw std::runtime_error("batch_inputs text_dir contains no .txt, .md, or .json files: " + dir.string());

}

items.reserve(paths.size());

for (const auto & path : paths) {

const auto text = read_batch_text_payload(path);

if (text.empty()) {

throw std::runtime_error("batch_inputs text file contains no non-whitespace text: " + path.string());

}

items.push_back({

{"id", safe_workflow_id(path.stem().string())},

{"text_path", path.string()},

{"text", text},

});

}

}

validate_unique_batch_ids(items, id);

context.values[id + ".count"] = std::to_string(items.size());

context.batches[id] = std::move(items);

std::cout << "workflow_step=" << id << "\n";

std::cout << "batch_items=" << context.values[id + ".count"] << "\n";

std::ostringstream out;

out << program;

for (const std::string & arg : args) {

out << ' ' << arg;

}

return out.str();

#if defined(_WIN32)

std::vector<std::wstring> wide_args;

wide_args.reserve(args.size() + 1);

wide_args.push_back(std::filesystem::path(program).wstring());

for (const std::string & arg : args) {

wide_args.push_back(std::filesystem::path(arg).wstring());

}

std::vector<const wchar_t *> argv;

argv.reserve(wide_args.size() + 1);

for (const std::wstring & arg : wide_args) {

argv.push_back(arg.c_str());

}

argv.push_back(nullptr);

const intptr_t rc = _wspawnvp(_P_WAIT, wide_args.front().c_str(), argv.data());

if (rc == -1) {

throw std::runtime_error(

"failed to launch process: " + format_process_args(program, args) + ": " + std::strerror(errno));

}

return static_cast<int>(rc);

#else

std::vector<char *> argv;

argv.reserve(args.size() + 2);

argv.push_back(const_cast<char *>(program.c_str()));

for (const std::string & arg : args) {

argv.push_back(const_cast<char *>(arg.c_str()));

}

argv.push_back(nullptr);

pid_t pid = 0;

const int spawn_rc = posix_spawnp(&pid, program.c_str(), nullptr, nullptr, argv.data(), environ);

if (spawn_rc != 0) {

throw std::runtime_error(

"failed to launch process: " + format_process_args(program, args) + ": " + std::strerror(spawn_rc));

}

int status = 0;

while (waitpid(pid, &status, 0) < 0) {

if (errno != EINTR) {

throw std::runtime_error("failed waiting for process: " + program + ": " + std::strerror(errno));

}

}

if (WIFEXITED(status)) {

return WEXITSTATUS(status);

}

if (WIFSIGNALED(status)) {

return 128 + WTERMSIG(status);

}

throw std::runtime_error("process exited unexpectedly: " + program);

#endif

const engine::io::json::Value & step,

const WorkflowRunOptions & options,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const auto input = resolve_workflow_path(workflow_string(step, "input"), context);

const auto output = resolve_workflow_path(workflow_string(step, "output"), context);

if (!output.parent_path().empty()) {

std::filesystem::create_directories(output.parent_path());

}

const int sample_rate = static_cast<int>(workflow_optional_number(step, "sample_rate").value_or(48000.0));

const int channels = static_cast<int>(workflow_optional_number(step, "channels").value_or(2.0));

if (sample_rate <= 0 || channels <= 0) {

throw std::runtime_error("convert_audio step requires positive sample_rate and channels");

}

std::vector<std::string> args{

"-y",

"-hide_banner",

"-loglevel",

"error",

};

if (const auto start = workflow_optional_number(step, "start_seconds")) {

args.push_back("-ss");

args.push_back(std::to_string(*start));

}

if (const auto duration = workflow_optional_number(step, "duration_seconds")) {

if (!(*duration > 0.0)) {

throw std::runtime_error("convert_audio duration_seconds must be positive");

}

args.push_back("-t");

args.push_back(std::to_string(*duration));

}

args.push_back("-i");

args.push_back(path_arg(input));

args.push_back("-ac");

args.push_back(std::to_string(channels));

args.push_back("-ar");

args.push_back(std::to_string(sample_rate));

args.push_back(path_arg(output));

const int rc = run_process(options.audio_converter, args);

if (rc != 0) {

throw std::runtime_error("convert_audio step failed: " + id);

}

context.values[id + ".audio_path"] = output.string();

std::cout << "workflow_step=" << id << "\n";

std::cout << "audio_path=" << output.string() << "\n";

const std::filesystem::path & path,

const engine::runtime::AudioBuffer & audio) {

if (!path.parent_path().empty()) {

std::filesystem::create_directories(path.parent_path());

}

engine::audio::WavPcm16Sink().write(path, engine::audio::AudioBuffer{

audio.sample_rate,

audio.channels,

audio.samples,

});

if (value.is_string()) {

return engine::io::json::Value::make_string(expand_value(value.as_string(), context));

}

if (value.is_array()) {

engine::io::json::Value::Array out;

for (const auto & item : value.as_array()) {

out.push_back(expanded_json(item, context));

}

return engine::io::json::Value::make_array(std::move(out));

}

if (value.is_object()) {

engine::io::json::Value::Object out;

for (const auto & [key, item] : value.as_object()) {

out.emplace(key, expanded_json(item, context));

}

return engine::io::json::Value::make_object(std::move(out));

}

return value;

const std::unordered_map<std::string, std::string> & base,

const engine::io::json::Value * value,

const WorkflowContext & context) {

auto out = base;

if (value == nullptr || value->is_null()) {

return out;

}

for (const auto & [key, child] : value->as_object()) {

out[key] = expand_value(minitts::cli::json_option_string(child), context);

}

return out;

const std::string & id,

const engine::runtime::TaskResult & result,

const std::optional<std::filesystem::path> & audio_out,

const std::optional<std::filesystem::path> & text_out,

const std::optional<std::filesystem::path> & words_out,

const std::filesystem::path & step_dir,

WorkflowContext & context) {

if (result.audio_output.has_value() && audio_out.has_value()) {

context.values[id + ".audio_path"] = audio_out->string();

}

for (const auto & output : result.named_audio_outputs) {

const auto path = step_dir / (output.id + ".wav");

context.values[id + "." + output.id + "_path"] = path.string();

}

if (result.text_output.has_value()) {

context.values[id + ".text"] = result.text_output->text;

}

if (text_out.has_value()) {

context.values[id + ".text_path"] = text_out->string();

}

if (words_out.has_value()) {

context.values[id + ".words_path"] = words_out->string();

}

const engine::runtime::TaskResult & result,

const std::filesystem::path & path,

const std::string & id) {

if (!result.text_output.has_value()) {

throw std::runtime_error("workflow model step requested text_out but produced no text: " + id);

}

if (!path.parent_path().empty()) {

std::filesystem::create_directories(path.parent_path());

}

std::ofstream out(path);

out << result.text_output->text << "\n";

std::cout << "text_out=" << path.string() << "\n";

const engine::runtime::TaskResult & result,

const std::filesystem::path & path,

const std::string & id) {

if (result.word_timestamps.empty()) {

throw std::runtime_error("workflow model step requested words_out but produced no word timestamps: " + id);

}

if (!path.parent_path().empty()) {

std::filesystem::create_directories(path.parent_path());

}

std::ofstream out(path);

out << word_timestamps_to_json(result.word_timestamps) << "\n";

std::cout << "words_out=" << path.string() << "\n";

const engine::io::json::Value & step,

engine::core::BackendConfig fallback) {

if (const auto backend = workflow_optional_string(step, "backend")) {

fallback.type = minitts::cli::parse_backend(*backend);

}

if (const auto device = workflow_optional_number(step, "device")) {

fallback.device = static_cast<int>(*device);

}

if (const auto threads = workflow_optional_number(step, "threads")) {

fallback.threads = static_cast<int>(*threads);

}

if (fallback.threads <= 0) {

throw std::runtime_error("workflow step threads must be positive");

}

return fallback;

const engine::runtime::ModelRegistry & registry,

const engine::io::json::Value & step,

const WorkflowRunOptions & options,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const auto request_value = step.find("request");

if (request_value == nullptr || request_value->is_null()) {

throw std::runtime_error("model step requires request object: " + id);

}

engine::runtime::ModelLoadRequest load_request;

load_request.model_path = resolve_workflow_path(workflow_string(step, "model"), context);

load_request.model_spec_override = options.model_spec_override;

if (const auto family = workflow_optional_string(step, "family")) {

load_request.family_hint = *family;

}

if (const auto config = workflow_optional_string(step, "config")) {

load_request.config_id = *config;

}

if (const auto weight = workflow_optional_string(step, "weight")) {

load_request.weight_id = *weight;

}

load_request.options = merged_options(options.load_options, step.find("load_options"), context);

engine::runtime::SessionOptions session_options;

session_options.backend = parse_step_backend(step, options.backend);

session_options.options = merged_options(options.session_options, step.find("session_options"), context);

const engine::runtime::TaskSpec task_spec{

engine::runtime::parse_voice_task_kind(workflow_string(step, "task")),

engine::runtime::parse_run_mode(workflow_string_or(step, "mode", "offline")),

};

if (task_spec.mode != engine::runtime::RunMode::Offline) {

throw std::runtime_error("workflow model steps currently require offline mode: " + id);

}

auto model = registry.load(load_request);

auto session = model->create_task_session(task_spec, session_options);

auto * offline = dynamic_cast<engine::runtime::IOfflineVoiceTaskSession *>(session.get());

if (offline == nullptr) {

throw std::runtime_error("workflow model step does not support offline execution: " + id);

}

const auto expanded_request = expanded_json(*request_value, context);

auto request = minitts::cli::build_request_from_json(expanded_request, context.workflow_dir);

const auto text_out_value = workflow_optional_string(step, "text_out");

const auto words_out_value = workflow_optional_string(step, "words_out");

std::optional<std::filesystem::path> text_out;

std::optional<std::filesystem::path> words_out;

if (text_out_value.has_value()) {

text_out = resolve_workflow_path(*text_out_value, context);

}

if (words_out_value.has_value()) {

words_out = resolve_workflow_path(*words_out_value, context);

request.options["return_timestamps"] = "true";

}

session->prepare(engine::runtime::build_preparation_request(request));

const auto result = offline->run(request);

std::filesystem::path step_dir = context.output_dir / id;

if (const auto out_dir = workflow_optional_string(step, "out_dir")) {

step_dir = resolve_workflow_path(*out_dir, context);

}

std::optional<std::filesystem::path> audio_out;

if (const auto audio = workflow_optional_string(step, "audio_out")) {

audio_out = resolve_workflow_path(*audio, context);

} else if (result.audio_output.has_value() || result.named_audio_outputs.size() == 1) {

audio_out = step_dir / (id + ".wav");

}

emit_task_result(result, audio_out, step_dir, step_dir, std::nullopt, std::nullopt, std::nullopt);

if (text_out.has_value()) {

write_text_result(result, *text_out, id);

}

if (words_out.has_value()) {

write_words_result(result, *words_out, id);

}

record_result_paths(id, result, audio_out, text_out, words_out, step_dir, context);

std::cout << "workflow_step=" << id << "\n";

const auto root = engine::io::json::parse_file(path);

if (!root.is_array()) {

throw std::runtime_error("word timestamp input must be a JSON array: " + path.string());

}

std::vector<engine::runtime::WordTimestamp> words;

for (const auto & item : root.as_array()) {

engine::runtime::WordTimestamp word;

word.span.start_sample = engine::io::json::require_i64(item, "start_sample");

word.span.end_sample = engine::io::json::require_i64(item, "end_sample");

word.word = engine::io::json::require_string(item, "word");

const auto * confidence = item.find("confidence");

if (confidence != nullptr && !confidence->is_null()) {

word.confidence = confidence->as_f32();

}

words.push_back(std::move(word));

}

return words;

const engine::io::json::Value & step,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const std::string format = workflow_string(step, "format");

if (format != "srt") {

throw std::runtime_error("format_subtitles currently supports only format=srt: " + id);

}

const auto words_path = resolve_workflow_path(workflow_string(step, "words"), context);

const auto output = resolve_workflow_path(workflow_string(step, "output"), context);

const auto sample_rate_value = workflow_optional_number(step, "sample_rate");

if (!sample_rate_value.has_value()) {

throw std::runtime_error("format_subtitles requires sample_rate: " + id);

}

const int sample_rate = static_cast<int>(*sample_rate_value);

engine::text::SubtitleFormatOptions format_options;

format_options.sample_rate = sample_rate;

format_options.max_chars_per_line =

static_cast<int>(workflow_optional_number(step, "max_chars_per_line").value_or(42.0));

format_options.max_lines =

static_cast<int>(workflow_optional_number(step, "max_lines").value_or(2.0));

format_options.max_block_seconds = workflow_optional_number(step, "max_block_seconds").value_or(6.0);

format_options.max_gap_seconds = workflow_optional_number(step, "max_gap_seconds").value_or(1.0);

const auto words = read_word_timestamps_json(words_path);

const auto srt = engine::text::format_srt(words, format_options);

if (!output.parent_path().empty()) {

std::filesystem::create_directories(output.parent_path());

}

std::ofstream out(output);

out << srt;

context.values[id + ".srt_path"] = output.string();

std::cout << "workflow_step=" << id << "\n";

std::cout << "srt_out=" << output.string() << "\n";

const engine::runtime::ModelRegistry & registry,

const engine::io::json::Value & step,

const WorkflowRunOptions & options,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const auto request_value = step.find("request");

if (request_value == nullptr || request_value->is_null()) {

throw std::runtime_error("chunked_model step requires request object: " + id);

}

const std::string audio_key = workflow_string_or(step, "audio_key", "audio");

engine::audio::WavData input_wav;

if (workflow_has_field(*request_value, "audio_data")) {

auto input_audio_data = workflow_buffer(*request_value, "audio_data");

input_wav = engine::audio::read_wav_f32(input_audio_data);

} else {

const auto input_audio_path = resolve_workflow_path(workflow_string(*request_value, audio_key), context);

input_wav = engine::audio::read_wav_f32(input_audio_path);

}

if (input_wav.channels <= 0 || input_wav.sample_rate <= 0 || input_wav.samples.empty()) {

throw std::runtime_error("chunked_model input audio is invalid: " + id);

}

const int64_t total_frames = static_cast<int64_t>(input_wav.samples.size() / static_cast<size_t>(input_wav.channels));

const double chunk_seconds = workflow_optional_number(step, "chunk_seconds").value_or(30.0);

const double overlap_seconds = workflow_optional_number(step, "overlap_seconds").value_or(0.0);

if (!(chunk_seconds > 0.0) || overlap_seconds != 0.0) {

throw std::runtime_error("chunked_model requires chunk_seconds > 0 and overlap_seconds == 0");

}

const int64_t chunk_frames = static_cast<int64_t>(chunk_seconds * static_cast<double>(input_wav.sample_rate));

const int64_t step_frames = chunk_frames;

if (chunk_frames <= 0 || step_frames <= 0) {

throw std::runtime_error("chunked_model computed invalid chunk frame count");

}

std::optional<engine::audio::WavData> timing_wav;

if (const auto timing_audio = workflow_optional_string(step, "timing_audio")) {

timing_wav = engine::audio::read_wav_f32(resolve_workflow_path(*timing_audio, context));

if (timing_wav->sample_rate != input_wav.sample_rate || timing_wav->channels != input_wav.channels) {

throw std::runtime_error("chunked_model timing_audio must match the source sample rate and channel count");

}

const int64_t timing_frames =

static_cast<int64_t>(timing_wav->samples.size() / static_cast<size_t>(timing_wav->channels));

if (timing_frames < total_frames) {

throw std::runtime_error("chunked_model timing_audio is shorter than the source audio");

}

}

const bool use_timing_repaint_window = workflow_optional_bool(step, "set_repaint_window_from_timing", false);

if (use_timing_repaint_window && !timing_wav.has_value()) {

throw std::runtime_error("chunked_model set_repaint_window_from_timing requires timing_audio");

}

const float timing_threshold_dbfs =

static_cast<float>(workflow_optional_number(step, "timing_threshold_dbfs").value_or(-40.0));

const double timing_window_seconds = workflow_optional_number(step, "timing_window_seconds").value_or(0.1);

const double timing_margin_seconds = workflow_optional_number(step, "timing_margin_seconds").value_or(0.25);

engine::runtime::ModelLoadRequest load_request;

load_request.model_path = resolve_workflow_path(workflow_string(step, "model"), context);

load_request.model_spec_override = options.model_spec_override;

if (const auto family = workflow_optional_string(step, "family")) {

load_request.family_hint = *family;

}

load_request.options = merged_options(options.load_options, step.find("load_options"), context);

engine::runtime::SessionOptions session_options;

session_options.backend = parse_step_backend(step, options.backend);

session_options.options = merged_options(options.session_options, step.find("session_options"), context);

const engine::runtime::TaskSpec task_spec{

engine::runtime::parse_voice_task_kind(workflow_string(step, "task")),

engine::runtime::parse_run_mode(workflow_string_or(step, "mode", "offline")),

};

if (task_spec.mode != engine::runtime::RunMode::Offline) {

throw std::runtime_error("chunked_model steps require offline mode: " + id);

}

auto model = registry.load(load_request);

auto session = model->create_task_session(task_spec, session_options);

auto * offline = dynamic_cast<engine::runtime::IOfflineVoiceTaskSession *>(session.get());

if (offline == nullptr) {

throw std::runtime_error("chunked_model step does not support offline execution: " + id);

}

const auto step_dir = context.output_dir / id;

const auto chunk_dir = step_dir / "chunks";

std::filesystem::create_directories(chunk_dir);

engine::runtime::AudioBuffer merged;

std::vector<std::string> text_chunks;

std::string text_language;

int64_t chunk_index = 0;

for (int64_t start = 0; start < total_frames; start += step_frames) {

const int64_t frames = std::min(chunk_frames, total_frames - start);

if (frames <= 0) {

break;

}

engine::runtime::AudioBuffer chunk_audio;

chunk_audio.sample_rate = input_wav.sample_rate;

chunk_audio.channels = input_wav.channels;

chunk_audio.samples.resize(static_cast<size_t>(frames * input_wav.channels), 0.0F);

const auto copy_begin = input_wav.samples.begin() + static_cast<std::ptrdiff_t>(start * input_wav.channels);

const auto copy_end = copy_begin + static_cast<std::ptrdiff_t>(frames * input_wav.channels);

std::copy(copy_begin, copy_end, chunk_audio.samples.begin());

const auto chunk_path = chunk_dir / ("chunk_" + std::to_string(chunk_index) + ".wav");

write_runtime_audio(chunk_path, chunk_audio);

auto request_object = expanded_json(*request_value, context).as_object();

request_object[audio_key] = engine::io::json::Value::make_string(chunk_path.string());

request_object["duration_seconds"] =

engine::io::json::Value::make_number(static_cast<double>(frames) / static_cast<double>(input_wav.sample_rate));

if (use_timing_repaint_window) {

const auto active = engine::audio::find_interleaved_audio_activity_region(

timing_wav->samples,

timing_wav->channels,

timing_wav->sample_rate,

start,

frames,

timing_threshold_dbfs,

timing_window_seconds,

timing_margin_seconds);

const double start_seconds = active.has_activity

? static_cast<double>(active.start_frame - start) / static_cast<double>(input_wav.sample_rate)

: static_cast<double>(frames) / static_cast<double>(input_wav.sample_rate);

const double end_seconds = active.has_activity

? static_cast<double>(active.end_frame - start) / static_cast<double>(input_wav.sample_rate)

: static_cast<double>(frames) / static_cast<double>(input_wav.sample_rate);

request_object["repaint_start"] = engine::io::json::Value::make_number(start_seconds);

request_object["repaint_end"] = engine::io::json::Value::make_number(end_seconds);

} else if (workflow_optional_bool(step, "set_repaint_window", false)) {

request_object["repaint_start"] = engine::io::json::Value::make_number(0.0);

request_object["repaint_end"] = request_object["duration_seconds"];

}

auto request = minitts::cli::build_request_from_json(

engine::io::json::Value::make_object(std::move(request_object)),

context.workflow_dir);

if (chunk_index == 0) {

session->prepare(engine::runtime::build_preparation_request(request));

}

auto result = offline->run(request);

if (result.text_output.has_value()) {

text_chunks.push_back(result.text_output->text);

if (text_language.empty()) {

text_language = result.text_output->language;

}

}

if (result.audio_output.has_value()) {

const auto & audio = *result.audio_output;

if (merged.sample_rate == 0) {

merged.sample_rate = audio.sample_rate;

merged.channels = audio.channels;

}

if (merged.sample_rate != audio.sample_rate || merged.channels != audio.channels) {

throw std::runtime_error("chunked_model output sample rate/channel count changed across chunks: " + id);

}

merged.samples.insert(merged.samples.end(), audio.samples.begin(), audio.samples.end());

}

++chunk_index;

}

if (merged.samples.empty() && text_chunks.empty()) {

throw std::runtime_error("chunked_model produced no audio or text: " + id);

}

if (!merged.samples.empty()) {

const auto * audio_out_value = step.find("audio_out");

if (audio_out_value == nullptr || audio_out_value->is_null()) {

throw std::runtime_error("chunked_model audio output requires audio_out: " + id);

}

const auto audio_out = resolve_workflow_path(audio_out_value->as_string(), context);

write_runtime_audio(audio_out, merged);

context.values[id + ".audio_path"] = audio_out.string();

std::cout << "audio_out=" << audio_out.string() << "\n";

}

if (!text_chunks.empty()) {

std::ostringstream joined;

for (size_t index = 0; index < text_chunks.size(); ++index) {

if (index != 0) {

joined << ' ';

}

joined << text_chunks[index];

}

const std::string text = joined.str();

context.values[id + ".text"] = text;

if (!text_language.empty()) {

context.values[id + ".language"] = text_language;

}

const auto text_out_option = workflow_optional_string(step, "text_out");

const auto text_out = text_out_option.has_value()

? resolve_workflow_path(*text_out_option, context)

: step_dir / (id + ".txt");

if (!text_out.parent_path().empty()) {

std::filesystem::create_directories(text_out.parent_path());

}

std::ofstream out(text_out);

out << text << "\n";

context.values[id + ".text_path"] = text_out.string();

std::cout << "text_out=" << text_out.string() << "\n";

}

std::cout << "workflow_step=" << id << "\n";

const engine::io::json::Value & step,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const auto * inputs_value = step.find("inputs");

if (inputs_value == nullptr || !inputs_value->is_array()) {

throw std::runtime_error("mix_audio step requires inputs array: " + id);

}

std::vector<engine::audio::AudioMixInput> inputs;

for (const auto & item : inputs_value->as_array()) {

engine::audio::AudioMixInput input;

input.audio = engine::audio::read_wav_f32(resolve_workflow_path(workflow_string(item, "path"), context));

input.gain = static_cast<float>(workflow_optional_number(item, "gain").value_or(1.0));

if (const auto activity_ref = workflow_optional_string(item, "activity_ref")) {

input.activity_reference = engine::audio::read_wav_f32(resolve_workflow_path(*activity_ref, context));

input.activity_threshold_dbfs =

static_cast<float>(workflow_optional_number(item, "activity_threshold_dbfs").value_or(-40.0));

input.activity_window_seconds = workflow_optional_number(item, "activity_window_seconds").value_or(0.1);

input.activity_margin_seconds = workflow_optional_number(item, "activity_margin_seconds").value_or(0.35);

input.activity_fade_seconds = workflow_optional_number(item, "activity_fade_seconds").value_or(0.03);

}

inputs.push_back(std::move(input));

}

const bool normalize = workflow_optional_bool(step, "normalize_peak", true);

const auto mixed = engine::audio::mix_audio_to_reference_shape(inputs, normalize);

const auto output = resolve_workflow_path(workflow_string(step, "output"), context);

if (!output.parent_path().empty()) {

std::filesystem::create_directories(output.parent_path());

}

engine::audio::WavPcm16Sink().write(output, engine::audio::AudioBuffer{

mixed.sample_rate,

mixed.channels,

mixed.samples,

});

context.values[id + ".audio_path"] = output.string();

std::cout << "workflow_step=" << id << "\n";

std::cout << "audio_path=" << output.string() << "\n";

std::string value = workflow_string(step, "foreach");

if (value.size() >= 3 && value.front() == '$' && value[1] == '{' && value.back() == '}') {

value = value.substr(2, value.size() - 3);

}

const std::string suffix = ".items";

if (value.size() <= suffix.size() ||

value.compare(value.size() - suffix.size(), suffix.size(), suffix) != 0) {

throw std::runtime_error("workflow foreach must reference ${step_id.items}: " + value);

}

return value.substr(0, value.size() - suffix.size());

const engine::runtime::ModelRegistry & registry,

const engine::io::json::Value & step,

const WorkflowRunOptions & options,

WorkflowContext & context) {

const std::string id = workflow_string(step, "id");

const std::string source_id = parse_foreach_source(step);

if (source_id == id) {

throw std::runtime_error("workflow foreach cannot reference the current step: " + id);

}

const auto source_it = context.batches.find(source_id);

if (source_it == context.batches.end()) {

throw std::runtime_error("workflow foreach source is not a batch: " + source_id);

}

const auto source_items = source_it->second;

const auto * request_value = step.find("request");

if (request_value == nullptr || request_value->is_null()) {

throw std::runtime_error("model step requires request object: " + id);

}

engine::runtime::ModelLoadRequest load_request;

load_request.model_path = resolve_workflow_path(workflow_string(step, "model"), context);

load_request.model_spec_override = options.model_spec_override;

if (const auto family = workflow_optional_string(step, "family")) {

load_request.family_hint = *family;

}

if (const auto config = workflow_optional_string(step, "config")) {

load_request.config_id = *config;

}

if (const auto weight = workflow_optional_string(step, "weight")) {

load_request.weight_id = *weight;

}

load_request.options = merged_options(options.load_options, step.find("load_options"), context);