zaloopipe/pipeline-runner/src/runner.cpp

#include "runner.h"

#include "pipeline.h"

#include <piprocess.h>
#include <pisystemtime.h>
#include <pithread.h>
#include <pitime.h>

PipelineRunner::PipelineRunner(const PIString & pipelines_dir, const PIString & logs_dir)
	: pipelines_dir_(pipelines_dir)
	, logs_dir_(logs_dir) {}

PipelineRunner::~PipelineRunner() {
	for (int i = 0; i < threads_.size(); ++i) {
		if (threads_[i]) {
			threads_[i]->waitForFinish(PISystemTime::fromSeconds(30));
			delete threads_[i];
		}
	}
}

PIString PipelineRunner::startRun(const PIString & pipeline_id) {
	Pipeline pipeline = findPipeline(pipelines_dir_, pipeline_id);
	if (pipeline.id.isEmpty()) return PIString();

	PIString run_id = generateUUID();

	RunState state;
	state.run_id       = run_id;
	state.pipeline_id  = pipeline_id;
	state.status       = RunStatus::Running;
	state.current_step = 0;

	// Initialize step results
	for (int i = 0; i < pipeline.prompts.size(); ++i) {
		StepResult sr;
		sr.step_index = i;
		sr.title      = pipeline.prompts[i].title;
		sr.status     = RunStatus::Pending;
		state.steps << sr;
	}

	{
		PIMutexLocker ml(mutex_);
		runs_.insert(run_id, state);
		active_runs_.insert(run_id, true);
	}

	appendLog(run_id, "Starting pipeline: " + pipeline.name + " (run_id: " + run_id + ")");

	// Execute in a separate thread
	PIString rid      = run_id;
	PIThread * thread = new PIThread([this, rid, pipeline]() { executePipeline(rid, pipeline); });
	thread->startOnce();
	threads_ << thread;

	return run_id;
}

RunState PipelineRunner::getRunState(const PIString & run_id) {
	PIMutexLocker ml(mutex_);
	if (runs_.contains(run_id)) {
		return runs_.value(run_id);
	}
	return RunState();
}

bool PipelineRunner::isRunActive(const PIString & run_id) {
	PIMutexLocker ml(mutex_);
	return active_runs_.contains(run_id);
}

void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & pipeline) {
	int total = pipeline.prompts.size();

#ifdef FAKE
	// Fake mode: simulate all steps and update state atomically
	for (int i = 0; i < total; ++i) {
		const Prompt & prompt = pipeline.prompts[i];
		appendLog(run_id, "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] Starting: " + prompt.title);
		appendLog(run_id, "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] Prompt: " + prompt.text);
		piMSleep(100);
		appendLog(run_id, "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] COMPLETED (returncode: 0)");
	}

	// Update all step results and run status in a single mutex section
	{
		PIMutexLocker ml(mutex_);
		if (runs_.contains(run_id)) {
			for (int i = 0; i < total; ++i) {
				runs_[run_id].current_step        = i;
				runs_[run_id].steps[i].status     = RunStatus::Completed;
				runs_[run_id].steps[i].returncode = 0;
				runs_[run_id].steps[i].output     = "FAKE: executed step " + PIString::fromNumber(i) + ": " + pipeline.prompts[i].title;
				runs_[run_id].steps[i].error      = "";
			}
			runs_[run_id].status = RunStatus::Completed;
		}
		active_runs_.remove(run_id);
	}

	appendLog(run_id, "Pipeline finished");
#else
	for (int i = 0; i < total; ++i) {
		const Prompt & prompt = pipeline.prompts[i];
		appendLog(run_id, "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] Starting: " + prompt.title);
		appendLog(run_id, "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] Prompt: " + prompt.text);

		// Update step status to running
		{
			PIMutexLocker ml(mutex_);
			if (runs_.contains(run_id)) {
				runs_[run_id].current_step    = i;
				runs_[run_id].steps[i].status = RunStatus::Running;
			}
		}

		// Build command: opencode run <text> --title <title>
		PIStringList args;
		args << "run" << prompt.text << "--title" << prompt.title;

		ProcessResult result = ProcessExecutor::run("opencode", args, 300, pipeline.working_dir);
		int rc               = result.exitCode;
		PIString output      = result.output;
		PIString error       = result.error;

		RunStatus stepStatus = (rc == 0) ? RunStatus::Completed : RunStatus::Error;
		appendLog(run_id,
		          "[Step " + PIString::fromNumber(i + 1) + "/" + PIString::fromNumber(total) + "] " +
		              runStatusToString(stepStatus).toUpperCase() + " (returncode: " + PIString::fromNumber(rc) + ")");

		// Update step result
		{
			PIMutexLocker ml(mutex_);
			if (runs_.contains(run_id)) {
				runs_[run_id].steps[i].status     = stepStatus;
				runs_[run_id].steps[i].returncode = rc;
				runs_[run_id].steps[i].output     = output;
				runs_[run_id].steps[i].error      = error;
			}
		}

		if (rc != 0) {
			appendLog(run_id, "Pipeline failed at step " + PIString::fromNumber(i + 1));
			{
				PIMutexLocker ml(mutex_);
				if (runs_.contains(run_id)) {
					runs_[run_id].status = RunStatus::Error;
				}
			}
			break;
		}
	}

	// Mark as completed if all steps passed
	{
		PIMutexLocker ml(mutex_);
		if (runs_.contains(run_id) && runs_[run_id].status == RunStatus::Running) {
			runs_[run_id].status = RunStatus::Completed;
		}
		active_runs_.remove(run_id);
	}

	appendLog(run_id, "Pipeline finished");
#endif
}

void PipelineRunner::appendLog(const PIString & run_id, const PIString & line) {
	::appendLog(logs_dir_, run_id, line);
}