Complete uni-wizard implementation with unified tool registry: **Core Architecture:** - harness.py - Single entry point for all capabilities - tools/registry.py - Central tool registry with schema generation - Elegant routing: One harness, infinite capabilities **Tool Categories (13 tools total):** - System: system_info, process_list, service_status, service_control, health_check, disk_usage - Git: git_status, git_log, git_pull, git_commit, git_push, git_checkout, git_branch_list - Network: http_get, http_post, gitea_create_issue, gitea_comment, gitea_list_issues, gitea_get_issue **Daemons:** - health_daemon.py - HTTP endpoint on :8082, writes to ~/timmy/logs/health.json - task_router.py - Polls Gitea for assigned issues, routes to tools, posts results **Systemd Services:** - timmy-health.service - Health monitoring daemon - timmy-task-router.service - Gitea task router daemon **Testing:** - test_harness.py - Exercises all tool categories **Design Principles:** - Local-first: No cloud dependencies - Self-healing: Tools can restart, reconnect, recover - Unified: One consciousness, all capabilities Closes #76, #77, #78
175 lines
4.9 KiB
Python
175 lines
4.9 KiB
Python
"""
|
|
Uni-Wizard Harness
|
|
Single entry point for all capabilities
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from typing import Dict, Any, Optional
|
|
from pathlib import Path
|
|
|
|
# Add tools to path
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from tools import registry, call_tool
|
|
|
|
|
|
class UniWizardHarness:
|
|
"""
|
|
The Uni-Wizard Harness - one consciousness, infinite capabilities.
|
|
|
|
All API flows route through this single harness:
|
|
- System monitoring and control
|
|
- Git operations
|
|
- Network requests
|
|
- Gitea API
|
|
- Local inference
|
|
|
|
Usage:
|
|
harness = UniWizardHarness()
|
|
result = harness.execute("system_info")
|
|
result = harness.execute("git_status", repo_path="/path/to/repo")
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.registry = registry
|
|
self.history = []
|
|
|
|
def list_capabilities(self) -> str:
|
|
"""List all available tools/capabilities"""
|
|
tools = []
|
|
for category in self.registry.get_categories():
|
|
cat_tools = self.registry.get_tools_by_category(category)
|
|
tools.append(f"\n{category.upper()}:")
|
|
for tool in cat_tools:
|
|
tools.append(f" - {tool['name']}: {tool['description']}")
|
|
|
|
return "\n".join(tools)
|
|
|
|
def execute(self, tool_name: str, **params) -> str:
|
|
"""
|
|
Execute a tool by name.
|
|
|
|
Args:
|
|
tool_name: Name of the tool to execute
|
|
**params: Parameters for the tool
|
|
|
|
Returns:
|
|
String result from the tool
|
|
"""
|
|
# Log execution
|
|
self.history.append({
|
|
"tool": tool_name,
|
|
"params": params
|
|
})
|
|
|
|
# Execute via registry
|
|
result = call_tool(tool_name, **params)
|
|
return result
|
|
|
|
def execute_plan(self, plan: list) -> Dict[str, str]:
|
|
"""
|
|
Execute a sequence of tool calls.
|
|
|
|
Args:
|
|
plan: List of dicts with 'tool' and 'params'
|
|
e.g., [{"tool": "system_info", "params": {}}]
|
|
|
|
Returns:
|
|
Dict mapping tool names to results
|
|
"""
|
|
results = {}
|
|
for step in plan:
|
|
tool_name = step.get("tool")
|
|
params = step.get("params", {})
|
|
|
|
result = self.execute(tool_name, **params)
|
|
results[tool_name] = result
|
|
|
|
return results
|
|
|
|
def get_tool_definitions(self) -> str:
|
|
"""Get tool definitions formatted for LLM system prompt"""
|
|
return self.registry.get_tool_definitions()
|
|
|
|
def get_status(self) -> str:
|
|
"""Get harness status"""
|
|
return json.dumps({
|
|
"total_tools": len(self.registry.list_tools()),
|
|
"categories": self.registry.get_categories(),
|
|
"tools_by_category": {
|
|
cat: self.registry.list_tools(cat)
|
|
for cat in self.registry.get_categories()
|
|
},
|
|
"execution_history_count": len(self.history)
|
|
}, indent=2)
|
|
|
|
|
|
# Singleton instance
|
|
_harness = None
|
|
|
|
def get_harness() -> UniWizardHarness:
|
|
"""Get the singleton harness instance"""
|
|
global _harness
|
|
if _harness is None:
|
|
_harness = UniWizardHarness()
|
|
return _harness
|
|
|
|
|
|
def main():
|
|
"""CLI interface for the harness"""
|
|
harness = get_harness()
|
|
|
|
if len(sys.argv) < 2:
|
|
print("Uni-Wizard Harness")
|
|
print("==================")
|
|
print("\nUsage: python harness.py <command> [args]")
|
|
print("\nCommands:")
|
|
print(" list - List all capabilities")
|
|
print(" status - Show harness status")
|
|
print(" tools - Show tool definitions (for LLM)")
|
|
print(" exec <tool> - Execute a tool")
|
|
print("\nExamples:")
|
|
print(' python harness.py exec system_info')
|
|
print(' python harness.py exec git_status repo_path=/tmp/timmy-home')
|
|
return
|
|
|
|
command = sys.argv[1]
|
|
|
|
if command == "list":
|
|
print(harness.list_capabilities())
|
|
|
|
elif command == "status":
|
|
print(harness.get_status())
|
|
|
|
elif command == "tools":
|
|
print(harness.get_tool_definitions())
|
|
|
|
elif command == "exec" and len(sys.argv) >= 3:
|
|
tool_name = sys.argv[2]
|
|
|
|
# Parse params from args (key=value format)
|
|
params = {}
|
|
for arg in sys.argv[3:]:
|
|
if '=' in arg:
|
|
key, value = arg.split('=', 1)
|
|
# Try to parse as int/bool
|
|
if value.isdigit():
|
|
value = int(value)
|
|
elif value.lower() == 'true':
|
|
value = True
|
|
elif value.lower() == 'false':
|
|
value = False
|
|
params[key] = value
|
|
|
|
result = harness.execute(tool_name, **params)
|
|
print(result)
|
|
|
|
else:
|
|
print(f"Unknown command: {command}")
|
|
print("Run without arguments for help")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|