feat: orchestrator MVP — webhooks, agent launcher, QG checks

src/agents/launcher.py

@@ -0,0 +1,105 @@
+import subprocess
+import os
+from ..config import settings
+from ..db import get_db
+
+
+class AgentLauncher:
+    """Launch Claude CLI agents for specific tasks."""
+
+    AGENT_CONFIGS = {
+        "analyst": {
+            "system_prompt": ".openclaw/agents/analyst.md",
+            "task_file": ".task.md",
+            "allowed_tools": "Read,Write,Edit,Bash",
+        },
+        "architect": {
+            "system_prompt": ".openclaw/agents/architect.md",
+            "task_file": ".task-arch.md",
+            "allowed_tools": "Read,Write,Edit,Bash",
+        },
+        "developer": {
+            "system_prompt": ".openclaw/agents/developer.md",
+            "task_file": ".task-dev.md",
+            "allowed_tools": "Read,Write,Edit,Bash",
+        },
+        "reviewer": {
+            "system_prompt": ".openclaw/agents/reviewer.md",
+            "task_file": ".task-review.md",
+            "allowed_tools": "Read,Write,Edit,Bash",
+        },
+        "tester": {
+            "system_prompt": ".openclaw/agents/tester.md",
+            "task_file": ".task-test.md",
+            "allowed_tools": "Read,Write,Edit,Bash",
+        },
+    }
+
+    def launch(self, agent: str, repo: str, task_content: str = None) -> int:
+        """
+        Launch a Claude CLI agent.
+
+        Args:
+            agent: Agent role (analyst, architect, developer, reviewer, tester)
+            repo: Repository name
+            task_content: Optional task content to write to task file
+
+        Returns:
+            agent_run_id from DB
+        """
+        config = self.AGENT_CONFIGS.get(agent)
+        if not config:
+            raise ValueError(f"Unknown agent: {agent}")
+
+        repo_path = os.path.join(settings.repos_dir, repo)
+        if not os.path.isdir(repo_path):
+            raise FileNotFoundError(f"Repo not found: {repo_path}")
+
+        # Write task file if content provided
+        if task_content:
+            task_path = os.path.join(repo_path, config["task_file"])
+            with open(task_path, "w") as f:
+                f.write(task_content)
+
+        # Record run in DB
+        conn = get_db()
+        cursor = conn.execute(
+            "INSERT INTO agent_runs (task_id, agent) VALUES (NULL, ?)",
+            (agent,),
+        )
+        run_id = cursor.lastrowid
+        conn.commit()
+
+        # Prepare output log
+        output_path = f"/app/data/runs/{run_id}.log"
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+
+        # Build shell command
+        cmd = (
+            f'cd {repo_path} && {settings.claude_bin} --print '
+            f'"$(cat {config["task_file"]})" '
+            f'--system-prompt "$(cat {config["system_prompt"]})" '
+            f'--allowedTools {config["allowed_tools"]}'
+        )
+
+        # Launch as background process
+        with open(output_path, "w") as log_file:
+            subprocess.Popen(
+                ["bash", "-c", cmd],
+                stdout=log_file,
+                stderr=subprocess.STDOUT,
+                cwd=repo_path,
+            )
+
+        # Update DB with output path
+        conn.execute(
+            "UPDATE agent_runs SET output_path = ? WHERE id = ?",
+            (output_path, run_id),
+        )
+        conn.commit()
+        conn.close()
+
+        return run_id
+
+
+launcher = AgentLauncher()

src/config.py

@@ -0,0 +1,28 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    # Plane
+    plane_api_url: str = "http://localhost:8091"
+    plane_api_token: str = ""
+    plane_workspace_slug: str = ""
+    plane_webhook_secret: str = ""
+
+    # Gitea
+    gitea_url: str = "http://localhost:3000"
+    gitea_token: str = ""
+    gitea_webhook_secret: str = ""
+
+    # Claude CLI
+    claude_bin: str = "/usr/bin/claude"
+    repos_dir: str = "/home/slin/repos"
+
+    # DB
+    db_path: str = "/app/data/orchestrator.db"
+
+    class Config:
+        env_prefix = "ORCH_"
+        env_file = ".env"
+
+
+settings = Settings()

src/db.py

