diff --git a/archon-poc/README.md b/archon-poc/README.md new file mode 100644 index 0000000..d3bc4ed --- /dev/null +++ b/archon-poc/README.md @@ -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 diff --git a/archon-poc/__pycache__/ollama_client.cpython-312.pyc b/archon-poc/__pycache__/ollama_client.cpython-312.pyc new file mode 100644 index 0000000..50d557a Binary files /dev/null and b/archon-poc/__pycache__/ollama_client.cpython-312.pyc differ diff --git a/archon-poc/ollama_client.py b/archon-poc/ollama_client.py new file mode 100644 index 0000000..d09f6a3 --- /dev/null +++ b/archon-poc/ollama_client.py @@ -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 ") + print("Example: ollama_client.py 'What is the Archon architecture?'") + + +if __name__ == "__main__": + main() diff --git a/archon-poc/profile.yaml b/archon-poc/profile.yaml new file mode 100644 index 0000000..555b5aa --- /dev/null +++ b/archon-poc/profile.yaml @@ -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" diff --git a/archon-poc/runtime/__pycache__/harness.cpython-312.pyc b/archon-poc/runtime/__pycache__/harness.cpython-312.pyc new file mode 100644 index 0000000..940f546 Binary files /dev/null and b/archon-poc/runtime/__pycache__/harness.cpython-312.pyc differ diff --git a/archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc b/archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc new file mode 100644 index 0000000..3173914 Binary files /dev/null and b/archon-poc/runtime/__pycache__/tool_registry.cpython-312.pyc differ diff --git a/archon-poc/runtime/harness.py b/archon-poc/runtime/harness.py new file mode 100644 index 0000000..0784ae6 --- /dev/null +++ b/archon-poc/runtime/harness.py @@ -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() diff --git a/archon-poc/runtime/tool_registry.py b/archon-poc/runtime/tool_registry.py new file mode 100644 index 0000000..7e8b931 --- /dev/null +++ b/archon-poc/runtime/tool_registry.py @@ -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() diff --git a/archon-poc/test_integration.py b/archon-poc/test_integration.py new file mode 100644 index 0000000..9bb9f46 --- /dev/null +++ b/archon-poc/test_integration.py @@ -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())