Merge PR #4: Sovereign Real-time Learning System

agent/knowledge_ingester.py

@@ -0,0 +1,69 @@
+"""Sovereign Knowledge Ingester for Hermes Agent.
+
+Uses Gemini 3.1 Pro to learn from Google Search in real-time and
+persists the knowledge to Timmy's sovereign memory.
+"""
+
+import logging
+import base64
+from typing import Any, Dict, List, Optional
+from agent.gemini_adapter import GeminiAdapter
+from tools.gitea_client import GiteaClient
+
+logger = logging.getLogger(__name__)
+
+class KnowledgeIngester:
+    def __init__(self):
+        self.adapter = GeminiAdapter()
+        self.gitea = GiteaClient()
+
+    def learn_about(self, topic: str) -> str:
+        """Searches Google, analyzes the results, and saves the knowledge."""
+        logger.info(f"Learning about: {topic}")
+        
+        # 1. Search and Analyze
+        prompt = f"""
+Please perform a deep dive into the following topic: {topic}
+
+Use Google Search to find the most recent and relevant information.
+Analyze the findings and provide a structured 'Knowledge Fragment' in Markdown format.
+Include:
+- Summary of the topic
+- Key facts and recent developments
+- Implications for Timmy's sovereign mission
+- References (URLs)
+"""
+        result = self.adapter.generate(
+            model="gemini-3.1-pro-preview",
+            prompt=prompt,
+            system_instruction="You are Timmy's Sovereign Knowledge Ingester. Your goal is to find and synthesize high-fidelity information from Google Search.",
+            grounding=True,
+            thinking=True
+        )
+        
+        knowledge_fragment = result["text"]
+        
+        # 2. Persist to Timmy's Memory
+        repo = "Timmy_Foundation/timmy-config"
+        filename = f"memories/realtime_learning/{topic.lower().replace(' ', '_')}.md"
+        
+        try:
+            # Check if file exists to get SHA
+            sha = None
+            try:
+                existing = self.gitea.get_file(repo, filename)
+                sha = existing.get("sha")
+            except:
+                pass
+            
+            content_b64 = base64.b64encode(knowledge_fragment.encode()).decode()
+            
+            if sha:
+                self.gitea.update_file(repo, filename, content_b64, f"Update knowledge on {topic}", sha)
+            else:
+                self.gitea.create_file(repo, filename, content_b64, f"Initial knowledge on {topic}")
+                
+            return f"Successfully learned about {topic} and updated Timmy's memory at {filename}"
+        except Exception as e:
+            logger.error(f"Failed to persist knowledge: {e}")
+            return f"Learned about {topic}, but failed to save to memory: {e}\n\n{knowledge_fragment}"

skills/research/realtime_learning.py

@@ -0,0 +1,22 @@
+"""
+---
+title: Real-time Learning
+description: Allows Timmy to learn about any topic in real-time using Google Search and persist it to his sovereign memory.
+conditions:
+  - New information required
+  - Real-time events or trends
+  - Knowledge base expansion
+---
+"""
+
+from agent.knowledge_ingester import KnowledgeIngester
+
+def learn(topic: str) -> str:
+    """
+    Performs real-time learning on a topic and updates Timmy's memory.
+    
+    Args:
+        topic: The topic to learn about (e.g., 'recent advancements in quantum computing').
+    """
+    ingester = KnowledgeIngester()
+    return ingester.learn_about(topic)

tools/gitea_client.py

@@ -0,0 +1,59 @@
+"""
+Gitea API Client — typed, sovereign, zero-dependency.
+
+Enables the agent to interact with Timmy's sovereign Gitea instance
+for issue tracking, PR management, and knowledge persistence.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.request
+import urllib.error
+import urllib.parse
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Optional, Dict, List
+
+class GiteaClient:
+    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
+        self.base_url = base_url or os.environ.get("GITEA_URL", "http://143.198.27.163:3000")
+        self.token = token or os.environ.get("GITEA_TOKEN")
+        self.api = f"{self.base_url.rstrip('/')}/api/v1"
+
+    def _request(self, method: str, path: str, data: Optional[dict] = None) -> Any:
+        url = f"{self.api}{path}"
+        body = json.dumps(data).encode() if data else None
+        req = urllib.request.Request(url, data=body, method=method)
+        if self.token:
+            req.add_header("Authorization", f"token {self.token}")
+        req.add_header("Content-Type", "application/json")
+        req.add_header("Accept", "application/json")
+
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                raw = resp.read().decode()
+                return json.loads(raw) if raw else {}
+        except urllib.error.HTTPError as e:
+            raise Exception(f"Gitea {e.code}: {e.read().decode()}") from e
+
+    def get_file(self, repo: str, path: str, ref: str = "main") -> Dict[str, Any]:
+        return self._request("GET", f"/repos/{repo}/contents/{path}?ref={ref}")
+
+    def create_file(self, repo: str, path: str, content: str, message: str, branch: str = "main") -> Dict[str, Any]:
+        data = {
+            "branch": branch,
+            "content": content, # Base64 encoded
+            "message": message
+        }
+        return self._request("POST", f"/repos/{repo}/contents/{path}", data)
+
+    def update_file(self, repo: str, path: str, content: str, message: str, sha: str, branch: str = "main") -> Dict[str, Any]:
+        data = {
+            "branch": branch,
+            "content": content, # Base64 encoded
+            "message": message,
+            "sha": sha
+        }
+        return self._request("PUT", f"/repos/{repo}/contents/{path}", data)