diff --git a/.gitignore b/.gitignore index dcc8cab..fcd4947 100644 --- a/.gitignore +++ b/.gitignore @@ -1,55 +1,8 @@ -# Python -__pycache__/ -*.py[cod] -*$py.class -*.so -*.egg -*.egg-info/ -dist/ -build/ -*.whl - -# Virtual environments -venv/ -.venv/ -env/ -.env/ - -# IDE -.idea/ -.vscode/ -*.swp -*.swo -*~ - -# OS -.DS_Store -Thumbs.db - -# dotenv -.env -.env.* -!env.example - -# Logs -*.log -logs/ - -# Jupyter -.ipynb_checkpoints/ - -# Coverage -htmlcov/ -.coverage -.coverage.* - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pytest -.pytest_cache/ - -# Ruff -.ruff_cache/ +*.user +/build* +/*/build* +/bin +/release +*.orig +/.* +*test-results.xml diff --git a/AGENTS.md b/AGENTS.md index ef873fe..7296801 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,26 +1,38 @@ # AGENTS.md ## Repo layout -- `pipeline-runner/` — single Python package (FastAPI + uvicorn). All code lives here. -- `pipeline-runner/app.py` — FastAPI app entrypoint with REST API + WebSocket + static file serving. -- `pipeline-runner/main.py` — uvicorn dev server launcher (`python main.py`). -- `pipeline-runner/static/index.html` — frontend UI. +- `pipeline-runner/` — C++ 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. +- `pipeline-runner/src/runner.h/.cpp` — async pipeline execution (PIThread + PIProcess per step). +- `pipeline-runner/src/messageutils.h/.cpp` — HTTP response helpers (JSON, error, success). +- `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/storage/pipelines/` — JSON files for persisted pipelines. - `pipeline-runner/storage/logs/` — execution logs (one file per run). ## Dev commands -- Install deps: `cd pipeline-runner && pip install -r requirements.txt` -- Start dev server: `cd pipeline-runner && python main.py` (uvicorn with reload, port 8000) -- Deploy with auto-reload: `cd pipeline-runner && ./deploy.sh` (activates venv, installs deps, starts uvicorn with `--reload`) -- Run tests: `cd pipeline-runner && python -m pytest` - - Tests use `httpx`-backed `TestClient` from FastAPI. - - Tests that exercise pipeline execution (`TestExecutePipeline`) monkeypatch `asyncio.create_subprocess_exec` to avoid calling `opencode` — they do not require `opencode` to be installed. - - The `clean_storage` fixture auto-removes `storage/pipelines/` and `storage/logs/` before/after each test. +- Build: `cd pipeline-runner/build && cmake .. && make` +- 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` + +## API endpoints +| Method | Path | Description | +|--------|------|-------------| +| GET | `/api/pipelines` | List all pipelines | +| POST | `/api/pipelines` | Create pipeline (JSON body) | +| GET | `/api/pipelines/{id}` | Get single pipeline | +| DELETE | `/api/pipelines/{id}` | Delete pipeline | +| POST | `/api/runs` | Start execution (`{"pipeline_id": "..."}`) | +| GET | `/api/runs/{run_id}/status` | Poll run status + step results | +| GET | `/api/runs/{run_id}/log` | Get log file contents | ## Key behaviors -- Pipeline execution runs `opencode run --title ` per step via `asyncio.create_subprocess_exec`. +- Pipeline execution runs `opencode run <text> --title <title>` per step via `PIProcess`. - Pipeline stops on first step error (non-zero return code). -- Real-time logs are streamed over WebSocket at `/ws/{run_id}`. -- CORS is open (`allow_origins=["*"]`) for dev convenience. -- No lint, typecheck, or formatter tooling is configured. - +- Each run executes in a separate `PIThread`. +- CORS headers are added to all responses (`Access-Control-Allow-Origin: *`). +- Config is loaded from `pipeline-runner.conf` (JSON format). diff --git a/pipeline-runner/.clang-format b/pipeline-runner/.clang-format new file mode 100644 index 0000000..ab89af9 --- /dev/null +++ b/pipeline-runner/.clang-format @@ -0,0 +1,224 @@ +--- +Language: Cpp +AccessModifierOffset: -4 +AlignAfterOpenBracket: Align +AlignArrayOfStructures: Left +AlignConsecutiveAssignments: + Enabled: true + AcrossEmptyLines: true + AcrossComments: true + AlignCompound: false + PadOperators: true +AlignConsecutiveBitFields: + Enabled: true + AcrossEmptyLines: false + AcrossComments: true + AlignCompound: false + PadOperators: true +AlignConsecutiveDeclarations: + Enabled: false + AcrossEmptyLines: false + AcrossComments: false + AlignCompound: false + PadOperators: false +AlignConsecutiveMacros: + Enabled: true + AcrossEmptyLines: true + AcrossComments: true + AlignCompound: false + PadOperators: true +AlignEscapedNewlines: Left +AlignOperands: Align +AlignTrailingComments: true +AllowAllArgumentsOnNextLine: false +AllowAllParametersOfDeclarationOnNextLine: false +AllowShortEnumsOnASingleLine: false +AllowShortBlocksOnASingleLine: Empty +AllowShortCaseLabelsOnASingleLine: true +AllowShortFunctionsOnASingleLine: Inline +AllowShortLambdasOnASingleLine: All +AllowShortIfStatementsOnASingleLine: WithoutElse +AllowShortLoopsOnASingleLine: false +AlwaysBreakAfterDefinitionReturnType: None +AlwaysBreakAfterReturnType: None +AlwaysBreakBeforeMultilineStrings: false +AlwaysBreakTemplateDeclarations: Yes +AttributeMacros: + - __capability +BinPackArguments: false +BinPackParameters: false +BraceWrapping: + AfterCaseLabel: false + AfterClass: false + AfterControlStatement: Never + AfterEnum: false + AfterFunction: false + AfterNamespace: false + AfterObjCDeclaration: false + AfterStruct: false + AfterUnion: false + AfterExternBlock: false + BeforeCatch: false + BeforeElse: false + BeforeLambdaBody: false + BeforeWhile: false + IndentBraces: false + SplitEmptyFunction: false + SplitEmptyRecord: false + SplitEmptyNamespace: true +BreakBeforeBinaryOperators: None +BreakBeforeConceptDeclarations: Always +BreakBeforeBraces: Attach +BreakInheritanceList: BeforeComma +BreakBeforeTernaryOperators: true +BreakConstructorInitializers: BeforeComma +BreakAfterJavaFieldAnnotations: false +BreakStringLiterals: true +ColumnLimit: 140 +CommentPragmas: '^ IWYU pragma:' +QualifierAlignment: Leave +CompactNamespaces: false +ConstructorInitializerIndentWidth: 4 +ContinuationIndentWidth: 4 +Cpp11BracedListStyle: true +DeriveLineEnding: false +DerivePointerAlignment: false +DisableFormat: false +EmptyLineAfterAccessModifier: Never +EmptyLineBeforeAccessModifier: Always +ExperimentalAutoDetectBinPacking: false +PackConstructorInitializers: CurrentLine +BasedOnStyle: '' +ConstructorInitializerAllOnOneLineOrOnePerLine: true +AllowAllConstructorInitializersOnNextLine: true +FixNamespaceComments: true +ForEachMacros: + - foreach + - Q_FOREACH + - BOOST_FOREACH + - piForeach + - piForeachC + - piForeachR + - piForeachRC + - piForeachCR +IfMacros: + - KJ_IF_MAYBE +IncludeBlocks: Regroup +IncludeCategories: + - Regex: '^"(llvm|llvm-c|clang|clang-c)/' + Priority: 2 + SortPriority: 0 + CaseSensitive: false + - Regex: '^(<|"(gtest|gmock|isl|json)/)' + Priority: 3 + SortPriority: 0 + CaseSensitive: false + - Regex: '.*' + Priority: 1 + SortPriority: 0 + CaseSensitive: false +IncludeIsMainRegex: '(Test)?$' +IncludeIsMainSourceRegex: '' +IndentAccessModifiers: false +IndentCaseLabels: false +IndentCaseBlocks: false +IndentGotoLabels: false +IndentPPDirectives: AfterHash +IndentExternBlock: NoIndent +IndentRequiresClause: true +IndentWidth: 4 +IndentWrappedFunctionNames: false +InsertBraces: false +InsertTrailingCommas: Wrapped +JavaScriptQuotes: Leave +JavaScriptWrapImports: true +KeepEmptyLinesAtTheStartOfBlocks: false +LambdaBodyIndentation: Signature +MacroBlockBegin: "PRIVATE_DEFINITION_START|STATIC_INITIALIZER_BEGIN" +MacroBlockEnd: "PRIVATE_DEFINITION_END|STATIC_INITIALIZER_END" +MaxEmptyLinesToKeep: 2 +NamespaceIndentation: None +ObjCBinPackProtocolList: Auto +ObjCBlockIndentWidth: 2 +ObjCBreakBeforeNestedBlockParam: true +ObjCSpaceAfterProperty: false +ObjCSpaceBeforeProtocolList: true +PenaltyBreakAssignment: 2 +PenaltyBreakBeforeFirstCallParameter: 19 +PenaltyBreakComment: 300 +PenaltyBreakFirstLessLess: 120 +PenaltyBreakOpenParenthesis: 0 +PenaltyBreakString: 1000 +PenaltyBreakTemplateDeclaration: 10 +PenaltyExcessCharacter: 1000000 +PenaltyReturnTypeOnItsOwnLine: 60 +PenaltyIndentedWhitespace: 0 +PointerAlignment: Middle +PPIndentWidth: 2 +ReferenceAlignment: Middle +ReflowComments: true +RemoveBracesLLVM: false +RequiresClausePosition: OwnLine +SeparateDefinitionBlocks: Leave +ShortNamespaceLines: 1 +SortIncludes: CaseSensitive +SortJavaStaticImport: Before +SortUsingDeclarations: true +SpaceAfterCStyleCast: false +SpaceAfterLogicalNot: false +SpaceAfterTemplateKeyword: false +SpaceBeforeAssignmentOperators: true +SpaceBeforeCaseColon: false +SpaceBeforeCpp11BracedList: false +SpaceBeforeCtorInitializerColon: false +SpaceBeforeInheritanceColon: false +SpaceBeforeParens: ControlStatementsExceptControlMacros +SpaceBeforeParensOptions: + AfterControlStatements: true + AfterForeachMacros: false + AfterFunctionDefinitionName: false + AfterFunctionDeclarationName: false + AfterIfMacros: false + AfterOverloadedOperator: false + AfterRequiresInClause: false + AfterRequiresInExpression: false + BeforeNonEmptyParentheses: false +SpaceAroundPointerQualifiers: Both +SpaceBeforeRangeBasedForLoopColon: false +SpaceInEmptyBlock: false +SpaceInEmptyParentheses: false +SpacesBeforeTrailingComments: 1 +SpacesInAngles: Never +SpacesInConditionalStatement: false +SpacesInContainerLiterals: false +SpacesInCStyleCastParentheses: false +SpacesInLineCommentPrefix: + Minimum: 1 + Maximum: -1 +SpacesInParentheses: false +SpacesInSquareBrackets: false +SpaceBeforeSquareBrackets: false +BitFieldColonSpacing: After +Standard: c++11 +StatementAttributeLikeMacros: + - Q_EMIT + - PIMETA +StatementMacros: + - Q_UNUSED + - QT_REQUIRE_VERSION + - PRIVATE_DECLARATION + - NO_COPY_CLASS + - FOREVER_WAIT + - WAIT_FOREVER +TabWidth: 4 +UseCRLF: false +UseTab: AlignWithSpaces +WhitespaceSensitiveMacros: + - STRINGIZE + - PP_STRINGIZE + - BOOST_PP_STRINGIZE + - NS_SWIFT_NAME + - CF_SWIFT_NAME + - PIMETA +... + diff --git a/pipeline-runner/CMakeLists.txt b/pipeline-runner/CMakeLists.txt new file mode 100644 index 0000000..c858acf --- /dev/null +++ b/pipeline-runner/CMakeLists.txt @@ -0,0 +1,47 @@ +cmake_minimum_required(VERSION 3.10) + +project(PipelineRunner) +set(CMAKE_CXX_STANDARD 11) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_EXPORT_COMPILE_COMMANDS ON) + +# PIP +find_package(PIP REQUIRED) + +# Google Test +include(FetchContent) +FetchContent_Declare( + googletest + GIT_REPOSITORY https://git.shstk.ru/mirrors/googletest.git + GIT_TAG v1.14.0 +) +set(gtest_force_shared_crt ON CACHE BOOL "" FORCE) +FetchContent_MakeAvailable(googletest) + +# Source files +file(GLOB ALL_SRCS "src/*.cpp") +file(GLOB HDRS "src/*.h") +list(FILTER ALL_SRCS EXCLUDE REGEX ".*main\\.cpp$") +set(SRCS ${ALL_SRCS}) + +# Main executable +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) + +set_target_properties(pipeline-runner PROPERTIES + INSTALL_RPATH "\$ORIGIN;\$ORIGIN/lib" + BUILD_RPATH "\$ORIGIN;\$ORIGIN/lib" +) + +# Tests +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) + +include(GoogleTest) +gtest_discover_tests(test-pipeline-runner) + +# Install +install(TARGETS pipeline-runner DESTINATION bin) +install(FILES pipeline-runner.conf DESTINATION bin) diff --git a/pipeline-runner/README.md b/pipeline-runner/README.md deleted file mode 100644 index cea4805..0000000 --- a/pipeline-runner/README.md +++ /dev/null @@ -1,49 +0,0 @@ -# Pipeline Runner - -Веб-интерфейс для создания и выполнения pipeline промптов через `opencode run`. - -## Установка - -```bash -cd pipeline-runner -pip install -r requirements.txt -``` - -## Запуск - -```bash -python main.py -``` - -Откройте http://localhost:8000 в браузере. - -### Авто-релоад - -```bash -./deploy.sh -``` - -Скрипт активирует виртуальное окружение, установит зависимости и запустит uvicorn с авто-релоадом — сервер перезапустится автоматически после изменений в коде. - -## Функционал - -1. **Создание Pipeline**: - - Добавьте промпты с текстом и заголовком - - Сохраните pipeline с именем - -2. **Запуск**: - - Выберите сохраненный pipeline - - Нажмите "Run Pipeline" - - Следите за выполнением в реальном времени через WebSocket - -3. **Результаты**: - - Каждый шаг отображается в отдельной панели - - Статусы: pending, running, completed, error - - Логи сохраняются в `storage/logs/` (один файл на запуск) - -## Структура - -- `/api/pipelines` - REST API для управления pipeline -- `/ws/{run_id}` - WebSocket для real-time логов -- `storage/pipelines/` - JSON файлы сохраненных pipeline -- `storage/logs/` - Логи выполнений diff --git a/pipeline-runner/app.py b/pipeline-runner/app.py deleted file mode 100644 index 28ad531..0000000 --- a/pipeline-runner/app.py +++ /dev/null @@ -1,226 +0,0 @@ -import asyncio -import json -import uuid -from datetime import datetime, timezone -from pathlib import Path -from typing import Optional - -from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException -from fastapi.middleware.cors import CORSMiddleware -from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel - -app = FastAPI(title="Pipeline Runner") - -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - -BASE_DIR = Path(__file__).parent -PIPELINES_DIR = BASE_DIR / "storage" / "pipelines" -LOGS_DIR = BASE_DIR / "storage" / "logs" - -PIPELINES_DIR.mkdir(parents=True, exist_ok=True) -LOGS_DIR.mkdir(parents=True, exist_ok=True) - -active_connections: dict[str, set[WebSocket]] = {} -active_runs: dict[str, asyncio.Task] = {} - - -class Prompt(BaseModel): - id: str - text: str - title: str - order: int - - -class Pipeline(BaseModel): - id: str - name: str - prompts: list[Prompt] - created_at: str - updated_at: str - - -def load_pipelines() -> list[Pipeline]: - pipelines = [] - for f in PIPELINES_DIR.glob("*.json"): - try: - with open(f) as file: - data = json.load(file) - pipelines.append(Pipeline(**data)) - except (json.JSONDecodeError, OSError): - continue - return sorted(pipelines, key=lambda x: x.updated_at, reverse=True) - - -def save_pipeline(pipeline: Pipeline): - pipeline.updated_at = datetime.now(timezone.utc).isoformat() - try: - with open(PIPELINES_DIR / f"{pipeline.id}.json", "w") as f: - json.dump(pipeline.model_dump(), f, indent=2) - except OSError as e: - raise HTTPException(status_code=500, detail=f"Failed to save pipeline: {e}") - - -def delete_pipeline(pipeline_id: str): - path = PIPELINES_DIR / f"{pipeline_id}.json" - if path.exists(): - try: - path.unlink() - except OSError as e: - raise HTTPException(status_code=500, detail=f"Failed to delete pipeline: {e}") - - -@app.get("/api/pipelines", response_model=list[Pipeline]) -async def list_pipelines(): - return load_pipelines() - - -@app.post("/api/pipelines", response_model=Pipeline) -async def create_pipeline(pipeline: Pipeline): - existing = load_pipelines() - if any(p.id == pipeline.id for p in existing): - raise HTTPException(status_code=409, detail="Pipeline already exists") - save_pipeline(pipeline) - return pipeline - - -@app.delete("/api/pipelines/{pipeline_id}") -async def delete(pipeline_id: str): - pipelines = load_pipelines() - if not any(p.id == pipeline_id for p in pipelines): - raise HTTPException(status_code=404, detail="Pipeline not found") - delete_pipeline(pipeline_id) - return {"deleted": pipeline_id} - - -@app.post("/api/pipelines/{pipeline_id}/execute") -async def execute_pipeline(pipeline_id: str): - pipelines = load_pipelines() - pipeline = next((p for p in pipelines if p.id == pipeline_id), None) - if not pipeline: - raise HTTPException(status_code=404, detail="Pipeline not found") - - run_id = str(uuid.uuid4()) - log_file = LOGS_DIR / f"{run_id}.log" - - active_connections[run_id] = set() - task = asyncio.create_task(run_pipeline_task(pipeline, run_id, log_file)) - active_runs[run_id] = task - - return {"run_id": run_id, "log_file": str(log_file)} - - -async def run_pipeline_task(pipeline: Pipeline, run_id: str, log_file: Path): - async def write_log(line: str): - with open(log_file, "a") as f: - f.write(f"[{datetime.now(timezone.utc).isoformat()}] {line}\n") - for ws in list(active_connections.get(run_id, set())): - try: - await ws.send_json({"type": "log", "data": line}) - except: - pass - - async def stream_reader(stream, prefix, output_collector): - while True: - line = await stream.readline() - if not line: - break - text = line.decode("utf-8", errors="replace").rstrip("\n") - if text: - output_collector.append(text) - await write_log(f"{prefix} {text}") - - await write_log(f"Starting pipeline: {pipeline.name} (run_id: {run_id})") - - try: - for i, prompt in enumerate(pipeline.prompts, 1): - await write_log(f"[Step {i}/{len(pipeline.prompts)}] Starting: {prompt.title}") - await write_log(f"[Step {i}/{len(pipeline.prompts)}] Prompt: {prompt.text}") - - for ws in list(active_connections.get(run_id, set())): - try: - await ws.send_json({ - "type": "status", - "step_index": i - 1, - "status": "running", - "title": prompt.title - }) - except: - pass - - cmd = ["opencode", "run", prompt.text, "--title", prompt.title] - process = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - - stdout_lines: list[str] = [] - stderr_lines: list[str] = [] - - await asyncio.gather( - stream_reader(process.stdout, f"[Step {i}/{len(pipeline.prompts)}]", stdout_lines), - stream_reader(process.stderr, f"[Step {i}/{len(pipeline.prompts)}]", stderr_lines) - ) - - await process.wait() - - output = "\n".join(stdout_lines) - error = "\n".join(stderr_lines) - - status = "completed" if process.returncode == 0 else "error" - result = {"returncode": process.returncode, "output": output, "error": error} - - await write_log(f"[Step {i}/{len(pipeline.prompts)}] {status.upper()} (returncode: {process.returncode})") - - for ws in list(active_connections.get(run_id, set())): - try: - await ws.send_json({ - "type": "status", - "step_index": i - 1, - "status": status, - "title": prompt.title, - "result": result - }) - except: - pass - - if process.returncode != 0: - await write_log(f"Pipeline failed at step {i}") - break - - await write_log("Pipeline finished") - - finally: - if run_id in active_runs: - del active_runs[run_id] - - -@app.websocket("/ws/{run_id}") -async def websocket_endpoint(websocket: WebSocket, run_id: str): - await websocket.accept() - if run_id not in active_connections: - active_connections[run_id] = set() - active_connections[run_id].add(websocket) - - try: - while True: - await websocket.receive_text() - except Exception: - active_connections[run_id].discard(websocket) - if not active_connections[run_id]: - del active_connections[run_id] - - -if (BASE_DIR / "static").exists(): - app.mount("/static", StaticFiles(directory="static"), name="static") - -@app.get("/") -async def root(): - return {"message": "Pipeline Runner API", "ws_endpoint": "/ws/{run_id}"} diff --git a/pipeline-runner/deploy.sh b/pipeline-runner/deploy.sh deleted file mode 100755 index 0b2ec48..0000000 --- a/pipeline-runner/deploy.sh +++ /dev/null @@ -1,12 +0,0 @@ -#!/bin/bash - -cd "$(dirname "$0")" - -# Activate virtual environment -source .venv/bin/activate - -# Install dependencies -pip install -q -r requirements.txt - -# Start uvicorn with auto-reload -uvicorn app:app --host 0.0.0.0 --port 8000 --reload \ No newline at end of file diff --git a/pipeline-runner/main.py b/pipeline-runner/main.py deleted file mode 100644 index 5f330aa..0000000 --- a/pipeline-runner/main.py +++ /dev/null @@ -1,4 +0,0 @@ -import uvicorn - -if __name__ == "__main__": - uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) diff --git a/pipeline-runner/pipeline-runner.conf b/pipeline-runner/pipeline-runner.conf new file mode 100644 index 0000000..de4a447 --- /dev/null +++ b/pipeline-runner/pipeline-runner.conf @@ -0,0 +1,5 @@ +{ + "port": 8000, + "pipelines_dir": "./storage/pipelines/", + "logs_dir": "./storage/logs/" +} diff --git a/pipeline-runner/requirements.txt b/pipeline-runner/requirements.txt deleted file mode 100644 index d8e5065..0000000 --- a/pipeline-runner/requirements.txt +++ /dev/null @@ -1,7 +0,0 @@ -fastapi>=0.115.0 -uvicorn[standard]>=0.34.0 -websockets>=14.0 - -# Testing -pytest>=8.0 -httpx2>=0.28.0 diff --git a/pipeline-runner/src/main.cpp b/pipeline-runner/src/main.cpp new file mode 100644 index 0000000..93823e0 --- /dev/null +++ b/pipeline-runner/src/main.cpp @@ -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; +} diff --git a/pipeline-runner/src/messageutils.cpp b/pipeline-runner/src/messageutils.cpp new file mode 100644 index 0000000..e908882 --- /dev/null +++ b/pipeline-runner/src/messageutils.cpp @@ -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 diff --git a/pipeline-runner/src/messageutils.h b/pipeline-runner/src/messageutils.h new file mode 100644 index 0000000..a4365af --- /dev/null +++ b/pipeline-runner/src/messageutils.h @@ -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 diff --git a/pipeline-runner/src/pipeline.cpp b/pipeline-runner/src/pipeline.cpp new file mode 100644 index 0000000..dba333e --- /dev/null +++ b/pipeline-runner/src/pipeline.cpp @@ -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"; +} diff --git a/pipeline-runner/src/pipeline.h b/pipeline-runner/src/pipeline.h new file mode 100644 index 0000000..0b1c904 --- /dev/null +++ b/pipeline-runner/src/pipeline.h @@ -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 diff --git a/pipeline-runner/src/runner.cpp b/pipeline-runner/src/runner.cpp new file mode 100644 index 0000000..da19770 --- /dev/null +++ b/pipeline-runner/src/runner.cpp @@ -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); +} diff --git a/pipeline-runner/src/runner.h b/pipeline-runner/src/runner.h new file mode 100644 index 0000000..bbd471e --- /dev/null +++ b/pipeline-runner/src/runner.h @@ -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 diff --git a/pipeline-runner/src/server.cpp b/pipeline-runner/src/server.cpp new file mode 100644 index 0000000..165762e --- /dev/null +++ b/pipeline-runner/src/server.cpp @@ -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"); +} diff --git a/pipeline-runner/src/server.h b/pipeline-runner/src/server.h new file mode 100644 index 0000000..fbf5bbe --- /dev/null +++ b/pipeline-runner/src/server.h @@ -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 diff --git a/pipeline-runner/static/index.html b/pipeline-runner/static/index.html deleted file mode 100644 index 6ba4d25..0000000 --- a/pipeline-runner/static/index.html +++ /dev/null @@ -1,432 +0,0 @@ -<!DOCTYPE html> -<html lang="en"> -<head> - <meta charset="UTF-8"> - <meta name="viewport" content="width=device-width, initial-scale=1.0"> - <title>Pipeline Runner - - - - -
-

