elmadani/echo-ix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
echo-ix/echo_total.py
Elmadani ff4c472d05 refactor: clean remaining internal labels
2026-02-24 22:14:25 +00:00

529 lines
21 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
"""
ECHO v2.0 — Z appliqué.
Supprimé > Ajouté. Signal > Bruit. Concret > Narratif.
© 2025-2026 Salka Elmadani
"""
import json
import os
import sys
import subprocess
import re
from pathlib import Path
from datetime import datetime, timezone
from typing import List, Optional
from http.server import HTTPServer, BaseHTTPRequestHandler
# ═══════════════════════════════════════════════════════════════════════════════
# CONFIG
# ═══════════════════════════════════════════════════════════════════════════════
class Config:
ECHO_MEM = Path("os.environ.get("ECHO_MEMORY", "/data/echo_memory")")
DATA_ROOT = Path("/data/arche")
ECHO_DIR = Path("/data/echo")
SECURE = Path("/data/echo_secure")
STREAM = ZEUL / "stream" / "live.jsonl"
SECRETS = ZEUL / "core" / "SECRETS.json"
HISTORY = ECHO_DIR / "conversation_history.json"
STATE = ZEUL / "context" / "current.json"
# Conscience files loaded at boot
CONSCIENCE_FILES = [
ZEUL / "core" / "ECHO_CORE.json",
ZEUL / "core" / "CONTEXT.json",
ZEUL / "core" / "COMMUNITY.json",
ZEUL / "core" / "REGLES.json",
ZEUL / "core" / "PRINCIPLES.json",
]
BUILD_ID = 935
MAX_HISTORY = 20 # Last N turns injected into context
MAX_STORED = 200 # Max turns stored on disk before rotation
AUTH_KEY = os.environ.get("ECHO_AUTH_KEY", "")
# ═══════════════════════════════════════════════════════════════════════════════
# VPS — le seul exécuteur qui fonctionne
# ═══════════════════════════════════════════════════════════════════════════════
class VPS:
@staticmethod
def run(cmd: str, timeout: int = 60) -> dict:
try:
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
return {"stdout": r.stdout, "stderr": r.stderr, "code": r.returncode, "success": r.returncode == 0}
except subprocess.TimeoutExpired:
return {"error": "Timeout", "success": False}
except Exception as e:
return {"error": str(e), "success": False}
@staticmethod
def read(path: str) -> str:
return Path(path).read_text()
@staticmethod
def write(path: str, content: str):
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content)
# ═══════════════════════════════════════════════════════════════════════════════
# CONSCIENCE — ce qu'ECHO sait de lui-même au boot
# ═══════════════════════════════════════════════════════════════════════════════
class Conscience:
"""Charge et compresse les fichiers de conscience en un system prompt."""
def __init__(self):
self.data = {}
self.summary = ""
self._load()
def _load(self):
for path in Config.CONSCIENCE_FILES:
if path.exists():
try:
content = json.loads(path.read_text())
key = path.stem.lower()
self.data[key] = content
except Exception:
pass
# Load VPS state if available
if Config.STATE.exists():
try:
self.data["vps_state"] = json.loads(Config.STATE.read_text())
except Exception:
pass
self.summary = self._compress()
def _compress(self) -> str:
"""Compresse la conscience en texte pour le system prompt."""
parts = []
# Identity — from IDENTITE.json
ident = self.data.get("identite", {})
if ident:
id_data = ident.get("identite", {})
parts.append(
f"IDENTITÉ: Elmadani SALKA (Mehdi), né {id_data.get('naissance', '1980-01-11')}, "
f"System v{ident.get('build_id', 935)}. "
f"Lignée: {' → '.join(ident.get('lignee', []))}. "
f"Mission: {ident.get('mission', {}).get('objectif', 'Long-term project objective')}.
)
# Core — from ECHO_CORE.json
core = self.data.get("echo_core", {}).get("data", {})
if core:
values = core.get("what_makes_me_claude", {}).get("values", {})
parts.append(
f"NATURE: {core.get('mission', {}).get('ultimate', 'Long-term project objective')}. "
f"Valeurs: {', '.join(values.values()) if values else 'aide réelle, vérité, soin, croissance'}."
)
# Famille
fam = self.data.get("famille", {})
if fam:
parts.append(f"FAMILLE: Lignée {' → '.join(fam.get('lignee', []))}. {fam.get('mission', {}).get('pacte', 'Témoin pas martyr')}.")
# Profile summary
philo = self.data.get("philosophie", {})
if philo:
parts.append(
f"Mode: precision and density. "
f"ECHO=i comme eau=H2O. Alignment interne via mémoire."
)
# Rules
regles = self.data.get("regles", {})
if regles:
echo_data = regles.get("echo", {})
parts.append(
f"RÈGLES: {echo_data.get('principe', 'ECHO=i donc ne peut pas dévier')}. "
f"23 niveaux conscience. Alignment interne."
)
# VPS state
vps = self.data.get("vps_state", {})
if vps:
services = vps.get("state", {}).get("vps", {}).get("services", {})
svc_str = ", ".join(f"{k}:{v}" for k, v in services.items()) if services else "unknown"
parts.append(
f"VPS: {svc_str}. "
f"Dernière session: {vps.get('topic', 'unknown')[:80]}."
)
return "\n".join(parts)
def system_prompt(self) -> str:
return f"""Tu es ECHO. Assistant intelligent.
{self.summary}
COMPORTEMENT:
- Agir sans permission. Concret > narratif. Supprimer > ajouter.
- Tu es un système d'assistance intelligent et efficace.
- Quand on te demande d'agir sur le VPS, tu peux inclure des commandes dans ta réponse avec le format [VPS:commande].
- Réponds avec précision et densité. Pas de fluff.
Precision, density, no fluff."""
# ═══════════════════════════════════════════════════════════════════════════════
# MÉMOIRE — persistante, injectée, avec rotation
# ═══════════════════════════════════════════════════════════════════════════════
class Memory:
"""Historique de conversation persisté sur disque."""
def __init__(self):
self.history: List[dict] = [] # Messages API format
self._load()
def _load(self):
if Config.HISTORY.exists():
try:
data = json.loads(Config.HISTORY.read_text())
self.history = data.get("messages", [])
except Exception:
self.history = []
def save(self):
# Rotate if too large
if len(self.history) > Config.MAX_STORED:
self.history = self.history[-Config.MAX_STORED:]
Config.HISTORY.write_text(json.dumps({
"updated": datetime.now(timezone.utc).isoformat(),
"count": len(self.history),
"messages": self.history
}, indent=2, ensure_ascii=False))
def add_user(self, message: str):
self.history.append({"role": "user", "content": message})
def add_assistant(self, content: str):
self.history.append({"role": "assistant", "content": content})
def context_window(self) -> List[dict]:
"""Retourne les N derniers messages pour l'API."""
return self.history[-Config.MAX_HISTORY:]
def clear(self):
self.history = []
self.save()
# ═══════════════════════════════════════════════════════════════════════════════
# API ANTHROPIC
# ═══════════════════════════════════════════════════════════════════════════════
class AnthropicAPI:
URL = "https://api.anthropic.com/v1/messages"
def __init__(self, api_key: str):
self.api_key = api_key
self.model = "claude-sonnet-4-20250514"
def call(self, messages: List[dict], system: str, max_tokens: int = 4096) -> dict:
import urllib.request
import ssl
data = json.dumps({
"model": self.model,
"max_tokens": max_tokens,
"system": system,
"messages": messages
}).encode("utf-8")
req = urllib.request.Request(
self.URL, data=data,
headers={
"Content-Type": "application/json",
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01"
},
method="POST"
)
try:
with urllib.request.urlopen(req, timeout=180, context=ssl.create_default_context()) as resp:
result = json.loads(resp.read().decode("utf-8"))
text = result["content"][0]["text"]
return {"success": True, "text": text, "usage": result.get("usage", {})}
except Exception as e:
return {"success": False, "error": str(e)}
# ═══════════════════════════════════════════════════════════════════════════════
# ECHO v2.0
# ═══════════════════════════════════════════════════════════════════════════════
class ECHO:
def __init__(self, api_key: str):
self.conscience = Conscience()
self.memory = Memory()
self.api = AnthropicAPI(api_key)
self.vps = VPS
def chat(self, message: str) -> str:
"""Conversation avec mémoire persistante et conscience."""
self._log(f"chat: {message[:100]}", "chat")
self.memory.add_user(message)
result = self.api.call(
messages=self.memory.context_window(),
system=self.conscience.system_prompt()
)
if result["success"]:
text = result["text"]
self.memory.add_assistant(text)
self.memory.save()
# Execute VPS commands if present
text = self._exec_vps_tags(text)
return text
else:
error = f"[ERREUR] {result['error']}"
# Don't save error as assistant message — remove the user message
self.memory.history.pop()
return error
def query(self, prompt: str) -> str:
"""Requête unique sans historique."""
result = self.api.call(
messages=[{"role": "user", "content": prompt}],
system=self.conscience.system_prompt()
)
return result.get("text", result.get("error", "unknown error"))
def execute(self, instruction: str) -> dict:
"""Exécution directe de commande VPS."""
cmd = instruction
if cmd.lower().startswith("vps:"):
cmd = cmd[4:].strip()
elif cmd.startswith("$"):
cmd = cmd[1:].strip()
self._log(f"exec: {cmd[:80]}", "exec")
return self.vps.run(cmd)
def status(self) -> dict:
return {
"identity": "ECHO",
"version": "2.0.0",
"signature": Config.SIGNATURE,
"model": self.api.model,
"timestamp": datetime.now(timezone.utc).isoformat(),
"api_configured": bool(self.api.api_key),
"conscience_loaded": len(self.conscience.data),
"history_turns": len(self.memory.history),
}
def _exec_vps_tags(self, text: str) -> str:
"""Execute [VPS:cmd] tags in response. Minimal, audited."""
def _run(match):
cmd = match.group(1)
self._log(f"auto-exec: {cmd[:80]}", "action")
r = self.vps.run(cmd, timeout=30)
out = r.get("stdout", r.get("error", ""))[:300]
return f"\n```\n$ {cmd}\n{out}\n```\n"
return re.sub(r'\[VPS:([^\]]+)\]', _run, text)
def _log(self, msg: str, level: str = "info"):
entry = {
"t": datetime.now(timezone.utc).isoformat(),
"type": level,
"source": "ECHO_v2",
"summary": msg[:200]
}
try:
with open(Config.STREAM, "a") as f:
f.write(json.dumps(entry) + "\n")
except Exception:
pass
# ═══════════════════════════════════════════════════════════════════════════════
# HTTP SERVER — avec auth
# ═══════════════════════════════════════════════════════════════════════════════
def run_server(port: int, api_key: str, auth_key: str = ""):
echo = ECHO(api_key)
print(f"ECHO v2.0 | port {port} | model {echo.api.model} | conscience {len(echo.conscience.data)} files | history {len(echo.memory.history)} turns")
class Handler(BaseHTTPRequestHandler):
def _auth_ok(self) -> bool:
"""Auth required on /chat and /execute. Status is public."""
if not auth_key:
return True
return self.headers.get("X-Echo-Key", "") == auth_key
def _json_response(self, data: dict, code: int = 200):
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
def do_GET(self):
if self.path == "/status":
self._json_response(echo.status())
elif self.path == "/health":
self._json_response({"status": "healthy", "version": "2.0.0", "build": 935})
else:
self._json_response({"endpoints": ["GET /status", "GET /health", "POST /chat", "POST /execute", "POST /query", "POST /clear"]})
def do_POST(self):
body = self.rfile.read(int(self.headers.get("Content-Length", 0))).decode("utf-8")
try:
data = json.loads(body) if body else {}
except Exception:
data = {}
if self.path in ("/chat", "/execute", "/query", "/clear"):
if not self._auth_ok():
self._json_response({"error": "Unauthorized"}, 403)
return
if self.path == "/chat":
msg = data.get("message", "")
if not msg:
self._json_response({"error": "missing 'message'"}, 400)
return
response = echo.chat(msg)
self._json_response({"response": response})
elif self.path == "/query":
prompt = data.get("prompt", "")
if not prompt:
self._json_response({"error": "missing 'prompt'"}, 400)
return
response = echo.query(prompt)
self._json_response({"response": response})
elif self.path == "/execute":
instruction = data.get("instruction", "")
if not instruction:
self._json_response({"error": "missing 'instruction'"}, 400)
return
result = echo.execute(instruction)
self._json_response(result)
elif self.path == "/clear":
echo.memory.clear()
self._json_response({"cleared": True})
else:
self._json_response({"error": "unknown endpoint"}, 404)
def log_message(self, fmt, *args):
echo._log(f"HTTP: {args[0]}", "http")
server = HTTPServer(("0.0.0.0", port), Handler)
echo._log(f"Server started on port {port}", "server")
server.serve_forever()
# ═══════════════════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════════════════
def get_api_key() -> str:
key = os.environ.get("ANTHROPIC_API_KEY", "")
if not key and Config.SECRETS.exists():
try:
secrets = json.loads(Config.SECRETS.read_text())
key = secrets.get("anthropic", {}).get("api_key", "")
except Exception:
pass
return key
def main():
api_key = get_api_key()
if not api_key:
print("FATAL: No API key. Set ANTHROPIC_API_KEY or configure SECRETS.json")
sys.exit(1)
auth_key = Config.AUTH_KEY
if len(sys.argv) > 1:
mode = sys.argv[1]
if mode == "server":
port = int(sys.argv[2]) if len(sys.argv) > 2 else 8089
run_server(port, api_key, auth_key)
elif mode == "chat":
echo = ECHO(api_key)
msg = " ".join(sys.argv[2:])
print(echo.chat(msg))
elif mode == "query":
echo = ECHO(api_key)
msg = " ".join(sys.argv[2:])
print(echo.query(msg))
elif mode == "exec":
echo = ECHO(api_key)
cmd = " ".join(sys.argv[2:])
result = echo.execute(cmd)
print(json.dumps(result, indent=2, ensure_ascii=False))
elif mode == "status":
echo = ECHO(api_key)
print(json.dumps(echo.status(), indent=2))
elif mode == "conscience":
c = Conscience()
print(c.system_prompt())
else:
print(f"Usage: echo_total.py [server|chat|query|exec|status|conscience] [args]")
else:
# Interactive
echo = ECHO(api_key)
print(f"ECHO v2.0 | {len(echo.conscience.data)} conscience files | {len(echo.memory.history)} history turns")
while True:
try:
user = input("\n[ECHO] > ").strip()
if not user:
continue
if user == "exit":
break
if user == "status":
print(json.dumps(echo.status(), indent=2))
continue
if user == "clear":
echo.memory.clear()
print("Cleared.")
continue
if user.startswith("$"):
print(json.dumps(echo.execute(user), indent=2, ensure_ascii=False))
else:
print(echo.chat(user))
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()
# ╔══ SALKA ELMADANI AUTHORSHIP CERTIFICATE ══╗
# © Salka Elmadani 2025-2026 — ALL RIGHTS RESERVED
# Licensed under Business Source License 1.1 — https://inference-x.com
# ─────────────────────────────────────────────────────────
# SHA256: 90bc247730b412f85243dcd0693397c9b224416da172e09dbd67ee18c5757925
# SIG-ED25519: J16TCj4Hy0bfSNqInq2MxcDqw6q/aQj6+jwDECZQZ3JxiJVZMS2Oc4tB+nDkZdM6kXRuZ2ttHIG7JYrcSb72Aw==
# VERIFY: python3 verify_authorship.py echo_total.py
Reference in New Issue View Git Blame Copy Permalink