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

Merge remote-tracking branch 'origin/main' 

# Conflicts:
#	data/kanban/0f822987.json
#	data/kanban/8893ad14.json
This commit is contained in:
Austin 2026-07-19 02:11:36 -07:00
parent 1a8b17d15e d7ea76cf99
commit 82af4268af
16 changed files with 1388 additions and 238 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,6 +1,10 @@
__pycache__/
*.pyc
.env
.venv/
.pytest_cache/
.coverage
htmlcov/
*.egg-info/
dist/
node_modules/

34
README.md
View File

@ -91,6 +91,8 @@ A locally-hosted operating system for AI agents — an open-source GitHub reposi
## 🚀 Quick Start
### Linux/macOS
```bash
git clone https://github.com/modimihir07/agentic-os.git
cd agentic-os
@ -99,6 +101,16 @@ chmod +x install.sh && ./install.sh
# Open http://127.0.0.1:8080
```
### Windows PowerShell
```powershell
git clone https://github.com/modimihir07/agentic-os.git
cd agentic-os
.\install.ps1
.\start.ps1
# Open http://127.0.0.1:8080
```
---
## 📋 Prerequisites
@ -107,12 +119,14 @@ chmod +x install.sh && ./install.sh
| Tool | Required? | Install |
|------|-----------|---------|
| Python 3.10+ | ✅ Required | Linux/macOS: install from your package manager or `python.org`. Windows: install from `python.org` and select **Add Python to PATH**. Verify with `python --version` or `py -3 --version`. |
| Node.js 18+ | ⚠ For opencode | `curl -fsSL https://deb.nodesource.com/setup_20.x \| sudo bash - && sudo apt install -y nodejs` |
| Python 3.10+ | ✅ Required | Install from your package manager or [python.org](https://www.python.org/downloads/). Verify with `python3 --version`. |
| Node.js 18+ | ⚠ For opencode and Gemini CLI | `curl -fsSL https://deb.nodesource.com/setup_20.x \| sudo bash - && sudo apt install -y nodejs` |
| opencode | ⚠ For code tasks | `npm install -g @opencode/cli` |
| Hermes Agent | ⚠ For memory/scheduling | `curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent/main/scripts/install.sh \| bash` |
| Hermes Agent | ⚠ For memory/scheduling | Follow the upstream Hermes Agent documentation. If native Windows support is unavailable, use WSL for Hermes. |
| Gemini CLI | ⚠ For Google AI | `npm install -g @google/gemini-cli` |
> ⚠ = Optional — the dashboard starts and core pages work without all agent CLIs installed. Agent-specific chat, routing, health, and skill execution features will show offline or warning status until the relevant CLI is installed and authenticated.
### Windows
| Tool | Required? | Install | Verify / Next Step |
@ -125,16 +139,6 @@ chmod +x install.sh && ./install.sh
> ⚠ = Optional — the dashboard works with any subset of installed agents.
### Windows Quick Start
```powershell
git clone https://github.com/modimihir07/agentic-os.git
cd agentic-os
.\install.ps1
.\start.ps1
# Open http://127.0.0.1:8080
```
The PowerShell launchers resolve Python in this order: `py -3.10`, `py -3`, then `python`. For manual commands, prefer `python -m pip install -r requirements.txt` and `python server.py --port 8080` so the same interpreter runs both dependency installation and the server.
> **PowerShell execution policy:** if you use PowerShell scripts, allow local scripts for your user with `Set-ExecutionPolicy -Scope CurrentUser RemoteSigned`, or run a single installer invocation with `powershell -ExecutionPolicy Bypass -File .\install.ps1`.
@ -172,7 +176,7 @@ Edit `data/settings.json`:
agentic-os/
├── server.py # FastAPI backend (REST API)
├── requirements.txt # Python dependencies
├── install.sh # One-command installer
├── install.sh / install.ps1 # Platform installers
├── start.sh # Launch dashboard
├── backup.sh / restore.sh # Disaster recovery
│
@ -330,7 +334,7 @@ Browse opencode session logs by date and size. Click "Replay" to view all messag
## 🧪 Tested On
- **OS**: Linux (Ubuntu 22.04+), macOS
- **OS**: Linux (Ubuntu 22.04+), macOS; Windows dashboard setup supported via PowerShell, with Hermes best used through WSL if native support is unavailable
- **Python**: 3.10, 3.11, 3.12
- **Browsers**: Chrome, Firefox, Edge
- **Agents**: opencode v0.8+, Hermes Agent v1.0+, Gemini CLI v1.0+

39
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)}`),
@ -34,6 +23,11 @@ const api = {
getSkill: (name) => api.get(`/api/skills/${encodeURIComponent(name)}`),
runSkill: (name, input = '', agent = 'auto') => api.post(`/api/skills/${encodeURIComponent(name)}/run`, { input, agent }),
getSkillEval: (name) => api.get(`/api/skills/${encodeURIComponent(name)}/eval`),
createSkill: (name, skillMd) => api.post('/api/skills', { name, skill_md: skillMd }),
updateSkill: (name, skillMd) => api.put(`/api/skills/${encodeURIComponent(name)}`, { skill_md: skillMd }),
getSkillContextFile: (name, filename) => api.get(`/api/skills/${encodeURIComponent(name)}/context/${encodeURIComponent(filename)}`),
putSkillContextFile: (name, filename, content) => api.put(`/api/skills/${encodeURIComponent(name)}/context/${encodeURIComponent(filename)}`, { content }),
deleteSkillContextFile: (name, filename) => api.del(`/api/skills/${encodeURIComponent(name)}/context/${encodeURIComponent(filename)}`),
getJobs: () => api.get('/api/scheduler/jobs'),
createJob: (job) => api.post('/api/scheduler/jobs', job),
deleteJob: (id) => api.del(`/api/scheduler/jobs/${encodeURIComponent(id)}`),
@ -55,6 +49,7 @@ const api = {
// Kanban
getKanbanBoard: (status) => api.get(status ? `/api/kanban/board?status=${encodeURIComponent(status)}` : '/api/kanban/board'),
getKanbanTask: (id) => api.get(`/api/kanban/tasks/${encodeURIComponent(id)}`),
deleteKanbanTask: (id) => api.del(`/api/kanban/tasks/${encodeURIComponent(id)}`),
createKanbanTask: (data) => api.post('/api/kanban/tasks', data),
updateKanbanTask: (id, data) => api.patch(`/api/kanban/tasks/${encodeURIComponent(id)}`, data),
completeKanbanTask: (id, summary) => api.post(`/api/kanban/tasks/${encodeURIComponent(id)}/complete`, { summary }),

156
dashboard/pages/skills.js
View File

@ -6,8 +6,9 @@ async function renderSkills() {
<h1 class="page-title">Skills Hub</h1>
<p class="page-subtitle">Browse, run, and monitor skill performance</p>
</div>
<div class="btn-group">
<div class="btn-group" id="skillActions">
<input id="skillFilter" class="form-input" style="width:200px" placeholder="Filter skills..." oninput="filterSkills()">
<button class="btn btn-primary" onclick="showAddSkill()">+ New Skill</button>
</div>
</div>
<div class="tabs" id="skillTabs">
@ -82,7 +83,7 @@ function filterSkills() {
async function showSkillDetail(name) {
document.getElementById('skillsContainer').style.display = 'none';
document.getElementById('skillTabs').style.display = 'none';
document.getElementById('skillFilter').style.display = 'none';
document.getElementById('skillActions').style.display = 'none';
const detail = document.getElementById('skillDetail');
detail.style.display = 'block';
detail.innerHTML = `<div class="loading"><div class="loading-spinner"></div></div>`;
@ -93,6 +94,8 @@ async function showSkillDetail(name) {
const lastScore = scores.length > 0 ? scores[scores.length - 1] : null;
const avg = lastScore && lastScore.criteria_scores ? (lastScore.criteria_scores.reduce((a, b) => a + b, 0) / lastScore.criteria_scores.length) : null;
window._currentSkillName = name;
detail.innerHTML = `
<div style="margin-bottom:16px">
<button class="btn btn-ghost" onclick="backToSkills()">← Back to Skills</button>
@ -100,8 +103,11 @@ async function showSkillDetail(name) {
</div>
<div class="grid grid-2">
<div class="card">
<div class="card-header"><span class="card-title">📄 SKILL.md</span></div>
<pre style="max-height:400px;overflow:auto;font-size:12px">${escapeHtml(skill.skill || 'No SKILL.md')}</pre>
<div class="card-header">
<span class="card-title">📄 SKILL.md</span>
<button class="btn btn-sm btn-ghost" style="margin-left:auto" onclick="editSkillMd('${name}')">✎ Edit</button>
</div>
<pre id="skillMdView" style="max-height:400px;overflow:auto;font-size:12px">${escapeHtml(skill.skill || 'No SKILL.md')}</pre>
</div>
<div class="card">
<div class="card-header"><span class="card-title">📖 Learnings</span></div>
@ -120,10 +126,20 @@ async function showSkillDetail(name) {
` : '<div style="color:var(--text-muted);font-size:13px">No evaluation scores yet</div>'}
</div>
<div class="card">
<div class="card-header"><span class="card-title">📁 Context Files</span></div>
${skill.context && skill.context.length > 0
? `<div style="display:flex;flex-wrap:wrap;gap:6px">${skill.context.map(f => `<span class="badge badge-info">${f}</span>`).join('')}</div>`
: '<div style="color:var(--text-muted);font-size:13px">No context files</div>'}
<div class="card-header">
<span class="card-title">📁 Context Files</span>
<button class="btn btn-sm btn-ghost" style="margin-left:auto" onclick="addSkillContextFile('${name}')">+ Add File</button>
</div>
<div id="skillContextList">
${skill.context && skill.context.length > 0
? skill.context.map(f => `
<div style="display:flex;align-items:center;gap:8px;padding:6px 0;border-bottom:1px solid var(--border);font-size:13px">
<span style="flex:1">${escapeHtml(f)}</span>
<button class="btn btn-sm btn-ghost" onclick="editSkillContextFile('${name}','${escapeHtml(f)}')">✎</button>
<button class="btn btn-sm btn-ghost" style="color:var(--red)" onclick="deleteSkillContextFile('${name}','${escapeHtml(f)}')">🗑</button>
</div>`).join('')
: '<div style="color:var(--text-muted);font-size:13px">No context files</div>'}
</div>
${skill.eval && skill.eval.criteria ? `<div style="margin-top:12px"><strong style="font-size:12px">Eval Criteria:</strong><div style="display:flex;flex-wrap:wrap;gap:4px;margin-top:6px">${skill.eval.criteria.map(c => `<span class="badge badge-accent">${c}</span>`).join('')}</div></div>` : ''}
</div>
</div>
@ -136,7 +152,7 @@ async function showSkillDetail(name) {
function backToSkills() {
document.getElementById('skillsContainer').style.display = '';
document.getElementById('skillTabs').style.display = '';
document.getElementById('skillFilter').style.display = '';
document.getElementById('skillActions').style.display = '';
document.getElementById('skillDetail').style.display = 'none';
}
@ -195,3 +211,125 @@ async function executeSkillRun(name) {
if (runBtn) { runBtn.textContent = '▶ Run'; runBtn.disabled = false; }
}
}
function showAddSkill() {
showModal('New Skill', `
<div class="form-group">
<label class="form-label">Name</label>
<input id="newSkillName" class="form-input" placeholder="e.g., my-custom-skill">
<div class="form-hint">Letters, numbers, dashes and underscores only.</div>
</div>
<div class="form-group">
<label class="form-label">SKILL.md</label>
<textarea id="newSkillMd" class="form-textarea" rows="12" placeholder="# My Custom Skill&#10;&#10;Describe what this skill does and how an agent should perform it..."></textarea>
</div>
`, `
<button class="btn btn-ghost" onclick="closeModal()">Cancel</button>
<button class="btn btn-primary" onclick="submitNewSkill()">Create Skill</button>
`);
}
async function submitNewSkill() {
const name = document.getElementById('newSkillName').value.trim();
const skillMd = document.getElementById('newSkillMd').value;
if (!name) { showToast('Skill name is required', 'error'); return; }
try {
await api.createSkill(name, skillMd);
showToast('Skill created!', 'success');
closeModal();
await renderSkills();
showSkillDetail(name);
} catch (err) {
showToast('Failed to create skill: ' + err.message, 'error');
}
}
function editSkillMd(name) {
const view = document.getElementById('skillMdView');
const current = view ? view.textContent : '';
view.outerHTML = `
<textarea id="skillMdEdit" class="form-textarea" rows="16" style="font-family:var(--font-mono);font-size:12px">${escapeHtml(current)}</textarea>
<div style="display:flex;gap:8px;margin-top:8px">
<button class="btn btn-sm btn-primary" onclick="saveSkillMd('${name}')">Save</button>
<button class="btn btn-sm btn-ghost" onclick="showSkillDetail('${name}')">Cancel</button>
</div>
`;
}
async function saveSkillMd(name) {
const content = document.getElementById('skillMdEdit').value;
try {
await api.updateSkill(name, content);
showToast('SKILL.md saved', 'success');
showSkillDetail(name);
} catch (err) {
showToast('Failed to save: ' + err.message, 'error');
}
}
function addSkillContextFile(name) {
showModal('Add Context File', `
<div class="form-group">
<label class="form-label">File Name</label>
<input id="newContextFilename" class="form-input" placeholder="e.g., reference.md">
</div>
<div class="form-group">
<label class="form-label">Content</label>
<textarea id="newContextContent" class="form-textarea" rows="10"></textarea>
</div>
`, `
<button class="btn btn-ghost" onclick="closeModal()">Cancel</button>
<button class="btn btn-primary" onclick="submitNewContextFile('${name}')">Add File</button>
`);
}
async function submitNewContextFile(name) {
const filename = document.getElementById('newContextFilename').value.trim();
const content = document.getElementById('newContextContent').value;
if (!filename) { showToast('File name is required', 'error'); return; }
try {
await api.putSkillContextFile(name, filename, content);
showToast('Context file added', 'success');
closeModal();
showSkillDetail(name);
} catch (err) {
showToast('Failed to add file: ' + err.message, 'error');
}
}
async function editSkillContextFile(name, filename) {
try {
const file = await api.getSkillContextFile(name, filename);
showModal(`Edit: ${filename}`, `
<textarea id="editContextContent" class="form-textarea" rows="14">${escapeHtml(file.content)}</textarea>
`, `
<button class="btn btn-ghost" onclick="closeModal()">Cancel</button>
<button class="btn btn-primary" onclick="saveSkillContextFile('${name}','${filename}')">Save</button>
`);
} catch (err) {
showToast('Failed to load file: ' + err.message, 'error');
}
}
async function saveSkillContextFile(name, filename) {
const content = document.getElementById('editContextContent').value;
try {
await api.putSkillContextFile(name, filename, content);
showToast('File saved', 'success');
closeModal();
showSkillDetail(name);
} catch (err) {
showToast('Failed to save: ' + err.message, 'error');
}
}
async function deleteSkillContextFile(name, filename) {
if (!confirm(`Delete "${filename}"?`)) return;
try {
await api.deleteSkillContextFile(name, filename);
showToast('File deleted', 'info');
showSkillDetail(name);
} catch (err) {
showToast('Failed to delete: ' + err.message, 'error');
}
}

12
data/kanban/0f822987.json
View File

@ -1,12 +0,0 @@
{
"id": "0f822987",
"title": "Fix login bug",
"body": "The login page has a race condition",
"status": "done",
"priority": "high",
"assignee": "opencode",
"comments": [],
"links": [],
"created": "2026-06-05T09:52:18.236452+00:00",
"updated": "2026-07-18T03:44:13.215700+00:00"
}

15
data/kanban/8893ad14.json
View File

@ -1,15 +0,0 @@
{
"id": "8893ad14",
"title": "Test kanban task",
"body": "",
"status": "done",
"priority": "medium",
"assignee": "",
"comments": [],
"links": [],
"created": "2026-06-05T09:50:30.293643+00:00",
"updated": "2026-07-05T21:29:57.461011+00:00",
"summary": "Test done",
"completed_at": "2026-06-05T09:50:46.473612+00:00",
"block_reason": "Waiting for API review"
}

13
install.ps1
View File

@ -22,7 +22,7 @@ function Resolve-Python {
}
}
throw "Python 3.10+ is required. Install it from https://www.python.org/downloads/ and check 'Add Python to PATH'."
throw "Python 3.10+ is required. Install it from https://www.python.org/downloads/ (check 'Add Python to PATH') or run: winget install Python.Python.3.12"
}
$python = Resolve-Python
@ -34,11 +34,11 @@ Write-Host "Python: $pythonVersion"
Write-Host "Installing Python dependencies..."
& $pythonCommand @pythonArgs -m pip install -r requirements.txt --quiet
# Check Node.js (for opencode)
# Check Node.js (for opencode and Gemini CLI)
if (Get-Command node -ErrorAction SilentlyContinue) {
Write-Host "Node.js: $(node --version)"
} else {
Write-Warning "Node.js not found. opencode requires Node 18+. Install from https://nodejs.org/."
Write-Warning "Node.js not found. opencode and Gemini CLI require Node 18+. Install from https://nodejs.org/ or run: winget install OpenJS.NodeJS.LTS"
}
# Check opencode
@ -53,7 +53,7 @@ if (Get-Command opencode -ErrorAction SilentlyContinue) {
if (Get-Command hermes -ErrorAction SilentlyContinue) {
Write-Host "Hermes: found"
} else {
Write-Warning "Hermes Agent not found. See the Hermes Agent documentation for Windows installation guidance."
Write-Warning "Hermes Agent not found. Native Windows support is not confirmed by this project - check the upstream Hermes Agent documentation, and use WSL if no native installer is available."
}
# Check Gemini CLI
@ -100,3 +100,8 @@ Write-Host " 1. Edit data/settings.json with your API keys"
Write-Host " 2. Double-click the 'Agentic OS Dashboard' shortcut on your Desktop"
Write-Host " (or run .\start.ps1 / .\Launch-Dashboard.bat manually)"
Write-Host " 3. Your browser will open automatically once the server is ready"
Write-Host ""
Write-Host "Optional agent CLI reminders:"
Write-Host " opencode: npm install -g @opencode/cli"
Write-Host " Gemini: npm install -g @google/gemini-cli"
Write-Host " Hermes: Use upstream docs for native Windows support, or WSL if native support is unavailable."

4
pytest.ini Normal file
View File

@ -0,0 +1,4 @@
[pytest]
testpaths = tests
python_files = test_*.py
addopts = -q

4
requirements-dev.txt Normal file
View File

@ -0,0 +1,4 @@
-r requirements.txt
pytest>=8.0.0
pytest-cov>=5.0.0
httpx>=0.27.0

40
scheduler/scheduler.py
View File

@ -24,24 +24,40 @@ def run_skill(skill_name: str):
"skill": skill_name,
"timestamp": datetime.now(timezone.utc).isoformat(),
}
with open(audit_file, "a") as f:
f.write(json.dumps(entry) + "\n")
try:
audit_file.parent.mkdir(parents=True, exist_ok=True)
with open(audit_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
except OSError as e:
print(f" [audit] failed to record run of {skill_name!r}: {e}")
print(f"[{datetime.now().isoformat()}] Ran skill: {skill_name}")
def load_jobs(scheduler: BackgroundScheduler):
"""Load job definitions from jobs/ directory."""
"""Load job definitions from jobs/ directory.
A single malformed job file is logged and skipped rather than being allowed
to abort loading of every other job.
"""
for job_file in JOBS_DIR.glob("*.json"):
data = json.loads(job_file.read_text())
try:
data = json.loads(job_file.read_text())
except (json.JSONDecodeError, OSError) as e:
print(f" Skipping {job_file.name}: could not read job ({e})")
continue
if not data.get("enabled", True):
continue
scheduler.add_job(
run_skill,
CronTrigger.from_crontab(data["cron"]),
args=[data["skill"]],
id=data.get("id", data["name"]),
name=data["name"],
replace_existing=True,
)
try:
scheduler.add_job(
run_skill,
CronTrigger.from_crontab(data["cron"]),
args=[data["skill"]],
id=data.get("id", data["name"]),
name=data["name"],
replace_existing=True,
)
except (KeyError, ValueError) as e:
print(f" Skipping {job_file.name}: invalid job definition ({e})")
continue
print(f" Scheduled: {data['name']} ({data['cron']})")
def main():

525
server.py
View File

@ -8,6 +8,7 @@ import asyncio
import json
import os
import re
import shlex
import shutil
import signal
import subprocess
@ -27,6 +28,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
@ -87,6 +91,16 @@ class SkillRunRequest(BaseModel):
input: Optional[str] = ""
agent: Optional[str] = "auto"
class SkillCreate(BaseModel):
name: str
skill_md: str = ""
class SkillUpdate(BaseModel):
skill_md: str
class SkillContextFileWrite(BaseModel):
content: str = ""
class ScheduleJobRequest(BaseModel):
name: str
skill: str
@ -119,36 +133,120 @@ 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, best_effort=False):
"""Load JSON from path, returning ``default`` when the file is missing.
On corrupt or unreadable content a descriptive ``HTTPException(500)`` is
raised so the error is propagated to the client instead of surfacing as an
opaque 500. Aggregate/listing callers can set ``best_effort=True`` to
tolerate one bad file: the corruption is logged and ``default`` is returned
instead of aborting the whole view.
"""
if not path.exists():
return default
try:
return json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, UnicodeDecodeError) as e:
if best_effort:
print(f"[load] skipping corrupt {path.name}: {e}")
return default
raise HTTPException(500, f"Failed to read {path.name}: {e}")
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]
with open(audit_file, "a") as f:
f.write(json.dumps(entry) + "\n")
entry["id"] = new_id()
try:
audit_file.parent.mkdir(parents=True, exist_ok=True)
with open(audit_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
except OSError as e:
# Auditing is best-effort: never let a logging failure abort the
# underlying operation, but surface it on the server console.
print(f"[audit] failed to write entry {entry.get('action')!r}: {e}")
# ─── Agent Discovery (instant filesystem checks) ────────────────────
def check_agent(name: str) -> dict:
"""Instant filesystem-based check. No subprocess needed."""
def _cli_has_subcommand(base_args: list, subcommand: str) -> bool:
try:
if name == "opencode":
exists = shutil.which("opencode") is not None
status = "online" if exists else "offline"
elif name == "hermes":
exists = shutil.which("hermes") is not None
status = "online" if exists else "offline"
elif name == "gemini":
# Gemini has valid OAuth tokens logged in
oauth = Path.home() / ".gemini" / "oauth_creds.json"
exists = shutil.which("gemini") is not None
logged_in = oauth.exists() and "ya29" in oauth.read_text()
status = "online" if exists and logged_in else "offline" if not exists else "warning"
else:
status = "offline"
r = subprocess.run([*base_args, "--help"], capture_output=True, text=True, timeout=10)
return subcommand in ((r.stdout or "") + (r.stderr or ""))
except Exception:
return False
def hermes_cli_args(*args: str) -> list:
"""Build the command to invoke Hermes, bridging through WSL if the real agent only lives there.
The dashboard commonly runs as a native Windows process while Hermes (whose official
installer is Bash-only) lives inside WSL - a plain PATH lookup on Windows will never find it
there. Windows machines can also have an unrelated tool also named 'hermes' on PATH (e.g. the
academic softwarepub/HERMES metadata-publishing project, which coincidentally shares the name),
so don't just trust that a native 'hermes' is the right one - confirm it exposes the
NousResearch agent's `chat` subcommand before using it, falling back to WSL otherwise.
"""
if shutil.which("hermes") is not None and _cli_has_subcommand(["hermes"], "chat"):
return ["hermes", *args]
if shutil.which("wsl") is not None:
quoted = " ".join(shlex.quote(a) for a in args)
return ["wsl", "-e", "bash", "-lc", f"hermes {quoted}"]
return ["hermes", *args]
_hermes_available_cache = {"checked_at": 0.0, "result": False}
HERMES_AVAILABLE_CACHE_TTL = 60
def hermes_available() -> bool:
"""Cached: this spawns a subprocess (possibly via WSL), and /api/status is polled every 15s."""
now = time.time()
if now - _hermes_available_cache["checked_at"] < HERMES_AVAILABLE_CACHE_TTL:
return _hermes_available_cache["result"]
try:
r = subprocess.run(hermes_cli_args("--version"), capture_output=True, text=True, timeout=10)
result = r.returncode == 0
except Exception:
result = False
_hermes_available_cache["checked_at"] = now
_hermes_available_cache["result"] = result
return result
def check_agent(name: str) -> dict:
"""Filesystem-based check for opencode/gemini; hermes needs a real subprocess since it may live inside WSL."""
if name == "opencode":
status = "online" if shutil.which("opencode") is not None else "offline"
elif name == "hermes":
status = "online" if hermes_available() else "offline"
elif name == "gemini":
exists = shutil.which("gemini") is not None
# Gemini needs a valid OAuth token on disk to be usable.
logged_in = False
oauth = Path.home() / ".gemini" / "oauth_creds.json"
if oauth.exists():
try:
logged_in = "ya29" in oauth.read_text()
except OSError as e:
# Distinguish an unreadable credential file from "not logged in"
# instead of silently reporting the agent as offline.
print(f"[agent-health] could not read gemini credentials: {e}")
status = "online" if exists and logged_in else "offline" if not exists else "warning"
else:
status = "offline"
return {"name": name, "status": status}
@ -156,7 +254,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",
@ -193,49 +291,122 @@ def update_brain_file(file_name: str, data: BrainUpdate):
# ─── Routes: Skills ───────────────────────────────────────────────
SKILL_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
SKILL_CONTEXT_FILENAME_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_.-]{0,127}$")
def skill_dir_path(name: str) -> Path:
if not SKILL_NAME_RE.fullmatch(name or ""):
raise HTTPException(400, "Invalid skill name")
base = (BASE_DIR / "skills").resolve()
candidate = (base / name).resolve()
if candidate.parent != base:
raise HTTPException(400, "Invalid skill name")
return candidate
def resolve_skill_dir(name: str) -> Path:
"""Return the directory of an existing skill by matching ``name`` against the
actual directory entries.
Using the entry from ``iterdir()`` (rather than a path built from ``name``)
means traversal input can never escape the skills directory. Raises 404 if no
skill matches.
"""
base = BASE_DIR / "skills"
if base.exists():
for entry in base.iterdir():
if entry.is_dir() and entry.name == name:
return entry
raise HTTPException(404, "Skill not found")
def skill_context_file_path(name: str, filename: str) -> Path:
if not SKILL_CONTEXT_FILENAME_RE.fullmatch(filename or ""):
raise HTTPException(400, "Invalid file name")
context_dir = (skill_dir_path(name) / "context").resolve()
candidate = (context_dir / filename).resolve()
if candidate.parent != context_dir:
raise HTTPException(400, "Invalid file name")
return candidate
@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", {}, best_effort=True)
score_history = read_json(d / "score-history.json", [], best_effort=True)
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}")
def get_skill(name: str):
path = BASE_DIR / "skills" / name
if not path.exists():
raise HTTPException(404, "Skill not found")
path = resolve_skill_dir(name)
return {
"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", default={}),
"score_history": read_json(path / "score-history.json", default=[]),
"context": [f.name for f in (path / "context").iterdir()] if (path / "context").exists() else [],
}
@app.post("/api/skills/{name}/run")
def run_skill(name: str, req: Optional[SkillRunRequest] = None):
path = BASE_DIR / "skills" / name
@app.post("/api/skills")
def create_skill(data: SkillCreate):
path = skill_dir_path(data.name)
if path.exists():
raise HTTPException(409, "Skill already exists")
path.mkdir(parents=True)
(path / "context").mkdir()
(path / "SKILL.md").write_text(data.skill_md, encoding="utf-8")
(path / "learnings.md").write_text("", encoding="utf-8")
append_audit({"action": "skill_created", "skill": data.name})
return {"name": data.name}
@app.put("/api/skills/{name}")
def update_skill(name: str, data: SkillUpdate):
path = skill_dir_path(name)
if not path.exists():
raise HTTPException(404, "Skill not found")
(path / "SKILL.md").write_text(data.skill_md, encoding="utf-8")
append_audit({"action": "skill_updated", "skill": name})
return {"status": "ok"}
@app.get("/api/skills/{name}/context/{filename}")
def get_skill_context_file(name: str, filename: str):
path = skill_context_file_path(name, filename)
if not path.exists():
raise HTTPException(404, "File not found")
return {"filename": filename, "content": read_file(path)}
@app.put("/api/skills/{name}/context/{filename}")
def put_skill_context_file(name: str, filename: str, data: SkillContextFileWrite):
skill_path = skill_dir_path(name)
if not skill_path.exists():
raise HTTPException(404, "Skill not found")
path = skill_context_file_path(name, filename)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(data.content, encoding="utf-8")
append_audit({"action": "skill_context_updated", "skill": name, "file": filename})
return {"status": "ok"}
@app.delete("/api/skills/{name}/context/{filename}")
def delete_skill_context_file(name: str, filename: str):
path = skill_context_file_path(name, filename)
if not path.exists():
raise HTTPException(404, "File not found")
path.unlink()
append_audit({"action": "skill_context_deleted", "skill": name, "file": filename})
return {"status": "deleted"}
@app.post("/api/skills/{name}/run")
def run_skill(name: str, req: Optional[SkillRunRequest] = None):
path = resolve_skill_dir(name)
agent_choice = req.agent if req else "auto"
skill_input = req.input if req else ""
@ -273,7 +444,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:
@ -316,10 +487,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())}
path = resolve_skill_dir(name) / "score-history.json"
return {"scores": read_json(path, default=[])}
# ─── Routes: Scheduler ────────────────────────────────────────────
@ -328,7 +497,9 @@ def list_jobs():
jobs_dir = BASE_DIR / "scheduler" / "jobs"
jobs = []
for f in sorted(jobs_dir.glob("*.json")):
jobs.append(json.loads(f.read_text()))
job = read_json(f, default=None, best_effort=True)
if job is not None:
jobs.append(job)
return jobs
@app.post("/api/scheduler/jobs")
@ -336,7 +507,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,
@ -355,8 +526,8 @@ def create_job(job: ScheduleJobRequest):
def delete_job(job_id: str):
jobs_dir = BASE_DIR / "scheduler" / "jobs"
for f in jobs_dir.glob("*.json"):
data = json.loads(f.read_text())
if data.get("id") == job_id:
data = read_json(f, default=None, best_effort=True)
if data and data.get("id") == job_id:
f.unlink()
append_audit({"action": "job_deleted", "job_id": job_id})
return {"status": "deleted"}
@ -370,7 +541,15 @@ def get_audit(limit: int = Query(100, le=500)):
if not audit_file.exists():
return {"entries": []}
lines = audit_file.read_text().strip().split("\n")
entries = [json.loads(l) for l in lines if l.strip()]
entries = []
for l in lines:
if not l.strip():
continue
try:
entries.append(json.loads(l))
except json.JSONDecodeError:
# Skip a corrupt line rather than failing the whole audit view.
continue
return {"entries": entries[-limit:]}
# ─── Routes: Cost Analytics ───────────────────────────────────────
@ -378,15 +557,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"),
@ -394,7 +570,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 ─────────────────────────────────────
@ -402,9 +578,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):
@ -412,7 +586,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({
@ -420,7 +594,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}
@ -452,13 +626,42 @@ def create_backup():
append_audit({"action": "backup_created", "file": backup_file.name})
return {"status": "ok", "file": backup_file.name, "size": backup_file.stat().st_size}
def _resolve_backup_file(name: str) -> Path:
"""Resolve a restore request to a real .tar.gz inside backups/, rejecting traversal.
The returned path is taken from the directory listing (never built from the raw
request value), so a caller can only ever select an existing backup file.
"""
if not name or name != Path(name).name or not name.endswith(".tar.gz"):
raise HTTPException(400, "Invalid backup file name")
backup_dir = (BASE_DIR / "backups").resolve()
for candidate in backup_dir.glob("*.tar.gz"):
if candidate.name == name:
return candidate
raise HTTPException(404, "Backup file not found")
def _safe_extractall(tar: tarfile.TarFile, dest: Path):
"""Extract a tar archive, refusing members that would escape dest (CVE-2007-4559)."""
dest = dest.resolve()
for member in tar.getmembers():
target = (dest / member.name).resolve()
if target != dest and dest not in target.parents:
raise HTTPException(400, f"Unsafe path in archive: {member.name}")
if member.issym() or member.islnk():
link_target = (target.parent / member.linkname).resolve()
if link_target != dest and dest not in link_target.parents:
raise HTTPException(400, f"Unsafe link in archive: {member.name}")
tar.extractall(path=dest, filter="data")
@app.post("/api/backup/restore")
def restore_backup(data: BackupRestoreRequest):
backup_file = BASE_DIR / "backups" / data.file
backup_file = _resolve_backup_file(data.file)
if not backup_file.exists():
raise HTTPException(404, "Backup file not found")
with tarfile.open(backup_file, "r:gz") as tar:
tar.extractall(path=BASE_DIR)
_safe_extractall(tar, BASE_DIR)
append_audit({"action": "backup_restored", "file": data.file})
return {"status": "restored"}
@ -477,17 +680,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"}
@ -519,16 +720,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)
@ -588,7 +787,7 @@ def execute_agent(agent: str, message: str) -> str:
elif agent == "hermes":
try:
code, out, err = run_cli(["hermes", "chat", "-q", message], timeout=180)
code, out, err = run_cli(hermes_cli_args("chat", "-q", message), timeout=180)
except subprocess.TimeoutExpired:
return f"⏱ Hermes timed out.\n\nThe model took too long to respond. Try a shorter query or check your OpenRouter rate limits.\n\n**Message:** {message[:100]}"
if code == 0:
@ -635,11 +834,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,
@ -650,7 +849,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,
@ -729,9 +928,20 @@ class PtySession:
os.kill(self.pid, signal.SIGKILL)
except OSError:
pass
try:
os.waitpid(self.pid, 0) # reap the killed child so it doesn't stay a zombie
except ChildProcessError:
pass
@app.websocket("/ws/terminal")
async def ws_terminal(websocket: WebSocket):
# CORSMiddleware does not protect WebSocket handshakes, so this endpoint - which spawns a
# full interactive shell - must check the Origin header itself, or any webpage could open
# this socket and get command execution on the machine running the dashboard.
origin = websocket.headers.get("origin")
if origin not in get_cors_origins():
await websocket.close(code=1008)
return
await websocket.accept()
session = PtySession()
try:
@ -849,7 +1059,9 @@ def load_kanban_tasks():
ensure_dir(KANBAN_DIR)
tasks = []
for f in sorted(KANBAN_DIR.glob("*.json")):
tasks.append(json.loads(f.read_text()))
task = read_json(f, default=None, best_effort=True)
if task is not None:
tasks.append(task)
return tasks
KANBAN_ID_RE = re.compile(r"^[0-9a-f]{6,16}$")
@ -868,7 +1080,7 @@ def save_kanban_task(task: dict):
ensure_dir(KANBAN_DIR)
kanban_task_path(task["id"]).write_text(json.dumps(task, indent=2))
KANBAN_AGENTS = {"opencode", "hermes", "gemini"}
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."""
@ -887,41 +1099,55 @@ def dispatch_kanban_task(task_id: str):
threading.Thread(target=_run_kanban_agent, args=(task_id,), daemon=True).start()
def _run_kanban_agent(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
return
task = json.loads(path.read_text())
agent = task.get("assignee")
prompt = task["title"] if not task.get("body") else f"{task['title']}\n\n{task['body']}"
# Runs in a daemon thread: any unhandled exception would be lost and leave
# the task stuck in "in_progress" forever, so catch failures and surface
# them by marking the task blocked with the error.
try:
path = kanban_task_path(task_id)
if not path.exists():
return
task = json.loads(path.read_text())
agent = task.get("assignee")
prompt = task["title"] if not task.get("body") else f"{task['title']}\n\n{task['body']}"
response = execute_agent(agent, prompt)
failed = response.startswith(("⏱", "⚠", "Unknown agent"))
response = execute_agent(agent, prompt)
failed = response.startswith(("⏱", "⚠", "Unknown agent"))
task = json.loads(path.read_text()) # reload in case it changed while the agent ran
task.setdefault("comments", []).append({
"id": str(uuid.uuid4())[:8],
"message": f"🤖 **{agent}**\n\n{response}",
"timestamp": get_timestamp(),
})
if failed:
task["status"] = "blocked"
task["block_reason"] = response[:300]
append_audit({"action": "kanban_task_dispatch_failed", "task_id": task_id, "agent": agent})
else:
task["status"] = "done"
task["summary"] = response[:300]
task["completed_at"] = get_timestamp()
append_audit({"action": "kanban_task_dispatch_completed", "task_id": task_id, "agent": agent})
task["updated"] = get_timestamp()
save_kanban_task(task)
task = json.loads(path.read_text()) # reload in case it changed while the agent ran
task.setdefault("comments", []).append({
"id": new_id(),
"message": f"🤖 **{agent}**\n\n{response}",
"timestamp": get_timestamp(),
})
if failed:
task["status"] = "blocked"
task["block_reason"] = response[:300]
append_audit({"action": "kanban_task_dispatch_failed", "task_id": task_id, "agent": agent})
else:
task["status"] = "done"
task["summary"] = response[:300]
task["completed_at"] = get_timestamp()
append_audit({"action": "kanban_task_dispatch_completed", "task_id": task_id, "agent": agent})
task["updated"] = get_timestamp()
save_kanban_task(task)
except Exception as e:
print(f"[kanban] dispatch for task {task_id} crashed: {e}")
try:
path = kanban_task_path(task_id)
task = json.loads(path.read_text())
task["status"] = "blocked"
task["block_reason"] = f"Dispatch crashed: {e}"[:300]
task["updated"] = get_timestamp()
save_kanban_task(task)
append_audit({"action": "kanban_task_dispatch_error", "task_id": task_id, "error": str(e)[:200]})
except Exception as inner:
print(f"[kanban] could not mark task {task_id} as blocked: {inner}")
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) ────────────────────────
@ -947,11 +1173,20 @@ def kanban_get_task(task_id: str):
raise HTTPException(404, "Task not found")
return json.loads(path.read_text())
@app.delete("/api/kanban/tasks/{task_id}")
def kanban_delete_task(task_id: str):
path = kanban_task_path(task_id)
if not path.exists():
raise HTTPException(404, "Task not found")
path.unlink()
append_audit({"action": "kanban_task_deleted", "task_id": task_id})
return {"status": "deleted", "task_id": task_id}
@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,
@ -1048,7 +1283,7 @@ def kanban_add_comment(task_id: str, data: KanbanCommentCreate):
raise HTTPException(404, "Task not found")
task = json.loads(path.read_text())
comment = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"message": data.message,
"timestamp": get_timestamp(),
}
@ -1118,7 +1353,7 @@ def kanban_decompose_task(task_id: str):
subtask = subtask.strip().lstrip("-* ")
if subtask:
child = {
"id": str(uuid.uuid4())[:8],
"id": new_id(),
"title": subtask[:80],
"body": subtask,
"status": "todo",
@ -1148,7 +1383,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,
@ -1260,7 +1495,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
@ -1273,7 +1508,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 {
@ -1294,7 +1529,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"})
@ -1332,7 +1567,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 {
@ -1349,22 +1584,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)}
@ -1372,18 +1602,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)}
@ -1457,7 +1684,7 @@ def index():
content = content.replace('src="app.js"', 'src="/dashboard/app.js"')
content = content.replace('pages/', '/dashboard/pages/')
return HTMLResponse(content=content)
return HTMLResponse("<h1>Agentic OS</h1><p>Dashboard not built yet. Run <code>./install.sh</code> first.</p>")
return HTMLResponse("<h1>Agentic OS</h1><p>Dashboard not built yet. Run the installer for your platform first (<code>./install.sh</code> on Linux/macOS or <code>.\\install.ps1</code> on Windows).</p>")
# ─── Favicon ──────────────────────────────────────────────────────

51
tests/conftest.py Normal file
View File

@ -0,0 +1,51 @@
"""Shared pytest fixtures for the Agentic OS test suite.
The application code (``server.py``) computes a number of module-level path
constants from ``BASE_DIR`` at import time. To keep tests hermetic — no writes
to the real repository — the fixtures below redirect every one of those
constants at a temporary directory and rebuild the minimal folder layout the
endpoints expect.
"""
import importlib
import sys
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
@pytest.fixture()
def server_module(tmp_path, monkeypatch):
"""Import ``server`` with all filesystem paths redirected to ``tmp_path``.
Returns the imported module with its path globals patched so tests can
exercise endpoints without touching the real project directories.
"""
server = importlib.import_module("server")
base = tmp_path
for sub in ["data", "data/kanban", "brain", "brain/journal", "audit",
"skills", "scheduler/jobs", "registry", "standards",
"prompts", "backups", "agents"]:
(base / sub).mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(server, "BASE_DIR", base)
monkeypatch.setattr(server, "KANBAN_DIR", base / "data" / "kanban")
monkeypatch.setattr(server, "GOALS_FILE", base / "data" / "goals.json")
monkeypatch.setattr(server, "JOURNAL_DIR", base / "brain" / "journal")
monkeypatch.setattr(server, "CHAT_HISTORY_FILE",
base / "data" / "chat-history.json")
monkeypatch.setattr(server, "_terminal_cwd", str(base))
return server
@pytest.fixture()
def client(server_module):
from fastapi.testclient import TestClient
with TestClient(server_module.app) as c:
yield c

65
tests/test_scheduler.py Normal file
View File

@ -0,0 +1,65 @@
"""Unit tests for ``scheduler/scheduler.py``."""
import importlib
import json
import sys
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
SCHEDULER_DIR = REPO_ROOT / "scheduler"
if str(SCHEDULER_DIR) not in sys.path:
sys.path.insert(0, str(SCHEDULER_DIR))
@pytest.fixture()
def scheduler_module(tmp_path, monkeypatch):
sched = importlib.import_module("scheduler")
base = tmp_path / "scheduler"
(base / "jobs").mkdir(parents=True)
(tmp_path / "audit").mkdir()
monkeypatch.setattr(sched, "BASE_DIR", base)
monkeypatch.setattr(sched, "JOBS_DIR", base / "jobs")
return sched
def test_run_skill_appends_audit(scheduler_module, capsys):
scheduler_module.run_skill("heartbeat")
audit_file = scheduler_module.BASE_DIR.parent / "audit" / "audit.log"
entry = json.loads(audit_file.read_text().strip())
assert entry["action"] == "scheduler_run"
assert entry["skill"] == "heartbeat"
assert "timestamp" in entry
assert "Ran skill: heartbeat" in capsys.readouterr().out
def test_load_jobs_registers_enabled(scheduler_module):
(scheduler_module.JOBS_DIR / "hb.json").write_text(json.dumps({
"id": "hb1", "name": "Heartbeat", "skill": "heartbeat",
"cron": "*/5 * * * *", "enabled": True,
}))
sched = scheduler_module.BackgroundScheduler()
scheduler_module.load_jobs(sched)
jobs = sched.get_jobs()
assert len(jobs) == 1
assert jobs[0].id == "hb1"
assert jobs[0].name == "Heartbeat"
def test_load_jobs_skips_disabled(scheduler_module):
(scheduler_module.JOBS_DIR / "off.json").write_text(json.dumps({
"id": "off1", "name": "Disabled", "skill": "x",
"cron": "0 0 * * *", "enabled": False,
}))
sched = scheduler_module.BackgroundScheduler()
scheduler_module.load_jobs(sched)
assert sched.get_jobs() == []
def test_load_jobs_falls_back_to_name_as_id(scheduler_module):
(scheduler_module.JOBS_DIR / "noid.json").write_text(json.dumps({
"name": "NoId", "skill": "x", "cron": "0 0 * * *", "enabled": True,
}))
sched = scheduler_module.BackgroundScheduler()
scheduler_module.load_jobs(sched)
assert sched.get_jobs()[0].id == "NoId"

359
tests/test_server_endpoints.py Normal file
View File

@ -0,0 +1,359 @@
"""Endpoint tests for ``server.py`` exercised through FastAPI's TestClient."""
import json
import pytest
# ─── Status / static ──────────────────────────────────────────────
def test_status_ok(client, server_module, monkeypatch):
monkeypatch.setattr(server_module.shutil, "which", lambda name: None)
r = client.get("/api/status")
assert r.status_code == 200
body = r.json()
assert body["status"] == "healthy"
assert {a["name"] for a in body["agents"]} == {"opencode", "hermes", "gemini"}
assert body["skills_count"] == 0
def test_index_without_dashboard(client):
r = client.get("/")
assert r.status_code == 200
assert "Agentic OS" in r.text
def test_favicon_endpoints(client):
for path in ("/favicon.ico", "/favicon.svg"):
r = client.get(path)
assert r.status_code == 200
assert r.headers["content-type"].startswith("image/svg")
# ─── Brain ─────────────────────────────────────────────────────────
def test_brain_list_and_get_and_update(client, server_module):
brain = server_module.BASE_DIR / "brain"
(brain / "memory.md").write_text("remember this")
r = client.get("/api/brain")
assert r.json()["memory.md"] == "remember this"
r = client.get("/api/brain/memory.md")
assert r.json() == {"name": "memory.md", "content": "remember this"}
r = client.put("/api/brain/memory.md", json={"content": "new content"})
assert r.status_code == 200
assert (brain / "memory.md").read_text() == "new content"
def test_brain_get_missing_404(client):
assert client.get("/api/brain/ghost.md").status_code == 404
# ─── Skills ────────────────────────────────────────────────────────
def _make_skill(server_module, name, skill_md="", eval_data=None, scores=None):
d = server_module.BASE_DIR / "skills" / name
d.mkdir(parents=True, exist_ok=True)
if skill_md:
(d / "SKILL.md").write_text(skill_md)
if eval_data is not None:
(d / "eval.json").write_text(json.dumps(eval_data))
if scores is not None:
(d / "score-history.json").write_text(json.dumps(scores))
return d
def test_list_skills(client, server_module):
_make_skill(server_module, "code-review", skill_md="Review code",
eval_data={"criteria": ["clarity"]}, scores=[{"score": 8}])
_make_skill(server_module, "_template", skill_md="ignored")
skills = client.get("/api/skills").json()
names = [s["name"] for s in skills]
assert "code-review" in names
assert "_template" not in names
cr = next(s for s in skills if s["name"] == "code-review")
assert cr["eval_criteria"] == ["clarity"]
assert cr["scores"] == [{"score": 8}]
def test_get_skill_and_missing(client, server_module):
_make_skill(server_module, "brainstorming", skill_md="Ideas",
eval_data={"criteria": []}, scores=[{"score": 5}])
r = client.get("/api/skills/brainstorming")
assert r.json()["skill"] == "Ideas"
assert client.get("/api/skills/nope").status_code == 404
def test_get_skill_eval(client, server_module):
_make_skill(server_module, "tdd-cycle", scores=[{"score": 7}])
assert client.get("/api/skills/tdd-cycle/eval").json() == {"scores": [{"score": 7}]}
assert client.get("/api/skills/other/eval").json() == {"scores": []}
@pytest.mark.parametrize("name,expected_agent", [
("devops-audit", "opencode"),
("research-synthesis", "gemini"),
("generic-skill", "opencode"),
])
def test_run_skill_auto_routes(client, server_module, monkeypatch, name, expected_agent):
_make_skill(server_module, name, skill_md="do stuff")
captured = {}
def fake_exec(agent, prompt):
captured["agent"] = agent
return "done"
monkeypatch.setattr(server_module, "execute_agent", fake_exec)
r = client.post(f"/api/skills/{name}/run", json={"input": "go", "agent": "auto"})
assert r.status_code == 200
assert r.json()["agent"] == expected_agent
assert captured["agent"] == expected_agent
def test_run_skill_uses_skill_md_primary(client, server_module, monkeypatch):
_make_skill(server_module, "meeting-notes", skill_md="Primary: hermes\nrest")
monkeypatch.setattr(server_module, "execute_agent", lambda a, p: "ok")
r = client.post("/api/skills/meeting-notes/run", json={"agent": "auto"})
assert r.json()["agent"] == "hermes"
def test_run_skill_missing_404(client, server_module, monkeypatch):
monkeypatch.setattr(server_module, "execute_agent", lambda a, p: "ok")
assert client.post("/api/skills/ghost/run", json={}).status_code == 404
# ─── Scheduler jobs ────────────────────────────────────────────────
def test_scheduler_job_crud(client, server_module):
assert client.get("/api/scheduler/jobs").json() == []
created = client.post("/api/scheduler/jobs", json={
"name": "nightly audit", "skill": "devops-audit", "cron": "0 0 * * *",
}).json()
assert created["name"] == "nightly audit"
jobs = client.get("/api/scheduler/jobs").json()
assert len(jobs) == 1
# file name derives from the job name with spaces replaced.
assert (server_module.BASE_DIR / "scheduler" / "jobs" / "nightly_audit.json").exists()
assert client.delete(f"/api/scheduler/jobs/{created['id']}").json() == {"status": "deleted"}
assert client.get("/api/scheduler/jobs").json() == []
def test_delete_missing_job_404(client):
assert client.delete("/api/scheduler/jobs/deadbeef").status_code == 404
# ─── Audit ─────────────────────────────────────────────────────────
def test_audit_empty_and_limit(client, server_module):
assert client.get("/api/audit").json() == {"entries": []}
for i in range(5):
server_module.append_audit({"action": "act", "n": i})
entries = client.get("/api/audit?limit=2").json()["entries"]
assert len(entries) == 2
assert entries[-1]["n"] == 4
# ─── Cost ──────────────────────────────────────────────────────────
def test_cost_empty_then_record(client):
assert client.get("/api/cost").json()["entries"] == []
client.post("/api/cost/record", json={"agent": "gemini", "tokens": 10, "cost": 0.0, "model": "flash"})
data = client.get("/api/cost").json()
assert data["entries"][0]["agent"] == "gemini"
assert data["entries"][0]["tokens"] == 10
# ─── Plugins ───────────────────────────────────────────────────────
def test_plugins_install_flow(client):
assert client.get("/api/plugins").json() == {"plugins": []}
assert client.post("/api/plugins/install", json={"name": "cool-plugin"}).json()["status"] == "installed"
assert client.post("/api/plugins/install", json={"name": "cool-plugin"}).json()["status"] == "already_installed"
assert client.post("/api/plugins/install", json={"name": ""}).status_code == 400
assert client.get("/api/plugins").json()["plugins"][0]["name"] == "cool-plugin"
# ─── Settings ──────────────────────────────────────────────────────
def test_settings_get_empty_and_update_merges(client, server_module):
assert client.get("/api/settings").json() == {}
client.put("/api/settings", json={"settings": {"theme": "dark"}})
client.put("/api/settings", json={"settings": {"port": 9000}})
result = client.get("/api/settings").json()
assert result == {"theme": "dark", "port": 9000}
# ─── Standards ─────────────────────────────────────────────────────
def test_standards_list_and_discover(client, server_module):
std = server_module.BASE_DIR / "standards"
(std / "naming.md").write_text("use snake_case")
(std / "index.yml").write_text("standards: [naming]")
body = client.get("/api/standards").json()
assert any(s["name"] == "naming" for s in body["standards"])
assert "snake_case" in body["standards"][0]["content"]
assert client.post("/api/standards/discover").json()["status"] == "discovery_started"
# ─── Prompts ───────────────────────────────────────────────────────
def test_prompts_list(client, server_module):
(server_module.BASE_DIR / "prompts" / "code-review.md").write_text("template body")
assert client.get("/api/prompts").json()["code-review"] == "template body"
# ─── Backups ───────────────────────────────────────────────────────
def test_backup_create_list_restore(client, server_module):
(server_module.BASE_DIR / "brain" / "memory.md").write_text("data")
created = client.post("/api/backup").json()
assert created["status"] == "ok"
listed = client.get("/api/backups").json()
assert any(b["name"] == created["file"] for b in listed)
assert client.post("/api/backup/restore", json={"file": created["file"]}).json()["status"] == "restored"
def test_restore_missing_backup_404(client):
assert client.post("/api/backup/restore", json={"file": "nope.tar.gz"}).status_code == 404
# ─── Chat ──────────────────────────────────────────────────────────
def test_chat_invalid_agent(client):
assert client.post("/api/chat", json={"agent": "bogus", "message": "hi"}).status_code == 400
def test_chat_records_history(client, server_module, monkeypatch):
monkeypatch.setattr(server_module, "execute_agent", lambda a, m: "hello from agent")
r = client.post("/api/chat", json={"agent": "Gemini", "message": "hi"})
assert r.status_code == 200
assert r.json()["response"]["content"] == "hello from agent"
history = client.get("/api/chat/history").json()["messages"]
assert history[0]["role"] == "user"
assert history[1]["role"] == "assistant"
# ─── Terminal ──────────────────────────────────────────────────────
def test_terminal_session_and_blank_command(client, server_module):
assert client.get("/api/terminal/session").json()["cwd"]
r = client.post("/api/terminal/run", json={"command": " "})
assert r.json()["returncode"] == 0
def test_terminal_cd_invalid_dir(client):
r = client.post("/api/terminal/run", json={"command": "cd /this/does/not/exist"})
assert r.json()["returncode"] == 1
assert "no such directory" in r.json()["stderr"]
def test_terminal_run_echo(client):
r = client.post("/api/terminal/run", json={"command": "echo hello-term"})
assert r.json()["returncode"] == 0
assert "hello-term" in r.json()["stdout"]
# ─── Goals ─────────────────────────────────────────────────────────
def test_goals_crud(client, server_module):
assert client.get("/api/goals").json() == {"goals": []}
created = client.post("/api/goals", json={"title": "Ship v1", "description": "launch"}).json()
gid = created["id"]
assert created["status"] == "active"
updated = client.put(f"/api/goals/{gid}", json={"progress": 50}).json()
assert updated["progress"] == 50
assert client.put("/api/goals/missing", json={"progress": 1}).status_code == 404
assert client.delete(f"/api/goals/{gid}").json() == {"status": "deleted"}
assert client.get("/api/goals").json() == {"goals": []}
def test_goal_creation_syncs_active_projects(client, server_module):
active = server_module.BASE_DIR / "brain" / "active-projects.md"
active.write_text("# Projects\n")
client.post("/api/goals", json={"title": "Docs", "description": "write docs"})
assert "Docs" in active.read_text()
# ─── Journal ───────────────────────────────────────────────────────
def test_journal_save_get_list_search(client, server_module):
assert client.get("/api/journal/entries").json() == {"entries": []}
client.put("/api/journal/entries/2026-07-08", json={"content": "Today I tested code"})
assert client.get("/api/journal/entries/2026-07-08").json()["content"] == "Today I tested code"
entries = client.get("/api/journal/entries").json()["entries"]
assert entries[0]["date"] == "2026-07-08"
found = client.get("/api/journal/search?q=tested").json()["results"]
assert found[0]["date"] == "2026-07-08"
assert client.get("/api/journal/search?q=").json() == {"results": []}
# ─── Agent health ──────────────────────────────────────────────────
def test_agent_health(client, server_module, monkeypatch):
monkeypatch.setattr(server_module.shutil, "which", lambda n: None)
body = client.get("/api/agents/health").json()
assert len(body["agents"]) == 3
assert body["agents"][0]["success_rate"] == 100
def test_agent_stats_valid_and_invalid(client, server_module, monkeypatch):
monkeypatch.setattr(server_module.shutil, "which", lambda n: None)
assert client.get("/api/agents/opencode/stats").json()["name"] == "opencode"
assert client.get("/api/agents/nope/stats").status_code == 400
def test_agent_health_refresh(client, server_module, monkeypatch):
monkeypatch.setattr(server_module.shutil, "which", lambda n: None)
assert len(client.post("/api/agents/health/refresh").json()["agents"]) == 3
# ─── Smart router ──────────────────────────────────────────────────
def test_router_suggest_high_confidence(client):
body = client.post("/api/router/suggest", json={"task": "deploy docker infra with terraform"}).json()
assert body["suggested_agent"] == "opencode"
assert body["confidence"] == "high"
def test_router_suggest_low_confidence(client):
body = client.post("/api/router/suggest", json={"task": "xyzzy"}).json()
assert body["confidence"] == "low"
def test_router_route_valid_and_invalid(client):
assert client.post("/api/router/route", json={"task": "t", "agent": "Hermes"}).json()["status"] == "routed"
assert client.post("/api/router/route", json={"task": "t", "agent": "bad"}).json()["status"] == "error"
# ─── Analytics ─────────────────────────────────────────────────────
def test_skill_analytics(client, server_module):
_make_skill(server_module, "code-review", scores=[{"score": 5}, {"score": 8}])
_make_skill(server_module, "brainstorming", scores=[])
body = client.get("/api/analytics/skills").json()["skills"]
cr = next(s for s in body if s["name"] == "code-review")
assert cr["total_runs"] == 2
assert cr["avg_score"] == 6.5
assert cr["trend"] == "up"
def test_trend_analytics(client, server_module):
_make_skill(server_module, "tdd-cycle", scores=[{"score": 3, "date": "d1"}])
body = client.get("/api/analytics/trends").json()["trends"]
assert body[0]["name"] == "tdd-cycle"
assert body[0]["scores"] == [3]
# ─── Session replay ────────────────────────────────────────────────
def test_sessions_list_empty(client, server_module, monkeypatch, tmp_path):
monkeypatch.setattr(server_module.Path, "home", staticmethod(lambda: tmp_path))
assert client.get("/api/sessions/list").json() == {"sessions": []}
def test_session_replay_not_found(client, server_module, monkeypatch, tmp_path):
monkeypatch.setattr(server_module.Path, "home", staticmethod(lambda: tmp_path))
body = client.get("/api/sessions/some-id/replay").json()
assert body["content"] == "Session log not found"

147
tests/test_server_helpers.py Normal file
View File

@ -0,0 +1,147 @@
"""Unit tests for the pure helper functions in ``server.py``."""
import json
import pytest
def test_read_file_missing_returns_empty(server_module, tmp_path):
assert server_module.read_file(tmp_path / "nope.txt") == ""
def test_read_write_file_roundtrip(server_module, tmp_path):
target = tmp_path / "note.txt"
assert server_module.write_file(target, "hello") is True
assert server_module.read_file(target) == "hello"
def test_list_dir_missing_returns_empty(server_module, tmp_path):
assert server_module.list_dir(tmp_path / "absent") == []
def test_list_dir_skips_hidden_and_sorts(server_module, tmp_path):
d = tmp_path / "things"
d.mkdir()
(d / "b.txt").write_text("")
(d / "a.txt").write_text("")
(d / ".hidden").write_text("")
assert server_module.list_dir(d) == ["a.txt", "b.txt"]
def test_get_timestamp_is_iso_utc(server_module):
ts = server_module.get_timestamp()
# datetime.fromisoformat round-trips a valid ISO 8601 string.
from datetime import datetime
parsed = datetime.fromisoformat(ts)
assert parsed.tzinfo is not None
def test_append_audit_writes_entry(server_module):
server_module.append_audit({"action": "unit_test"})
audit_file = server_module.BASE_DIR / "audit" / "audit.log"
lines = audit_file.read_text().strip().splitlines()
assert len(lines) == 1
entry = json.loads(lines[0])
assert entry["action"] == "unit_test"
assert "timestamp" in entry
assert len(entry["id"]) == 8
def test_get_cors_origins_defaults(server_module, monkeypatch):
monkeypatch.delenv("AGENTIC_OS_CORS_ORIGINS", raising=False)
origins = server_module.get_cors_origins()
assert "http://localhost:8080" in origins
assert "http://127.0.0.1:8080" in origins
def test_get_cors_origins_reads_port_from_settings(server_module, monkeypatch):
monkeypatch.delenv("AGENTIC_OS_CORS_ORIGINS", raising=False)
settings = server_module.BASE_DIR / "data" / "settings.json"
settings.write_text(json.dumps({"dashboard": {"port": 9000}}))
origins = server_module.get_cors_origins()
assert "http://localhost:9000" in origins
assert "http://127.0.0.1:9000" in origins
def test_get_cors_origins_includes_env_extras(server_module, monkeypatch):
monkeypatch.setenv("AGENTIC_OS_CORS_ORIGINS", "https://a.example, https://b.example ")
origins = server_module.get_cors_origins()
assert any(o == "https://a.example" for o in origins)
assert any(o == "https://b.example" for o in origins)
def test_get_cors_origins_bad_settings_falls_back(server_module):
settings = server_module.BASE_DIR / "data" / "settings.json"
settings.write_text("{ not valid json")
origins = server_module.get_cors_origins()
assert "http://localhost:8080" in origins
@pytest.mark.parametrize("which_result,expected", [(None, "offline"), ("/usr/bin/opencode", "online")])
def test_check_agent_opencode(server_module, monkeypatch, which_result, expected):
monkeypatch.setattr(server_module.shutil, "which", lambda name: which_result)
assert server_module.check_agent("opencode") == {"name": "opencode", "status": expected}
def test_check_agent_unknown_is_offline(server_module):
assert server_module.check_agent("mystery")["status"] == "offline"
def test_check_agent_gemini_warning_when_not_logged_in(server_module, monkeypatch, tmp_path):
monkeypatch.setattr(server_module.shutil, "which", lambda name: "/usr/bin/gemini")
monkeypatch.setattr(server_module.Path, "home", staticmethod(lambda: tmp_path))
# No oauth creds file -> installed but not logged in -> warning.
assert server_module.check_agent("gemini")["status"] == "warning"
def test_check_agent_gemini_online_when_logged_in(server_module, monkeypatch, tmp_path):
monkeypatch.setattr(server_module.shutil, "which", lambda name: "/usr/bin/gemini")
monkeypatch.setattr(server_module.Path, "home", staticmethod(lambda: tmp_path))
creds = tmp_path / ".gemini" / "oauth_creds.json"
creds.parent.mkdir(parents=True)
creds.write_text('{"token": "ya29.abc"}')
assert server_module.check_agent("gemini")["status"] == "online"
def test_clean_hermes_output_empty(server_module):
assert server_module.clean_hermes_output("") == ""
def test_clean_hermes_output_extracts_box_content(server_module):
raw = "Query: hi\n╭─ box\nHello there\nSecond line\n╰─ end\nDuration: 1s"
assert server_module.clean_hermes_output(raw) == "Hello there\nSecond line"
def test_clean_hermes_output_fallback_without_box(server_module):
raw = "Query: hi\nInitializing...\nActual answer here"
assert "Actual answer here" in server_module.clean_hermes_output(raw)
def test_kanban_task_path_rejects_bad_id(server_module):
with pytest.raises(server_module.HTTPException):
server_module.kanban_task_path("../../etc/passwd")
def test_kanban_task_path_rejects_empty(server_module):
with pytest.raises(server_module.HTTPException):
server_module.kanban_task_path("")
def test_kanban_task_path_accepts_valid_id(server_module):
path = server_module.kanban_task_path("abc123")
assert path.name == "abc123.json"
assert path.parent == server_module.KANBAN_DIR.resolve()
def test_load_save_chat_history_roundtrip(server_module):
assert server_module.load_chat_history() == {"messages": []}
server_module.save_chat_message({"content": "hi"})
assert server_module.load_chat_history()["messages"][-1]["content"] == "hi"
def test_save_chat_message_caps_at_200(server_module):
for i in range(210):
server_module.save_chat_message({"content": str(i)})
history = server_module.load_chat_history()
assert len(history["messages"]) == 200
assert history["messages"][-1]["content"] == "209"

158
tests/test_server_kanban.py Normal file
View File

@ -0,0 +1,158 @@
"""Tests for the Kanban board endpoints and autonomous dispatch logic."""
import json
import pytest
@pytest.fixture()
def sync_threads(server_module, monkeypatch):
"""Run ``threading.Thread`` targets synchronously so dispatch is deterministic."""
class _SyncThread:
def __init__(self, target=None, args=(), kwargs=None, daemon=None):
self._target = target
self._args = args
self._kwargs = kwargs or {}
def start(self):
if self._target:
self._target(*self._args, **self._kwargs)
monkeypatch.setattr(server_module.threading, "Thread", _SyncThread)
return server_module
def _create(client, **kw):
payload = {"title": "T", "body": "", "status": "triage",
"priority": "medium", "assignee": ""}
payload.update(kw)
return client.post("/api/kanban/tasks", json=payload).json()
def test_board_empty(client):
body = client.get("/api/kanban/board").json()
assert body["total"] == 0
assert set(body["columns"]) == {"triage", "todo", "ready", "in_progress", "blocked", "done"}
def test_create_and_get_task(client):
task = _create(client, title="Write tests", assignee="")
assert task["title"] == "Write tests"
assert task["status"] == "triage"
fetched = client.get(f"/api/kanban/tasks/{task['id']}").json()
assert fetched["id"] == task["id"]
def test_get_missing_task_404(client):
assert client.get("/api/kanban/tasks/abc123").status_code == 404
def test_board_groups_by_status(client):
_create(client, status="todo")
_create(client, status="done")
board = client.get("/api/kanban/board").json()
assert len(board["columns"]["todo"]) == 1
assert len(board["columns"]["done"]) == 1
assert board["total"] == 2
# filter by status query
filtered = client.get("/api/kanban/board?status=todo").json()
assert filtered["total"] == 1
def test_update_task(client):
task = _create(client)
updated = client.patch(f"/api/kanban/tasks/{task['id']}",
json={"title": "renamed", "priority": "high"}).json()
assert updated["title"] == "renamed"
assert updated["priority"] == "high"
def test_update_missing_404(client):
assert client.patch("/api/kanban/tasks/abc123", json={"title": "x"}).status_code == 404
def test_complete_block_unblock(client):
task = _create(client)
tid = task["id"]
assert client.post(f"/api/kanban/tasks/{tid}/complete", json={"summary": "done!"}).json()["status"] == "done"
assert client.post(f"/api/kanban/tasks/{tid}/block", json={"reason": "stuck"}).json()["status"] == "blocked"
unblocked = client.post(f"/api/kanban/tasks/{tid}/unblock").json()
assert unblocked["status"] == "ready"
assert unblocked["block_reason"] == ""
def test_comments(client):
task = _create(client)
updated = client.post(f"/api/kanban/tasks/{task['id']}/comments", json={"message": "hi there"}).json()
assert updated["comments"][0]["message"] == "hi there"
def test_links_add_and_remove(client):
parent = _create(client)
child = _create(client)
r = client.post("/api/kanban/links", json={"parent_id": parent["id"], "child_id": child["id"]})
assert r.json() == {"status": "linked"}
linked = client.get(f"/api/kanban/tasks/{parent['id']}").json()
assert {"parent": parent["id"], "child": child["id"]} in linked["links"]
r = client.delete(f"/api/kanban/links?parent_id={parent['id']}&child_id={child['id']}")
assert r.json() == {"status": "unlinked"}
unlinked = client.get(f"/api/kanban/tasks/{parent['id']}").json()
assert unlinked["links"] == []
def test_link_missing_task_404(client):
parent = _create(client)
assert client.post("/api/kanban/links",
json={"parent_id": parent["id"], "child_id": "abcdef"}).status_code == 404
def test_specify_moves_triage_to_todo(client):
task = _create(client, status="triage")
assert client.post(f"/api/kanban/tasks/{task['id']}/specify").json()["status"] == "todo"
def test_decompose_creates_children(client):
task = _create(client, body="- first subtask\n- second subtask\n\n")
body = client.post(f"/api/kanban/tasks/{task['id']}/decompose").json()
assert body["parent"] == task["id"]
assert len(body["children"]) == 2
titles = [c["title"] for c in body["children"]]
assert "first subtask" in titles
def test_dispatch_requires_valid_assignee(client):
task = _create(client, assignee="")
assert client.post(f"/api/kanban/tasks/{task['id']}/dispatch").status_code == 400
def test_create_with_agent_assignee_dispatches(client, sync_threads, monkeypatch):
monkeypatch.setattr(sync_threads, "execute_agent", lambda agent, prompt: "agent finished the job")
task = _create(client, title="Do work", assignee="opencode")
# With synchronous dispatch, the task runs to completion immediately.
assert task["status"] == "done"
assert task["summary"].startswith("agent finished")
assert any("opencode" in c["message"] for c in task["comments"])
def test_dispatch_failure_blocks_task(client, sync_threads, monkeypatch):
monkeypatch.setattr(sync_threads, "execute_agent", lambda agent, prompt: "⚠ Agent not installed")
task = _create(client, title="Do work", assignee="hermes")
assert task["status"] == "blocked"
assert "not installed" in task["block_reason"]
def test_bulk_dispatch_endpoint(client, sync_threads, monkeypatch):
monkeypatch.setattr(sync_threads, "execute_agent", lambda agent, prompt: "ok done")
# Write task files directly so they start eligible (todo/ready with an agent
# assignee) — going through the create endpoint would auto-dispatch them.
def _seed(tid, status, assignee):
sync_threads.save_kanban_task({
"id": tid, "title": "t", "body": "", "status": status,
"priority": "medium", "assignee": assignee, "comments": [], "links": [],
})
_seed("aaaaaa", "todo", "gemini")
_seed("bbbbbb", "ready", "opencode")
_seed("cccccc", "triage", "") # not eligible
body = client.post("/api/kanban/dispatch").json()
assert len(body["dispatched"]) == 2