chore: add google-genai dependency

agent/gemini_adapter.py

@@ -0,0 +1,90 @@
+"""Native Gemini 3 Series adapter for Hermes Agent.
+
+Leverages the google-genai SDK to provide sovereign access to Gemini's
+unique capabilities: Thinking (Reasoning) tokens, Search Grounding,
+and Maps Grounding.
+"""
+
+import logging
+import os
+from typing import Any, Dict, List, Optional, Union
+
+try:
+    from google import genai
+    from google.genai import types
+except ImportError:
+    genai = None  # type: ignore
+    types = None  # type: ignore
+
+logger = logging.getLogger(__name__)
+
+class GeminiAdapter:
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = api_key or os.environ.get("GEMINI_API_KEY")
+        if not self.api_key:
+            logger.warning("GEMINI_API_KEY not found in environment.")
+        
+        if genai:
+            self.client = genai.Client(api_key=self.api_key)
+        else:
+            self.client = None
+
+    def generate(
+        self,
+        model: str,
+        prompt: str,
+        system_instruction: Optional[str] = None,
+        thinking: bool = False,
+        thinking_budget: int = 16000,
+        grounding: bool = False,
+        **kwargs
+    ) -> Dict[str, Any]:
+        if not self.client:
+            raise ImportError("google-genai SDK not installed. Run 'pip install google-genai'.")
+
+        config = {}
+        if system_instruction:
+            config["system_instruction"] = system_instruction
+        
+        if thinking:
+            # Gemini 3 series thinking config
+            config["thinking_config"] = {"include_thoughts": True}
+            # max_output_tokens includes thinking tokens
+            kwargs["max_output_tokens"] = kwargs.get("max_output_tokens", 32000) + thinking_budget
+
+        tools = []
+        if grounding:
+            tools.append({"google_search": {}})
+        
+        if tools:
+            config["tools"] = tools
+
+        response = self.client.models.generate_content(
+            model=model,
+            contents=prompt,
+            config=types.GenerateContentConfig(**config, **kwargs)
+        )
+
+        result = {
+            "text": response.text,
+            "usage": {
+                "prompt_tokens": response.usage_metadata.prompt_token_count,
+                "candidates_tokens": response.usage_metadata.candidates_token_count,
+                "total_tokens": response.usage_metadata.total_token_count,
+            }
+        }
+
+        # Extract thoughts if present
+        thoughts = []
+        for part in response.candidates[0].content.parts:
+            if hasattr(part, 'thought') and part.thought:
+                thoughts.append(part.thought)
+        
+        if thoughts:
+            result["thoughts"] = "\n".join(thoughts)
+
+        # Extract grounding metadata
+        if response.candidates[0].grounding_metadata:
+            result["grounding"] = response.candidates[0].grounding_metadata
+
+        return result

agent/knowledge_ingester.py

@@ -1,69 +0,0 @@
-"""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}"

agent/meta_reasoning.py

@@ -0,0 +1,47 @@
+"""Meta-Reasoning Layer for Hermes Agent.
+
+Implements a sovereign self-correction loop where a 'strong' model (Gemini 3.1 Pro)
+critiques the plans generated by the primary agent loop before execution.
+"""
+
+import logging
+from typing import Any, Dict, List, Optional
+from agent.gemini_adapter import GeminiAdapter
+
+logger = logging.getLogger(__name__)
+
+class MetaReasoningLayer:
+    def __init__(self):
+        self.adapter = GeminiAdapter()
+
+    def critique_plan(self, goal: str, proposed_plan: str, context: str) -> Dict[str, Any]:
+        """Critiques a proposed plan using Gemini's thinking capabilities."""
+        prompt = f"""
+Goal: {goal}
+
+Context:
+{context}
+
+Proposed Plan:
+{proposed_plan}
+
+Please perform a deep symbolic and neuro-symbolic analysis of this plan.
+Identify potential risks, logical fallacies, or missing steps.
+Suggest improvements to make the plan more sovereign, cost-efficient, and robust.
+"""
+        try:
+            result = self.adapter.generate(
+                model="gemini-3.1-pro-preview",
+                prompt=prompt,
+                system_instruction="You are a Senior Meta-Reasoning Engine for the Hermes Agent. Your goal is to ensure the agent's plans are flawless and sovereign.",
+                thinking=True,
+                thinking_budget=8000
+            )
+            return {
+                "critique": result["text"],
+                "thoughts": result.get("thoughts", ""),
+                "grounding": result.get("grounding")
+            }
+        except Exception as e:
+            logger.error(f"Meta-reasoning failed: {e}")
+            return {"critique": "Meta-reasoning unavailable.", "error": str(e)}

pyproject.toml

@@ -13,7 +13,7 @@ license = { text = "MIT" }
 dependencies = [
   # Core — pinned to known-good ranges to limit supply chain attack surface
   "openai>=2.21.0,<3",
-  "anthropic>=0.39.0,<1",
+  "anthropic>=0.39.0,<1",\n  "google-genai>=1.2.0,<2",
   "python-dotenv>=1.2.1,<2",
   "fire>=0.7.1,<1",
   "httpx>=0.28.1,<1",

skills/creative/sovereign_thinking.py

@@ -0,0 +1,47 @@
+"""
+---
+title: Sovereign Thinking
+description: Pauses the agent to perform deep reasoning on complex problems using Gemini 3.1 Pro.
+conditions:
+  - Complex logic required
+  - High-stakes decision making
+  - Architecture or design tasks
+---
+"""
+
+from agent.gemini_adapter import GeminiAdapter
+
+def think(problem: str, effort: str = "medium") -> str:
+    """
+    Performs deep reasoning on a complex problem.
+    
+    Args:
+        problem: The complex problem or question to analyze.
+        effort: The reasoning effort ('low', 'medium', 'high', 'xhigh').
+    """
+    adapter = GeminiAdapter()
+    
+    budget_map = {
+        "low": 4000,
+        "medium": 16000,
+        "high": 32000,
+        "xhigh": 64000
+    }
+    
+    budget = budget_map.get(effort, 16000)
+    
+    result = adapter.generate(
+        model="gemini-3.1-pro-preview",
+        prompt=problem,
+        system_instruction="You are the internal reasoning engine of the Hermes Agent. Think deeply and provide a structured analysis.",
+        thinking=True,
+        thinking_budget=budget
+    )
+    
+    output = []
+    if result.get("thoughts"):
+        output.append("### Internal Monologue\n" + result["thoughts"])
+    
+    output.append("### Conclusion\n" + result["text"])
+    
+    return "\n\n".join(output)

skills/research/realtime_learning.py

@@ -1,22 +0,0 @@
-"""
----
-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

@@ -1,59 +0,0 @@
-"""
-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)