RomanNum3ral
/
agentic-os
mirror of https://github.com/modimihir07/agentic-os.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor duplicated patterns into shared utilities 

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
zumayaaustin 2026-07-08 23:10:05 +00:00
parent fb3a1979dc
commit 2e7bb73807
2 changed files with 119 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
dashboard/api.js
View File

@ -1,31 +1,20 @@
const api = {
async get(path) {
const r = await fetch(path);
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || `Request failed: ${r.status}`); }
return r.json();
},
async post(path, body = {}, controller) {
const opts = { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) };
async request(path, { method = 'GET', body, controller } = {}) {
const opts = { method };
if (body !== undefined) {
opts.headers = { 'Content-Type': 'application/json' };
opts.body = JSON.stringify(body);
}
if (controller) opts.signal = controller.signal;
const r = await fetch(path, opts);
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || `Request failed: ${r.status}`); }
return r.json();
},
async put(path, body = {}) {
const r = await fetch(path, { method: 'PUT', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) });
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || `Request failed: ${r.status}`); }
return r.json();
},
async patch(path, body = {}) {
const r = await fetch(path, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) });
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || `Request failed: ${r.status}`); }
return r.json();
},
async del(path) {
const r = await fetch(path, { method: 'DELETE' });
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || `Request failed: ${r.status}`); }
return r.json();
},
get(path) { return api.request(path); },
post(path, body = {}, controller) { return api.request(path, { method: 'POST', body, controller }); },
put(path, body = {}) { return api.request(path, { method: 'PUT', body }); },
patch(path, body = {}) { return api.request(path, { method: 'PATCH', body }); },
del(path) { return api.request(path, { method: 'DELETE' }); },
getStatus: () => api.get('/api/status'),
getBrain: () => api.get('/api/brain'),
getBrainFile: (name) => api.get(`/api/brain/${encodeURIComponent(name)}`),

240
server.py
View File

