Compare commits
2 Commits
archon-see
...
feature/ar
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
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