ezra/gemma-spectrum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Archon Architecture POC (Epic #370, Dispatch #371)

Browse Source 
Three-layer architecture implementation:
- Layer 1: Thin Hermes profile (profile.yaml, < 50 lines)
- Layer 2: Claw Code runtime (harness.py, tool_registry.py)
- Layer 3: Gemma via Ollama (ollama_client.py)

Features:
- Tool registry with pattern matching (/time, /status, /echo, /ollama_list)
- Full routing between tool execution and intelligence layer
- Ollama client with streaming and chat support
- Comprehensive integration test suite (12 tests)
- Connection to localhost:11434 using gemma3:4b model
This commit is contained in:
Ezra
2026-04-02 19:47:00 +00:00
parent 3ad94ff05d
commit 1b3bca9902
9 changed files with 898 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
archon-poc/README.md Normal file
View 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
View File

Binary file not shown.

226
archon-poc/ollama_client.py Normal file
View 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
View 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
View File

Binary file not shown.

BIN
archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc Normal file
View File

Binary file not shown.

177
archon-poc/runtime/harness.py Normal file
View 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
View 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
View 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())