@@ -0,0 +1,42 @@
+import sqlite3
+from .config import settings
+
+
+def get_db() -> sqlite3.Connection:
+    conn = sqlite3.connect(settings.db_path)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def init_db():
+    conn = get_db()
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS events (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            timestamp TEXT DEFAULT (datetime('now')),
+            source TEXT NOT NULL,
+            event_type TEXT NOT NULL,
+            payload TEXT NOT NULL,
+            processed INTEGER DEFAULT 0
+        );
+        CREATE TABLE IF NOT EXISTS tasks (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            plane_id TEXT,
+            repo TEXT NOT NULL,
+            branch TEXT,
+            stage TEXT DEFAULT 'created',
+            agent_running TEXT,
+            created_at TEXT DEFAULT (datetime('now')),
+            updated_at TEXT DEFAULT (datetime('now'))
+        );
+        CREATE TABLE IF NOT EXISTS agent_runs (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            task_id INTEGER REFERENCES tasks(id),
+            agent TEXT NOT NULL,
+            started_at TEXT DEFAULT (datetime('now')),
+            finished_at TEXT,
+            exit_code INTEGER,
+            output_path TEXT
+        );
+    """)
+    conn.close()

src/main.py

@@ -0,0 +1,32 @@
+from fastapi import FastAPI
+from contextlib import asynccontextmanager
+from .db import init_db
+from .webhooks.plane import router as plane_router
+from .webhooks.gitea import router as gitea_router
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    init_db()
+    yield
+
+
+app = FastAPI(title="Multi-Agent Orchestrator", lifespan=lifespan)
+app.include_router(plane_router, prefix="/webhook")
+app.include_router(gitea_router, prefix="/webhook")
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "service": "orchestrator"}
+
+
+@app.get("/status")
+async def status():
+    from .db import get_db
+    conn = get_db()
+    tasks = conn.execute(
+        "SELECT * FROM tasks WHERE stage != 'done' ORDER BY created_at DESC LIMIT 10"
+    ).fetchall()
+    conn.close()
+    return {"active_tasks": [dict(t) for t in tasks]}

src/qg/checks.py

@@ -0,0 +1,26 @@
+# Quality Gate checks placeholder
+# Will be expanded as pipeline matures
+
+
+def check_analysis_complete(task_id: int) -> bool:
+    """Check if analysis artifacts exist."""
+    # TODO: verify .task-arch.md exists in repo
+    return True
+
+
+def check_architecture_approved(task_id: int) -> bool:
+    """Check if architecture was approved in Plane."""
+    # TODO: check Plane comment for :approved:
+    return False
+
+
+def check_ci_green(repo: str, branch: str) -> bool:
+    """Check if CI status is green for branch."""
+    # TODO: query Gitea commit status API
+    return False
+
+
+def check_review_approved(repo: str, pr_number: int) -> bool:
+    """Check if PR has approved review."""
+    # TODO: query Gitea PR reviews API
+    return False

src/webhooks/gitea.py

@@ -0,0 +1,54 @@
+from fastapi import APIRouter, Request
+import json
+from ..db import get_db
+
+router = APIRouter()
+
+
+@router.post("/gitea")
+async def gitea_webhook(request: Request):
+    """Handle Gitea webhook events."""
+    body = await request.body()
+    payload = json.loads(body)
+
+    # Log event
+    conn = get_db()
+    event_type = request.headers.get("X-Gitea-Event", "unknown")
+    conn.execute(
+        "INSERT INTO events (source, event_type, payload) VALUES (?, ?, ?)",
+        ("gitea", event_type, body.decode()),
+    )
+    conn.commit()
+
+    if event_type == "push":
+        await handle_push(payload, conn)
+    elif event_type == "pull_request":
+        await handle_pr(payload, conn)
+    elif event_type == "status":
+        await handle_ci_status(payload, conn)
+
+    conn.close()
+    return {"status": "accepted"}
+
+
+async def handle_push(payload: dict, conn):
+    """Push event — log for now."""
+    pass
+
+
+async def handle_pr(payload: dict, conn):
+    """PR event — check reviews, CI status."""
+    action = payload.get("action", "")
+    pr = payload.get("pull_request", {})
+
+    if action == "reviewed" and pr.get("state") == "approved":
+        # TODO: QG-5 check -> launch Tester
+        pass
+
+
+async def handle_ci_status(payload: dict, conn):
+    """CI status update — check if all green -> advance."""
+    state = payload.get("state", "")
+    if state == "success":
+        # TODO: Check all required contexts green -> advance stage
+        pass

src/webhooks/plane.py

@@ -0,0 +1,49 @@
+from fastapi import APIRouter, Request
+import json
+from ..db import get_db
+
+router = APIRouter()
+
+
+@router.post("/plane")
+async def plane_webhook(request: Request):
+    """Handle Plane webhook events."""
+    body = await request.body()
+    payload = json.loads(body)
+
+    # Log event
+    conn = get_db()
+    conn.execute(
+        "INSERT INTO events (source, event_type, payload) VALUES (?, ?, ?)",
+        ("plane", payload.get("event", "unknown"), body.decode()),
+    )
+    conn.commit()
+
+    event = payload.get("event")
+    data = payload.get("data", {})
+
+    if event == "work_item.created":
+        await handle_work_item_created(data, conn)
+    elif event == "comment.created":
+        await handle_comment(data, conn)
+
+    conn.close()
+    return {"status": "accepted"}
+
+
+async def handle_work_item_created(data: dict, conn):
+    """New work item -> create task record."""
+    plane_id = data.get("id", "")
+    conn.execute(
+        "INSERT INTO tasks (plane_id, repo, stage) VALUES (?, ?, ?)",
+        (plane_id, "enduro-trails", "analysis"),
+    )
+    conn.commit()
+
+
+async def handle_comment(data: dict, conn):
+    """Check for :approved: reaction -> advance stage."""
+    comment_body = data.get("comment", "")
+    if ":approved:" in comment_body:
+        # TODO: Determine which task, advance QG
+        pass