andrey/zaloopipe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add ProcessExecutor, FAKE mode, fix thread lifecycle crash

Browse Source 
- Add ProcessExecutor class wrapping PIProcess with proper cleanup
- Add CMake FAKE option (-DFAKE=ON) for testing without opencode
- Add 3 FAKE-mode tests for full pipeline execution, status transitions, output
- Fix PIThread lifecycle: use startOnce() instead of start(), track threads,
  wait + delete in destructor to prevent use-after-free of PIDeque<PIChar>
- Update build.sh with --fake flag and automatic cache invalidation
- Upgrade CMake to C++17 standard
- Update AGENTS.md with build commands and FAKE mode
This commit is contained in:
andrey
2026-07-16 09:51:39 +03:00
parent 46e12ef5fe
commit 5ca6927baa
8 changed files with 361 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pipeline-runner/CMakeLists.txt
+15 -1
View File
@@ -1,10 +1,18 @@
cmake_minimum_required(VERSION 3.10)
project(PipelineRunner)
set(CMAKE_CXX_STANDARD 11)
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
option(FAKE "Fake mode for testing without opencode" OFF)
if(FAKE)
message(STATUS "FAKE mode ON — using fake process executor")
else()
message(STATUS "FAKE mode OFF — using real process executor")
endif()
# PIP
find_package(PIP REQUIRED)
@@ -28,6 +36,9 @@ set(SRCS ${ALL_SRCS})
add_executable(pipeline-runner src/main.cpp ${SRCS} ${HDRS})
target_include_directories(pipeline-runner PRIVATE src)
target_link_libraries(pipeline-runner PIP PIP::HTTPServer PIP::Crypt PIP::Console)
if(FAKE)
target_compile_definitions(pipeline-runner PRIVATE FAKE)
endif()
set_target_properties(pipeline-runner PROPERTIES
INSTALL_RPATH "\$ORIGIN;\$ORIGIN/lib"
@@ -38,6 +49,9 @@ set_target_properties(pipeline-runner PROPERTIES
add_executable(test-pipeline-runner tests/test_pipeline.cpp tests/test_runner.cpp ${SRCS})
target_include_directories(test-pipeline-runner PRIVATE src tests)
target_link_libraries(test-pipeline-runner PIP PIP::HTTPServer PIP::Crypt PIP::Console GTest::gtest_main)
if(FAKE)
target_compile_definitions(test-pipeline-runner PRIVATE FAKE)
endif()
include(GoogleTest)
gtest_discover_tests(test-pipeline-runner)
pipeline-runner/build.sh
Executable
+58
View File
@@ -0,0 +1,58 @@
#!/bin/bash
set -e
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
BUILD_DIR="${SCRIPT_DIR}/build"
REBUILD=false
FAKE=false
for arg in "$@"; do
case "$arg" in
--rebuild) REBUILD=true ;;
--fake) FAKE=true ;;
esac
done
CMAKE_ARGS=""
if [ "${FAKE}" = true ]; then
CMAKE_ARGS="-DFAKE=ON"
fi
if [ "${REBUILD}" = true ]; then
echo "==> Clean build directory"
rm -rf "${BUILD_DIR}"
mkdir -p "${BUILD_DIR}"
echo "==> Configure"
cmake -S "${SCRIPT_DIR}" -B "${BUILD_DIR}" ${CMAKE_ARGS}
fi
mkdir -p "${BUILD_DIR}"
if [ -f "${BUILD_DIR}/CMakeCache.txt" ]; then
CACHED_FAKE=$(grep -o "FAKE:BOOL=.\+" "${BUILD_DIR}/CMakeCache.txt" 2>/dev/null || echo "")
if [ "${FAKE}" = true ] && [ "${CACHED_FAKE}" != "FAKE:BOOL=ON" ]; then
echo "==> FAKE flag changed, cleaning build"
rm -rf "${BUILD_DIR}"
mkdir -p "${BUILD_DIR}"
elif [ "${FAKE}" = false ] && [ "${CACHED_FAKE}" = "FAKE:BOOL=ON" ]; then
echo "==> FAKE flag changed, cleaning build"
rm -rf "${BUILD_DIR}"
mkdir -p "${BUILD_DIR}"
fi
fi
if [ ! -f "${BUILD_DIR}/CMakeCache.txt" ]; then
echo "==> Configure"
cmake -S "${SCRIPT_DIR}" -B "${BUILD_DIR}" ${CMAKE_ARGS}
fi
echo "==> Build"
cmake --build "${BUILD_DIR}" -j$(nproc)
echo "==> Run tests"
rm -rf /tmp/pipeline-runner-test/
"${BUILD_DIR}/test-pipeline-runner" \
--gtest_brief=1 \
--gtest_output=xml:"${BUILD_DIR}/test-results.xml"
echo "==> Done"
pipeline-runner/src/processexecutor.cpp
+30
View File
@@ -0,0 +1,30 @@
#include "processexecutor.h"
#include <pibytearray.h>
#include <piprocess.h>
#include <pisystemtime.h>
ProcessResult ProcessExecutor::run(const PIString & command, const PIStringList & args, int timeoutSeconds, const PIString & workingDir) {
PIProcess proc;
proc.enableReadStdOut(true);
proc.enableReadStdErr(true);
if (!workingDir.isEmpty()) {
proc.setWorkingDirectory(workingDir);
}
proc.exec(command, args);
proc.waitForFinish(PISystemTime::fromSeconds(timeoutSeconds));
PIByteArray stdoutData = proc.readOutput();
PIByteArray stderrData = proc.readError();
int rc = proc.exitCode();
proc.stopAndWait();
ProcessResult result;
result.exitCode = rc;
result.output = PIString::fromUTF8(stdoutData);
result.error = PIString::fromUTF8(stderrData);
return result;
}
pipeline-runner/src/processexecutor.h
+20
View File
@@ -0,0 +1,20 @@
#ifndef PROCESSEXECUTOR_H
#define PROCESSEXECUTOR_H
#include <pistring.h>
#include <pistringlist.h>
struct ProcessResult {
int exitCode;
PIString output;
PIString error;
};
class ProcessExecutor {
public:
// Execute a command with args, wait up to timeoutSeconds, return result.
// Only one process runs at a time — synchronous, stack-based PIProcess.
static ProcessResult run(const PIString & command, const PIStringList & args, int timeoutSeconds, const PIString & workingDir);
};
#endif
pipeline-runner/src/runner.cpp
+47 -25
View File
@@ -4,12 +4,21 @@
#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() {}
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);
@@ -41,12 +50,10 @@ PIString PipelineRunner::startRun(const PIString & pipeline_id) {
appendLog(run_id, "Starting pipeline: " + pipeline.name + " (run_id: " + run_id + ")");
// Execute in a separate thread
// Use a lambda that captures a copy of the pipeline and run_id
PIString rid = run_id;
PIString pdir = pipelines_dir_;
PIString ldir = logs_dir_;
PIThread * thread = new PIThread([this, rid, pipeline, pdir, ldir]() { executePipeline(rid, pipeline); });
thread->start();
PIThread * thread = new PIThread([this, rid, pipeline]() { executePipeline(rid, pipeline); });
thread->startOnce();
threads_ << thread;
return run_id;
}
@@ -67,6 +74,34 @@ bool PipelineRunner::isRunActive(const PIString & 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);
@@ -82,29 +117,15 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
}
// Build command: opencode run <text> --title <title>
PIProcess proc;
proc.enableReadStdOut(true);
proc.enableReadStdErr(true);
if (!pipeline.working_dir.isEmpty()) {
proc.setWorkingDirectory(pipeline.working_dir);
}
PIStringList args;
args << "run" << prompt.text << "--title" << prompt.title;
proc.exec("opencode", args);
// Wait for process to finish (max 300 seconds per step)
proc.waitForFinish(PISystemTime::fromSeconds(300));
ProcessResult result = ProcessExecutor::run("opencode", args, 300, pipeline.working_dir);
int rc = result.exitCode;
PIString output = result.output;
PIString error = result.error;
PIByteArray stdoutData = proc.readOutput();
PIByteArray stderrData = proc.readError();
int rc = proc.exitCode();
PIString output = PIString::fromUTF8(stdoutData);
PIString error = PIString::fromUTF8(stderrData);
RunStatus stepStatus = (rc == 0) ? RunStatus::Completed : RunStatus::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) + ")");
@@ -142,6 +163,7 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
}
appendLog(run_id, "Pipeline finished");
#endif
}
void PipelineRunner::appendLog(const PIString & run_id, const PIString & line) {
pipeline-runner/src/runner.h
+4 -1
View File
@@ -1,12 +1,13 @@
#ifndef RUNNER_H
#define RUNNER_H
#include "pipeline.h"
#include "processexecutor.h"
#include <pimap.h>
#include <pimutex.h>
#include <piobject.h>
#include <pistring.h>
#include <pithread.h>
#include <pivector.h>
enum class RunStatus {
@@ -69,6 +70,8 @@ private:
PIMutex mutex_;
PIMap<PIString, RunState> runs_;
PIMap<PIString, bool> active_runs_;
PIVector<PIThread *> threads_;
};
#endif
pipeline-runner/tests/test_runner.cpp
+160
View File
@@ -6,6 +6,7 @@
#include <pifile.h>
#include <piprocess.h>
#include <pisystemtime.h>
#include <pitime.h>
static const PIString testDir = "/tmp/pipeline-runner-test/";
static const PIString testPipelinesDir = testDir + "pipelines/";
@@ -31,6 +32,16 @@ static void cleanupDirs() {
PIDir::remove(testDir);
}
static void waitForRun(PipelineRunner & runner, const PIString & runId, int timeoutMs = 5000) {
for (int i = 0; i < timeoutMs / 50; ++i) {
RunState state = runner.getRunState(runId);
if (state.status == RunStatus::Completed || state.status == RunStatus::Error) {
return;
}
piMSleep(50);
}
}
static Pipeline makeEchoPipeline() {
Pipeline pl;
pl.id = "echo-test";
@@ -90,6 +101,8 @@ TEST(RunnerTest, StartRunAndGetState) {
EXPECT_EQ(state.steps.size(), 1);
EXPECT_EQ(state.steps[0].title, "Step 1");
// Wait for background thread to finish before cleanup
waitForRun(runner, runId);
cleanupDirs();
}
@@ -126,6 +139,8 @@ TEST(RunnerTest, IsRunActive) {
PIString runId = runner.startRun("active-test");
EXPECT_TRUE(runner.isRunActive(runId));
// Wait for background thread to finish before cleanup
waitForRun(runner, runId);
cleanupDirs();
}
@@ -135,3 +150,148 @@ TEST(RunnerTest, RunStatusToString) {
EXPECT_EQ(runStatusToString(RunStatus::Completed), "completed");
EXPECT_EQ(runStatusToString(RunStatus::Error), "error");
}
#ifdef FAKE
static Pipeline makeFakePipeline(const PIString & id, const PIString & name, int numSteps) {
Pipeline pl;
pl.id = id;
pl.name = name;
pl.working_dir = "/tmp";
pl.created_at = nowISO();
pl.updated_at = pl.created_at;
for (int i = 0; i < numSteps; ++i) {
Prompt p;
p.id = "p" + PIString::fromNumber(i);
p.text = "fake text " + PIString::fromNumber(i);
p.title = "Fake Step " + PIString::fromNumber(i);
p.order = i;
pl.prompts << p;
}
return pl;
}
TEST(FakeRunnerTest, FullPipelineCompletes) {
setupDirs();
PipelineRunner runner(testPipelinesDir, testLogsDir);
Pipeline pl = makeFakePipeline("fake-full", "Fake Full Pipeline", 3);
savePipeline(testPipelinesDir, pl);
PIString runId = runner.startRun("fake-full");
EXPECT_FALSE(runId.isEmpty());
// Wait for pipeline to finish (3 steps * 100ms = 300ms, allow extra time)
for (int i = 0; i < 100; ++i) {
piMSleep(50);
RunState state = runner.getRunState(runId);
if (state.status == RunStatus::Completed) {
bool allDone = true;
for (int j = 0; j < state.steps.size(); ++j) {
if (state.steps[j].status != RunStatus::Completed) {
allDone = false;
break;
}
}
if (allDone) break;
}
if (state.status == RunStatus::Error) break;
}
RunState state = runner.getRunState(runId);
EXPECT_EQ(state.status, RunStatus::Completed);
EXPECT_EQ(state.steps.size(), 3);
for (int i = 0; i < 3; ++i) {
EXPECT_EQ(state.steps[i].status, RunStatus::Completed);
EXPECT_EQ(state.steps[i].returncode, 0);
}
// Wait for background thread to finish before cleanup
waitForRun(runner, runId, 2000);
cleanupDirs();
}
TEST(FakeRunnerTest, RunStatusTransitions) {
setupDirs();
PipelineRunner runner(testPipelinesDir, testLogsDir);
Pipeline pl = makeFakePipeline("fake-transition", "Fake Transition Pipeline", 2);
savePipeline(testPipelinesDir, pl);
PIString runId = runner.startRun("fake-transition");
EXPECT_FALSE(runId.isEmpty());
// Initially running
RunState state = runner.getRunState(runId);
EXPECT_EQ(state.status, RunStatus::Running);
// Wait for completion (2 steps * 100ms = 200ms, allow extra time)
for (int i = 0; i < 100; ++i) {
piMSleep(50);
state = runner.getRunState(runId);
if (state.status == RunStatus::Completed) {
bool allDone = true;
for (int j = 0; j < state.steps.size(); ++j) {
if (state.steps[j].status != RunStatus::Completed) {
allDone = false;
break;
}
}
if (allDone) break;
}
if (state.status == RunStatus::Error) break;
}
state = runner.getRunState(runId);
EXPECT_EQ(state.status, RunStatus::Completed);
EXPECT_FALSE(runner.isRunActive(runId));
// Wait for background thread to finish before cleanup
waitForRun(runner, runId, 2000);
cleanupDirs();
}
TEST(FakeRunnerTest, StepOutputContainsFakeMarker) {
setupDirs();
PipelineRunner runner(testPipelinesDir, testLogsDir);
Pipeline pl = makeFakePipeline("fake-output", "Fake Output Pipeline", 2);
savePipeline(testPipelinesDir, pl);
PIString runId = runner.startRun("fake-output");
EXPECT_FALSE(runId.isEmpty());
// Wait for completion (2 steps * 100ms = 200ms, allow extra time)
for (int i = 0; i < 100; ++i) {
piMSleep(50);
RunState state = runner.getRunState(runId);
if (state.status == RunStatus::Completed) {
bool allDone = true;
for (int j = 0; j < state.steps.size(); ++j) {
if (state.steps[j].status != RunStatus::Completed) {
allDone = false;
break;
}
}
if (allDone) break;
}
if (state.status == RunStatus::Error) break;
}
RunState state = runner.getRunState(runId);
EXPECT_EQ(state.status, RunStatus::Completed);
for (int i = 0; i < state.steps.size(); ++i) {
EXPECT_TRUE(state.steps[i].output.contains("FAKE"));
EXPECT_TRUE(state.steps[i].output.contains("Fake Step " + PIString::fromNumber(i)));
EXPECT_TRUE(state.steps[i].error.isEmpty());
}
// Wait for background thread to finish before cleanup
waitForRun(runner, runId, 2000);
cleanupDirs();
}
#endif