From 0375b20bb453a34d23518412ed0b8ed9fbccb2a4 Mon Sep 17 00:00:00 2001 From: nearxos Date: Sun, 14 Jun 2026 22:27:24 +0300 Subject: [PATCH] Enhance MCP development features and introduce skills management - Added configuration options for requiring human approval before applying LLM-generated MCP patches. - Updated Docker setup to include skills directory. - Integrated skills management into the backend, allowing for procedural guides and skill matching. - Refactored database initialization to apply Alembic migrations. - Enhanced task approval process to handle MCP patch applications with optional approval. - Introduced new schemas for skills and updated existing APIs to support skills functionality. This commit lays the groundwork for improved agent capabilities and better management of MCP development processes. --- .env.example | 7 + backend/Dockerfile | 1 + backend/alembic.ini | 44 +++ backend/alembic/env.py | 62 ++++ backend/alembic/script.py.mako | 25 ++ .../alembic/versions/001_initial_schema.py | 92 ++++++ backend/app/agent/__init__.py | 4 +- backend/app/agent/graph.py | 44 +++ backend/app/agent/mcp_development.py | 168 ++++++----- backend/app/api/skills.py | 97 +++++++ backend/app/api/tasks.py | 24 +- backend/app/config.py | 6 + backend/app/database.py | 63 ++--- backend/app/main.py | 4 +- backend/app/mcp_manager/commands.py | 30 +- backend/app/schemas.py | 31 ++ backend/app/services/device_defaults.py | 4 +- backend/app/services/mcp_approvals.py | 208 ++++++++++++++ backend/app/services/skill_registry.py | 264 ++++++++++++++++++ backend/app/worker.py | 2 +- backend/tests/conftest.py | 13 + backend/tests/test_device_defaults.py | 74 ++--- backend/tests/test_mcp_approvals.py | 106 +++++++ backend/tests/test_skill_registry.py | 60 ++++ docker-compose.yml | 2 + frontend/app/skills/page.tsx | 199 +++++++++++++ frontend/components/AppShell.tsx | 1 + skills/pfsense-change-safe/SKILL.md | 71 +++++ skills/vessel-dns-portal/SKILL.md | 81 ++++++ skills/vessel-voip/SKILL.md | 97 +++++++ 30 files changed, 1733 insertions(+), 151 deletions(-) create mode 100644 backend/alembic.ini create mode 100644 backend/alembic/env.py create mode 100644 backend/alembic/script.py.mako create mode 100644 backend/alembic/versions/001_initial_schema.py create mode 100644 backend/app/api/skills.py create mode 100644 backend/app/services/mcp_approvals.py create mode 100644 backend/app/services/skill_registry.py create mode 100644 backend/tests/conftest.py create mode 100644 backend/tests/test_mcp_approvals.py create mode 100644 backend/tests/test_skill_registry.py create mode 100644 frontend/app/skills/page.tsx create mode 100644 skills/pfsense-change-safe/SKILL.md create mode 100644 skills/vessel-dns-portal/SKILL.md create mode 100644 skills/vessel-voip/SKILL.md diff --git a/.env.example b/.env.example index 0e60e3a..e50eedb 100644 --- a/.env.example +++ b/.env.example @@ -68,6 +68,8 @@ LLM_CLOUD_PROVIDER_ORDER=gemini,deepseek,openrouter # ---- MCP self-development (agent patches MCP source on failures) ------ MCP_DEVELOPMENT_ENABLED=true +# Require human approval before LLM-generated MCP source is applied/reloaded (recommended) +MCP_DEV_REQUIRE_APPROVAL=true # Push to Gitea automatically after patch (false = approval required on task) MCP_DEV_AUTO_PUSH=false MCP_DEV_MODEL_TIER=premium @@ -116,6 +118,11 @@ OBSIDIAN_SEARCH_PATHS=agentic-os-tasks,Meta/Agent-Memory # Troubleshooting decision rules (YAML) TROUBLESHOOTING_RULES_PATH=/app/rules/troubleshooting.yaml +# Agent skills (procedural guides — skills/*/SKILL.md) +SKILLS_PATH=/app/skills +SKILLS_MAX_MATCHED=2 +SKILLS_MAX_BODY_CHARS=8000 + # ---- Frontend --------------------------------------------------------- # Public URL the browser uses to reach the backend API NEXT_PUBLIC_API_BASE_URL=http://localhost:8000 diff --git a/backend/Dockerfile b/backend/Dockerfile index fcf2add..5c5da61 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -17,6 +17,7 @@ RUN pip install -r requirements.txt COPY backend/ . COPY rules/ /app/rules/ +COPY skills/ /app/skills/ EXPOSE 8000 diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000..23ef524 --- /dev/null +++ b/backend/alembic.ini @@ -0,0 +1,44 @@ +# Alembic configuration for Agentic OS (backend package root). + +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os + +sqlalchemy.url = driver://user:pass@localhost/dbname + +[post_write_hooks] + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/backend/alembic/env.py b/backend/alembic/env.py new file mode 100644 index 0000000..bc51ad2 --- /dev/null +++ b/backend/alembic/env.py @@ -0,0 +1,62 @@ +"""Alembic migration environment (async SQLAlchemy).""" +from __future__ import annotations + +import asyncio +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config + +from app.config import settings +from app.database import Base + +# Import models so Base.metadata is populated. +from app import models # noqa: F401 + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +config.set_main_option("sqlalchemy.url", settings.database_url) +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + connectable = async_engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/backend/alembic/script.py.mako b/backend/alembic/script.py.mako new file mode 100644 index 0000000..17dcba0 --- /dev/null +++ b/backend/alembic/script.py.mako @@ -0,0 +1,25 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision: str = ${repr(up_revision)} +down_revision: Union[str, None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/backend/alembic/versions/001_initial_schema.py b/backend/alembic/versions/001_initial_schema.py new file mode 100644 index 0000000..1a436d0 --- /dev/null +++ b/backend/alembic/versions/001_initial_schema.py @@ -0,0 +1,92 @@ +"""Initial schema + legacy location→vessel renames. + +Revision ID: 001_initial +Revises: +Create Date: 2026-06-14 +""" +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op + +revision: str = "001_initial" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Legacy dev DBs may still use locations/location_id — rename before create_all. + op.execute( + """ + DO $$ + BEGIN + IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'locations') + AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'vessels') + THEN + ALTER TABLE locations RENAME TO vessels; + END IF; + IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'devices' AND column_name = 'location_id') + THEN + ALTER TABLE devices RENAME COLUMN location_id TO vessel_id; + END IF; + IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'tasks' AND column_name = 'location_id') + THEN + ALTER TABLE tasks RENAME COLUMN location_id TO vessel_id; + END IF; + END $$; + """ + ) + + from app import models # noqa: F401 + from app.database import Base + + bind = op.get_bind() + Base.metadata.create_all(bind) + + for stmt in ( + "CREATE INDEX IF NOT EXISTS ix_tasks_vessel_id ON tasks (vessel_id)", + "CREATE INDEX IF NOT EXISTS ix_task_events_task_id_id ON task_events (task_id, id)", + "CREATE INDEX IF NOT EXISTS ix_approval_requests_task_status " + "ON approval_requests (task_id, status)", + ): + op.execute(stmt) + + # catalog_key added on older DBs that predate the column. + op.execute( + """ + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_name = 'devices' AND column_name = 'catalog_key' + ) THEN + ALTER TABLE devices ADD COLUMN catalog_key VARCHAR(64); + END IF; + END $$; + """ + ) + + +def downgrade() -> None: + op.drop_index("ix_approval_requests_task_status", table_name="approval_requests") + op.drop_index("ix_task_events_task_id_id", table_name="task_events") + op.drop_index("ix_tasks_vessel_id", table_name="tasks") + + for table in ( + "approval_requests", + "task_events", + "tasks", + "devices", + "vessels", + "audit_logs", + "balance_snapshots", + "users", + ): + op.execute(f"DROP TABLE IF EXISTS {table} CASCADE") + + op.execute("DROP TYPE IF EXISTS approvalstatus") + op.execute("DROP TYPE IF EXISTS taskstatus") + op.execute("DROP TYPE IF EXISTS devicetype") + op.execute("DROP TYPE IF EXISTS role") diff --git a/backend/app/agent/__init__.py b/backend/app/agent/__init__.py index 97af5a5..b7018ac 100644 --- a/backend/app/agent/__init__.py +++ b/backend/app/agent/__init__.py @@ -1,3 +1,3 @@ -from app.agent.graph import run_task_agent +"""Agent package — import submodules directly (e.g. app.agent.graph).""" -__all__ = ["run_task_agent"] +__all__: list[str] = [] diff --git a/backend/app/agent/graph.py b/backend/app/agent/graph.py index 0b9ec15..4a7a1ee 100644 --- a/backend/app/agent/graph.py +++ b/backend/app/agent/graph.py @@ -37,6 +37,7 @@ from app.services.device_defaults import resolve_slot_defaults from app.services.investigation_context import gather_investigation_context from app.services.approval_filter import split_proposed_fixes from app.services.troubleshooting_rules import rule_guidance_block +from app.services.skill_registry import match_skills, skill_guidance_block from app.services.llm import triage_model, triage_model_info from app.services.llm_usage import LlmUsageTracker, merge_run_history from app.services.model_config import get_routing_config @@ -75,6 +76,7 @@ class AgentState(TypedDict, total=False): prior_context: dict | None investigation_context: dict | None matched_rule: dict | None + matched_skills: list[dict] | None follow_up: str | None run_number: int prior_runs: list[dict] @@ -348,6 +350,7 @@ async def triage_node(state: AgentState) -> AgentState: f"Onboard devices to check: {json.dumps(device_summary)}" f"{_investigation_context_block(state.get('investigation_context'))}" f"{rule_guidance_block(state.get('matched_rule'))}" + f"{skill_guidance_block(state.get('matched_skills'))}" f"{_prior_context_block(state.get('prior_context'), state.get('follow_up'))}" ) triage_info = await triage_model_info() @@ -449,6 +452,7 @@ async def reason_node(state: AgentState) -> AgentState: f"Title: {state['title']}\nIssue: {state['issue']}\nTriage: {json.dumps(triage)}\n\n{diag_text}" f"{_investigation_context_block(state.get('investigation_context'))}" f"{rule_guidance_block(state.get('matched_rule'))}" + f"{skill_guidance_block(state.get('matched_skills'))}" f"{_prior_context_block(state.get('prior_context'), state.get('follow_up'))}" ) messages = [SystemMessage(content=prompt), HumanMessage(content=ctx)] @@ -604,6 +608,10 @@ async def report_node(state: AgentState) -> AgentState: } if state.get("matched_rule"): report["matched_rule"] = state["matched_rule"] + if state.get("matched_skills"): + report["matched_skills"] = [ + {k: v for k, v in s.items() if k != "body"} for s in state["matched_skills"] + ] report = merge_run_history(state.get("prior_runs") or [], run_number, report, llm_summary) @@ -786,6 +794,7 @@ async def run_task_agent(task_id: int) -> None: devices: list[dict] = [] matched_rule: dict | None = None + matched_skills: list[dict] = [] vessel_name: str | None = None vessel_site: str | None = None vessel_public_ip: str | None = None @@ -795,6 +804,18 @@ async def run_task_agent(task_id: int) -> None: all_devices, routing_text, title=task.title ) matched_rule = rule.as_dict() if rule else None + rule_id = matched_rule.get("id") if matched_rule else None + matched_skills = [ + { + "id": s.id, + "name": s.name, + "description": s.description, + "body": s.body, + "score": s.score, + "priority": s.priority, + } + for s in match_skills(task.title, routing_text, matched_rule_id=rule_id) + ] vres = await db.execute(select(Vessel).where(Vessel.id == task.vessel_id)) vessel = vres.scalar_one_or_none() if vessel: @@ -807,6 +828,20 @@ async def run_task_agent(task_id: int) -> None: if d: devices = [_device_dict(d)] + if not matched_skills: + rule_id = matched_rule.get("id") if matched_rule else None + matched_skills = [ + { + "id": s.id, + "name": s.name, + "description": s.description, + "body": s.body, + "score": s.score, + "priority": s.priority, + } + for s in match_skills(task.title, routing_text, matched_rule_id=rule_id) + ] + task.status = TaskStatus.running task.details = details await db.commit() @@ -842,6 +877,14 @@ async def run_task_agent(task_id: int) -> None: f"Matched rule: {matched_rule.get('name')} → {', '.join(matched_rule.get('devices') or [])}", matched_rule, ) + if matched_skills: + names = ", ".join(s["name"] for s in matched_skills) + await _emit( + task_id, + "skill", + f"Matched skills: {names}", + {"skills": [{k: v for k, v in s.items() if k != "body"} for s in matched_skills]}, + ) remaining = await balance.latest_remaining_usd() from app.config import settings @@ -873,6 +916,7 @@ async def run_task_agent(task_id: int) -> None: "run_number": run_number, "prior_runs": prior_runs if continue_mode else [], "matched_rule": matched_rule, + "matched_skills": matched_skills or None, } try: diff --git a/backend/app/agent/mcp_development.py b/backend/app/agent/mcp_development.py index c3feab1..81b2fce 100644 --- a/backend/app/agent/mcp_development.py +++ b/backend/app/agent/mcp_development.py @@ -10,6 +10,10 @@ from langchain_core.messages import HumanMessage, SystemMessage from app.agent.json_utils import parse_json as _parse_json from app.config import settings from app.services import mcp_workspace +from app.services.mcp_approvals import ( + backup_proposal_files, + execute_approved_patch, +) from app.services.model_config import get_routing_config, resolve_tier_model from app.services.model_router import build_chat_model @@ -103,16 +107,67 @@ async def propose_mcp_patch( return {"ok": True, **data} -async def apply_and_reload_mcp( +async def _queue_apply_patch_approval( + *, + task_id: int, server: str, - edits: list[dict], - manager, + proposal: dict, + tool: str, + args: dict, + device: str | None, + emit: EmitFn, ) -> dict[str, Any]: - written = await mcp_workspace.apply_file_edits(server, edits) - if not written: - return {"ok": False, "error": "No files written"} - await manager.reload_server(server) - return {"ok": True, "files": written, "tools": mcp_workspace.list_registered_tools(server)} + from app.database import SessionLocal + from app.models.enums import ApprovalStatus + from app.models.task import ApprovalRequest + + edits = proposal.get("files") or [] + backups = await backup_proposal_files(server, edits) + commit_msg = proposal.get("commit_message") or f"agentic-os: fix {tool} on {server}" + + approval_args = { + "server": server, + "files": edits, + "backups": backups, + "commit_message": commit_msg, + "summary": proposal.get("summary"), + "retry_context": { + "device": device, + "server": server, + "tool": tool, + "args": args, + }, + } + + async with SessionLocal() as db: + db.add( + ApprovalRequest( + task_id=task_id, + tool_name="mcp_apply_patch", + tool_args=approval_args, + risk=proposal.get("summary"), + status=ApprovalStatus.pending, + ) + ) + await db.commit() + + await emit( + task_id, + "approval", + f"MCP patch proposed for {server} — approve to apply and reload", + { + "server": server, + "tool_name": "mcp_apply_patch", + "files": [e.get("path") for e in edits], + "summary": proposal.get("summary"), + }, + ) + return { + "ok": True, + "pending_approval": True, + "approval_type": "mcp_apply_patch", + "summary": proposal.get("summary"), + } async def attempt_mcp_repair( @@ -128,8 +183,9 @@ async def attempt_mcp_repair( emit: EmitFn, task_title: str = "", push: bool | None = None, + device: str | None = None, ) -> dict[str, Any]: - """Full repair loop: analyze → patch → reload → optional git push.""" + """Full repair loop: analyze → (approval?) → patch → reload → optional git push.""" if not settings.mcp_development_enabled: return {"ok": False, "skipped": True, "reason": "MCP development disabled"} @@ -152,6 +208,17 @@ async def attempt_mcp_repair( await emit(task_id, "mcp_dev", f"MCP patch proposal failed: {proposal.get('error')}", proposal) return proposal + if settings.mcp_dev_require_approval and task_id > 0: + return await _queue_apply_patch_approval( + task_id=task_id, + server=server, + proposal=proposal, + tool=tool, + args=args, + device=device, + emit=emit, + ) + await emit( task_id, "mcp_dev", @@ -166,66 +233,25 @@ async def attempt_mcp_repair( {"server": server, "phase": "reload"}, ) - reload_result = await apply_and_reload_mcp(server, proposal.get("files") or [], manager) - if not reload_result.get("ok"): - await emit(task_id, "error", reload_result.get("error", "MCP reload failed"), reload_result) - return reload_result - - await emit( - task_id, - "mcp_dev", - f"MCP '{server}' reloaded — {len(reload_result.get('tools') or [])} tools registered", - {"server": server, "phase": "reload_done", "tools": reload_result.get("tools")}, + return await execute_approved_patch( + task_id=task_id, + tool_args={ + "server": server, + "files": proposal.get("files") or [], + "commit_message": proposal.get("commit_message") or f"agentic-os: fix {tool} on {server}", + "summary": proposal.get("summary"), + "retry_context": { + "device": device, + "server": server, + "tool": tool, + "args": args, + }, + }, + manager=manager, + emit=emit, + push=push, ) - should_push = push if push is not None else settings.mcp_dev_auto_push - push_result: dict[str, Any] | None = None - commit_msg = proposal.get("commit_message") or f"agentic-os: fix {tool} on {server}" - written = reload_result.get("files") or [] - - if should_push and settings.gitea_token: - push_result = await mcp_workspace.commit_and_push(server, commit_msg, written) - await emit( - task_id, - "mcp_dev", - f"Git push {'ok' if push_result.get('ok') else 'failed'}: {push_result.get('commit') or push_result.get('error')}", - {"server": server, "phase": "push", **push_result}, - ) - elif settings.gitea_token and written and task_id > 0: - from app.database import SessionLocal - from app.models.enums import ApprovalStatus - from app.models.task import ApprovalRequest - - async with SessionLocal() as db: - db.add( - ApprovalRequest( - task_id=task_id, - tool_name="mcp_git_push", - tool_args={ - "server": server, - "files": written, - "commit_message": commit_msg, - }, - risk=proposal.get("summary"), - status=ApprovalStatus.pending, - ) - ) - await db.commit() - await emit( - task_id, - "approval", - f"MCP patch ready — approve to push {server} to Gitea", - {"server": server, "tool_name": "mcp_git_push", "files": written}, - ) - - return { - "ok": True, - "summary": proposal.get("summary"), - "reload": reload_result, - "push": push_result, - "tools": reload_result.get("tools"), - } - async def maybe_repair_after_tool_failure( *, @@ -259,10 +285,14 @@ async def maybe_repair_after_tool_failure( emit=emit, task_title=task_title, push=settings.mcp_dev_auto_push, + device=device, ) if not result.get("ok"): return None + if result.get("pending_approval"): + return None + if result.get("push") and not result["push"].get("ok") and result["push"].get("runtime_only"): await emit( task_id, @@ -271,6 +301,10 @@ async def maybe_repair_after_tool_failure( result["push"], ) + retry = result.get("retry") + if retry and retry.get("result") is not None: + return retry["result"], bool(retry.get("ok")) + from app.agent.mcp_helpers import tool_result_ok from app.agent.tool_events import tool_event_payload from app.mcp_manager.recovery import call_tool_resilient diff --git a/backend/app/api/skills.py b/backend/app/api/skills.py new file mode 100644 index 0000000..9b1a96d --- /dev/null +++ b/backend/app/api/skills.py @@ -0,0 +1,97 @@ +"""Agent skills API — list, view, edit, and preview skill matching.""" +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException + +from app.core.deps import require_admin, require_readonly +from app.database import get_db +from app.models.user import User +from app.schemas import SkillOut, SkillUpdate, SkillsListOut +from app.services.audit import record_audit +from app.services.skill_registry import ( + get_skill, + list_skills_summary, + match_skills, + read_skill_raw, + save_skill_raw, +) +from sqlalchemy.ext.asyncio import AsyncSession + +router = APIRouter(prefix="/api/skills", tags=["skills"]) + + +@router.get("", response_model=SkillsListOut) +async def list_skills(_: User = Depends(require_readonly)): + return SkillsListOut( + skills=list_skills_summary(), + path_hint="skills/", + ) + + +@router.post("/preview") +async def preview_skill_match( + body: dict, + _: User = Depends(require_readonly), +): + """Preview which skills match a task title + issue (and optional rule id).""" + title = str(body.get("title") or "") + issue = str(body.get("issue") or "") + matched_rule_id = body.get("matched_rule_id") + if matched_rule_id is not None: + matched_rule_id = str(matched_rule_id) + matched = match_skills(title, issue, matched_rule_id=matched_rule_id) + return {"matched": [s.as_dict() for s in matched]} + + +@router.get("/{skill_id}", response_model=SkillOut) +async def get_skill_detail(skill_id: str, _: User = Depends(require_readonly)): + skill = get_skill(skill_id) + if not skill: + raise HTTPException(status_code=404, detail="Skill not found") + return SkillOut( + id=skill["id"], + name=skill["name"], + description=skill["description"], + priority=skill.get("priority") or 0, + rule_ids=list(skill.get("rule_ids") or []), + match=skill.get("match") or {}, + markdown=read_skill_raw(skill_id), + path_hint=f"skills/{skill_id}/SKILL.md", + ) + + +@router.put("/{skill_id}", response_model=SkillOut) +async def update_skill( + skill_id: str, + body: SkillUpdate, + db: AsyncSession = Depends(get_db), + user: User = Depends(require_admin), +): + try: + save_skill_raw(skill_id, body.markdown) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=400, detail=f"Invalid skill: {exc}") from exc + + await record_audit( + db, + user, + "skills.update", + "skills", + 0, + {"skill_id": skill_id, "bytes": len(body.markdown)}, + ) + skill = get_skill(skill_id) + if not skill: + raise HTTPException(status_code=404, detail="Skill not found") + return SkillOut( + id=skill["id"], + name=skill["name"], + description=skill["description"], + priority=skill.get("priority") or 0, + rule_ids=list(skill.get("rule_ids") or []), + match=skill.get("match") or {}, + markdown=read_skill_raw(skill_id), + path_hint=f"skills/{skill_id}/SKILL.md", + ) diff --git a/backend/app/api/tasks.py b/backend/app/api/tasks.py index ae0f5c6..bedf92d 100644 --- a/backend/app/api/tasks.py +++ b/backend/app/api/tasks.py @@ -351,7 +351,29 @@ async def decide_approval( approval.decided_by_id = user.id approval.decided_at = datetime.now(timezone.utc) - if body.approve and approval.tool_name == "mcp_git_push" and approval.tool_args: + if body.approve and approval.tool_name == "mcp_apply_patch" and approval.tool_args: + from app.mcp_manager.commands import enqueue_mcp_command + + args = approval.tool_args + result = await enqueue_mcp_command( + "apply_patch", + server=args.get("server"), + timeout_seconds=180, + task_id=task_id, + tool_args=args, + ) + if not result.get("ok"): + raise HTTPException( + status_code=502, + detail=result.get("error", "MCP patch apply failed"), + ) + await publish_event( + task_id, + "mcp_dev", + result.get("summary") or f"MCP patch applied to {args.get('server')}", + {"kind": "mcp_apply_patch", **result}, + ) + elif body.approve and approval.tool_name == "mcp_git_push" and approval.tool_args: from app.services import mcp_workspace args = approval.tool_args diff --git a/backend/app/config.py b/backend/app/config.py index bcd541b..f8f0d21 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -75,11 +75,17 @@ class Settings(BaseSettings): # Troubleshooting decision rules (YAML, editable via Admin → Rules) troubleshooting_rules_path: str = "/app/rules/troubleshooting.yaml" + # Agent skills (procedural guides in skills/*/SKILL.md) + skills_path: str = "/app/skills" + skills_max_matched: int = 2 + skills_max_body_chars: int = 8000 + # Runtime paths (inside container) runtime_dir: str = "/runtime" # MCP self-development: agent patches MCP source on tool failure / missing capability mcp_development_enabled: bool = True + mcp_dev_require_approval: bool = True # when true, apply/reload waits for human approval mcp_dev_auto_push: bool = False # when false, patch applies + reload; git push needs approval mcp_dev_model_tier: str = "premium" diff --git a/backend/app/database.py b/backend/app/database.py index 3662635..5e434bf 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -1,13 +1,18 @@ -"""Async SQLAlchemy engine, session factory, and Base model.""" +"""Database schema migrations via Alembic.""" from __future__ import annotations +import asyncio +import logging from collections.abc import AsyncGenerator +from pathlib import Path from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase from app.config import settings +logger = logging.getLogger(__name__) + engine = create_async_engine(settings.database_url, echo=False, pool_pre_ping=True) SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) @@ -21,45 +26,19 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]: yield session -async def init_db() -> None: - """Create tables and apply lightweight dev migrations.""" - from app import models # noqa: F401 - from sqlalchemy import text +def _run_alembic_upgrade() -> None: + from alembic import command + from alembic.config import Config - async with engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - # Dev migration: locations → vessels (rename legacy table/columns) - await conn.execute( - text( - """ - DO $$ - BEGIN - IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'locations') - AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'vessels') - THEN - ALTER TABLE locations RENAME TO vessels; - END IF; - IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'devices' AND column_name = 'location_id') - THEN - ALTER TABLE devices RENAME COLUMN location_id TO vessel_id; - END IF; - IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'tasks' AND column_name = 'location_id') - THEN - ALTER TABLE tasks RENAME COLUMN location_id TO vessel_id; - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'devices' AND column_name = 'catalog_key') - THEN - ALTER TABLE devices ADD COLUMN catalog_key VARCHAR(64); - END IF; - END $$; - """ - ) - ) - # Indexes for hot query paths (idempotent; covers pre-existing DBs). - for stmt in ( - "CREATE INDEX IF NOT EXISTS ix_tasks_vessel_id ON tasks (vessel_id)", - "CREATE INDEX IF NOT EXISTS ix_task_events_task_id_id ON task_events (task_id, id)", - "CREATE INDEX IF NOT EXISTS ix_approval_requests_task_status " - "ON approval_requests (task_id, status)", - ): - await conn.execute(text(stmt)) + ini = Path(__file__).resolve().parent.parent / "alembic.ini" + cfg = Config(str(ini)) + command.upgrade(cfg, "head") + + +async def init_db() -> None: + """Apply Alembic migrations to head.""" + try: + await asyncio.to_thread(_run_alembic_upgrade) + except Exception as exc: # noqa: BLE001 + logger.exception("Database migration failed") + raise RuntimeError("Database migration failed") from exc diff --git a/backend/app/main.py b/backend/app/main.py index 2dd93ef..af3ea50 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,7 +9,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from sqlalchemy import select -from app.api import auth, balance, devices, llm, mcp, rules, tasks, users, vessels +from app.api import auth, balance, devices, llm, mcp, rules, skills, tasks, users, vessels from app.config import settings from app.core.security import hash_password from app.database import SessionLocal, init_db @@ -56,7 +56,7 @@ app.add_middleware( allow_headers=["*"], ) -for r in (auth.router, users.router, vessels.router, devices.router, tasks.router, mcp.router, balance.router, llm.router, rules.router): +for r in (auth.router, users.router, vessels.router, devices.router, tasks.router, mcp.router, balance.router, llm.router, rules.router, skills.router): app.include_router(r) diff --git a/backend/app/mcp_manager/commands.py b/backend/app/mcp_manager/commands.py index 3b8f5f9..9974eee 100644 --- a/backend/app/mcp_manager/commands.py +++ b/backend/app/mcp_manager/commands.py @@ -80,6 +80,10 @@ async def process_mcp_command(cmd: dict) -> dict: if not server: raise ValueError("server name required") return await _execute_develop_command(cmd, manager) + elif action == "apply_patch": + if not server: + raise ValueError("server name required") + return await _execute_apply_patch_command(cmd, manager) else: raise ValueError(f"Unknown action: {action}") return {"ok": True, "action": action, "server": server} @@ -139,6 +143,7 @@ async def _finalize_develop_task(task_id: int, result: dict, server: str) -> Non "tools": result.get("tools"), "push": result.get("push"), "resolved": not has_pending, + "pending_approval": result.get("pending_approval"), } task.error = None task.status = TaskStatus.waiting_approval if has_pending else TaskStatus.succeeded @@ -149,7 +154,9 @@ async def _finalize_develop_task(task_id: int, result: dict, server: str) -> Non if result.get("ok"): msg = result.get("summary") or f"MCP develop finished for {server}" - if has_pending: + if result.get("pending_approval"): + msg += " — approve MCP patch on this task." + elif has_pending: msg += " — approve git push on this task." await publish_event(task_id, "report", msg, {"kind": "mcp_develop", **result}, persist=True) else: @@ -180,7 +187,7 @@ async def _develop_on_worker(cmd: dict, manager) -> dict: task_id=task_id, server=server, tool=cmd.get("tool") or "unknown", - args={}, + args=cmd.get("args") or {}, issue=cmd.get("issue") or "", error=cmd.get("error"), output=cmd.get("output"), @@ -189,3 +196,22 @@ async def _develop_on_worker(cmd: dict, manager) -> dict: task_title=cmd.get("title") or f"MCP develop: {server}", push=cmd.get("push"), ) + + +async def _execute_apply_patch_command(cmd: dict, manager) -> dict: + from app.services.events import publish_event + from app.services.mcp_approvals import execute_approved_patch + + task_id = int(cmd.get("task_id") or 0) + tool_args = cmd.get("tool_args") or {} + + async def emit(tid: int, kind: str, message: str, payload: dict | None = None): + if tid: + await publish_event(tid, kind, message, payload) + + return await execute_approved_patch( + task_id=task_id, + tool_args=tool_args, + manager=manager, + emit=emit, + ) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index f071746..098fb7d 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -313,3 +313,34 @@ class TroubleshootingRulesOut(BaseModel): yaml: str parsed: dict[str, Any] path_hint: str + + +# ---- Agent skills ----------------------------------------------------- +class SkillSummaryOut(BaseModel): + id: str + name: str + description: str + priority: int + rule_ids: list[str] + match: dict[str, Any] + path_hint: str + + +class SkillsListOut(BaseModel): + skills: list[SkillSummaryOut] + path_hint: str + + +class SkillOut(BaseModel): + id: str + name: str + description: str + priority: int + rule_ids: list[str] + match: dict[str, Any] + markdown: str + path_hint: str + + +class SkillUpdate(BaseModel): + markdown: str diff --git a/backend/app/services/device_defaults.py b/backend/app/services/device_defaults.py index c59bb11..cbdf797 100644 --- a/backend/app/services/device_defaults.py +++ b/backend/app/services/device_defaults.py @@ -29,8 +29,10 @@ from __future__ import annotations import os from dataclasses import dataclass +from typing import TYPE_CHECKING -from app.inventory_catalog import CatalogEntry +if TYPE_CHECKING: + from app.inventory_catalog import CatalogEntry # catalog_key → how secrets are stored / resolved SLOT_AUTH: dict[str, dict] = { diff --git a/backend/app/services/mcp_approvals.py b/backend/app/services/mcp_approvals.py new file mode 100644 index 0000000..e74c23c --- /dev/null +++ b/backend/app/services/mcp_approvals.py @@ -0,0 +1,208 @@ +"""Apply or roll back MCP patches after human approval (worker-side).""" +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +from app.config import settings +from app.services import mcp_workspace + +logger = logging.getLogger(__name__) + +EmitFn = Callable[[int, str, str, dict | None], Awaitable[None]] + + +async def backup_proposal_files(server: str, edits: list[dict]) -> dict[str, str]: + """Snapshot current workspace content before a proposed patch is applied.""" + backups: dict[str, str] = {} + for edit in edits: + path = edit.get("path") or edit.get("file") + if not path: + continue + try: + backups[path] = await mcp_workspace.read_source_file(server, path) + except OSError: + backups[path] = "" + return backups + + +async def restore_file_backups(server: str, backups: dict[str, str]) -> list[str]: + restored: list[str] = [] + for path, content in backups.items(): + await mcp_workspace.write_source_file(server, path, content) + restored.append(path) + return restored + + +async def apply_and_reload_mcp(server: str, edits: list[dict], manager) -> dict[str, Any]: + written = await mcp_workspace.apply_file_edits(server, edits) + if not written: + return {"ok": False, "error": "No files written"} + await manager.reload_server(server) + return {"ok": True, "files": written, "tools": mcp_workspace.list_registered_tools(server)} + + +async def queue_git_push_approval( + *, + task_id: int, + server: str, + written: list[str], + commit_msg: str, + summary: str | None, + emit: EmitFn, +) -> dict[str, Any] | None: + if not settings.gitea_token or not written or task_id <= 0: + return None + + from app.database import SessionLocal + from app.models.enums import ApprovalStatus + from app.models.task import ApprovalRequest + + async with SessionLocal() as db: + db.add( + ApprovalRequest( + task_id=task_id, + tool_name="mcp_git_push", + tool_args={ + "server": server, + "files": written, + "commit_message": commit_msg, + }, + risk=summary, + status=ApprovalStatus.pending, + ) + ) + await db.commit() + + await emit( + task_id, + "approval", + f"MCP patch applied — approve to push {server} to Gitea", + {"server": server, "tool_name": "mcp_git_push", "files": written}, + ) + return {"queued": True, "tool_name": "mcp_git_push"} + + +async def maybe_retry_tool_after_patch( + *, + task_id: int, + retry_context: dict | None, + manager, + emit: EmitFn, +) -> dict[str, Any] | None: + if not retry_context: + return None + + device = retry_context.get("device") or "device" + server = retry_context.get("server") or "" + tool = retry_context.get("tool") or "" + args = retry_context.get("args") or {} + if not tool: + return None + + from app.agent.mcp_helpers import tool_result_ok + from app.agent.tool_events import tool_event_payload + from app.mcp_manager.recovery import call_tool_resilient + + await emit(task_id, "log", f"Retrying {tool} after approved MCP patch…", {"server": server, "tool": tool}) + res = await call_tool_resilient(manager, server, tool, args, task_id=task_id, emit=emit) + ok = tool_result_ok(res) + await emit( + task_id, + "tool_call", + f"{device} › {tool} (after approved MCP patch)", + tool_event_payload( + device=device, + tool=tool, + args=args, + status="done", + ok=ok, + output=res.get("text", ""), + ), + ) + return {"ok": ok, "result": res} + + +async def execute_approved_patch( + *, + task_id: int, + tool_args: dict, + manager, + emit: EmitFn | None = None, + push: bool | None = None, +) -> dict[str, Any]: + """Apply a previously approved MCP patch, reload the server, optionally push/retry.""" + from app.services.events import publish_event + + async def _emit(tid: int, kind: str, message: str, payload: dict | None = None) -> None: + if emit: + await emit(tid, kind, message, payload) + elif tid: + await publish_event(tid, kind, message, payload) + + server = tool_args.get("server") or "" + edits = tool_args.get("files") or [] + commit_msg = tool_args.get("commit_message") or "agentic-os: MCP patch" + summary = tool_args.get("summary") + retry_context = tool_args.get("retry_context") + + if not server or not edits: + return {"ok": False, "error": "Invalid MCP patch approval payload"} + + await _emit( + task_id, + "mcp_dev", + f"Applying approved MCP patch to {server}…", + {"server": server, "phase": "apply", "files": [e.get("path") for e in edits]}, + ) + + reload_result = await apply_and_reload_mcp(server, edits, manager) + if not reload_result.get("ok"): + await _emit(task_id, "error", reload_result.get("error", "MCP reload failed"), reload_result) + return reload_result + + await _emit( + task_id, + "mcp_dev", + f"MCP '{server}' reloaded — {len(reload_result.get('tools') or [])} tools registered", + {"server": server, "phase": "reload_done", "tools": reload_result.get("tools")}, + ) + + written = reload_result.get("files") or [] + should_push = push if push is not None else settings.mcp_dev_auto_push + push_result: dict[str, Any] | None = None + + if should_push and settings.gitea_token: + push_result = await mcp_workspace.commit_and_push(server, commit_msg, written) + await _emit( + task_id, + "mcp_dev", + f"Git push {'ok' if push_result.get('ok') else 'failed'}: {push_result.get('commit') or push_result.get('error')}", + {"server": server, "phase": "push", **push_result}, + ) + elif settings.gitea_token and written and task_id > 0: + push_result = await queue_git_push_approval( + task_id=task_id, + server=server, + written=written, + commit_msg=commit_msg, + summary=summary, + emit=_emit, + ) + + retry_result = await maybe_retry_tool_after_patch( + task_id=task_id, + retry_context=retry_context, + manager=manager, + emit=_emit, + ) + + return { + "ok": True, + "summary": summary, + "reload": reload_result, + "push": push_result, + "retry": retry_result, + "tools": reload_result.get("tools"), + } diff --git a/backend/app/services/skill_registry.py b/backend/app/services/skill_registry.py new file mode 100644 index 0000000..92f621d --- /dev/null +++ b/backend/app/services/skill_registry.py @@ -0,0 +1,264 @@ +"""Agent skills — procedural guides loaded from skills/*/SKILL.md on disk. + +Skills complement troubleshooting rules: rules route devices and supply short hints; +skills inject deeper procedural knowledge into triage and reasoning prompts. +""" +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass + +import yaml + +from app.config import settings +from app.services.troubleshooting_rules import _text_has_term + +logger = logging.getLogger(__name__) + +_skill_cache: dict[str, dict] = {} +_skill_dir_mtime: float = -1.0 + + +@dataclass +class MatchedSkill: + id: str + name: str + description: str + body: str + priority: int + score: int + + def as_dict(self) -> dict: + return { + "id": self.id, + "name": self.name, + "description": self.description, + "priority": self.priority, + "score": self.score, + "body_chars": len(self.body), + } + + +def skills_dir() -> str: + return os.environ.get("SKILLS_PATH", settings.skills_path) + + +def invalidate_skills_cache() -> None: + global _skill_cache, _skill_dir_mtime + _skill_cache = {} + _skill_dir_mtime = -1.0 + + +def _parse_skill_file(path: str) -> dict | None: + try: + with open(path, encoding="utf-8") as fh: + raw = fh.read() + except OSError: + return None + + meta: dict = {} + body = raw.strip() + if raw.startswith("---"): + parts = raw.split("---", 2) + if len(parts) >= 3: + try: + parsed = yaml.safe_load(parts[1]) + meta = parsed if isinstance(parsed, dict) else {} + except yaml.YAMLError as exc: + logger.warning("invalid skill frontmatter %s: %s", path, exc) + meta = {} + body = parts[2].strip() + + folder = os.path.basename(os.path.dirname(path)) + skill_id = str(meta.get("id") or folder) + return { + "id": skill_id, + "name": str(meta.get("name") or skill_id), + "description": str(meta.get("description") or ""), + "priority": int(meta.get("priority") or 0), + "match": meta.get("match") or {}, + "rule_ids": list(meta.get("rule_ids") or []), + "body": body, + "path": path, + } + + +def _dir_latest_mtime(root: str) -> float: + latest = -1.0 + if not os.path.isdir(root): + return latest + for dirpath, _dirnames, filenames in os.walk(root): + for name in filenames: + if name != "SKILL.md": + continue + try: + latest = max(latest, os.path.getmtime(os.path.join(dirpath, name))) + except OSError: + continue + return latest + + +def load_skills(*, force: bool = False) -> dict[str, dict]: + """Return skill_id → skill record (metadata + body).""" + global _skill_cache, _skill_dir_mtime + root = skills_dir() + mtime = _dir_latest_mtime(root) + + if not force and _skill_cache and mtime == _skill_dir_mtime: + return _skill_cache + + skills: dict[str, dict] = {} + if os.path.isdir(root): + for entry in sorted(os.listdir(root)): + skill_path = os.path.join(root, entry, "SKILL.md") + if not os.path.isfile(skill_path): + continue + record = _parse_skill_file(skill_path) + if record: + skills[record["id"]] = record + + _skill_cache = skills + _skill_dir_mtime = mtime + return skills + + +def list_skills_summary() -> list[dict]: + return [ + { + "id": s["id"], + "name": s["name"], + "description": s["description"], + "priority": s["priority"], + "rule_ids": s.get("rule_ids") or [], + "match": s.get("match") or {}, + "path_hint": f"skills/{s['id']}/SKILL.md", + } + for s in sorted(load_skills().values(), key=lambda x: (-x["priority"], x["id"])) + ] + + +def get_skill(skill_id: str) -> dict | None: + return load_skills().get(skill_id) + + +def read_skill_raw(skill_id: str) -> str: + skill = get_skill(skill_id) + if not skill: + return "" + try: + with open(skill["path"], encoding="utf-8") as fh: + return fh.read() + except OSError: + return "" + + +def save_skill_raw(skill_id: str, content: str) -> dict: + skill = get_skill(skill_id) + if not skill: + raise ValueError(f"Unknown skill: {skill_id}") + # Validate frontmatter parses before writing. + if content.startswith("---"): + parts = content.split("---", 2) + if len(parts) >= 3: + parsed = yaml.safe_load(parts[1]) + if not isinstance(parsed, dict): + raise ValueError("Skill frontmatter must be a YAML mapping") + path = skill["path"] + with open(path, "w", encoding="utf-8") as fh: + fh.write(content if content.endswith("\n") else content + "\n") + invalidate_skills_cache() + updated = get_skill(skill_id) + if not updated: + raise ValueError(f"Skill {skill_id} missing after save") + return updated + + +def _skill_match_score(text: str, skill: dict, *, matched_rule_id: str | None) -> int: + match = skill.get("match") or {} + keywords = match.get("keywords") or [] + all_of = match.get("all_of") or [] + + if all_of and not all(_text_has_term(text, t) for t in all_of): + keyword_score = 0 + else: + keyword_score = 0 + for kw in keywords: + if _text_has_term(text, kw): + keyword_score += 2 if " " in str(kw).strip() else 1 + if all_of: + keyword_score += len(all_of) * 2 + + rule_boost = 0 + rule_ids = skill.get("rule_ids") or [] + if matched_rule_id and matched_rule_id in rule_ids: + rule_boost = 10 + + return keyword_score + rule_boost + + +def _truncate_body(body: str, max_chars: int) -> str: + if max_chars <= 0 or len(body) <= max_chars: + return body + return body[: max_chars - 20].rstrip() + "\n\n… (truncated)" + + +def match_skills( + title: str, + issue: str, + *, + matched_rule_id: str | None = None, + max_skills: int | None = None, +) -> list[MatchedSkill]: + """Return up to *max_skills* best-matching procedural guides.""" + limit = max_skills if max_skills is not None else settings.skills_max_matched + max_body = settings.skills_max_body_chars + text = f"{title} {issue}".lower() + skills = load_skills() + + scored: list[tuple[int, int, dict]] = [] + for skill in skills.values(): + score = _skill_match_score(text, skill, matched_rule_id=matched_rule_id) + if score <= 0: + continue + scored.append((score, skill.get("priority") or 0, skill)) + + scored.sort(key=lambda t: (-t[0], -t[1], t[2]["id"])) + out: list[MatchedSkill] = [] + for score, _prio, skill in scored[:limit]: + out.append( + MatchedSkill( + id=skill["id"], + name=skill["name"], + description=skill["description"], + body=_truncate_body(skill["body"], max_body), + priority=skill.get("priority") or 0, + score=score, + ) + ) + return out + + +def skill_guidance_block(matched: list[MatchedSkill | dict] | None) -> str: + if not matched: + return "" + lines = ["\n\n--- Agent skills (procedural guides) ---"] + for item in matched: + if isinstance(item, dict): + name = item.get("name", "") + sid = item.get("id", "") + description = item.get("description", "") + body = item.get("body", "") + else: + name, sid, description, body = item.name, item.id, item.description, item.body + lines.append(f"\n### Skill: {name} ({sid})") + if description: + lines.append(description) + if body: + lines.append(body) + lines.append( + "\nFollow these guides when investigating. Rules define device scope; " + "skills provide domain-specific procedure and proven root causes." + ) + return "\n".join(lines) diff --git a/backend/app/worker.py b/backend/app/worker.py index c680fc5..b517897 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -10,7 +10,7 @@ import logging from sqlalchemy import select -from app.agent import run_task_agent +from app.agent.graph import run_task_agent from app.config import settings from app.database import SessionLocal, init_db from app.mcp_manager import get_manager diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..2a21ee9 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,13 @@ +"""Shared pytest fixtures.""" +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _clear_settings_cache(): + from app.config import get_settings + + get_settings.cache_clear() + yield + get_settings.cache_clear() diff --git a/backend/tests/test_device_defaults.py b/backend/tests/test_device_defaults.py index c019a0d..e7141a8 100644 --- a/backend/tests/test_device_defaults.py +++ b/backend/tests/test_device_defaults.py @@ -3,25 +3,30 @@ from __future__ import annotations import os -os.environ.setdefault("SECRET_KEY", "test-secret-key-for-unit-tests-only") -os.environ["DEVICE_SSH_USERNAME"] = "shipadmin" -os.environ["DEVICE_SSH_PASSWORD"] = "ssh-secret" -os.environ["DEVICE_PFSENSE_PORT"] = "40443" -os.environ["DEVICE_PFSENSE_API_KEY"] = "pf-key-123" -os.environ["DEVICE_PROXMOX_PASSWORD"] = "pve-pass" -os.environ["DEVICE_PROXMOX_TOKEN_ID"] = "root@pam!agentic" -os.environ["DEVICE_PROXMOX_TOKEN_SECRET"] = "token-uuid" +import pytest -from app.inventory_catalog import catalog_by_key # noqa: E402 -from app.services.device_defaults import ( # noqa: E402 - merge_selection_with_defaults, - proxmox_api_configured, - resolve_proxmox_api_defaults, - resolve_slot_defaults, -) +os.environ.setdefault("SECRET_KEY", "test-secret-key-for-unit-tests-only") + + +@pytest.fixture(autouse=True) +def device_env(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("DEVICE_SSH_USERNAME", "shipadmin") + monkeypatch.setenv("DEVICE_SSH_PASSWORD", "ssh-secret") + monkeypatch.delenv("DEVICE_PROXMOX_USERNAME", raising=False) + monkeypatch.setenv("DEVICE_PFSENSE_PORT", "40443") + monkeypatch.setenv("DEVICE_PFSENSE_API_KEY", "pf-key-123") + monkeypatch.setenv("DEVICE_PROXMOX_PASSWORD", "pve-pass") + monkeypatch.setenv("DEVICE_PROXMOX_TOKEN_ID", "root@pam!agentic") + monkeypatch.setenv("DEVICE_PROXMOX_TOKEN_SECRET", "token-uuid") + from app.config import get_settings + + get_settings.cache_clear() def test_ssh_slot_uses_global_fallback(): + from app.inventory_catalog import catalog_by_key + from app.services.device_defaults import resolve_slot_defaults + entry = catalog_by_key()["docker_vm"] d = resolve_slot_defaults("docker_vm", entry) assert d.username == "shipadmin" @@ -30,6 +35,9 @@ def test_ssh_slot_uses_global_fallback(): def test_per_slot_overrides_global(): + from app.inventory_catalog import catalog_by_key + from app.services.device_defaults import resolve_slot_defaults + entry = catalog_by_key()["proxmox"] d = resolve_slot_defaults("proxmox", entry) assert d.secret == "pve-pass" @@ -37,32 +45,32 @@ def test_per_slot_overrides_global(): def test_api_key_slot(): + from app.inventory_catalog import catalog_by_key + from app.services.device_defaults import resolve_slot_defaults + entry = catalog_by_key()["pfsense"] d = resolve_slot_defaults("pfsense", entry) - assert d.port == 40443 assert d.secret == "pf-key-123" assert d.secret_kind == "api_key" + assert d.port == 40443 -def test_merge_applies_env_when_blank(): - entry = catalog_by_key()["pfsense"] +def test_merge_keeps_existing_secret(): + from app.inventory_catalog import catalog_by_key + from app.services.device_defaults import merge_selection_with_defaults + + entry = catalog_by_key()["docker_vm"] port, user, secret = merge_selection_with_defaults( - "pfsense", entry, port=None, username=None, secret=None + "docker_vm", entry, port=None, username=None, secret=None, existing_secret="stored" ) - assert port == 40443 - assert secret == "pf-key-123" + assert secret == "stored" + assert user == "shipadmin" -def test_proxmox_api_env(): - api = resolve_proxmox_api_defaults("10.1.2.3") - assert api["proxmox_token_id"] == "root@pam!agentic" - assert api["proxmox_token_secret"] == "token-uuid" - assert proxmox_api_configured(api) +def test_proxmox_api_defaults_from_env(): + from app.services.device_defaults import proxmox_api_configured, resolve_proxmox_api_defaults - -def test_merge_ui_override_wins(): - entry = catalog_by_key()["pfsense"] - _, _, secret = merge_selection_with_defaults( - "pfsense", entry, port=8443, username="admin", secret="custom-key" - ) - assert secret == "custom-key" + ctx = resolve_proxmox_api_defaults("10.20.30.254") + assert ctx["proxmox_api_url"] == "https://10.20.30.254:8006" + assert ctx["proxmox_token_id"] == "root@pam!agentic" + assert proxmox_api_configured(ctx) diff --git a/backend/tests/test_mcp_approvals.py b/backend/tests/test_mcp_approvals.py new file mode 100644 index 0000000..4ac26bb --- /dev/null +++ b/backend/tests/test_mcp_approvals.py @@ -0,0 +1,106 @@ +"""Tests for MCP patch approval gating.""" +from __future__ import annotations + +import asyncio +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("SECRET_KEY", "test-secret-key-for-unit-tests-only") + + +def test_repair_queues_apply_patch_when_approval_required(monkeypatch): + from app.config import get_settings, settings + + get_settings.cache_clear() + + proposal = { + "ok": True, + "summary": "Add missing tool", + "commit_message": "fix: add tool", + "files": [{"path": "server.py", "content": "# patched"}], + } + + mock_db = MagicMock() + mock_db.commit = AsyncMock() + mock_session = MagicMock() + mock_session.__aenter__ = AsyncMock(return_value=mock_db) + mock_session.__aexit__ = AsyncMock(return_value=None) + + async def _run(): + with ( + patch.object(settings, "mcp_dev_require_approval", True), + patch("app.agent.mcp_development.propose_mcp_patch", new=AsyncMock(return_value=proposal)), + patch( + "app.agent.mcp_development.backup_proposal_files", + new=AsyncMock(return_value={"server.py": "# old"}), + ), + patch("app.agent.mcp_development.execute_approved_patch", new=AsyncMock()) as mock_apply, + patch("app.database.SessionLocal", return_value=mock_session), + ): + from app.agent.mcp_development import attempt_mcp_repair + + emit = AsyncMock() + result = await attempt_mcp_repair( + task_id=42, + server="pfsense-mcp", + tool="pfsense_list_rules", + args={"interface": "wan"}, + issue="tool missing", + error="unknown tool", + output=None, + manager=AsyncMock(), + emit=emit, + device="pfsense", + ) + return result, mock_apply, emit, mock_db + + result, mock_apply, emit, mock_db = asyncio.run(_run()) + + assert result.get("pending_approval") is True + assert result.get("approval_type") == "mcp_apply_patch" + mock_apply.assert_not_called() + mock_db.add.assert_called_once() + mock_db.commit.assert_awaited_once() + assert any(c.args[1] == "approval" for c in emit.call_args_list) + + +def test_repair_applies_immediately_when_approval_not_required(monkeypatch): + from app.config import get_settings, settings + + get_settings.cache_clear() + + proposal = { + "ok": True, + "summary": "Fix crash", + "files": [{"path": "server.py", "content": "# patched"}], + } + + async def _run(): + with ( + patch.object(settings, "mcp_dev_require_approval", False), + patch("app.agent.mcp_development.propose_mcp_patch", new=AsyncMock(return_value=proposal)), + patch( + "app.agent.mcp_development.execute_approved_patch", + new=AsyncMock(return_value={"ok": True, "tools": ["a"]}), + ) as mock_apply, + ): + from app.agent.mcp_development import attempt_mcp_repair + + return await attempt_mcp_repair( + task_id=1, + server="pfsense-mcp", + tool="t", + args={}, + issue="x", + error="err", + output=None, + manager=AsyncMock(), + emit=AsyncMock(), + ), mock_apply + + result, mock_apply = asyncio.run(_run()) + + assert result.get("ok") is True + mock_apply.assert_awaited_once() diff --git a/backend/tests/test_skill_registry.py b/backend/tests/test_skill_registry.py new file mode 100644 index 0000000..a65abfa --- /dev/null +++ b/backend/tests/test_skill_registry.py @@ -0,0 +1,60 @@ +"""Tests for agent skill registry.""" +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("SECRET_KEY", "test-secret-key-for-unit-tests-only") +_test_dir = Path(__file__).resolve().parent +_backend_root = _test_dir.parent +_skills_dir = _backend_root.parent / "skills" +if not _skills_dir.is_dir(): + _skills_dir = _backend_root / "skills" +os.environ["SKILLS_PATH"] = str(_skills_dir) + +from app.services.skill_registry import ( # noqa: E402 + invalidate_skills_cache, + load_skills, + match_skills, + skill_guidance_block, +) + + +def test_load_skills_finds_seeded_skills(): + invalidate_skills_cache() + skills = load_skills(force=True) + assert "vessel-voip" in skills + assert "vessel-dns-portal" in skills + assert skills["vessel-voip"]["body"] + + +def test_match_voip_skill_by_keywords(): + invalidate_skills_cache() + matched = match_skills("VoIP issue", "one-way audio on SIP calls") + ids = [s.id for s in matched] + assert "vessel-voip" in ids + + +def test_match_skill_by_rule_id_boost(): + invalidate_skills_cache() + matched = match_skills("Mystery", "something vague", matched_rule_id="crew-dns-slow") + ids = [s.id for s in matched] + assert "vessel-dns-portal" in ids + + +def test_match_respects_max_skills(): + invalidate_skills_cache() + matched = match_skills( + "VoIP DNS portal", + "slow dns captive portal sip registration one-way audio", + max_skills=1, + ) + assert len(matched) == 1 + + +def test_skill_guidance_block_includes_body(): + invalidate_skills_cache() + matched = match_skills("VoIP", "one-way audio rtp") + block = skill_guidance_block(matched) + assert "Agent skills" in block + assert "One-way audio playbook" in block diff --git a/docker-compose.yml b/docker-compose.yml index f403bf4..4b8b59c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -71,6 +71,7 @@ services: - mcp_runtime:/runtime - ./mcp-servers:/app/mcp-servers:ro - ./rules:/app/rules + - ./skills:/app/skills ports: - "8000:8000" command: ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] @@ -92,6 +93,7 @@ services: - mcp_runtime:/runtime - ./mcp-servers:/app/mcp-servers:ro - ./rules:/app/rules + - ./skills:/app/skills command: ["python", "-m", "app.worker"] frontend: diff --git a/frontend/app/skills/page.tsx b/frontend/app/skills/page.tsx new file mode 100644 index 0000000..b4ec30e --- /dev/null +++ b/frontend/app/skills/page.tsx @@ -0,0 +1,199 @@ +"use client"; + +import { useState } from "react"; +import useSWR from "swr"; +import AppShell from "@/components/AppShell"; +import { PageHeader } from "@/components/ui"; +import { api, fetcher } from "@/lib/api"; +import { useAuth } from "@/lib/auth"; + +type SkillSummary = { + id: string; + name: string; + description: string; + priority: number; + rule_ids: string[]; + path_hint: string; +}; + +type SkillsListResponse = { + skills: SkillSummary[]; + path_hint: string; +}; + +type SkillDetail = SkillSummary & { + markdown: string; + match: Record; +}; + +export default function SkillsPage() { + const { user } = useAuth(); + const isAdmin = user?.role === "admin"; + const { data: listData } = useSWR("/api/skills", fetcher); + + const [selectedId, setSelectedId] = useState(null); + + const [markdown, setMarkdown] = useState(null); + const [previewTitle, setPreviewTitle] = useState("VoIP issue"); + const [previewIssue, setPreviewIssue] = useState("one-way audio on SIP calls from crew phones"); + const [previewRuleId, setPreviewRuleId] = useState(""); + const [preview, setPreview] = useState<{ matched: Record[] } | null>(null); + const [saving, setSaving] = useState(false); + const [msg, setMsg] = useState(null); + const [msgKind, setMsgKind] = useState<"ok" | "err">("ok"); + + const skills = listData?.skills ?? []; + const activeId = selectedId ?? skills[0]?.id ?? null; + + const { data: skillData, mutate } = useSWR( + activeId ? `/api/skills/${activeId}` : null, + fetcher, + ); + + const content = markdown ?? skillData?.markdown ?? ""; + + const selectSkill = (id: string) => { + setSelectedId(id); + setMarkdown(null); + setMsg(null); + }; + + const save = async () => { + if (!activeId) return; + setSaving(true); + setMsg(null); + try { + await api(`/api/skills/${activeId}`, { + method: "PUT", + body: JSON.stringify({ markdown: content }), + }); + setMsgKind("ok"); + setMsg("Skill saved — next task run will use it."); + setMarkdown(null); + mutate(); + } catch (err: unknown) { + setMsgKind("err"); + setMsg(err instanceof Error ? err.message : "Save failed"); + } finally { + setSaving(false); + } + }; + + const runPreview = async () => { + setPreview(null); + try { + const body: Record = { + title: previewTitle, + issue: previewIssue, + }; + if (previewRuleId.trim()) body.matched_rule_id = previewRuleId.trim(); + const res = await api<{ matched: Record[] }>("/api/skills/preview", { + method: "POST", + body: JSON.stringify(body), + }); + setPreview(res); + } catch (err: unknown) { + setMsgKind("err"); + setMsg(err instanceof Error ? err.message : "Preview failed"); + } + }; + + return ( + + + +
+
+

Skill preview

+

Test which skills match a task title + issue.

+ setPreviewTitle(e.target.value)} + placeholder="Task title" + /> +