From 5ca6927baa9f8fc4c7486c870885a0aae37460b8 Mon Sep 17 00:00:00 2001 From: andrey Date: Thu, 16 Jul 2026 09:51:39 +0300 Subject: [PATCH] feat: add ProcessExecutor, FAKE mode, fix thread lifecycle crash - 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 - 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 --- AGENTS.md | 34 +++-- pipeline-runner/CMakeLists.txt | 16 ++- pipeline-runner/build.sh | 58 +++++++++ pipeline-runner/src/processexecutor.cpp | 30 +++++ pipeline-runner/src/processexecutor.h | 20 +++ pipeline-runner/src/runner.cpp | 72 +++++++---- pipeline-runner/src/runner.h | 5 +- pipeline-runner/tests/test_runner.cpp | 160 ++++++++++++++++++++++++ 8 files changed, 361 insertions(+), 34 deletions(-) create mode 100755 pipeline-runner/build.sh create mode 100644 pipeline-runner/src/processexecutor.cpp create mode 100644 pipeline-runner/src/processexecutor.h diff --git a/AGENTS.md b/AGENTS.md index 7296801..86be863 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,7 +1,7 @@ # AGENTS.md ## Repo layout -- `pipeline-runner/` — C++ application built with PIP (Platform Independent Primitives). +- `pipeline-runner/` — C++11 application built with PIP (Platform Independent Primitives). - `pipeline-runner/src/main.cpp` — entry point, config loading, server start. - `pipeline-runner/src/server.h/.cpp` — PIHTTPServer routes, all API endpoints. - `pipeline-runner/src/pipeline.h/.cpp` — Pipeline/Prompt structs, JSON file I/O, log functions. @@ -10,14 +10,29 @@ - `pipeline-runner/pipeline-runner.conf` — JSON config file (port, directories). - `pipeline-runner/tests/` — gtest tests for pipeline storage and runner. - `pipeline-runner/CMakeLists.txt` — CMake build config with PIP + gtest (FetchContent). +- `pipeline-runner/build.sh` — build + test script (incremental by default, `--rebuild` for clean). - `pipeline-runner/storage/pipelines/` — JSON files for persisted pipelines. - `pipeline-runner/storage/logs/` — execution logs (one file per run). ## Dev commands -- Build: `cd pipeline-runner/build && cmake .. && make` +- Build + test: `./pipeline-runner/build.sh` (incremental, auto-configures on first run). +- Clean build + test: `./pipeline-runner/build.sh --rebuild`. +- Build with FAKE mode: `./pipeline-runner/build.sh --rebuild --fake` (uses fake process executor, no opencode needed). +- Switching `--fake` on/off triggers a clean + reconfigure automatically. - Run: `cd pipeline-runner/build && ./pipeline-runner ../pipeline-runner.conf` -- Run tests: `cd pipeline-runner/build && ./test-pipeline-runner` -- Clean build: `rm -rf pipeline-runner/build && mkdir pipeline-runner/build && cd pipeline-runner/build && cmake .. && make` +- Test XML report: generated at `pipeline-runner/build/test-results.xml` after each test run. +- CMake uses `-S -B ` and `cmake --build -j$(nproc)` (no `cd build && make`). + +## Build requirements +- **PIP library** must be installed system-wide (`find_package(PIP REQUIRED)`). +- gtest is fetched via custom mirror: `https://git.shstk.ru/mirrors/googletest.git` (v1.14.0). +- `CMAKE_EXPORT_COMPILE_COMMANDS ON` — `compile_commands.json` is generated in build dir for LSP. +- `CMakeLists.txt` uses `file(GLOB ...)` for sources — new `.cpp` files require re-running `cmake ..`. +- RPATH is set to `$ORIGIN;$ORIGIN/lib` so the binary finds PIP shared libs from its directory. + +## Formatting +- Tabs (width 4), LF line endings, UTF-8 — enforced by `.editorconfig`. +- `.clang-format` column limit: 140. Use `clang-format -i` on changed files. ## API endpoints | Method | Path | Description | @@ -31,8 +46,13 @@ | GET | `/api/runs/{run_id}/log` | Get log file contents | ## Key behaviors -- Pipeline execution runs `opencode run --title ` per step via `PIProcess`. +- C++11 standard — no modern C++ features (no `auto` return type deduction, no `std::optional`, etc.). +- All source code is in the **global namespace** (no namespace wrapping). +- Pipeline execution runs `opencode run <text> --title <title>` per step via `PIProcess` (not a shell). +- Each step has a **300-second timeout** (`proc.waitForFinish(PISystemTime::fromSeconds(300))`). - Pipeline stops on first step error (non-zero return code). -- Each run executes in a separate `PIThread`. +- Each run executes in a separate `PIThread` (thread object is intentionally leaked; PIP manages lifetime). +- Run state is **in-memory only** — lost on server restart. Log files persist to disk. - CORS headers are added to all responses (`Access-Control-Allow-Origin: *`). -- Config is loaded from `pipeline-runner.conf` (JSON format). +- Config paths (`pipelines_dir`, `logs_dir`) are relative to the **process working directory**, not the binary path. +- Tests use `/tmp/pipeline-runner-test/` with manual setup/cleanup per test — **not isolated**, do not run in parallel. diff --git a/pipeline-runner/CMakeLists.txt b/pipeline-runner/CMakeLists.txt index c858acf..1b2d4de 100644 --- a/pipeline-runner/CMakeLists.txt +++ b/pipeline-runner/CMakeLists.txt @@ -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) diff --git a/pipeline-runner/build.sh b/pipeline-runner/build.sh new file mode 100755 index 0000000..a100ab2 --- /dev/null +++ b/pipeline-runner/build.sh @@ -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" diff --git a/pipeline-runner/src/processexecutor.cpp b/pipeline-runner/src/processexecutor.cpp new file mode 100644 index 0000000..05f9aa0 --- /dev/null +++ b/pipeline-runner/src/processexecutor.cpp @@ -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; +} diff --git a/pipeline-runner/src/processexecutor.h b/pipeline-runner/src/processexecutor.h new file mode 100644 index 0000000..5ce47d8 --- /dev/null +++ b/pipeline-runner/src/processexecutor.h @@ -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 diff --git a/pipeline-runner/src/runner.cpp b/pipeline-runner/src/runner.cpp index da19770..d471bd1 100644 --- a/pipeline-runner/src/runner.cpp +++ b/pipeline-runner/src/runner.cpp @@ -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) { diff --git a/pipeline-runner/src/runner.h b/pipeline-runner/src/runner.h index bbd471e..6da3add 100644 --- a/pipeline-runner/src/runner.h +++ b/pipeline-runner/src/runner.h @@ -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 diff --git a/pipeline-runner/tests/test_runner.cpp b/pipeline-runner/tests/test_runner.cpp index 943ceca..b8e04be 100644 --- a/pipeline-runner/tests/test_runner.cpp +++ b/pipeline-runner/tests/test_runner.cpp @@ -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