commit 5a29468b7e965d7ca7e1ebc772ec7c6724799864 Author: andrey.bychkov Date: Fri Jul 3 09:31:45 2026 +0300 feat: initial commit - pipeline runner with real-time streaming diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dcc8cab --- /dev/null +++ b/.gitignore @@ -0,0 +1,55 @@ +# 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/ diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..ef873fe --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,26 @@ +# 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/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. + +## Key behaviors +- Pipeline execution runs `opencode run --title ` per step via `asyncio.create_subprocess_exec`. +- 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. + diff --git a/pipeline-runner/README.md b/pipeline-runner/README.md new file mode 100644 index 0000000..cea4805 --- /dev/null +++ b/pipeline-runner/README.md @@ -0,0 +1,49 @@ +# 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 new file mode 100644 index 0000000..28ad531 --- /dev/null +++ b/pipeline-runner/app.py @@ -0,0 +1,226 @@ +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 new file mode 100755 index 0000000..0b2ec48 --- /dev/null +++ b/pipeline-runner/deploy.sh @@ -0,0 +1,12 @@ +#!/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 new file mode 100644 index 0000000..5f330aa --- /dev/null +++ b/pipeline-runner/main.py @@ -0,0 +1,4 @@ +import uvicorn + +if __name__ == "__main__": + uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) diff --git a/pipeline-runner/requirements.txt b/pipeline-runner/requirements.txt new file mode 100644 index 0000000..d8e5065 --- /dev/null +++ b/pipeline-runner/requirements.txt @@ -0,0 +1,7 @@ +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/static/index.html b/pipeline-runner/static/index.html new file mode 100644 index 0000000..6ba4d25 --- /dev/null +++ b/pipeline-runner/static/index.html @@ -0,0 +1,432 @@ +<!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 new file mode 100644 index 0000000..127a6b7 --- /dev/null +++ b/pipeline-runner/tests/test_app.py @@ -0,0 +1,189 @@ +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