@ -25,6 +25,9 @@ from pydantic import BaseModel
BASE_DIR = Path(__file__).parent.resolve()
# Agents supported across chat, routing, health, and kanban dispatch.
AGENTS = ["opencode", "hermes", "gemini"]
app = FastAPI(title="Agentic OS", version="1.1.0")
# Load OpenRouter API key from Hermes .env
@ -120,13 +123,35 @@ def list_dir(path: Path):
return []
return sorted([p.name for p in path.iterdir() if not p.name.startswith(".")])
def read_json(path: Path, default=None):
"""Load JSON from path, returning default when the file is missing."""
if not path.exists():
return default
return json.loads(path.read_text(encoding="utf-8"))
def write_json(path: Path, data, indent: int = 2):
"""Serialize data as pretty JSON to path."""
path.write_text(json.dumps(data, indent=indent), encoding="utf-8")
def iter_skill_dirs():
"""Yield skill directories, skipping hidden and underscore-prefixed ones."""
skills_dir = BASE_DIR / "skills"
if not skills_dir.exists():
return
for d in sorted(skills_dir.iterdir()):
if d.is_dir() and not d.name.startswith("_"):
yield d
def new_id():
return str(uuid.uuid4())[:8]
def get_timestamp():
return datetime.now(timezone.utc).isoformat()
def append_audit(entry: dict):
audit_file = BASE_DIR / "audit" / "audit.log"
entry["timestamp"] = get_timestamp()
entry["id"] = str(uuid.uuid4())[:8]
entry["id"] = new_id()
with open(audit_file, "a") as f:
f.write(json.dumps(entry) + "\n")
@ -157,7 +182,7 @@ def check_agent(name: str) -> dict:
@app.get("/api/status")
def get_status():
agents = [check_agent(a) for a in ["opencode", "hermes", "gemini"]]
agents = [check_agent(a) for a in AGENTS]
skills = list_dir(BASE_DIR / "skills")
return {
"status": "healthy",
@ -197,25 +222,18 @@ def update_brain_file(file_name: str, data: BrainUpdate):
@app.get("/api/skills")
def list_skills():
skills = []
for d in sorted((BASE_DIR / "skills").iterdir()):
if d.is_dir() and not d.name.startswith("_"):
skill_md = read_file(d / "SKILL.md")
learnings = read_file(d / "learnings.md")
eval_data = {}
eval_path = d / "eval.json"
if eval_path.exists():
eval_data = json.loads(eval_path.read_text())
score_history = []
score_path = d / "score-history.json"
if score_path.exists():
score_history = json.loads(score_path.read_text())
skills.append({
"name": d.name,
"description": skill_md[:200] if skill_md else "",
"has_learnings": bool(learnings),
"eval_criteria": eval_data.get("criteria", []),
"scores": score_history,
})
for d in iter_skill_dirs():
skill_md = read_file(d / "SKILL.md")
learnings = read_file(d / "learnings.md")
eval_data = read_json(d / "eval.json", {})
score_history = read_json(d / "score-history.json", [])
skills.append({
"name": d.name,
"description": skill_md[:200] if skill_md else "",
"has_learnings": bool(learnings),
"eval_criteria": eval_data.get("criteria", []),
"scores": score_history,
})
return skills
@app.get("/api/skills/{name}")
@ -227,8 +245,8 @@ def get_skill(name: str):
"name": name,
"skill": read_file(path / "SKILL.md"),
"learnings": read_file(path / "learnings.md"),
"eval": json.loads((path / "eval.json").read_text()) if (path / "eval.json").exists() else {},
"score_history": json.loads((path / "score-history.json").read_text()) if (path / "score-history.json").exists() else [],
"eval": read_json(path / "eval.json", {}),
"score_history": read_json(path / "score-history.json", []),
"context": [f.name for f in (path / "context").iterdir()] if (path / "context").exists() else [],
}
@ -274,7 +292,7 @@ def run_skill(name: str, req: Optional[SkillRunRequest] = None):
if skill_input:
prompt += f"## User Input\n{skill_input}"
run_id = str(uuid.uuid4())[:8]
run_id = new_id()
# Execute via agent
try:
@ -317,10 +335,8 @@ def run_skill(name: str, req: Optional[SkillRunRequest] = None):
@app.get("/api/skills/{name}/eval")
def get_skill_eval(name: str):
path = BASE_DIR / "skills" / name / "score-history.json"
if not path.exists():
return {"scores": []}
return {"scores": json.loads(path.read_text())}
scores = read_json(BASE_DIR / "skills" / name / "score-history.json", [])
return {"scores": scores}
# ─── Routes: Scheduler ────────────────────────────────────────────
@ -337,7 +353,7 @@ def create_job(job: ScheduleJobRequest):
jobs_dir = BASE_DIR / "scheduler" / "jobs"
jobs_dir.mkdir(parents=True, exist_ok=True)
job_data = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"name": job.name,
"skill": job.skill,
"cron": job.cron,
@ -346,9 +362,7 @@ def create_job(job: ScheduleJobRequest):
"last_run": None,
"next_run": None,
}
(jobs_dir / f"{job.name.replace(' ', '_')}.json").write_text(
json.dumps(job_data, indent=2)
)
write_json(jobs_dir / f"{job.name.replace(' ', '_')}.json", job_data)
append_audit({"action": "job_created", "job": job.name})
return job_data
@ -379,15 +393,12 @@ def get_audit(limit: int = Query(100, le=500)):
@app.get("/api/cost")
def get_cost():
cost_file = BASE_DIR / "data" / "cost-history.json"
if not cost_file.exists():
return {"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []}
return json.loads(cost_file.read_text())
return read_json(cost_file, {"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []})
@app.post("/api/cost/record")
def record_cost(data: dict):
cost_file = BASE_DIR / "data" / "cost-history.json"
cost_data = json.loads(cost_file.read_text()) if cost_file.exists() else \
{"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []}
cost_data = read_json(cost_file, {"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []})
cost_data["entries"].append({
"timestamp": get_timestamp(),
"agent": data.get("agent", "unknown"),
@ -395,7 +406,7 @@ def record_cost(data: dict):
"cost": data.get("cost", 0.0),
"model": data.get("model", "unknown"),
})
cost_file.write_text(json.dumps(cost_data, indent=2))
write_json(cost_file, cost_data)
return {"status": "recorded"}
# ─── Routes: Registry/Plugins ─────────────────────────────────────
@ -403,9 +414,7 @@ def record_cost(data: dict):
@app.get("/api/plugins")
def list_plugins():
reg_file = BASE_DIR / "registry" / "plugins.json"
if not reg_file.exists():
return {"plugins": []}
return json.loads(reg_file.read_text())
return read_json(reg_file, {"plugins": []})
@app.post("/api/plugins/install")
def install_plugin(data: dict):
@ -413,7 +422,7 @@ def install_plugin(data: dict):
if not name:
raise HTTPException(400, "Plugin name required")
reg_file = BASE_DIR / "registry" / "plugins.json"
reg = json.loads(reg_file.read_text()) if reg_file.exists() else {"plugins": []}
reg = read_json(reg_file, {"plugins": []})
if any(p["name"] == name for p in reg["plugins"]):
return {"status": "already_installed"}
reg["plugins"].append({
@ -421,7 +430,7 @@ def install_plugin(data: dict):
"installed": get_timestamp(),
"version": "1.0.0",
})
reg_file.write_text(json.dumps(reg, indent=2))
write_json(reg_file, reg)
append_audit({"action": "plugin_installed", "plugin": name})
return {"status": "installed", "plugin": name}
@ -478,17 +487,15 @@ def list_prompts():
@app.get("/api/settings")
def get_settings():
sf = BASE_DIR / "data" / "settings.json"
if not sf.exists():
return {}
return json.loads(sf.read_text())
return read_json(sf, {})
@app.put("/api/settings")
def update_settings(data: SettingsUpdate):
sf = BASE_DIR / "data" / "settings.json"
# Merge with existing
existing = json.loads(sf.read_text()) if sf.exists() else {}
existing = read_json(sf, {})
existing.update(data.settings)
sf.write_text(json.dumps(existing, indent=2))
write_json(sf, existing)
append_audit({"action": "settings_updated"})
return {"status": "ok"}
@ -520,16 +527,14 @@ def discover_standards():
CHAT_HISTORY_FILE = BASE_DIR / "data" / "chat-history.json"
def load_chat_history():
if CHAT_HISTORY_FILE.exists():
return json.loads(CHAT_HISTORY_FILE.read_text())
return {"messages": []}
return read_json(CHAT_HISTORY_FILE, {"messages": []})
def save_chat_message(msg: dict):
history = load_chat_history()
history["messages"].append(msg)
if len(history["messages"]) > 200:
history["messages"] = history["messages"][-200:]
CHAT_HISTORY_FILE.write_text(json.dumps(history, indent=2))
write_json(CHAT_HISTORY_FILE, history)
def run_cli(args: list, timeout: int = 30) -> tuple:
r = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
@ -636,11 +641,11 @@ def execute_agent(agent: str, message: str) -> str:
@app.post("/api/chat")
def chat(req: ChatRequest):
agent = req.agent.lower().strip()
if agent not in ["opencode", "hermes", "gemini"]:
if agent not in AGENTS:
raise HTTPException(400, "Agent must be one of: opencode, hermes, gemini")
user_msg = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"role": "user",
"agent": agent,
"content": req.message,
@ -651,7 +656,7 @@ def chat(req: ChatRequest):
response_text = execute_agent(agent, req.message)
agent_msg = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"role": "assistant",
"agent": agent,
"content": response_text,
@ -788,9 +793,16 @@ def kanban_task_path(task_id: str) -> Path:
def save_kanban_task(task: dict):
ensure_dir(KANBAN_DIR)
kanban_task_path(task["id"]).write_text(json.dumps(task, indent=2))
write_json(kanban_task_path(task["id"]), task)
KANBAN_AGENTS = {"opencode", "hermes", "gemini"}
def load_kanban_task_or_404(task_id: str):
"""Return (path, task) for a task id, raising 404 when it does not exist."""
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
return path, json.loads(path.read_text())
KANBAN_AGENTS = set(AGENTS)
def dispatch_kanban_task(task_id: str):
"""Move a task to in_progress and hand it to its assignee agent in the background."""
@ -821,7 +833,7 @@ def _run_kanban_agent(task_id: str):
task = json.loads(path.read_text()) # reload in case it changed while the agent ran
task.setdefault("comments", []).append({
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"message": f"🤖 **{agent}**\n\n{response}",
"timestamp": get_timestamp(),
})
@ -838,12 +850,10 @@ def _run_kanban_agent(task_id: str):
save_kanban_task(task)
def load_goals():
if GOALS_FILE.exists():
return json.loads(GOALS_FILE.read_text())
return []
return read_json(GOALS_FILE, [])
def save_goals(goals: list):
GOALS_FILE.write_text(json.dumps(goals, indent=2))
write_json(GOALS_FILE, goals)
# ─── Routes: Kanban Board (13 endpoints) ────────────────────────
@ -864,16 +874,14 @@ def kanban_board(status: Optional[str] = None):
@app.get("/api/kanban/tasks/{task_id}")
def kanban_get_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
return json.loads(path.read_text())
_, task = load_kanban_task_or_404(task_id)
return task
@app.post("/api/kanban/tasks")
def kanban_create_task(data: KanbanTaskCreate):
try:
task = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"title": data.title,
"body": data.body,
"status": data.status,
@ -895,10 +903,7 @@ def kanban_create_task(data: KanbanTaskCreate):
@app.patch("/api/kanban/tasks/{task_id}")
def kanban_update_task(task_id: str, data: KanbanTaskUpdate):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
assignee_changed = data.assignee is not None and data.assignee != task.get("assignee")
for field in ["title", "body", "status", "priority", "assignee"]:
val = getattr(data, field, None)
@ -914,10 +919,7 @@ def kanban_update_task(task_id: str, data: KanbanTaskUpdate):
@app.post("/api/kanban/tasks/{task_id}/dispatch")
def kanban_dispatch_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
if task.get("assignee") not in KANBAN_AGENTS:
raise HTTPException(400, "Task must be assigned to opencode, hermes, or gemini to dispatch")
dispatch_kanban_task(task_id)
@ -925,10 +927,7 @@ def kanban_dispatch_task(task_id: str):
@app.post("/api/kanban/tasks/{task_id}/complete")
def kanban_complete_task(task_id: str, data: KanbanComplete):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
task["status"] = "done"
task["summary"] = data.summary
task["completed_at"] = get_timestamp()
@ -939,10 +938,7 @@ def kanban_complete_task(task_id: str, data: KanbanComplete):
@app.post("/api/kanban/tasks/{task_id}/block")
def kanban_block_task(task_id: str, data: KanbanBlock):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
task["status"] = "blocked"
task["block_reason"] = data.reason
task["updated"] = get_timestamp()
@ -952,10 +948,7 @@ def kanban_block_task(task_id: str, data: KanbanBlock):
@app.post("/api/kanban/tasks/{task_id}/unblock")
def kanban_unblock_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
task["status"] = "ready"
task["block_reason"] = ""
task["updated"] = get_timestamp()
@ -965,12 +958,9 @@ def kanban_unblock_task(task_id: str):
@app.post("/api/kanban/tasks/{task_id}/comments")
def kanban_add_comment(task_id: str, data: KanbanCommentCreate):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
comment = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"message": data.message,
"timestamp": get_timestamp(),
}
@ -1019,10 +1009,7 @@ def kanban_dispatch():
@app.post("/api/kanban/tasks/{task_id}/specify")
def kanban_specify_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
if task.get("status") == "triage":
task["status"] = "todo"
task["updated"] = get_timestamp()
@ -1031,16 +1018,13 @@ def kanban_specify_task(task_id: str):
@app.post("/api/kanban/tasks/{task_id}/decompose")
def kanban_decompose_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
path, task = load_kanban_task_or_404(task_id)
children = []
for i, subtask in enumerate(task.get("body", "").split("\n")):
subtask = subtask.strip().lstrip("-* ")
if subtask:
child = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"title": subtask[:80],
"body": subtask,
"status": "todo",
@ -1070,7 +1054,7 @@ def create_goal(data: GoalCreate):
try:
goals = load_goals()
goal = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"title": data.title,
"description": data.description,
"category": data.category,
@ -1182,7 +1166,7 @@ def search_journal(q: str = Query("")):
def get_agent_health():
try:
agents = []
for name in ["opencode", "hermes", "gemini"]:
for name in AGENTS:
info = check_agent(name)
info["uptime"] = 0
info["success_rate"] = 100
@ -1195,7 +1179,7 @@ def get_agent_health():
@app.get("/api/agents/{name}/stats")
def get_agent_stats(name: str):
try:
if name not in ["opencode", "hermes", "gemini"]:
if name not in AGENTS:
raise HTTPException(400, "Invalid agent")
info = check_agent(name)
return {
@ -1216,7 +1200,7 @@ def get_agent_stats(name: str):
def refresh_agent_health():
try:
agents = []
for name in ["opencode", "hermes", "gemini"]:
for name in AGENTS:
info = check_agent(name)
agents.append(info)
append_audit({"action": "agent_health_refreshed"})
@ -1254,7 +1238,7 @@ def router_suggest(data: RouterSuggest):
def router_route(data: RouterRoute):
try:
agent = data.agent.lower()
if agent not in ["opencode", "hermes", "gemini"]:
if agent not in AGENTS:
return {"status": "error", "message": f"Invalid agent: {agent}"}
append_audit({"action": "task_routed", "agent": agent, "task_preview": data.task[:50]})
return {
@ -1271,22 +1255,17 @@ def router_route(data: RouterRoute):
@app.get("/api/analytics/skills")
def get_skill_analytics():
try:
skills_dir = BASE_DIR / "skills"
analytics = []
for d in sorted(skills_dir.iterdir()):
if d.is_dir() and not d.name.startswith("_"):
eval_path = d / "eval.json"
score_path = d / "score-history.json"
scores = json.loads(score_path.read_text()) if score_path.exists() else []
eval_data = json.loads(eval_path.read_text()) if eval_path.exists() else {}
avg_score = sum(s.get("score", 0) for s in scores) / len(scores) if scores else 0
analytics.append({
"name": d.name,
"total_runs": len(scores),
"avg_score": round(avg_score, 1),
"last_score": scores[-1].get("score", 0) if scores else 0,
"trend": "up" if len(scores) >= 2 and scores[-1].get("score", 0) > scores[-2].get("score", 0) else "down" if len(scores) >= 2 else "stable",
})
for d in iter_skill_dirs():
scores = read_json(d / "score-history.json", [])
avg_score = sum(s.get("score", 0) for s in scores) / len(scores) if scores else 0
analytics.append({
"name": d.name,
"total_runs": len(scores),
"avg_score": round(avg_score, 1),
"last_score": scores[-1].get("score", 0) if scores else 0,
"trend": "up" if len(scores) >= 2 and scores[-1].get("score", 0) > scores[-2].get("score", 0) else "down" if len(scores) >= 2 else "stable",
})
return {"skills": sorted(analytics, key=lambda x: x["total_runs"], reverse=True)}
except Exception as e:
return {"skills": [], "error": str(e)}
@ -1294,18 +1273,15 @@ def get_skill_analytics():
@app.get("/api/analytics/trends")
def get_trend_analytics():
try:
skills_dir = BASE_DIR / "skills"
trends = []
for d in sorted(skills_dir.iterdir()):
if d.is_dir() and not d.name.startswith("_"):
score_path = d / "score-history.json"
scores = json.loads(score_path.read_text()) if score_path.exists() else []
if scores:
trends.append({
"name": d.name,
"scores": [s.get("score", 0) for s in scores[-10:]],
"labels": [s.get("date", "") for s in scores[-10:]],
})
for d in iter_skill_dirs():
scores = read_json(d / "score-history.json", [])
if scores:
trends.append({
"name": d.name,
"scores": [s.get("score", 0) for s in scores[-10:]],
"labels": [s.get("date", "") for s in scores[-10:]],
})
return {"trends": trends}
except Exception as e:
return {"trends": [], "error": str(e)}