@@ -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 = `
`;
@@ -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 = `
@@ -100,8 +103,11 @@ async function showSkillDetail(name) {
-
-
${escapeHtml(skill.skill || 'No SKILL.md')}
+
+
${escapeHtml(skill.skill || 'No SKILL.md')}
@@ -120,10 +126,20 @@ async function showSkillDetail(name) {
` : '
No evaluation scores yet
'}
-
- ${skill.context && skill.context.length > 0
- ? `
${skill.context.map(f => `${f}`).join('')}
`
- : '
No context files
'}
+
+
+ ${skill.context && skill.context.length > 0
+ ? skill.context.map(f => `
+
+ ${escapeHtml(f)}
+
+
+
`).join('')
+ : '
No context files
'}
+
${skill.eval && skill.eval.criteria ? `
Eval Criteria:${skill.eval.criteria.map(c => `${c}`).join('')}
` : ''}
@@ -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', `
+
+
+
+
+
+ `, `
+
+
+ `);
+}
+
+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 = `
+
+
+
+
+
+ `;
+}
+
+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', `
+
+
+
+
+
+
+
+
+ `, `
+
+
+ `);
+}
+
+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}`, `
+
+ `, `
+
+
+ `);
+ } 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');
+ }
+}
diff --git a/data/kanban/0f822987.json b/data/kanban/0f822987.json
deleted file mode 100644
index aca84b9..0000000
--- a/data/kanban/0f822987.json
+++ /dev/null
@@ -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"
-}
\ No newline at end of file
diff --git a/data/kanban/8893ad14.json b/data/kanban/8893ad14.json
deleted file mode 100644
index 61c8b51..0000000
--- a/data/kanban/8893ad14.json
+++ /dev/null
@@ -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"
-}
\ No newline at end of file
diff --git a/install.ps1 b/install.ps1
index a9bab85..6379aee 100644
--- a/install.ps1
+++ b/install.ps1
@@ -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."
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..3b2c446
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+addopts = -q
diff --git a/requirements-dev.txt b/requirements-dev.txt
new file mode 100644
index 0000000..8cb33fb
--- /dev/null
+++ b/requirements-dev.txt
@@ -0,0 +1,4 @@
+-r requirements.txt
+pytest>=8.0.0
+pytest-cov>=5.0.0
+httpx>=0.27.0
diff --git a/scheduler/scheduler.py b/scheduler/scheduler.py
index 61638d3..b8cba8a 100644
--- a/scheduler/scheduler.py
+++ b/scheduler/scheduler.py
@@ -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():
diff --git a/server.py b/server.py
index f137ee7..4b0760b 100644
--- a/server.py
+++ b/server.py
@@ -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("
Agentic OS
Dashboard not built yet. Run ./install.sh first.
")
+ return HTMLResponse("
Agentic OS
Dashboard not built yet. Run the installer for your platform first (./install.sh on Linux/macOS or .\\install.ps1 on Windows).
")
# โโโ Favicon โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..e79162f
--- /dev/null
+++ b/tests/conftest.py
@@ -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
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
new file mode 100644
index 0000000..2a4ee5d
--- /dev/null
+++ b/tests/test_scheduler.py
@@ -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"
diff --git a/tests/test_server_endpoints.py b/tests/test_server_endpoints.py
new file mode 100644
index 0000000..7df0ac6
--- /dev/null
+++ b/tests/test_server_endpoints.py
@@ -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"
diff --git a/tests/test_server_helpers.py b/tests/test_server_helpers.py
new file mode 100644
index 0000000..3f4afa4
--- /dev/null
+++ b/tests/test_server_helpers.py
@@ -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"
diff --git a/tests/test_server_kanban.py b/tests/test_server_kanban.py
new file mode 100644
index 0000000..8c9c077
--- /dev/null
+++ b/tests/test_server_kanban.py
@@ -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