mirror of
https://github.com/wu736139669/OpsPilot.git
synced 2026-08-05 06:23:35 +00:00
wushenghuaandClaude Opus 4.7
92670cd667
A web dashboard for managing multiple Claude Code agents via tmux. Features: task step tracking, remote confirmation, live terminal, directory picker, real-time WebSocket updates. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
585 lines
20 KiB
Python
585 lines
20 KiB
Python
"""
|
|
OpsPilot - AI Coding Agent Control Center
|
|
Web dashboard on port 5016, manages multiple Claude Code agents via tmux.
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
|
|
from fastapi.responses import HTMLResponse
|
|
import uvicorn
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
STATE_DIR = BASE_DIR / "state"
|
|
LOGS_DIR = BASE_DIR / "logs"
|
|
FRONTEND_DIR = BASE_DIR / "frontend"
|
|
|
|
for d in [STATE_DIR, LOGS_DIR]:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
app = FastAPI(title="OpsPilot - Agent Control Center")
|
|
|
|
ws_clients: list[WebSocket] = []
|
|
log_buffer: list[dict] = []
|
|
MAX_LOGS = 500
|
|
|
|
|
|
def add_log(level: str, source: str, message: str):
|
|
entry = {
|
|
"time": datetime.now().strftime("%H:%M:%S"),
|
|
"ts": datetime.now().isoformat(),
|
|
"level": level,
|
|
"source": source,
|
|
"message": message,
|
|
}
|
|
log_buffer.append(entry)
|
|
if len(log_buffer) > MAX_LOGS:
|
|
log_buffer.pop(0)
|
|
with open(LOGS_DIR / "events.log", "a") as f:
|
|
f.write(f"[{entry['time']}] [{level}] [{source}] {message}\n")
|
|
|
|
|
|
def load_agents() -> dict:
|
|
"""Load agent registry from disk."""
|
|
path = STATE_DIR / "agents.json"
|
|
if path.exists():
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (json.JSONDecodeError, IOError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_agents(agents: dict):
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(STATE_DIR / "agents.json", "w") as f:
|
|
json.dump(agents, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def load_projects() -> dict:
|
|
path = STATE_DIR / "projects.json"
|
|
if path.exists():
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (json.JSONDecodeError, IOError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_projects(projects: dict):
|
|
with open(STATE_DIR / "projects.json", "w") as f:
|
|
json.dump(projects, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def load_tasks() -> dict:
|
|
"""Load task registry. Keyed by project name."""
|
|
path = STATE_DIR / "tasks.json"
|
|
if path.exists():
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (json.JSONDecodeError, IOError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_tasks(tasks: dict):
|
|
with open(STATE_DIR / "tasks.json", "w") as f:
|
|
json.dump(tasks, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def get_tmux_status(session_name: str) -> dict:
|
|
"""Check if a tmux session exists and get its windows."""
|
|
try:
|
|
result = subprocess.run(
|
|
["tmux", "list-windows", "-t", session_name, "-F", "#{window_index}:#{window_name}:#{window_active}"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
windows = []
|
|
active_window = None
|
|
for line in result.stdout.strip().split("\n"):
|
|
if line:
|
|
parts = line.split(":", 2)
|
|
if len(parts) == 3:
|
|
windows.append({"index": parts[0], "name": parts[1], "active": parts[2] == "1"})
|
|
if parts[2] == "1":
|
|
active_window = parts[0]
|
|
return {"running": True, "windows": windows, "active_window": active_window}
|
|
except Exception:
|
|
return {"running": False, "windows": [], "active_window": None}
|
|
|
|
|
|
async def broadcast(data: dict):
|
|
disconnected = []
|
|
for ws in ws_clients:
|
|
try:
|
|
await ws.send_json(data)
|
|
except Exception:
|
|
disconnected.append(ws)
|
|
for ws in disconnected:
|
|
ws_clients.remove(ws)
|
|
|
|
|
|
# ── REST API ──────────────────────────────────────────────
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def dashboard():
|
|
html_path = FRONTEND_DIR / "dashboard.html"
|
|
if html_path.exists():
|
|
return HTMLResponse(html_path.read_text(encoding="utf-8"))
|
|
return HTMLResponse("<h1>OpsPilot Dashboard</h1><p>dashboard.html not found</p>")
|
|
|
|
|
|
@app.get("/api/state")
|
|
async def get_state():
|
|
agents = load_agents()
|
|
projects = load_projects()
|
|
# Merge live tmux status
|
|
for name, agent in agents.items():
|
|
session = agent.get("tmux_session", "")
|
|
if session:
|
|
agent["tmux"] = get_tmux_status(session)
|
|
return {
|
|
"agents": agents,
|
|
"projects": projects,
|
|
"logs": log_buffer[-100:],
|
|
"time": datetime.now().isoformat(),
|
|
}
|
|
|
|
|
|
@app.post("/api/agents")
|
|
async def register_agent(data: dict):
|
|
"""Register or update an agent."""
|
|
agents = load_agents()
|
|
name = data.get("name", "unnamed")
|
|
agents[name] = {
|
|
"name": name,
|
|
"project": data.get("project", ""),
|
|
"tmux_session": data.get("tmux_session", ""),
|
|
"status": data.get("status", "idle"),
|
|
"task": data.get("task", ""),
|
|
"last_active": datetime.now().isoformat(),
|
|
"items_processed": data.get("items_processed", 0),
|
|
**(data.get("meta", {})),
|
|
}
|
|
save_agents(agents)
|
|
add_log("info", name, f"Agent registered: {data.get('status', 'idle')}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "agent": agents[name]}
|
|
|
|
|
|
@app.post("/api/agents/{name}/status")
|
|
async def update_agent_status(name: str, data: dict):
|
|
agents = load_agents()
|
|
if name not in agents:
|
|
raise HTTPException(status_code=404, detail="Agent not found")
|
|
agents[name].update({
|
|
"status": data.get("status", agents[name]["status"]),
|
|
"task": data.get("task", agents[name].get("task", "")),
|
|
"last_active": datetime.now().isoformat(),
|
|
"items_processed": data.get("items_processed", agents[name].get("items_processed", 0)),
|
|
})
|
|
if "message" in data:
|
|
agents[name]["message"] = data["message"]
|
|
save_agents(agents)
|
|
add_log("info", name, f"Status: {data.get('status')} - {data.get('message', '')}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/api/projects")
|
|
async def upsert_project(data: dict):
|
|
projects = load_projects()
|
|
name = data.get("name", "unnamed")
|
|
# Merge with existing if present
|
|
existing = projects.get(name, {})
|
|
projects[name] = {
|
|
"name": name,
|
|
"path": data.get("path", existing.get("path", "")),
|
|
"status": data.get("status", existing.get("status", "active")),
|
|
"description": data.get("description", existing.get("description", "")),
|
|
"agents": data.get("agents", existing.get("agents", [])),
|
|
"kanban_status": data.get("kanban_status", existing.get("kanban_status", "in_progress")),
|
|
"updated": datetime.now().isoformat(),
|
|
}
|
|
save_projects(projects)
|
|
add_log("info", "project", f"Project updated: {name}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok"}
|
|
|
|
|
|
# ── Task & Step Management ────────────────────────
|
|
|
|
@app.post("/api/tasks")
|
|
async def create_task(data: dict):
|
|
"""Create a new task for a project."""
|
|
project_name = data.get("project", "")
|
|
tasks = load_tasks()
|
|
|
|
task_id = str(int(time.time() * 1000))
|
|
steps_raw = data.get("steps", [])
|
|
# Support both step formats: ["name", ...] or [{"name": "...", "status": "..."}, ...]
|
|
steps = []
|
|
for s in steps_raw:
|
|
if isinstance(s, str):
|
|
steps.append({"name": s, "status": "pending", "started_at": None, "completed_at": None, "duration_seconds": 0})
|
|
else:
|
|
steps.append({
|
|
"name": s.get("name", ""),
|
|
"status": s.get("status", "pending"),
|
|
"started_at": s.get("started_at"),
|
|
"completed_at": s.get("completed_at"),
|
|
"duration_seconds": s.get("duration_seconds", 0),
|
|
})
|
|
|
|
task = {
|
|
"id": task_id,
|
|
"project": project_name,
|
|
"title": data.get("title", "Untitled Task"),
|
|
"description": data.get("description", ""),
|
|
"status": "queued", # queued, running, completed, failed
|
|
"steps": steps,
|
|
"current_step": 0,
|
|
"created_at": datetime.now().isoformat(),
|
|
"started_at": None,
|
|
"completed_at": None,
|
|
}
|
|
tasks[task_id] = task
|
|
save_tasks(tasks)
|
|
add_log("info", "task", f"Task created: {task['title']} for {project_name} ({len(steps)} steps)")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "task": task}
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/start")
|
|
async def start_task(task_id: str):
|
|
"""Mark a task as running and start its first step."""
|
|
tasks = load_tasks()
|
|
if task_id not in tasks:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
task = tasks[task_id]
|
|
task["status"] = "running"
|
|
task["started_at"] = datetime.now().isoformat()
|
|
if task["steps"]:
|
|
task["steps"][0]["status"] = "running"
|
|
task["steps"][0]["started_at"] = datetime.now().isoformat()
|
|
task["current_step"] = 0
|
|
save_tasks(tasks)
|
|
add_log("info", "task", f"Task started: {task['title']}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "task": task}
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/steps/{step_index}/complete")
|
|
async def complete_step(task_id: str, step_index: int, data: dict = None):
|
|
"""Mark a step as completed and advance to the next."""
|
|
tasks = load_tasks()
|
|
if task_id not in tasks:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
task = tasks[task_id]
|
|
steps = task["steps"]
|
|
if step_index >= len(steps):
|
|
raise HTTPException(status_code=400, detail="Invalid step index")
|
|
|
|
now = datetime.now().isoformat()
|
|
step = steps[step_index]
|
|
step["status"] = "completed"
|
|
step["completed_at"] = now
|
|
if step.get("started_at"):
|
|
try:
|
|
start = datetime.fromisoformat(step["started_at"])
|
|
step["duration_seconds"] = int((datetime.now() - start).total_seconds())
|
|
except Exception:
|
|
pass
|
|
|
|
# Start next step if exists
|
|
next_idx = step_index + 1
|
|
if next_idx < len(steps):
|
|
steps[next_idx]["status"] = "running"
|
|
steps[next_idx]["started_at"] = now
|
|
task["current_step"] = next_idx
|
|
else:
|
|
# All steps done
|
|
task["status"] = "completed"
|
|
task["completed_at"] = now
|
|
task["current_step"] = len(steps)
|
|
|
|
save_tasks(tasks)
|
|
add_log("info", "task", f"Step {step_index+1}/{len(steps)} completed: {step['name']}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "task": task, "completed_step": step}
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/steps/{step_index}/fail")
|
|
async def fail_step(task_id: str, step_index: int):
|
|
"""Mark a step as failed."""
|
|
tasks = load_tasks()
|
|
if task_id not in tasks:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
task = tasks[task_id]
|
|
step = task["steps"][step_index]
|
|
step["status"] = "failed"
|
|
step["completed_at"] = datetime.now().isoformat()
|
|
if step.get("started_at"):
|
|
try:
|
|
step["duration_seconds"] = int((datetime.now() - datetime.fromisoformat(step["started_at"])).total_seconds())
|
|
except Exception:
|
|
pass
|
|
save_tasks(tasks)
|
|
add_log("error", "task", f"Step failed: {step['name']}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "step": step}
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/status")
|
|
async def update_task_status(task_id: str, data: dict):
|
|
"""Update task status."""
|
|
tasks = load_tasks()
|
|
if task_id not in tasks:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
task = tasks[task_id]
|
|
task["status"] = data.get("status", task["status"])
|
|
if data.get("status") == "completed":
|
|
task["completed_at"] = datetime.now().isoformat()
|
|
save_tasks(tasks)
|
|
add_log("info", "task", f"Task status: {task['status']}")
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok", "task": task}
|
|
|
|
|
|
# ── Tmux Control ──────────────────────────────────
|
|
|
|
@app.get("/api/fs/list")
|
|
async def list_directory(path: str = ""):
|
|
"""List subdirectories for the path picker."""
|
|
import os as _os
|
|
target = path or str(Path.home() / "Documents" / "Projects")
|
|
target = _os.path.expanduser(target)
|
|
if not _os.path.isdir(target):
|
|
return {"path": target, "parent": str(Path(target).parent), "dirs": [], "error": "Not a directory"}
|
|
try:
|
|
items = []
|
|
for name in sorted(_os.listdir(target)):
|
|
full = _os.path.join(target, name)
|
|
if _os.path.isdir(full) and not name.startswith('.'):
|
|
items.append({"name": name, "path": full})
|
|
parent = str(Path(target).parent) if target != "/" else "/"
|
|
return {"path": target, "parent": parent, "dirs": items}
|
|
except PermissionError:
|
|
return {"path": target, "parent": str(Path(target).parent), "dirs": [], "error": "Permission denied"}
|
|
|
|
|
|
@app.post("/api/tmux/send")
|
|
async def tmux_send_keys(data: dict):
|
|
"""Send keys to a tmux session/window."""
|
|
session = data.get("session", "")
|
|
window = data.get("window", "0")
|
|
keys = data.get("keys", "")
|
|
if not session or not keys:
|
|
raise HTTPException(status_code=400, detail="session and keys required")
|
|
target = f"{session}:{window}"
|
|
try:
|
|
subprocess.run(
|
|
["tmux", "send-keys", "-t", target, keys, "Enter"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
add_log("info", "tmux", f"Sent to {target}: {keys[:100]}")
|
|
return {"status": "ok", "target": target}
|
|
except Exception as e:
|
|
add_log("error", "tmux", str(e))
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.post("/api/tmux/session")
|
|
async def create_tmux_session(data: dict):
|
|
"""Create a new tmux session with a Claude Code agent."""
|
|
session_name = data.get("name", "odineye")
|
|
project_dir = data.get("project_dir", "")
|
|
initial_prompt = data.get("prompt", "")
|
|
|
|
# Check if session exists
|
|
existing = get_tmux_status(session_name)
|
|
if existing["running"]:
|
|
return {"status": "exists", "session": session_name, "tmux": existing}
|
|
|
|
try:
|
|
# Create detached session
|
|
subprocess.run(
|
|
["tmux", "new-session", "-d", "-s", session_name, "-n", "dev", "-c", project_dir or str(BASE_DIR)],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
# Send initial Claude Code command
|
|
if initial_prompt:
|
|
time.sleep(1) # Wait for shell to init
|
|
escaped = initial_prompt.replace("'", "'\\''")
|
|
subprocess.run(
|
|
["tmux", "send-keys", "-t", f"{session_name}:dev", "claude", "Enter"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
time.sleep(2) # Wait for Claude Code to start
|
|
subprocess.run(
|
|
["tmux", "send-keys", "-t", f"{session_name}:dev", escaped, "Enter"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
|
|
add_log("info", "tmux", f"Session {session_name} created, Claude Code started")
|
|
return {"status": "created", "session": session_name, "tmux": get_tmux_status(session_name)}
|
|
except Exception as e:
|
|
add_log("error", "tmux", str(e))
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.post("/api/tmux/capture")
|
|
async def tmux_capture(data: dict):
|
|
"""Capture pane content from a tmux session."""
|
|
session = data.get("session", "")
|
|
window = data.get("window", "0")
|
|
lines = int(data.get("lines", 50))
|
|
return _capture_tmux(session, window, lines)
|
|
|
|
|
|
@app.get("/api/tmux/live/{session}")
|
|
async def tmux_live(session: str, window: str = "0", lines: int = 80):
|
|
"""GET endpoint to capture tmux pane content (for live view)."""
|
|
return _capture_tmux(session, window, lines)
|
|
|
|
|
|
def _capture_tmux(session: str, window: str, lines: int) -> dict:
|
|
target = f"{session}:{window}"
|
|
try:
|
|
result = subprocess.run(
|
|
["tmux", "capture-pane", "-t", target, "-p", "-S", f"-{lines}"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
# Strip ANSI escape codes for clean display
|
|
import re
|
|
clean = re.sub(r'\x1b\[[0-9;]*[a-zA-Z]', '', result.stdout)
|
|
return {"status": "ok", "content": clean, "target": target}
|
|
except Exception as e:
|
|
return {"status": "error", "content": str(e), "target": target}
|
|
|
|
|
|
# Cache for live terminal outputs
|
|
_tmux_cache: dict = {}
|
|
|
|
async def _capture_all_tmux_sessions():
|
|
"""Periodically capture all tmux sessions for live display."""
|
|
agents = load_agents()
|
|
for name, agent in agents.items():
|
|
session = agent.get("tmux_session", "")
|
|
if session:
|
|
window = agent.get("tmux_window", "0")
|
|
cap = _capture_tmux(session, window, 80)
|
|
_tmux_cache[session] = cap
|
|
return _tmux_cache
|
|
|
|
|
|
@app.post("/api/tmux/confirm")
|
|
async def tmux_confirm(data: dict):
|
|
"""Send confirmation (Enter, Y, or custom key) to a tmux session. Used to handle Claude Code prompts."""
|
|
session = data.get("session", "odineye")
|
|
window = data.get("window", "0")
|
|
# "yes" sends "1" then Enter (selects first option), "enter" just presses Enter
|
|
action = data.get("action", "enter")
|
|
target = f"{session}:{window}"
|
|
key_map = {"enter": "Enter", "yes": "1", "no": "2", "y": "y", "n": "n", "escape": "Escape", "yes_allow_all": "2"}
|
|
key = key_map.get(action, action)
|
|
needs_enter = action in ("yes", "y", "no", "n", "yes_allow_all")
|
|
try:
|
|
cmd = ["tmux", "send-keys", "-t", target, key]
|
|
if needs_enter:
|
|
cmd.append("Enter")
|
|
subprocess.run(cmd, capture_output=True, text=True, timeout=5)
|
|
add_log("info", "tmux", f"Confirmation sent to {target}: {action} ({key})")
|
|
return {"status": "ok", "target": target, "action": action}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.post("/api/log")
|
|
async def add_log_entry(data: dict):
|
|
add_log(
|
|
data.get("level", "info"),
|
|
data.get("source", "system"),
|
|
data.get("message", "")
|
|
)
|
|
await broadcast(await _build_state())
|
|
return {"status": "ok"}
|
|
|
|
|
|
# ── WebSocket ─────────────────────────────────────────────
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_endpoint(websocket: WebSocket):
|
|
await websocket.accept()
|
|
ws_clients.append(websocket)
|
|
add_log("info", "ws", f"Client connected ({len(ws_clients)} total)")
|
|
try:
|
|
# Send initial state
|
|
await websocket.send_json(await _build_state())
|
|
while True:
|
|
msg = await websocket.receive_text()
|
|
if msg == "ping":
|
|
await websocket.send_json(await _build_state())
|
|
except WebSocketDisconnect:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
if websocket in ws_clients:
|
|
ws_clients.remove(websocket)
|
|
|
|
|
|
async def _build_state() -> dict:
|
|
agents = load_agents()
|
|
projects = load_projects()
|
|
for name, agent in agents.items():
|
|
session = agent.get("tmux_session", "")
|
|
if session:
|
|
agent["tmux"] = get_tmux_status(session)
|
|
# Include live terminal captures
|
|
await _capture_all_tmux_sessions()
|
|
# Find active task per project
|
|
all_tasks = load_tasks()
|
|
project_tasks = {}
|
|
for tid, t in all_tasks.items():
|
|
pn = t.get("project", "")
|
|
if pn not in project_tasks:
|
|
project_tasks[pn] = []
|
|
project_tasks[pn].append(t)
|
|
return {
|
|
"agents": agents,
|
|
"projects": projects,
|
|
"tasks": all_tasks,
|
|
"project_tasks": project_tasks,
|
|
"logs": log_buffer[-100:],
|
|
"terminals": _tmux_cache,
|
|
"time": datetime.now().isoformat(),
|
|
}
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
add_log("info", "system", "OpsPilot Control Center starting on port 5016")
|
|
asyncio.create_task(_periodic_broadcast())
|
|
|
|
|
|
async def _periodic_broadcast():
|
|
while True:
|
|
await asyncio.sleep(10)
|
|
if ws_clients:
|
|
try:
|
|
await broadcast(await _build_state())
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
add_log("info", "system", "Booting OpsPilot...")
|
|
uvicorn.run(app, host="0.0.0.0", port=5016, log_level="info")
|