From e37545ffff517e2f4747887b4c1c48189d21157f Mon Sep 17 00:00:00 2001 From: andrey Date: Thu, 16 Jul 2026 10:23:03 +0300 Subject: [PATCH] feat: persist run state to disk with recovery on startup --- pipeline-runner/src/runner.cpp | 159 +++++++++++++++++++++++++- pipeline-runner/src/runner.h | 10 ++ pipeline-runner/src/server.cpp | 3 + pipeline-runner/tests/test_runner.cpp | 142 +++++++++++++++++++++++ 4 files changed, 310 insertions(+), 4 deletions(-) diff --git a/pipeline-runner/src/runner.cpp b/pipeline-runner/src/runner.cpp index d471bd1..448c080 100644 --- a/pipeline-runner/src/runner.cpp +++ b/pipeline-runner/src/runner.cpp @@ -2,11 +2,90 @@ #include "pipeline.h" +#include +#include +#include #include #include #include #include +static PIString getStateFilePath(const PIString & logs_dir, const PIString & run_id) { + return logs_dir + run_id + ".state.json"; +} + +PIJSON PipelineRunner::stepResultToJSON(const StepResult & sr) { + PIJSON j = PIJSON::newObject(); + j["step_index"] = sr.step_index; + j["title"] = sr.title; + j["status"] = runStatusToString(sr.status); + j["returncode"] = sr.returncode; + j["output"] = sr.output; + j["error"] = sr.error; + return j; +} + +StepResult PipelineRunner::stepResultFromJSON(const PIJSON & j) { + StepResult sr; + sr.step_index = j["step_index"].toInt(); + sr.title = j["title"].toString(); + PIString statusStr = j["status"].toString(); + if (statusStr == "pending") + sr.status = RunStatus::Pending; + else if (statusStr == "running") + sr.status = RunStatus::Running; + else if (statusStr == "completed") + sr.status = RunStatus::Completed; + else + sr.status = RunStatus::Error; + sr.returncode = j["returncode"].toInt(); + sr.output = j["output"].toString(); + sr.error = j["error"].toString(); + return sr; +} + +PIJSON PipelineRunner::runStateToJSON(const RunState & state) { + PIJSON j = PIJSON::newObject(); + j["run_id"] = state.run_id; + j["pipeline_id"] = state.pipeline_id; + j["status"] = runStatusToString(state.status); + j["current_step"] = state.current_step; + + PIJSON stepsArr = PIJSON::newArray(); + for (int i = 0; i < state.steps.size(); ++i) { + stepsArr << stepResultToJSON(state.steps[i]); + } + j["steps"] = stepsArr; + return j; +} + +RunState PipelineRunner::runStateFromJSON(const PIJSON & j) { + RunState state; + state.run_id = j["run_id"].toString(); + state.pipeline_id = j["pipeline_id"].toString(); + + PIString statusStr = j["status"].toString(); + if (statusStr == "pending") + state.status = RunStatus::Pending; + else if (statusStr == "running") + state.status = RunStatus::Running; + else if (statusStr == "completed") + state.status = RunStatus::Completed; + else + state.status = RunStatus::Error; + + state.current_step = j["current_step"].toInt(); + + PIJSON stepsArrJSON = j["steps"]; + const auto & stepsArr = stepsArrJSON.array(); + for (int i = 0; i < stepsArr.size(); ++i) { + if (stepsArr[i].isObject()) { + state.steps << stepResultFromJSON(stepsArr[i]); + } + } + return state; +} + PipelineRunner::PipelineRunner(const PIString & pipelines_dir, const PIString & logs_dir) : pipelines_dir_(pipelines_dir) , logs_dir_(logs_dir) {} @@ -47,6 +126,7 @@ PIString PipelineRunner::startRun(const PIString & pipeline_id) { active_runs_.insert(run_id, true); } + saveRunState(run_id); appendLog(run_id, "Starting pipeline: " + pipeline.name + " (run_id: " + run_id + ")"); // Execute in a separate thread @@ -59,11 +139,19 @@ PIString PipelineRunner::startRun(const PIString & pipeline_id) { } RunState PipelineRunner::getRunState(const PIString & run_id) { - PIMutexLocker ml(mutex_); - if (runs_.contains(run_id)) { - return runs_.value(run_id); + { + PIMutexLocker ml(mutex_); + if (runs_.contains(run_id)) { + return runs_.value(run_id); + } } - return RunState(); + + RunState state = loadRunState(run_id); + if (!state.run_id.isEmpty()) { + PIMutexLocker ml(mutex_); + runs_.insert(run_id, state); + } + return state; } bool PipelineRunner::isRunActive(const PIString & run_id) { @@ -100,6 +188,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p active_runs_.remove(run_id); } + saveRunState(run_id); + appendLog(run_id, "Pipeline finished"); #else for (int i = 0; i < total; ++i) { @@ -116,6 +206,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p } } + saveRunState(run_id); + // Build command: opencode run --title PIStringList args; args << "run" << prompt.text << "--title" << prompt.title; @@ -141,6 +233,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p } } + saveRunState(run_id); + if (rc != 0) { appendLog(run_id, "Pipeline failed at step " + PIString::fromNumber(i + 1)); { @@ -149,6 +243,7 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p runs_[run_id].status = RunStatus::Error; } } + saveRunState(run_id); break; } } @@ -162,6 +257,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p active_runs_.remove(run_id); } + saveRunState(run_id); + appendLog(run_id, "Pipeline finished"); #endif } @@ -169,3 +266,57 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p void PipelineRunner::appendLog(const PIString & run_id, const PIString & line) { ::appendLog(logs_dir_, run_id, line); } + +void PipelineRunner::saveRunState(const PIString & run_id) { + PIMutexLocker ml(mutex_); + if (!runs_.contains(run_id)) return; + + RunState state = runs_.value(run_id); + PIJSON j = runStateToJSON(state); + + PIString path = getStateFilePath(logs_dir_, run_id); + PIString tmpPath = path + ".tmp"; + + if (PIFile::writeAll(tmpPath, j.toJSON(PIJSON::Tree).toUTF8())) { + PIFile::remove(path); + PIFile::rename(tmpPath, path); + } +} + +RunState PipelineRunner::loadRunState(const PIString & run_id) { + PIString path = getStateFilePath(logs_dir_, run_id); + if (!PIFile::isExists(path)) return RunState(); + + PIByteArray data = PIFile::readAll(path); + PIJSON j = PIJSON::fromJSON(PIString::fromUTF8(data)); + if (j.isObject()) { + return runStateFromJSON(j); + } + return RunState(); +} + +void PipelineRunner::recoverRuns() { + PIDir dir(logs_dir_); + if (!dir.isExists()) return; + + const auto entries = dir.entries(); + for (const auto & entry: entries) { + if (!entry.isFile() || entry.extension() != "json") continue; + PIString fileName = entry.name(); + if (!fileName.endsWith(".state.json")) continue; + + PIString run_id = fileName.left(fileName.length() - PIString(".state.json").length()); + RunState state = loadRunState(run_id); + + if (state.status == RunStatus::Running) { + PIMutexLocker ml(mutex_); + runs_.insert(run_id, state); + active_runs_.insert(run_id, true); + } + } +} + +void PipelineRunner::removeStateFile(const PIString & run_id) { + PIString path = getStateFilePath(logs_dir_, run_id); + PIFile::remove(path); +} diff --git a/pipeline-runner/src/runner.h b/pipeline-runner/src/runner.h index 6da3add..4a9acb1 100644 --- a/pipeline-runner/src/runner.h +++ b/pipeline-runner/src/runner.h @@ -60,7 +60,17 @@ public: // Check if run is still active bool isRunActive(const PIString & run_id); + // Persist run state to disk + void saveRunState(const PIString & run_id); + RunState loadRunState(const PIString & run_id); + void recoverRuns(); + void removeStateFile(const PIString & run_id); + private: + PIJSON runStateToJSON(const RunState & state); + PIJSON stepResultToJSON(const StepResult & sr); + RunState runStateFromJSON(const PIJSON & j); + StepResult stepResultFromJSON(const PIJSON & j); void executePipeline(const PIString & run_id, const Pipeline & pipeline); void appendLog(const PIString & run_id, const PIString & line); diff --git a/pipeline-runner/src/server.cpp b/pipeline-runner/src/server.cpp index 493174e..5ad3080 100644 --- a/pipeline-runner/src/server.cpp +++ b/pipeline-runner/src/server.cpp @@ -28,6 +28,9 @@ bool Server::start() { PIDir::make(pipelines_dir_); PIDir::make(logs_dir_); + // Recover running state from disk + runner_.recoverRuns(); + // Register routes httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Get, this, &Server::listPipelines); httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Post, this, &Server::createPipeline); diff --git a/pipeline-runner/tests/test_runner.cpp b/pipeline-runner/tests/test_runner.cpp index b8e04be..10ba676 100644 --- a/pipeline-runner/tests/test_runner.cpp +++ b/pipeline-runner/tests/test_runner.cpp @@ -151,6 +151,148 @@ TEST(RunnerTest, RunStatusToString) { EXPECT_EQ(runStatusToString(RunStatus::Error), "error"); } +TEST(PersistenceTest, SaveLoadRoundTrip) { + setupDirs(); + + PipelineRunner runner(testPipelinesDir, testLogsDir); + + // Test loadRunState on non-existent file + RunState empty = runner.loadRunState("nonexistent"); + EXPECT_TRUE(empty.run_id.isEmpty()); + + // Create state file with one step + PIString runId = "test-run-123"; + PIString statePath = testLogsDir + runId + ".state.json"; + + PIJSON j = PIJSON::newObject(); + j["run_id"] = runId; + j["pipeline_id"] = "test-pipeline"; + j["status"] = "running"; + j["current_step"] = 0; + + PIJSON stepsArr = PIJSON::newArray(); + PIJSON sj = PIJSON::newObject(); + sj["step_index"] = 0; + sj["title"] = "Step 1"; + sj["status"] = "completed"; + sj["returncode"] = 0; + sj["output"] = "output data"; + sj["error"] = ""; + stepsArr << sj; + j["steps"] = stepsArr; + + PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8()); + + // Load via runner + RunState loaded = runner.loadRunState(runId); + EXPECT_EQ(loaded.run_id, runId); + EXPECT_EQ(loaded.pipeline_id, "test-pipeline"); + EXPECT_EQ(loaded.status, RunStatus::Running); + EXPECT_EQ(loaded.steps.size(), 1); + EXPECT_EQ(loaded.steps[0].title, "Step 1"); + EXPECT_EQ(loaded.steps[0].status, RunStatus::Completed); + + cleanupDirs(); +} + +TEST(PersistenceTest, RecoverRuns) { + setupDirs(); + + // Create state file manually to simulate a running state from a previous session + PIString runId = "recovered-run-456"; + PIString statePath = testLogsDir + runId + ".state.json"; + + PIJSON j = PIJSON::newObject(); + j["run_id"] = runId; + j["pipeline_id"] = "some-pipeline"; + j["status"] = "running"; + j["current_step"] = 1; + + PIJSON stepsArr = PIJSON::newArray(); + PIJSON sj = PIJSON::newObject(); + sj["step_index"] = 0; + sj["title"] = "Step 1"; + sj["status"] = "completed"; + sj["returncode"] = 0; + sj["output"] = ""; + sj["error"] = ""; + stepsArr << sj; + + PIJSON sj2 = PIJSON::newObject(); + sj2["step_index"] = 1; + sj2["title"] = "Step 2"; + sj2["status"] = "pending"; + sj2["returncode"] = 0; + sj2["output"] = ""; + sj2["error"] = ""; + stepsArr << sj2; + + j["steps"] = stepsArr; + PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8()); + + // Also create a completed state file that should NOT be recovered + PIString completedRunId = "completed-run-789"; + PIString completedStatePath = testLogsDir + completedRunId + ".state.json"; + + PIJSON j2 = PIJSON::newObject(); + j2["run_id"] = completedRunId; + j2["pipeline_id"] = "some-pipeline"; + j2["status"] = "completed"; + j2["current_step"] = 0; + j2["steps"] = PIJSON::newArray(); + PIFile::writeAll(completedStatePath, j2.toJSON(PIJSON::Tree).toUTF8()); + + // New runner should recover only the running state + PipelineRunner runner(testPipelinesDir, testLogsDir); + runner.recoverRuns(); + + RunState recovered = runner.getRunState(runId); + EXPECT_EQ(recovered.run_id, runId); + EXPECT_EQ(recovered.status, RunStatus::Running); + EXPECT_EQ(recovered.steps.size(), 2); + + // Completed run should NOT be in memory after recovery + EXPECT_FALSE(runner.isRunActive(completedRunId)); + + cleanupDirs(); +} + +TEST(PersistenceTest, CompletedRunStatePersisted) { + setupDirs(); + + PipelineRunner runner(testPipelinesDir, testLogsDir); + + Pipeline pl; + pl.id = "persist-test"; + pl.name = "Persist Test"; + pl.working_dir = "/tmp"; + pl.created_at = nowISO(); + pl.updated_at = pl.created_at; + + Prompt p; + p.id = "p1"; + p.text = "test"; + p.title = "Step 1"; + p.order = 0; + pl.prompts << p; + + savePipeline(testPipelinesDir, pl); + + PIString runId = runner.startRun("persist-test"); + EXPECT_FALSE(runId.isEmpty()); + + waitForRun(runner, runId); + + // Verify state file exists and contains final status + PIString statePath = testLogsDir + runId + ".state.json"; + EXPECT_TRUE(PIFile::isExists(statePath)); + + RunState loaded = runner.loadRunState(runId); + EXPECT_EQ(loaded.run_id, runId); + + cleanupDirs(); +} + #ifdef FAKE static Pipeline makeFakePipeline(const PIString & id, const PIString & name, int numSteps) {