rewrite all to PIP
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
#include "server.h"
|
||||
|
||||
#include <picout.h>
|
||||
#include <pifile.h>
|
||||
#include <pijson.h>
|
||||
#include <pikbdlistener.h>
|
||||
#include <pisignals.h>
|
||||
|
||||
static PIString getConfigValue(const PIJSON & config, const PIString & key, const PIString & def = PIString()) {
|
||||
if (config.contains(key)) {
|
||||
return config[key].toString();
|
||||
}
|
||||
return def;
|
||||
}
|
||||
|
||||
static int getConfigInt(const PIJSON & config, const PIString & key, int def) {
|
||||
if (config.contains(key)) {
|
||||
return config[key].toInt();
|
||||
}
|
||||
return def;
|
||||
}
|
||||
|
||||
int main(int argc, char * argv[]) {
|
||||
// Load config file
|
||||
PIString configPath = "./pipeline-runner.conf";
|
||||
PIJSON config;
|
||||
|
||||
if (argc > 1) {
|
||||
configPath = PIString(argv[1]);
|
||||
}
|
||||
|
||||
if (PIFile::isExists(configPath)) {
|
||||
PIByteArray data = PIFile::readAll(configPath);
|
||||
config = PIJSON::fromJSON(PIString::fromUTF8(data));
|
||||
piCout << "Loaded config from: " << configPath;
|
||||
} else {
|
||||
piCout << "Config file not found: " << configPath << ", using defaults";
|
||||
}
|
||||
|
||||
// Parse config
|
||||
int port = getConfigInt(config, "port", 8000);
|
||||
PIString pipelinesDir = getConfigValue(config, "pipelines_dir", "./storage/pipelines/");
|
||||
PIString logsDir = getConfigValue(config, "logs_dir", "./storage/logs/");
|
||||
|
||||
piCout << "Port: " << port;
|
||||
piCout << "Pipelines dir: " << pipelinesDir;
|
||||
piCout << "Logs dir: " << logsDir;
|
||||
|
||||
// Handle signals
|
||||
PISignals::setSlot([](PISignals::Signal s) {
|
||||
piCout << "Interrupt received, shutting down...";
|
||||
PIKbdListener::exiting = true;
|
||||
PISignals::releaseSignals(s);
|
||||
});
|
||||
PISignals::grabSignals(PISignals::Interrupt);
|
||||
|
||||
// Start server
|
||||
Server server(static_cast<ushort>(port), pipelinesDir, logsDir);
|
||||
server.start();
|
||||
|
||||
// Wait for exit
|
||||
PIKbdListener listener;
|
||||
listener.start();
|
||||
WAIT_FOR_EXIT
|
||||
listener.stopAndWait();
|
||||
|
||||
piCout << "Server stopped";
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
#include "messageutils.h"
|
||||
|
||||
#include <pihttpserver.h>
|
||||
|
||||
namespace MessageUtils {
|
||||
|
||||
PIHTTP::MessageMutable jsonReply(PIHTTP::Code code, const PIJSON & json) {
|
||||
PIHTTP::MessageMutable msg;
|
||||
msg.setCode(code);
|
||||
msg.setBody(json.toJSON(PIJSON::Compact).toUTF8());
|
||||
msg.addHeader("Content-Type", "application/json; charset=utf-8");
|
||||
return msg;
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable errorReply(PIHTTP::Code code, const PIString & message) {
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["error"] = message;
|
||||
return jsonReply(code, j);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable successReply(const PIJSON & json) {
|
||||
return jsonReply(PIHTTP::Code::Ok, json);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable noContent() {
|
||||
PIHTTP::MessageMutable msg;
|
||||
msg.setCode(PIHTTP::Code::NoContent);
|
||||
return msg;
|
||||
}
|
||||
|
||||
} // namespace MessageUtils
|
||||
@@ -0,0 +1,17 @@
|
||||
#ifndef MESSAGEUTILS_H
|
||||
#define MESSAGEUTILS_H
|
||||
|
||||
#include <pihttptypes.h>
|
||||
#include <pijson.h>
|
||||
#include <pistring.h>
|
||||
|
||||
namespace MessageUtils {
|
||||
|
||||
PIHTTP::MessageMutable jsonReply(PIHTTP::Code code, const PIJSON & json);
|
||||
PIHTTP::MessageMutable errorReply(PIHTTP::Code code, const PIString & message);
|
||||
PIHTTP::MessageMutable successReply(const PIJSON & json);
|
||||
PIHTTP::MessageMutable noContent();
|
||||
|
||||
} // namespace MessageUtils
|
||||
|
||||
#endif
|
||||
@@ -0,0 +1,138 @@
|
||||
#include "pipeline.h"
|
||||
|
||||
#include <picrypt.h>
|
||||
#include <pidatetime.h>
|
||||
#include <pidir.h>
|
||||
#include <pifile.h>
|
||||
#include <piiodevice.h>
|
||||
|
||||
static PIJSON promptToJSON(const Prompt & p) {
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["id"] = p.id;
|
||||
j["text"] = p.text;
|
||||
j["title"] = p.title;
|
||||
j["order"] = p.order;
|
||||
return j;
|
||||
}
|
||||
|
||||
static Prompt promptFromJSON(const PIJSON & j) {
|
||||
Prompt p;
|
||||
p.id = j["id"].toString();
|
||||
p.text = j["text"].toString();
|
||||
p.title = j["title"].toString();
|
||||
p.order = j["order"].toInt();
|
||||
return p;
|
||||
}
|
||||
|
||||
static PIJSON pipelineToJSON(const Pipeline & pl) {
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["id"] = pl.id;
|
||||
j["name"] = pl.name;
|
||||
j["working_dir"] = pl.working_dir;
|
||||
j["created_at"] = pl.created_at;
|
||||
j["updated_at"] = pl.updated_at;
|
||||
|
||||
PIJSON arr = PIJSON::newArray();
|
||||
for (const auto & p: pl.prompts) {
|
||||
arr << promptToJSON(p);
|
||||
}
|
||||
j["prompts"] = arr;
|
||||
return j;
|
||||
}
|
||||
|
||||
static Pipeline pipelineFromJSON(const PIJSON & j) {
|
||||
Pipeline pl;
|
||||
pl.id = j["id"].toString();
|
||||
pl.name = j["name"].toString();
|
||||
pl.working_dir = j["working_dir"].toString();
|
||||
pl.created_at = j["created_at"].toString();
|
||||
pl.updated_at = j["updated_at"].toString();
|
||||
|
||||
PIJSON promptsJ = j["prompts"];
|
||||
if (promptsJ.isArray()) {
|
||||
const auto & promptsArr = promptsJ.array();
|
||||
for (int i = 0; i < promptsArr.size(); ++i) {
|
||||
if (promptsArr[i].isObject()) {
|
||||
pl.prompts << promptFromJSON(promptsArr[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
return pl;
|
||||
}
|
||||
|
||||
PIVector<Pipeline> loadPipelines(const PIString & pipelines_dir) {
|
||||
PIVector<Pipeline> result;
|
||||
PIDir dir(pipelines_dir);
|
||||
if (!dir.isExists()) return result;
|
||||
|
||||
const auto entries = dir.entries();
|
||||
for (const auto & entry: entries) {
|
||||
if (!entry.isFile() || entry.extension() != "json") continue;
|
||||
PIByteArray data = PIFile::readAll(entry.path);
|
||||
PIJSON j = PIJSON::fromJSON(PIString::fromUTF8(data));
|
||||
if (j.isObject()) {
|
||||
result << pipelineFromJSON(j);
|
||||
}
|
||||
}
|
||||
// Sort by updated_at descending
|
||||
result.sort([](const Pipeline & a, const Pipeline & b) { return a.updated_at > b.updated_at; });
|
||||
return result;
|
||||
}
|
||||
|
||||
bool savePipeline(const PIString & pipelines_dir, Pipeline & pipeline) {
|
||||
pipeline.updated_at = nowISO();
|
||||
PIJSON j = pipelineToJSON(pipeline);
|
||||
PIString path = pipelines_dir + pipeline.id + ".json";
|
||||
return PIFile::writeAll(path, j.toJSON(PIJSON::Tree).toUTF8());
|
||||
}
|
||||
|
||||
bool deletePipeline(const PIString & pipelines_dir, const PIString & pipeline_id) {
|
||||
PIString path = pipelines_dir + pipeline_id + ".json";
|
||||
return PIFile::remove(path);
|
||||
}
|
||||
|
||||
Pipeline findPipeline(const PIString & pipelines_dir, const PIString & pipeline_id) {
|
||||
PIString path = pipelines_dir + pipeline_id + ".json";
|
||||
if (!PIFile::isExists(path)) return Pipeline();
|
||||
PIByteArray data = PIFile::readAll(path);
|
||||
PIJSON j = PIJSON::fromJSON(PIString::fromUTF8(data));
|
||||
if (j.isObject()) {
|
||||
return pipelineFromJSON(j);
|
||||
}
|
||||
return Pipeline();
|
||||
}
|
||||
|
||||
void appendLog(const PIString & logs_dir, const PIString & run_id, const PIString & line) {
|
||||
PIString path = logs_dir + run_id + ".log";
|
||||
PIFile f(path, PIIODevice::ReadWrite);
|
||||
if (f.isOpened()) {
|
||||
f.seekToEnd();
|
||||
PIString ts = nowISO();
|
||||
f.write((ts + " " + line + "\n").toUTF8());
|
||||
f.close();
|
||||
}
|
||||
}
|
||||
|
||||
PIString readLog(const PIString & logs_dir, const PIString & run_id) {
|
||||
PIString path = logs_dir + run_id + ".log";
|
||||
if (!PIFile::isExists(path)) return PIString();
|
||||
PIByteArray data = PIFile::readAll(path);
|
||||
return PIString::fromUTF8(data);
|
||||
}
|
||||
|
||||
PIString generateUUID() {
|
||||
PIByteArray bytes = PICrypt::generateRandomBuff(16);
|
||||
PIString hex = bytes.toHex();
|
||||
// Format as UUID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
|
||||
return hex.left(8) + "-" + hex.mid(8, 4) + "-" + hex.mid(12, 4) + "-" + hex.mid(16, 4) + "-" + hex.mid(20, 12);
|
||||
}
|
||||
|
||||
static PIString pad2(int v) {
|
||||
return v < 10 ? "0" + PIString::fromNumber(v) : PIString::fromNumber(v);
|
||||
}
|
||||
|
||||
PIString nowISO() {
|
||||
PIDateTime dt = PIDateTime::current();
|
||||
return PIString::fromNumber(dt.year) + "-" + pad2(dt.month) + "-" + pad2(dt.day) + "T" + pad2(dt.hours) + ":" + pad2(dt.minutes) + ":" +
|
||||
pad2(dt.seconds) + "Z";
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
#ifndef PIPELINE_H
|
||||
#define PIPELINE_H
|
||||
|
||||
#include <pijson.h>
|
||||
#include <pimap.h>
|
||||
#include <pistring.h>
|
||||
#include <pivector.h>
|
||||
|
||||
struct Prompt {
|
||||
PIString id;
|
||||
PIString text;
|
||||
PIString title;
|
||||
int order = 0;
|
||||
};
|
||||
|
||||
struct Pipeline {
|
||||
PIString id;
|
||||
PIString name;
|
||||
PIVector<Prompt> prompts;
|
||||
PIString working_dir;
|
||||
PIString created_at;
|
||||
PIString updated_at;
|
||||
};
|
||||
|
||||
// Storage functions
|
||||
PIVector<Pipeline> loadPipelines(const PIString & pipelines_dir);
|
||||
bool savePipeline(const PIString & pipelines_dir, Pipeline & pipeline);
|
||||
bool deletePipeline(const PIString & pipelines_dir, const PIString & pipeline_id);
|
||||
Pipeline findPipeline(const PIString & pipelines_dir, const PIString & pipeline_id);
|
||||
|
||||
// Log functions
|
||||
void appendLog(const PIString & logs_dir, const PIString & run_id, const PIString & line);
|
||||
PIString readLog(const PIString & logs_dir, const PIString & run_id);
|
||||
|
||||
// Utility
|
||||
PIString generateUUID();
|
||||
PIString nowISO();
|
||||
|
||||
#endif
|
||||
@@ -0,0 +1,149 @@
|
||||
#include "runner.h"
|
||||
|
||||
#include "pipeline.h"
|
||||
|
||||
#include <piprocess.h>
|
||||
#include <pisystemtime.h>
|
||||
|
||||
PipelineRunner::PipelineRunner(const PIString & pipelines_dir, const PIString & logs_dir)
|
||||
: pipelines_dir_(pipelines_dir)
|
||||
, logs_dir_(logs_dir) {}
|
||||
|
||||
PipelineRunner::~PipelineRunner() {}
|
||||
|
||||
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
|
||||
// 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();
|
||||
|
||||
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();
|
||||
|
||||
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>
|
||||
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));
|
||||
|
||||
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;
|
||||
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");
|
||||
}
|
||||
|
||||
void PipelineRunner::appendLog(const PIString & run_id, const PIString & line) {
|
||||
::appendLog(logs_dir_, run_id, line);
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
#ifndef RUNNER_H
|
||||
#define RUNNER_H
|
||||
|
||||
#include "pipeline.h"
|
||||
|
||||
#include <pimap.h>
|
||||
#include <pimutex.h>
|
||||
#include <piobject.h>
|
||||
#include <pistring.h>
|
||||
#include <pivector.h>
|
||||
|
||||
enum class RunStatus {
|
||||
Pending,
|
||||
Running,
|
||||
Completed,
|
||||
Error
|
||||
};
|
||||
|
||||
static PIString runStatusToString(RunStatus s) {
|
||||
switch (s) {
|
||||
case RunStatus::Pending: return "pending";
|
||||
case RunStatus::Running: return "running";
|
||||
case RunStatus::Completed: return "completed";
|
||||
case RunStatus::Error: return "error";
|
||||
}
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
struct StepResult {
|
||||
int step_index = 0;
|
||||
PIString title;
|
||||
RunStatus status = RunStatus::Pending;
|
||||
int returncode = 0;
|
||||
PIString output;
|
||||
PIString error;
|
||||
};
|
||||
|
||||
struct RunState {
|
||||
PIString run_id;
|
||||
PIString pipeline_id;
|
||||
RunStatus status = RunStatus::Pending;
|
||||
int current_step = -1;
|
||||
PIVector<StepResult> steps;
|
||||
};
|
||||
|
||||
class PipelineRunner: public PIObject {
|
||||
PIOBJECT(PipelineRunner)
|
||||
|
||||
public:
|
||||
PipelineRunner(const PIString & pipelines_dir, const PIString & logs_dir);
|
||||
~PipelineRunner();
|
||||
|
||||
// Start a pipeline run, returns run_id or empty string on error
|
||||
PIString startRun(const PIString & pipeline_id);
|
||||
|
||||
// Get current state of a run
|
||||
RunState getRunState(const PIString & run_id);
|
||||
|
||||
// Check if run is still active
|
||||
bool isRunActive(const PIString & run_id);
|
||||
|
||||
private:
|
||||
void executePipeline(const PIString & run_id, const Pipeline & pipeline);
|
||||
void appendLog(const PIString & run_id, const PIString & line);
|
||||
|
||||
const PIString pipelines_dir_;
|
||||
const PIString logs_dir_;
|
||||
|
||||
PIMutex mutex_;
|
||||
PIMap<PIString, RunState> runs_;
|
||||
PIMap<PIString, bool> active_runs_;
|
||||
};
|
||||
|
||||
#endif
|
||||
@@ -0,0 +1,261 @@
|
||||
#include "server.h"
|
||||
|
||||
#include "messageutils.h"
|
||||
|
||||
#include <pidir.h>
|
||||
#include <pifile.h>
|
||||
#include <pihttpserver.h>
|
||||
|
||||
Server::Server(ushort port, const PIString & pipelines_dir, const PIString & logs_dir)
|
||||
: port_(port)
|
||||
, pipelines_dir_(pipelines_dir)
|
||||
, logs_dir_(logs_dir)
|
||||
, runner_(pipelines_dir, logs_dir) {
|
||||
httpserver_ = new PIHTTPServer();
|
||||
|
||||
// CORS headers
|
||||
httpserver_->addReplyHeader("Access-Control-Allow-Origin", "*");
|
||||
httpserver_->addReplyHeader("Access-Control-Allow-Methods", "GET, POST, DELETE, PUT, OPTIONS");
|
||||
httpserver_->addReplyHeader("Access-Control-Allow-Headers", "Content-Type");
|
||||
}
|
||||
|
||||
Server::~Server() {
|
||||
delete httpserver_;
|
||||
}
|
||||
|
||||
bool Server::start() {
|
||||
// Ensure storage directories exist
|
||||
PIDir::make(pipelines_dir_);
|
||||
PIDir::make(logs_dir_);
|
||||
|
||||
// Register routes
|
||||
httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Get, this, &Server::listPipelines);
|
||||
httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Post, this, &Server::createPipeline);
|
||||
httpserver_->registerPath("/api/pipelines/{id}", PIHTTP::Method::Get, this, &Server::getPipeline);
|
||||
httpserver_->registerPath("/api/pipelines/{id}", PIHTTP::Method::Delete, this, &Server::deletePipeline);
|
||||
httpserver_->registerPath("/api/runs", PIHTTP::Method::Post, this, &Server::startRun);
|
||||
httpserver_->registerPath("/api/runs/{run_id}/status", PIHTTP::Method::Get, this, &Server::getRunStatus);
|
||||
httpserver_->registerPath("/api/runs/{run_id}/log", PIHTTP::Method::Get, this, &Server::getRunLog);
|
||||
|
||||
// Unhandled request handler
|
||||
httpserver_->registerUnhandled(this, &Server::unhandledRequest);
|
||||
|
||||
bool ok = httpserver_->listenAll(port_);
|
||||
piCout << "Server started on port " << port_ << " (ok=" << ok << ")";
|
||||
return ok;
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::listPipelines(const PIHTTP::MessageConst & request) {
|
||||
piCout << "GET /api/pipelines";
|
||||
PIVector<Pipeline> pipelines = loadPipelines(pipelines_dir_);
|
||||
|
||||
PIJSON arr = PIJSON::newArray();
|
||||
for (const auto & pl: pipelines) {
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["id"] = pl.id;
|
||||
j["name"] = pl.name;
|
||||
j["working_dir"] = pl.working_dir;
|
||||
j["created_at"] = pl.created_at;
|
||||
j["updated_at"] = pl.updated_at;
|
||||
|
||||
PIJSON promptsArr = PIJSON::newArray();
|
||||
for (const auto & p: pl.prompts) {
|
||||
PIJSON pj = PIJSON::newObject();
|
||||
pj["id"] = p.id;
|
||||
pj["text"] = p.text;
|
||||
pj["title"] = p.title;
|
||||
pj["order"] = p.order;
|
||||
promptsArr << pj;
|
||||
}
|
||||
j["prompts"] = promptsArr;
|
||||
arr << j;
|
||||
}
|
||||
return MessageUtils::successReply(arr);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::createPipeline(const PIHTTP::MessageConst & request) {
|
||||
piCout << "POST /api/pipelines";
|
||||
|
||||
PIString body = PIString::fromUTF8(request.body());
|
||||
PIJSON j = PIJSON::fromJSON(body);
|
||||
if (!j.isObject()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::BadRequest, "Invalid JSON");
|
||||
}
|
||||
|
||||
Pipeline pl;
|
||||
pl.id = j["id"].toString();
|
||||
pl.name = j["name"].toString();
|
||||
pl.working_dir = j["working_dir"].toString();
|
||||
pl.created_at = j.contains("created_at") ? j["created_at"].toString() : nowISO();
|
||||
pl.updated_at = pl.created_at;
|
||||
|
||||
if (pl.id.isEmpty() || pl.name.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::BadRequest, "id and name are required");
|
||||
}
|
||||
|
||||
// Parse prompts
|
||||
const auto & promptsArr = j["prompts"].array();
|
||||
for (int i = 0; i < promptsArr.size(); ++i) {
|
||||
Prompt p;
|
||||
p.id = promptsArr[i]["id"].toString();
|
||||
p.text = promptsArr[i]["text"].toString();
|
||||
p.title = promptsArr[i]["title"].toString();
|
||||
p.order = promptsArr[i]["order"].toInt();
|
||||
pl.prompts << p;
|
||||
}
|
||||
|
||||
// Check for duplicate
|
||||
Pipeline existing = findPipeline(pipelines_dir_, pl.id);
|
||||
if (!existing.id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::Conflict, "Pipeline already exists");
|
||||
}
|
||||
|
||||
if (savePipeline(pipelines_dir_, pl)) {
|
||||
PIJSON resp = PIJSON::newObject();
|
||||
resp["id"] = pl.id;
|
||||
resp["name"] = pl.name;
|
||||
resp["working_dir"] = pl.working_dir;
|
||||
resp["created_at"] = pl.created_at;
|
||||
resp["updated_at"] = pl.updated_at;
|
||||
PIJSON pArr = PIJSON::newArray();
|
||||
for (const auto & p: pl.prompts) {
|
||||
PIJSON pj = PIJSON::newObject();
|
||||
pj["id"] = p.id;
|
||||
pj["text"] = p.text;
|
||||
pj["title"] = p.title;
|
||||
pj["order"] = p.order;
|
||||
pArr << pj;
|
||||
}
|
||||
resp["prompts"] = pArr;
|
||||
return MessageUtils::successReply(resp);
|
||||
}
|
||||
|
||||
return MessageUtils::errorReply(PIHTTP::Code::InternalServerError, "Failed to save pipeline");
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::getPipeline(const PIHTTP::MessageConst & request) {
|
||||
PIString id = request.pathArguments().value("id");
|
||||
piCout << "GET /api/pipelines/" << id;
|
||||
|
||||
Pipeline pl = findPipeline(pipelines_dir_, id);
|
||||
if (pl.id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::NotFound, "Pipeline not found");
|
||||
}
|
||||
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["id"] = pl.id;
|
||||
j["name"] = pl.name;
|
||||
j["working_dir"] = pl.working_dir;
|
||||
j["created_at"] = pl.created_at;
|
||||
j["updated_at"] = pl.updated_at;
|
||||
|
||||
PIJSON pArr = PIJSON::newArray();
|
||||
for (const auto & p: pl.prompts) {
|
||||
PIJSON pj = PIJSON::newObject();
|
||||
pj["id"] = p.id;
|
||||
pj["text"] = p.text;
|
||||
pj["title"] = p.title;
|
||||
pj["order"] = p.order;
|
||||
pArr << pj;
|
||||
}
|
||||
j["prompts"] = pArr;
|
||||
return MessageUtils::successReply(j);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::deletePipeline(const PIHTTP::MessageConst & request) {
|
||||
PIString id = request.pathArguments().value("id");
|
||||
piCout << "DELETE /api/pipelines/" << id;
|
||||
|
||||
Pipeline existing = findPipeline(pipelines_dir_, id);
|
||||
if (existing.id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::NotFound, "Pipeline not found");
|
||||
}
|
||||
|
||||
if (::deletePipeline(pipelines_dir_, id)) {
|
||||
PIJSON j = PIJSON::newObject();
|
||||
j["deleted"] = id;
|
||||
return MessageUtils::successReply(j);
|
||||
}
|
||||
|
||||
return MessageUtils::errorReply(PIHTTP::Code::InternalServerError, "Failed to delete pipeline");
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::startRun(const PIHTTP::MessageConst & request) {
|
||||
piCout << "POST /api/runs";
|
||||
|
||||
PIString body = PIString::fromUTF8(request.body());
|
||||
PIJSON j = PIJSON::fromJSON(body);
|
||||
if (!j.isObject()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::BadRequest, "Invalid JSON");
|
||||
}
|
||||
|
||||
PIString pipeline_id = j["pipeline_id"].toString();
|
||||
if (pipeline_id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::BadRequest, "pipeline_id is required");
|
||||
}
|
||||
|
||||
// Check pipeline exists
|
||||
Pipeline pl = findPipeline(pipelines_dir_, pipeline_id);
|
||||
if (pl.id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::NotFound, "Pipeline not found");
|
||||
}
|
||||
|
||||
PIString run_id = runner_.startRun(pipeline_id);
|
||||
if (run_id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::InternalServerError, "Failed to start run");
|
||||
}
|
||||
|
||||
PIJSON resp = PIJSON::newObject();
|
||||
resp["run_id"] = run_id;
|
||||
resp["pipeline_id"] = pipeline_id;
|
||||
resp["status"] = "running";
|
||||
return MessageUtils::successReply(resp);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::getRunStatus(const PIHTTP::MessageConst & request) {
|
||||
PIString run_id = request.pathArguments().value("run_id");
|
||||
piCout << "GET /api/runs/" << run_id << "/status";
|
||||
|
||||
RunState state = runner_.getRunState(run_id);
|
||||
if (state.run_id.isEmpty()) {
|
||||
return MessageUtils::errorReply(PIHTTP::Code::NotFound, "Run not found");
|
||||
}
|
||||
|
||||
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 (const auto & step: state.steps) {
|
||||
PIJSON sj = PIJSON::newObject();
|
||||
sj["step_index"] = step.step_index;
|
||||
sj["title"] = step.title;
|
||||
sj["status"] = runStatusToString(step.status);
|
||||
sj["returncode"] = step.returncode;
|
||||
sj["output"] = step.output;
|
||||
sj["error"] = step.error;
|
||||
stepsArr << sj;
|
||||
}
|
||||
j["steps"] = stepsArr;
|
||||
|
||||
return MessageUtils::successReply(j);
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::getRunLog(const PIHTTP::MessageConst & request) {
|
||||
PIString run_id = request.pathArguments().value("run_id");
|
||||
piCout << "GET /api/runs/" << run_id << "/log";
|
||||
|
||||
PIString log = ::readLog(logs_dir_, run_id);
|
||||
PIHTTP::MessageMutable msg;
|
||||
msg.setCode(PIHTTP::Code::Ok);
|
||||
msg.setBody(log.toUTF8());
|
||||
msg.addHeader("Content-Type", "text/plain; charset=utf-8");
|
||||
return msg;
|
||||
}
|
||||
|
||||
PIHTTP::MessageMutable Server::unhandledRequest(const PIHTTP::MessageConst & request) {
|
||||
piCout << "Unhandled: " << PIHTTP::methodName(request.method()) << " " << request.path();
|
||||
return MessageUtils::errorReply(PIHTTP::Code::NotFound, "Not found");
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
#ifndef SERVER_H
|
||||
#define SERVER_H
|
||||
|
||||
#include "pipeline.h"
|
||||
#include "runner.h"
|
||||
|
||||
#include <pihttptypes.h>
|
||||
#include <piobject.h>
|
||||
|
||||
class PIHTTPServer;
|
||||
|
||||
class Server: public PIObject {
|
||||
PIOBJECT(Server)
|
||||
|
||||
public:
|
||||
Server(ushort port, const PIString & pipelines_dir, const PIString & logs_dir);
|
||||
~Server();
|
||||
|
||||
bool start();
|
||||
|
||||
private:
|
||||
PIHTTP::MessageMutable listPipelines(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable createPipeline(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable deletePipeline(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable getPipeline(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable startRun(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable getRunStatus(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable getRunLog(const PIHTTP::MessageConst & request);
|
||||
PIHTTP::MessageMutable unhandledRequest(const PIHTTP::MessageConst & request);
|
||||
|
||||
ushort port_;
|
||||
PIString pipelines_dir_;
|
||||
PIString logs_dir_;
|
||||
PIHTTPServer * httpserver_;
|
||||
PipelineRunner runner_;
|
||||
};
|
||||
|
||||
#endif
|
||||
Reference in New Issue
Block a user