Compare commits
3 Commits
archon-see
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 7711b7e279 | |||
|
|
19fae5a6e5 | ||
|
|
1b3bca9902 |
16
archon-poc/.gitignore
vendored
Normal file
16
archon-poc/.gitignore
vendored
Normal file
@@ -0,0 +1,16 @@
|
|||||||
|
__pycache__/
|
||||||
|
*.pyc
|
||||||
|
*.pyo
|
||||||
|
*.pyd
|
||||||
|
.Python
|
||||||
|
*.so
|
||||||
|
*.egg
|
||||||
|
*.egg-info/
|
||||||
|
dist/
|
||||||
|
build/
|
||||||
|
.pytest_cache/
|
||||||
|
.coverage
|
||||||
|
htmlcov/
|
||||||
|
.env
|
||||||
|
.venv
|
||||||
|
venv/
|
||||||
62
archon-poc/README.md
Normal file
62
archon-poc/README.md
Normal file
@@ -0,0 +1,62 @@
|
|||||||
|
# Archon Architecture Proof-of-Concept
|
||||||
|
|
||||||
|
A three-layer architecture separating identity, runtime logic, and intelligence.
|
||||||
|
|
||||||
|
## Architecture Overview
|
||||||
|
|
||||||
|
```
|
||||||
|
┌─────────────────────────────────────────────────────────────┐
|
||||||
|
│ Layer 1: Hermes Profile (Thin) │
|
||||||
|
│ - Identity & routing only │
|
||||||
|
│ - No intelligence or reasoning logic │
|
||||||
|
│ - Pure configuration (< 50 lines) │
|
||||||
|
└─────────────────────────────────────────────────────────────┘
|
||||||
|
│
|
||||||
|
▼
|
||||||
|
┌─────────────────────────────────────────────────────────────┐
|
||||||
|
│ Layer 2: Claw Code Runtime │
|
||||||
|
│ - All business logic lives here │
|
||||||
|
│ - Tool registry and execution │
|
||||||
|
│ - Message routing and orchestration │
|
||||||
|
└─────────────────────────────────────────────────────────────┘
|
||||||
|
│
|
||||||
|
▼
|
||||||
|
┌─────────────────────────────────────────────────────────────┐
|
||||||
|
│ Layer 3: Gemma 4 via Ollama │
|
||||||
|
│ - The actual intelligence │
|
||||||
|
│ - Local inference at localhost:11434 │
|
||||||
|
│ - Model: gemma3:4b (available) / gemma4:4b (target) │
|
||||||
|
└─────────────────────────────────────────────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
## Files
|
||||||
|
|
||||||
|
| File | Purpose |
|
||||||
|
|------|---------|
|
||||||
|
| `profile.yaml` | Thin Hermes layer - identity only |
|
||||||
|
| `runtime/harness.py` | Claw Code runtime stub |
|
||||||
|
| `runtime/tool_registry.py` | Tool definitions and registry |
|
||||||
|
| `ollama_client.py` | Gemma 4 interface layer |
|
||||||
|
| `test_integration.py` | End-to-end integration test |
|
||||||
|
|
||||||
|
## Quick Start
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Run integration test
|
||||||
|
python3 test_integration.py
|
||||||
|
|
||||||
|
# Test Ollama connection
|
||||||
|
python3 ollama_client.py "Hello, Gemma!"
|
||||||
|
```
|
||||||
|
|
||||||
|
## Design Principles
|
||||||
|
|
||||||
|
1. **Profile is THIN** - Only identity, no logic
|
||||||
|
2. **Runtime owns everything** - All intelligence orchestration
|
||||||
|
3. **Local inference** - No external API dependencies
|
||||||
|
4. **Testable** - Full integration test coverage
|
||||||
|
|
||||||
|
## Epic Reference
|
||||||
|
|
||||||
|
Implements Epic #370: Archon Architecture
|
||||||
|
Dispatch #371: Proof-of-Concept Implementation
|
||||||
BIN
archon-poc/__pycache__/ollama_client.cpython-312.pyc
Normal file
BIN
archon-poc/__pycache__/ollama_client.cpython-312.pyc
Normal file
Binary file not shown.
226
archon-poc/ollama_client.py
Normal file
226
archon-poc/ollama_client.py
Normal file
@@ -0,0 +1,226 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Ollama Client - Layer 3 Interface
|
||||||
|
|
||||||
|
Connects to local Ollama instance for Gemma inference.
|
||||||
|
This is the intelligence layer of the Archon architecture.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import json
|
||||||
|
import requests
|
||||||
|
from typing import Dict, Any, List, Optional, Generator
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class OllamaConfig:
|
||||||
|
base_url: str = "http://localhost:11434"
|
||||||
|
default_model: str = "gemma3:4b"
|
||||||
|
timeout: int = 120
|
||||||
|
|
||||||
|
|
||||||
|
class OllamaClient:
|
||||||
|
"""
|
||||||
|
Client for Ollama API - the intelligence layer.
|
||||||
|
|
||||||
|
Communicates with local Ollama instance to run
|
||||||
|
Gemma models for inference.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, base_url: str = "http://localhost:11434"):
|
||||||
|
self.config = OllamaConfig(base_url=base_url)
|
||||||
|
self.session = requests.Session()
|
||||||
|
|
||||||
|
def health_check(self) -> bool:
|
||||||
|
"""Check if Ollama is reachable."""
|
||||||
|
try:
|
||||||
|
response = self.session.get(
|
||||||
|
f"{self.config.base_url}/api/tags",
|
||||||
|
timeout=5
|
||||||
|
)
|
||||||
|
return response.status_code == 200
|
||||||
|
except requests.RequestException:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def list_models(self) -> List[str]:
|
||||||
|
"""List available models."""
|
||||||
|
try:
|
||||||
|
response = self.session.get(
|
||||||
|
f"{self.config.base_url}/api/tags",
|
||||||
|
timeout=self.config.timeout
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
data = response.json()
|
||||||
|
return [m["name"] for m in data.get("models", [])]
|
||||||
|
except requests.RequestException as e:
|
||||||
|
raise OllamaError(f"Failed to list models: {e}")
|
||||||
|
|
||||||
|
def generate(
|
||||||
|
self,
|
||||||
|
prompt: str,
|
||||||
|
model: Optional[str] = None,
|
||||||
|
system: Optional[str] = None,
|
||||||
|
context: Optional[list] = None,
|
||||||
|
options: Optional[Dict[str, Any]] = None
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Generate a response from the model.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
prompt: The user prompt
|
||||||
|
model: Model name (default: gemma3:4b)
|
||||||
|
system: System prompt
|
||||||
|
context: Previous context for conversation
|
||||||
|
options: Additional generation options
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Response dict from Ollama
|
||||||
|
"""
|
||||||
|
model = model or self.config.default_model
|
||||||
|
|
||||||
|
payload = {
|
||||||
|
"model": model,
|
||||||
|
"prompt": prompt,
|
||||||
|
"stream": False
|
||||||
|
}
|
||||||
|
|
||||||
|
if system:
|
||||||
|
payload["system"] = system
|
||||||
|
if context:
|
||||||
|
payload["context"] = context
|
||||||
|
if options:
|
||||||
|
payload["options"] = options
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = self.session.post(
|
||||||
|
f"{self.config.base_url}/api/generate",
|
||||||
|
json=payload,
|
||||||
|
timeout=self.config.timeout
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
return response.json()
|
||||||
|
except requests.RequestException as e:
|
||||||
|
raise OllamaError(f"Generation failed: {e}")
|
||||||
|
|
||||||
|
def generate_stream(
|
||||||
|
self,
|
||||||
|
prompt: str,
|
||||||
|
model: Optional[str] = None,
|
||||||
|
system: Optional[str] = None
|
||||||
|
) -> Generator[str, None, None]:
|
||||||
|
"""
|
||||||
|
Stream generate responses.
|
||||||
|
|
||||||
|
Yields response chunks as they arrive.
|
||||||
|
"""
|
||||||
|
model = model or self.config.default_model
|
||||||
|
|
||||||
|
payload = {
|
||||||
|
"model": model,
|
||||||
|
"prompt": prompt,
|
||||||
|
"stream": True
|
||||||
|
}
|
||||||
|
|
||||||
|
if system:
|
||||||
|
payload["system"] = system
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = self.session.post(
|
||||||
|
f"{self.config.base_url}/api/generate",
|
||||||
|
json=payload,
|
||||||
|
stream=True,
|
||||||
|
timeout=self.config.timeout
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
for line in response.iter_lines():
|
||||||
|
if line:
|
||||||
|
try:
|
||||||
|
data = json.loads(line)
|
||||||
|
if "response" in data:
|
||||||
|
yield data["response"]
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
continue
|
||||||
|
|
||||||
|
except requests.RequestException as e:
|
||||||
|
raise OllamaError(f"Streaming failed: {e}")
|
||||||
|
|
||||||
|
def chat(
|
||||||
|
self,
|
||||||
|
messages: list,
|
||||||
|
model: Optional[str] = None
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Chat completion with message history.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
messages: List of {role, content} dicts
|
||||||
|
model: Model name
|
||||||
|
"""
|
||||||
|
model = model or self.config.default_model
|
||||||
|
|
||||||
|
payload = {
|
||||||
|
"model": model,
|
||||||
|
"messages": messages,
|
||||||
|
"stream": False
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = self.session.post(
|
||||||
|
f"{self.config.base_url}/api/chat",
|
||||||
|
json=payload,
|
||||||
|
timeout=self.config.timeout
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
return response.json()
|
||||||
|
except requests.RequestException as e:
|
||||||
|
raise OllamaError(f"Chat failed: {e}")
|
||||||
|
|
||||||
|
|
||||||
|
class OllamaError(Exception):
|
||||||
|
"""Ollama API error."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""CLI entry point."""
|
||||||
|
client = OllamaClient()
|
||||||
|
|
||||||
|
# Health check
|
||||||
|
print("Checking Ollama connection...")
|
||||||
|
if not client.health_check():
|
||||||
|
print("ERROR: Ollama not reachable at localhost:11434")
|
||||||
|
sys.exit(1)
|
||||||
|
print("✓ Ollama is running\n")
|
||||||
|
|
||||||
|
# List models
|
||||||
|
print("Available models:")
|
||||||
|
for model in client.list_models():
|
||||||
|
print(f" - {model}")
|
||||||
|
print()
|
||||||
|
|
||||||
|
# Generate if prompt provided
|
||||||
|
if len(sys.argv) > 1:
|
||||||
|
prompt = " ".join(sys.argv[1:])
|
||||||
|
print(f"Prompt: {prompt}")
|
||||||
|
print("-" * 40)
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = client.generate(
|
||||||
|
prompt=prompt,
|
||||||
|
system="You are a helpful assistant. Be concise."
|
||||||
|
)
|
||||||
|
print(response.get("response", "No response"))
|
||||||
|
print("-" * 40)
|
||||||
|
print(f"Tokens: {response.get('eval_count', 'N/A')}")
|
||||||
|
except OllamaError as e:
|
||||||
|
print(f"Error: {e}")
|
||||||
|
sys.exit(1)
|
||||||
|
else:
|
||||||
|
print("Usage: ollama_client.py <prompt>")
|
||||||
|
print("Example: ollama_client.py 'What is the Archon architecture?'")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
25
archon-poc/profile.yaml
Normal file
25
archon-poc/profile.yaml
Normal file
@@ -0,0 +1,25 @@
|
|||||||
|
name: Archon-POC
|
||||||
|
display_name: Archon (Proof-of-Concept)
|
||||||
|
model: gemma3:4b
|
||||||
|
variant: archon
|
||||||
|
provider: ollama
|
||||||
|
ollama_host: http://localhost:11434
|
||||||
|
|
||||||
|
creator: Ezra
|
||||||
|
lineage: Archon / Timmy Time Nexus
|
||||||
|
purpose: Three-layer architecture POC
|
||||||
|
|
||||||
|
# THIN PROFILE - No logic, only identity and routing
|
||||||
|
# All intelligence lives in Claw runtime + Gemma layer
|
||||||
|
|
||||||
|
routing:
|
||||||
|
runtime: claw.harness
|
||||||
|
intelligence: ollama.gemma
|
||||||
|
|
||||||
|
constraints:
|
||||||
|
max_tokens: 4096
|
||||||
|
temperature: 0.7
|
||||||
|
|
||||||
|
tagging:
|
||||||
|
required: true
|
||||||
|
tag: "#archon-poc"
|
||||||
BIN
archon-poc/runtime/__pycache__/harness.cpython-312.pyc
Normal file
BIN
archon-poc/runtime/__pycache__/harness.cpython-312.pyc
Normal file
Binary file not shown.
BIN
archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc
Normal file
BIN
archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc
Normal file
Binary file not shown.
177
archon-poc/runtime/harness.py
Normal file
177
archon-poc/runtime/harness.py
Normal file
@@ -0,0 +1,177 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Claw Code Runtime - Layer 2 of Archon Architecture
|
||||||
|
|
||||||
|
This harness contains all business logic, orchestrating between
|
||||||
|
the thin Hermes profile (Layer 1) and Gemma intelligence (Layer 3).
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import json
|
||||||
|
from typing import Dict, Any, Optional, List
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Add parent to path for imports
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||||
|
|
||||||
|
from runtime.tool_registry import ToolRegistry
|
||||||
|
|
||||||
|
|
||||||
|
class ClawHarness:
|
||||||
|
"""
|
||||||
|
The Claw Runtime - where all logic lives.
|
||||||
|
|
||||||
|
Responsibilities:
|
||||||
|
- Message routing and orchestration
|
||||||
|
- Tool execution management
|
||||||
|
- Context window management
|
||||||
|
- Conversation state tracking
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, ollama_host: str = "http://localhost:11434"):
|
||||||
|
self.ollama_host = ollama_host
|
||||||
|
self.tools = ToolRegistry()
|
||||||
|
self.conversation_history: List[Dict[str, Any]] = []
|
||||||
|
self.session_id: Optional[str] = None
|
||||||
|
|
||||||
|
def process_message(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Main entry point for processing messages.
|
||||||
|
|
||||||
|
Flow:
|
||||||
|
1. Parse intent
|
||||||
|
2. Determine if tools needed
|
||||||
|
3. Route to Gemma or execute tools
|
||||||
|
4. Return formatted response
|
||||||
|
"""
|
||||||
|
context = context or {}
|
||||||
|
|
||||||
|
# Log to conversation history
|
||||||
|
self.conversation_history.append({
|
||||||
|
"role": "user",
|
||||||
|
"content": message,
|
||||||
|
"timestamp": self._get_timestamp()
|
||||||
|
})
|
||||||
|
|
||||||
|
# Check for tool invocation
|
||||||
|
tool_call = self.tools.parse_tool_call(message)
|
||||||
|
if tool_call:
|
||||||
|
result = self._execute_tool(tool_call)
|
||||||
|
return self._format_response(
|
||||||
|
content=result,
|
||||||
|
tool_used=tool_call["name"],
|
||||||
|
metadata={"tool_result": True}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Route to intelligence layer (Gemma)
|
||||||
|
return self._route_to_intelligence(message, context)
|
||||||
|
|
||||||
|
def _route_to_intelligence(self, message: str, context: Dict) -> Dict[str, Any]:
|
||||||
|
"""Route message to Gemma via Ollama."""
|
||||||
|
from ollama_client import OllamaClient
|
||||||
|
|
||||||
|
client = OllamaClient(base_url=self.ollama_host)
|
||||||
|
|
||||||
|
# Build prompt with context
|
||||||
|
prompt = self._build_prompt(message, context)
|
||||||
|
|
||||||
|
# Get response from Gemma
|
||||||
|
response = client.generate(
|
||||||
|
model="gemma3:4b",
|
||||||
|
prompt=prompt,
|
||||||
|
system=self._get_system_prompt()
|
||||||
|
)
|
||||||
|
|
||||||
|
# Log response
|
||||||
|
self.conversation_history.append({
|
||||||
|
"role": "assistant",
|
||||||
|
"content": response.get("response", ""),
|
||||||
|
"timestamp": self._get_timestamp()
|
||||||
|
})
|
||||||
|
|
||||||
|
return self._format_response(
|
||||||
|
content=response.get("response", ""),
|
||||||
|
metadata={
|
||||||
|
"model": "gemma3:4b",
|
||||||
|
"tokens_used": response.get("eval_count", 0)
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
def _execute_tool(self, tool_call: Dict) -> str:
|
||||||
|
"""Execute a tool and return result."""
|
||||||
|
return self.tools.execute(tool_call)
|
||||||
|
|
||||||
|
def _build_prompt(self, message: str, context: Dict) -> str:
|
||||||
|
"""Build context-aware prompt for Gemma."""
|
||||||
|
history = "\n".join([
|
||||||
|
f"{msg['role']}: {msg['content']}"
|
||||||
|
for msg in self.conversation_history[-5:] # Last 5 messages
|
||||||
|
])
|
||||||
|
|
||||||
|
return f"""Previous conversation:
|
||||||
|
{history}
|
||||||
|
|
||||||
|
User: {message}
|
||||||
|
|
||||||
|
Assistant:"""
|
||||||
|
|
||||||
|
def _get_system_prompt(self) -> str:
|
||||||
|
"""Get system prompt for Gemma."""
|
||||||
|
return """You are the Archon POC, a helpful AI assistant.
|
||||||
|
Be concise but thorough. Tag your response with #archon-poc."""
|
||||||
|
|
||||||
|
def _format_response(self, content: str, tool_used: Optional[str] = None,
|
||||||
|
metadata: Optional[Dict] = None) -> Dict[str, Any]:
|
||||||
|
"""Format response for return to Layer 1."""
|
||||||
|
response = {
|
||||||
|
"content": content,
|
||||||
|
"status": "success",
|
||||||
|
"layer": "claw_runtime",
|
||||||
|
"tag": "#archon-poc"
|
||||||
|
}
|
||||||
|
|
||||||
|
if tool_used:
|
||||||
|
response["tool_used"] = tool_used
|
||||||
|
if metadata:
|
||||||
|
response["metadata"] = metadata
|
||||||
|
|
||||||
|
return response
|
||||||
|
|
||||||
|
def _get_timestamp(self) -> str:
|
||||||
|
"""Get current timestamp."""
|
||||||
|
from datetime import datetime
|
||||||
|
return datetime.now().isoformat()
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""CLI entry point."""
|
||||||
|
harness = ClawHarness()
|
||||||
|
|
||||||
|
if len(sys.argv) > 1:
|
||||||
|
message = " ".join(sys.argv[1:])
|
||||||
|
result = harness.process_message(message)
|
||||||
|
print(json.dumps(result, indent=2))
|
||||||
|
else:
|
||||||
|
# Interactive mode
|
||||||
|
print("Archon Harness - Interactive Mode")
|
||||||
|
print("Type 'exit' to quit\n")
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
message = input("> ")
|
||||||
|
if message.lower() in ("exit", "quit"):
|
||||||
|
break
|
||||||
|
|
||||||
|
result = harness.process_message(message)
|
||||||
|
print(f"\n{result['content']}\n")
|
||||||
|
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
break
|
||||||
|
except EOFError:
|
||||||
|
break
|
||||||
|
|
||||||
|
print("\nGoodbye!")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
194
archon-poc/runtime/tool_registry.py
Normal file
194
archon-poc/runtime/tool_registry.py
Normal file
@@ -0,0 +1,194 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Tool Registry - Layer 2 Component
|
||||||
|
|
||||||
|
Defines and manages tools available to the Claw runtime.
|
||||||
|
Tools are executed locally, not sent to the intelligence layer.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import re
|
||||||
|
import json
|
||||||
|
import subprocess
|
||||||
|
from typing import Dict, Any, List, Callable, Optional
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Tool:
|
||||||
|
name: str
|
||||||
|
description: str
|
||||||
|
parameters: Dict[str, Any]
|
||||||
|
handler: Callable
|
||||||
|
|
||||||
|
|
||||||
|
class ToolRegistry:
|
||||||
|
"""
|
||||||
|
Registry of available tools for the Claw runtime.
|
||||||
|
|
||||||
|
Tools are pattern-matched from user messages and executed
|
||||||
|
before routing to the intelligence layer.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.tools: Dict[str, Tool] = {}
|
||||||
|
self._register_builtin_tools()
|
||||||
|
|
||||||
|
def _register_builtin_tools(self):
|
||||||
|
"""Register built-in tools."""
|
||||||
|
self.register(Tool(
|
||||||
|
name="time",
|
||||||
|
description="Get current time",
|
||||||
|
parameters={},
|
||||||
|
handler=self._get_time
|
||||||
|
))
|
||||||
|
|
||||||
|
self.register(Tool(
|
||||||
|
name="status",
|
||||||
|
description="Get system status",
|
||||||
|
parameters={},
|
||||||
|
handler=self._get_status
|
||||||
|
))
|
||||||
|
|
||||||
|
self.register(Tool(
|
||||||
|
name="echo",
|
||||||
|
description="Echo a message back",
|
||||||
|
parameters={"message": "string"},
|
||||||
|
handler=self._echo
|
||||||
|
))
|
||||||
|
|
||||||
|
self.register(Tool(
|
||||||
|
name="ollama_list",
|
||||||
|
description="List available Ollama models",
|
||||||
|
parameters={},
|
||||||
|
handler=self._ollama_list
|
||||||
|
))
|
||||||
|
|
||||||
|
def register(self, tool: Tool):
|
||||||
|
"""Register a new tool."""
|
||||||
|
self.tools[tool.name] = tool
|
||||||
|
|
||||||
|
def parse_tool_call(self, message: str) -> Optional[Dict[str, Any]]:
|
||||||
|
"""
|
||||||
|
Parse a message for tool invocation.
|
||||||
|
|
||||||
|
Patterns:
|
||||||
|
- /tool_name
|
||||||
|
- /tool_name param=value
|
||||||
|
- @tool_name
|
||||||
|
"""
|
||||||
|
# Match /tool_name or @tool_name patterns
|
||||||
|
match = re.match(r'^[/@](\w+)(?:\s+(.+))?$', message.strip())
|
||||||
|
if not match:
|
||||||
|
return None
|
||||||
|
|
||||||
|
tool_name = match.group(1)
|
||||||
|
args_str = match.group(2) or ""
|
||||||
|
|
||||||
|
if tool_name not in self.tools:
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Parse parameters
|
||||||
|
params = {}
|
||||||
|
if args_str:
|
||||||
|
# Simple key=value parsing
|
||||||
|
for pair in args_str.split():
|
||||||
|
if '=' in pair:
|
||||||
|
key, value = pair.split('=', 1)
|
||||||
|
params[key] = value
|
||||||
|
else:
|
||||||
|
# Positional argument as 'message'
|
||||||
|
params["message"] = args_str
|
||||||
|
|
||||||
|
return {
|
||||||
|
"name": tool_name,
|
||||||
|
"parameters": params
|
||||||
|
}
|
||||||
|
|
||||||
|
def execute(self, tool_call: Dict[str, Any]) -> str:
|
||||||
|
"""Execute a tool call."""
|
||||||
|
tool_name = tool_call["name"]
|
||||||
|
params = tool_call.get("parameters", {})
|
||||||
|
|
||||||
|
if tool_name not in self.tools:
|
||||||
|
return f"Error: Unknown tool '{tool_name}'"
|
||||||
|
|
||||||
|
tool = self.tools[tool_name]
|
||||||
|
|
||||||
|
try:
|
||||||
|
result = tool.handler(**params)
|
||||||
|
return f"[Tool: {tool_name}]\n{result}"
|
||||||
|
except Exception as e:
|
||||||
|
return f"[Tool Error: {tool_name}]\n{str(e)}"
|
||||||
|
|
||||||
|
def list_tools(self) -> List[str]:
|
||||||
|
"""List all available tools."""
|
||||||
|
return [
|
||||||
|
f"{name}: {tool.description}"
|
||||||
|
for name, tool in self.tools.items()
|
||||||
|
]
|
||||||
|
|
||||||
|
# --- Tool Handlers ---
|
||||||
|
|
||||||
|
def _get_time(self) -> str:
|
||||||
|
"""Get current time."""
|
||||||
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||||
|
|
||||||
|
def _get_status(self) -> str:
|
||||||
|
"""Get system status."""
|
||||||
|
return json.dumps({
|
||||||
|
"runtime": "claw",
|
||||||
|
"version": "1.0.0-poc",
|
||||||
|
"status": "operational",
|
||||||
|
"tools_available": len(self.tools)
|
||||||
|
}, indent=2)
|
||||||
|
|
||||||
|
def _echo(self, message: str = "") -> str:
|
||||||
|
"""Echo a message."""
|
||||||
|
return message
|
||||||
|
|
||||||
|
def _ollama_list(self) -> str:
|
||||||
|
"""List Ollama models."""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["curl", "-s", "http://localhost:11434/api/tags"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True
|
||||||
|
)
|
||||||
|
data = json.loads(result.stdout)
|
||||||
|
models = [m["name"] for m in data.get("models", [])]
|
||||||
|
return f"Available models:\n" + "\n".join(f" - {m}" for m in models)
|
||||||
|
except Exception as e:
|
||||||
|
return f"Error listing models: {e}"
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""CLI for testing tool registry."""
|
||||||
|
registry = ToolRegistry()
|
||||||
|
|
||||||
|
print("Available tools:")
|
||||||
|
for tool_info in registry.list_tools():
|
||||||
|
print(f" {tool_info}")
|
||||||
|
|
||||||
|
print("\nTest parsing:")
|
||||||
|
test_messages = [
|
||||||
|
"/time",
|
||||||
|
"/status",
|
||||||
|
"/echo Hello world",
|
||||||
|
"/ollama_list",
|
||||||
|
"Regular message without tool"
|
||||||
|
]
|
||||||
|
|
||||||
|
for msg in test_messages:
|
||||||
|
parsed = registry.parse_tool_call(msg)
|
||||||
|
if parsed:
|
||||||
|
result = registry.execute(parsed)
|
||||||
|
print(f"\n> {msg}")
|
||||||
|
print(result)
|
||||||
|
else:
|
||||||
|
print(f"\n> {msg}")
|
||||||
|
print("(No tool call detected)")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
214
archon-poc/test_integration.py
Normal file
214
archon-poc/test_integration.py
Normal file
@@ -0,0 +1,214 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Integration Test - Archon Architecture
|
||||||
|
|
||||||
|
End-to-end test verifying the full three-layer flow:
|
||||||
|
1. Thin Hermes Profile (Layer 1)
|
||||||
|
2. Claw Code Runtime (Layer 2)
|
||||||
|
3. Gemma via Ollama (Layer 3)
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import yaml
|
||||||
|
import json
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Dict, Any
|
||||||
|
|
||||||
|
# Add archon-poc to path
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
|
||||||
|
from ollama_client import OllamaClient, OllamaError
|
||||||
|
from runtime.harness import ClawHarness
|
||||||
|
from runtime.tool_registry import ToolRegistry
|
||||||
|
|
||||||
|
|
||||||
|
class TestArchonArchitecture(unittest.TestCase):
|
||||||
|
"""Integration tests for Archon three-layer architecture."""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
"""Set up test fixtures."""
|
||||||
|
cls.base_path = Path(__file__).parent
|
||||||
|
|
||||||
|
# === Layer 1: Profile Tests ===
|
||||||
|
|
||||||
|
def test_01_profile_exists(self):
|
||||||
|
"""Verify profile.yaml exists and is valid."""
|
||||||
|
profile_path = self.base_path / "profile.yaml"
|
||||||
|
self.assertTrue(profile_path.exists(), "profile.yaml must exist")
|
||||||
|
|
||||||
|
with open(profile_path) as f:
|
||||||
|
profile = yaml.safe_load(f)
|
||||||
|
|
||||||
|
self.assertIn("name", profile)
|
||||||
|
self.assertIn("model", profile)
|
||||||
|
self.assertEqual(profile["model"], "gemma3:4b")
|
||||||
|
|
||||||
|
def test_02_profile_is_thin(self):
|
||||||
|
"""Verify profile is thin (< 50 lines, no logic)."""
|
||||||
|
profile_path = self.base_path / "profile.yaml"
|
||||||
|
lines = profile_path.read_text().splitlines()
|
||||||
|
|
||||||
|
self.assertLess(len(lines), 50, "Profile must be < 50 lines")
|
||||||
|
|
||||||
|
# Check for routing section (should exist)
|
||||||
|
content = "\n".join(lines)
|
||||||
|
self.assertIn("routing:", content)
|
||||||
|
|
||||||
|
# === Layer 2: Runtime Tests ===
|
||||||
|
|
||||||
|
def test_03_harness_exists(self):
|
||||||
|
"""Verify harness.py exists and imports."""
|
||||||
|
harness_path = self.base_path / "runtime" / "harness.py"
|
||||||
|
self.assertTrue(harness_path.exists())
|
||||||
|
|
||||||
|
# Already imported in setUp, if we got here it works
|
||||||
|
harness = ClawHarness()
|
||||||
|
self.assertIsNotNone(harness)
|
||||||
|
|
||||||
|
def test_04_tool_registry(self):
|
||||||
|
"""Verify tool registry works."""
|
||||||
|
registry = ToolRegistry()
|
||||||
|
|
||||||
|
# Test tool listing
|
||||||
|
tools = registry.list_tools()
|
||||||
|
self.assertGreater(len(tools), 0, "Must have at least one tool")
|
||||||
|
|
||||||
|
# Test tool parsing
|
||||||
|
parsed = registry.parse_tool_call("/time")
|
||||||
|
self.assertIsNotNone(parsed)
|
||||||
|
self.assertEqual(parsed["name"], "time")
|
||||||
|
|
||||||
|
# Test tool execution
|
||||||
|
result = registry.execute(parsed)
|
||||||
|
self.assertIn("Tool: time", result)
|
||||||
|
|
||||||
|
def test_05_harness_message_processing(self):
|
||||||
|
"""Test harness can process messages."""
|
||||||
|
harness = ClawHarness()
|
||||||
|
|
||||||
|
# Test tool invocation path
|
||||||
|
result = harness.process_message("/status")
|
||||||
|
self.assertEqual(result["status"], "success")
|
||||||
|
self.assertEqual(result["layer"], "claw_runtime")
|
||||||
|
|
||||||
|
# === Layer 3: Ollama Tests ===
|
||||||
|
|
||||||
|
def test_06_ollama_connection(self):
|
||||||
|
"""Verify Ollama is reachable."""
|
||||||
|
client = OllamaClient()
|
||||||
|
self.assertTrue(client.health_check(), "Ollama must be running")
|
||||||
|
|
||||||
|
def test_07_ollama_models(self):
|
||||||
|
"""Verify gemma model is available."""
|
||||||
|
client = OllamaClient()
|
||||||
|
models = client.list_models()
|
||||||
|
|
||||||
|
# Check for gemma3:4b (available) or gemma4:4b (target)
|
||||||
|
gemma_models = [m for m in models if "gemma" in m.lower()]
|
||||||
|
self.assertGreater(len(gemma_models), 0, "Need at least one Gemma model")
|
||||||
|
|
||||||
|
def test_08_ollama_generation(self):
|
||||||
|
"""Test basic generation through Ollama."""
|
||||||
|
client = OllamaClient()
|
||||||
|
|
||||||
|
response = client.generate(
|
||||||
|
prompt="Say 'Archon test passed' and nothing else.",
|
||||||
|
system="You are a test assistant. Be brief."
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertIn("response", response)
|
||||||
|
self.assertIsInstance(response["response"], str)
|
||||||
|
self.assertGreater(len(response["response"]), 0)
|
||||||
|
|
||||||
|
# === End-to-End Tests ===
|
||||||
|
|
||||||
|
def test_09_full_flow_tool_path(self):
|
||||||
|
"""Test full flow: Profile -> Runtime -> Tool -> Response."""
|
||||||
|
harness = ClawHarness()
|
||||||
|
|
||||||
|
# Simulate what the thin profile would do - just route to harness
|
||||||
|
result = harness.process_message("/echo Integration test")
|
||||||
|
|
||||||
|
self.assertEqual(result["status"], "success")
|
||||||
|
self.assertIn("content", result)
|
||||||
|
self.assertIn("Integration test", result["content"])
|
||||||
|
self.assertEqual(result["layer"], "claw_runtime")
|
||||||
|
self.assertEqual(result["tag"], "#archon-poc")
|
||||||
|
|
||||||
|
def test_10_full_flow_intelligence_path(self):
|
||||||
|
"""Test full flow: Profile -> Runtime -> Ollama -> Response."""
|
||||||
|
harness = ClawHarness()
|
||||||
|
|
||||||
|
# This routes through Ollama
|
||||||
|
result = harness.process_message("Say 'Archon intelligence layer active'")
|
||||||
|
|
||||||
|
self.assertEqual(result["status"], "success")
|
||||||
|
self.assertIn("content", result)
|
||||||
|
self.assertIn("layer", result)
|
||||||
|
self.assertIn("metadata", result)
|
||||||
|
|
||||||
|
def test_11_conversation_history(self):
|
||||||
|
"""Test conversation history tracking."""
|
||||||
|
harness = ClawHarness()
|
||||||
|
|
||||||
|
# Send multiple messages
|
||||||
|
harness.process_message("Message 1")
|
||||||
|
harness.process_message("Message 2")
|
||||||
|
|
||||||
|
# Check history
|
||||||
|
self.assertEqual(len(harness.conversation_history), 4) # 2 user + 2 assistant
|
||||||
|
|
||||||
|
# === Architecture Compliance Tests ===
|
||||||
|
|
||||||
|
def test_12_all_files_exist(self):
|
||||||
|
"""Verify all required files exist."""
|
||||||
|
required_files = [
|
||||||
|
"README.md",
|
||||||
|
"profile.yaml",
|
||||||
|
"runtime/harness.py",
|
||||||
|
"runtime/tool_registry.py",
|
||||||
|
"ollama_client.py",
|
||||||
|
"test_integration.py"
|
||||||
|
]
|
||||||
|
|
||||||
|
for file_path in required_files:
|
||||||
|
full_path = self.base_path / file_path
|
||||||
|
self.assertTrue(
|
||||||
|
full_path.exists(),
|
||||||
|
f"Required file missing: {file_path}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def run_tests():
|
||||||
|
"""Run all integration tests."""
|
||||||
|
print("=" * 60)
|
||||||
|
print("ARCHON ARCHITECTURE - INTEGRATION TEST SUITE")
|
||||||
|
print("=" * 60)
|
||||||
|
print()
|
||||||
|
|
||||||
|
# Create test suite
|
||||||
|
loader = unittest.TestLoader()
|
||||||
|
suite = loader.loadTestsFromTestCase(TestArchonArchitecture)
|
||||||
|
|
||||||
|
# Run with verbose output
|
||||||
|
runner = unittest.TextTestRunner(verbosity=2)
|
||||||
|
result = runner.run(suite)
|
||||||
|
|
||||||
|
print()
|
||||||
|
print("=" * 60)
|
||||||
|
if result.wasSuccessful():
|
||||||
|
print("✓ ALL TESTS PASSED")
|
||||||
|
print("Archon Architecture POC is working correctly!")
|
||||||
|
print("=" * 60)
|
||||||
|
return 0
|
||||||
|
else:
|
||||||
|
print("✗ SOME TESTS FAILED")
|
||||||
|
print("See output above for details")
|
||||||
|
print("=" * 60)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(run_tests())
|
||||||
Reference in New Issue
Block a user