🚀 Pipeline Runner

- -
- -
-
-

Create/Load Pipeline

- -
- - -
- -
- - -
- - -
    - -
  • - No prompts added yet -
    • -
- -
- - -
-
- -
-

Saved Pipelines

-
    - -
  • - No saved pipelines -
    • -
-
-
- - -
-
-

Execution

- -
-
- Selected: None -
- -
- -
-
- Run ID: | - Status: -
- - -
- -
-
-
-
-
-
- - - - diff --git a/pipeline-runner/tests/test_app.py b/pipeline-runner/tests/test_app.py deleted file mode 100644 index 127a6b7..0000000 --- a/pipeline-runner/tests/test_app.py +++ /dev/null @@ -1,189 +0,0 @@ -import json -import shutil -from pathlib import Path - -import pytest -from fastapi.testclient import TestClient - -from app import app, PIPELINES_DIR, LOGS_DIR - -BASE_DIR = Path(__file__).parent.parent - -@pytest.fixture(autouse=True) -def clean_storage(): - """Clean storage before and after each test.""" - if PIPELINES_DIR.exists(): - shutil.rmtree(PIPELINES_DIR) - if LOGS_DIR.exists(): - shutil.rmtree(LOGS_DIR) - PIPELINES_DIR.mkdir(parents=True, exist_ok=True) - LOGS_DIR.mkdir(parents=True, exist_ok=True) - yield - if PIPELINES_DIR.exists(): - shutil.rmtree(PIPELINES_DIR) - if LOGS_DIR.exists(): - shutil.rmtree(LOGS_DIR) - -def make_pipeline(name="test", prompt_count=1): - return { - "id": "test-pipeline-id", - "name": name, - "prompts": [ - { - "id": f"prompt-{i}", - "text": f"test prompt {i}", - "title": f"Test Prompt {i}", - "order": i - } - for i in range(prompt_count) - ], - "created_at": "2024-01-01T00:00:00+00:00", - "updated_at": "2024-01-01T00:00:00+00:00" - } - -@pytest.fixture -def client(): - return TestClient(app) - -class TestListPipelines: - def test_empty(self, client): - resp = client.get("/api/pipelines") - assert resp.status_code == 200 - assert resp.json() == [] - - def test_with_pipelines(self, client): - p = make_pipeline() - client.post("/api/pipelines", json=p) - resp = client.get("/api/pipelines") - assert resp.status_code == 200 - data = resp.json() - assert len(data) == 1 - assert data[0]["name"] == "test" - -class TestCreatePipeline: - def test_create(self, client): - p = make_pipeline() - resp = client.post("/api/pipelines", json=p) - assert resp.status_code == 200 - data = resp.json() - assert data["name"] == "test" - assert data["id"] == "test-pipeline-id" - - def test_duplicate_id(self, client): - p = make_pipeline() - client.post("/api/pipelines", json=p) - resp = client.post("/api/pipelines", json=p) - assert resp.status_code == 409 - -class TestDeletePipeline: - def test_delete(self, client): - p = make_pipeline() - client.post("/api/pipelines", json=p) - resp = client.delete("/api/pipelines/test-pipeline-id") - assert resp.status_code == 200 - assert resp.json()["deleted"] == "test-pipeline-id" - - def test_delete_not_found(self, client): - resp = client.delete("/api/pipelines/nonexistent") - assert resp.status_code == 404 - -class TestExecutePipeline: - def test_not_found(self, client): - resp = client.post("/api/pipelines/nonexistent/execute") - assert resp.status_code == 404 - - def test_execute(self, client, monkeypatch): - """Test pipeline execution by patching opencode with echo.""" - import asyncio - - class FakeStream: - async def readline(self): - return b"" - - class FakeProcess: - returncode = 0 - - async def wait(self): - return 0 - - async def fake_create_subprocess_exec(*args, **kwargs): - proc = FakeProcess() - proc.stdout = FakeStream() - proc.stderr = FakeStream() - return proc - - monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec) - - p = make_pipeline() - client.post("/api/pipelines", json=p) - - resp = client.post("/api/pipelines/test-pipeline-id/execute") - assert resp.status_code == 200 - data = resp.json() - assert "run_id" in data - assert "log_file" in data - - def test_log_file_created(self, client, monkeypatch): - """Test that log file is created after execution.""" - import asyncio - - class FakeStream: - async def readline(self): - return b"" - - class FakeProcess: - returncode = 0 - - async def wait(self): - return 0 - - async def fake_create_subprocess_exec(*args, **kwargs): - proc = FakeProcess() - proc.stdout = FakeStream() - proc.stderr = FakeStream() - return proc - - monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec) - - p = make_pipeline() - client.post("/api/pipelines", json=p) - - resp = client.post("/api/pipelines/test-pipeline-id/execute") - data = resp.json() - log_path = Path(data["log_file"]) - assert log_path.exists() - content = log_path.read_text() - assert "Starting pipeline" in content - assert "Pipeline finished" in content - - def test_error_step_stops_pipeline(self, client, monkeypatch): - """Test that pipeline stops on first error.""" - import asyncio - - class FakeStream: - async def readline(self): - return b"" - - class FakeProcess: - returncode = 1 - - async def wait(self): - return 1 - - async def fake_create_subprocess_exec(*args, **kwargs): - proc = FakeProcess() - proc.stdout = FakeStream() - proc.stderr = FakeStream() - return proc - - monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec) - - p = make_pipeline(prompt_count=3) - client.post("/api/pipelines", json=p) - - resp = client.post("/api/pipelines/test-pipeline-id/execute") - data = resp.json() - log_path = Path(data["log_file"]) - content = log_path.read_text() - assert "Pipeline failed at step 1" in content - assert "Step 2" not in content diff --git a/pipeline-runner/tests/test_pipeline.cpp b/pipeline-runner/tests/test_pipeline.cpp new file mode 100644 index 0000000..4c03199 --- /dev/null +++ b/pipeline-runner/tests/test_pipeline.cpp @@ -0,0 +1,166 @@ +#include "pipeline.h" + +#include +#include +#include + +static const PIString testDir = "/tmp/pipeline-runner-test/"; +static const PIString testPipelinesDir = testDir + "pipelines/"; +static const PIString testLogsDir = testDir + "logs/"; + +static void setupDirs() { + PIDir::make(testDir); + PIDir::make(testPipelinesDir); + PIDir::make(testLogsDir); +} + +static void cleanupDirs() { + // Remove all files first using their full path + PIDir pdir(testPipelinesDir); + for (const auto & e: pdir.entries()) { + if (e.isFile()) PIFile::remove(e.path); + } + PIDir ldir(testLogsDir); + for (const auto & e: ldir.entries()) { + if (e.isFile()) PIFile::remove(e.path); + } + PIDir::remove(testPipelinesDir); + PIDir::remove(testLogsDir); + PIDir::remove(testDir); +} + +static Pipeline makeTestPipeline(const PIString & id = "test-id") { + Pipeline pl; + pl.id = id; + pl.name = "Test Pipeline"; + pl.working_dir = "/tmp/workdir"; + pl.created_at = nowISO(); + pl.updated_at = pl.created_at; + + Prompt p1; + p1.id = "prompt-1"; + p1.text = "Do something"; + p1.title = "Step 1"; + p1.order = 0; + pl.prompts << p1; + + Prompt p2; + p2.id = "prompt-2"; + p2.text = "Do something else"; + p2.title = "Step 2"; + p2.order = 1; + pl.prompts << p2; + + return pl; +} + +TEST(PipelineTest, SaveAndLoad) { + setupDirs(); + Pipeline pl = makeTestPipeline(); + + EXPECT_TRUE(savePipeline(testPipelinesDir, pl)); + + PIVector loaded = loadPipelines(testPipelinesDir); + EXPECT_EQ(loaded.size(), 1); + EXPECT_EQ(loaded[0].id, pl.id); + EXPECT_EQ(loaded[0].name, pl.name); + EXPECT_EQ(loaded[0].working_dir, pl.working_dir); + EXPECT_EQ(loaded[0].prompts.size(), 2); + + cleanupDirs(); +} + +TEST(PipelineTest, FindPipeline) { + setupDirs(); + Pipeline pl = makeTestPipeline("find-me"); + savePipeline(testPipelinesDir, pl); + + Pipeline found = findPipeline(testPipelinesDir, "find-me"); + EXPECT_FALSE(found.id.isEmpty()); + EXPECT_EQ(found.id, "find-me"); + EXPECT_EQ(found.name, "Test Pipeline"); + + Pipeline notFound = findPipeline(testPipelinesDir, "nonexistent"); + EXPECT_TRUE(notFound.id.isEmpty()); + + cleanupDirs(); +} + +TEST(PipelineTest, DeletePipeline) { + setupDirs(); + Pipeline pl = makeTestPipeline("del-me"); + savePipeline(testPipelinesDir, pl); + + EXPECT_TRUE(::deletePipeline(testPipelinesDir, "del-me")); + EXPECT_TRUE(findPipeline(testPipelinesDir, "del-me").id.isEmpty()); + + EXPECT_FALSE(::deletePipeline(testPipelinesDir, "nonexistent")); + + cleanupDirs(); +} + +TEST(PipelineTest, MultiplePipelines) { + setupDirs(); + + Pipeline pl1 = makeTestPipeline("pl-1"); + pl1.name = "Pipeline 1"; + savePipeline(testPipelinesDir, pl1); + + Pipeline pl2 = makeTestPipeline("pl-2"); + pl2.name = "Pipeline 2"; + savePipeline(testPipelinesDir, pl2); + + PIVector loaded = loadPipelines(testPipelinesDir); + EXPECT_EQ(loaded.size(), 2); + + cleanupDirs(); +} + +TEST(PipelineTest, EmptyDir) { + setupDirs(); + PIVector loaded = loadPipelines(testPipelinesDir); + EXPECT_TRUE(loaded.isEmpty()); + cleanupDirs(); +} + +TEST(PipelineTest, LogAppendAndRead) { + setupDirs(); + + appendLog(testLogsDir, "run-1", "First log line"); + appendLog(testLogsDir, "run-1", "Second log line"); + appendLog(testLogsDir, "run-2", "Different run"); + + PIString log1 = readLog(testLogsDir, "run-1"); + EXPECT_TRUE(log1.contains("First log line")); + EXPECT_TRUE(log1.contains("Second log line")); + EXPECT_FALSE(log1.contains("Different run")); + + PIString log2 = readLog(testLogsDir, "run-2"); + EXPECT_TRUE(log2.contains("Different run")); + EXPECT_FALSE(log2.contains("First log line")); + + PIString emptyLog = readLog(testLogsDir, "nonexistent"); + EXPECT_TRUE(emptyLog.isEmpty()); + + cleanupDirs(); +} + +TEST(UtilTest, GenerateUUID) { + PIString uuid1 = generateUUID(); + PIString uuid2 = generateUUID(); + + EXPECT_FALSE(uuid1.isEmpty()); + EXPECT_NE(uuid1, uuid2); + + // UUID format: 8-4-4-4-12 + EXPECT_EQ(uuid1.find("-"), 8); +} + +TEST(UtilTest, NowISO) { + PIString ts = nowISO(); + EXPECT_FALSE(ts.isEmpty()); + // ISO 8601 format: YYYY-MM-DDTHH:MM:SSZ + EXPECT_EQ(ts.find("-"), 4); + EXPECT_TRUE(ts.contains("T")); + EXPECT_TRUE(ts.endsWith("Z")); +} diff --git a/pipeline-runner/tests/test_runner.cpp b/pipeline-runner/tests/test_runner.cpp new file mode 100644 index 0000000..943ceca --- /dev/null +++ b/pipeline-runner/tests/test_runner.cpp @@ -0,0 +1,137 @@ +#include "pipeline.h" +#include "runner.h" + +#include +#include +#include +#include +#include + +static const PIString testDir = "/tmp/pipeline-runner-test/"; +static const PIString testPipelinesDir = testDir + "pipelines/"; +static const PIString testLogsDir = testDir + "logs/"; + +static void setupDirs() { + PIDir::make(testDir); + PIDir::make(testPipelinesDir); + PIDir::make(testLogsDir); +} + +static void cleanupDirs() { + PIDir pdir(testPipelinesDir); + for (const auto & e: pdir.entries()) { + if (e.isFile()) PIFile::remove(e.path); + } + PIDir ldir(testLogsDir); + for (const auto & e: ldir.entries()) { + if (e.isFile()) PIFile::remove(e.path); + } + PIDir::remove(testPipelinesDir); + PIDir::remove(testLogsDir); + PIDir::remove(testDir); +} + +static Pipeline makeEchoPipeline() { + Pipeline pl; + pl.id = "echo-test"; + pl.name = "Echo Test"; + pl.working_dir = "/tmp"; + pl.created_at = nowISO(); + pl.updated_at = pl.created_at; + + Prompt p; + p.id = "prompt-1"; + p.text = "echo hello"; + p.title = "Echo Step"; + p.order = 0; + pl.prompts << p; + + return pl; +} + +TEST(RunnerTest, StartRunNotFound) { + setupDirs(); + PipelineRunner runner(testPipelinesDir, testLogsDir); + + PIString runId = runner.startRun("nonexistent"); + EXPECT_TRUE(runId.isEmpty()); + + cleanupDirs(); +} + +TEST(RunnerTest, StartRunAndGetState) { + setupDirs(); + PipelineRunner runner(testPipelinesDir, testLogsDir); + + // Create a pipeline that runs 'echo hello' (uses shell command, not opencode) + Pipeline pl; + pl.id = "state-test"; + pl.name = "State 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("state-test"); + EXPECT_FALSE(runId.isEmpty()); + + RunState state = runner.getRunState(runId); + EXPECT_EQ(state.run_id, runId); + EXPECT_EQ(state.pipeline_id, "state-test"); + EXPECT_EQ(state.status, RunStatus::Running); + EXPECT_EQ(state.steps.size(), 1); + EXPECT_EQ(state.steps[0].title, "Step 1"); + + cleanupDirs(); +} + +TEST(RunnerTest, GetStateNotFound) { + setupDirs(); + PipelineRunner runner(testPipelinesDir, testLogsDir); + + RunState state = runner.getRunState("nonexistent"); + EXPECT_TRUE(state.run_id.isEmpty()); + + cleanupDirs(); +} + +TEST(RunnerTest, IsRunActive) { + setupDirs(); + PipelineRunner runner(testPipelinesDir, testLogsDir); + + Pipeline pl; + pl.id = "active-test"; + pl.name = "Active 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("active-test"); + EXPECT_TRUE(runner.isRunActive(runId)); + + cleanupDirs(); +} + +TEST(RunnerTest, RunStatusToString) { + EXPECT_EQ(runStatusToString(RunStatus::Pending), "pending"); + EXPECT_EQ(runStatusToString(RunStatus::Running), "running"); + EXPECT_EQ(runStatusToString(RunStatus::Completed), "completed"); + EXPECT_EQ(runStatusToString(RunStatus::Error), "error"); +}