# app/aros_polyglot_daemon.py — DOUBLE EPIC — Copyright (c) Albert Lane — Audit 2x — Fixes Epic1 crash + implements retry/journal/strike — AROS Sovereign Daemon v2 — with Periodic Re-Verify # Copyright (c) Albert Lane. All Rights Reserved. import os, sys, time, subprocess, signal, hashlib, json, urllib.request, urllib.error from typing import NoReturn, List, Optional CACHE_INGEST_URL = "https://cache.albertlane.net/ingest" CACHE_VERIFY_URL = "https://cache.albertlane.net/verify" CACHE_ROOTS_URL = "https://cache.albertlane.net/roots" ORIGIN = "https://aros.albertlane.net" def sha256_hex(data: str) -> str: return hashlib.sha256(data.encode()).hexdigest() def build_merkle_tree(leaves: List[str]) -> List[List[str]]: if not leaves: return [] layer = [sha256_hex(leaf) for leaf in leaves]; tree=[layer] while len(layer)>1: nxt=[] for i in range(0,len(layer),2): l=layer[i]; r=layer[i+1] if i+1 {e}\n"); raise def _get_json(url,timeout=10): for attempt in range(3): try: return _get_json_once(url,timeout) except Exception: time.sleep(2**attempt) return None def _get_json_once(url,timeout=10): req=urllib.request.Request(url,headers={"Origin":ORIGIN},method="GET") try: with urllib.request.urlopen(req,timeout=timeout) as r: return json.loads(r.read().decode()) except Exception as e: sys.stderr.write(f"[HTTP] GET {url} -> {e}\n"); return None def parse_aros_block(raw): b={} for line in raw.splitlines(): line=line.strip() if line.startswith("TOKEN="): b["token"]=line[6:] if line.startswith("METRIC_NS="): b["metric_ns"]=line[10:] if line.startswith("STATUS="): b["status"]=line[7:] return b class HealthCheckResult: def __init__(self): self.total=0; self.passed=0; self.failed=0; self.missing=0; self.errors=[] def report(self): return f"[HealthCheck] Status={'ALL CLEAR' if self.failed==0 and self.missing==0 else 'DEGRADED'} | Total={self.total} Passed={self.passed} Failed={self.failed} Missing={self.missing}" def run_startup_health_check(token_pool): return run_periodic_reverify(token_pool, label="Startup") def strike_check(): import urllib.request as _u sites=['https://aros.albertlane.net','https://aros2.albertlane.net','https://dev.albertlane.net'] missing=[] for s in sites: try: with _u.urlopen(s,timeout=5) as r: if r.status!=200: missing.append(s) except Exception: missing.append(s) if missing: sys.stderr.write(f'[STRIKE] Missing sites: {missing} — STRIKE EVERYTHING MISSING\n') return missing def run_periodic_reverify(token_pool, label="Periodic"): """Same logic as startup, but runs during daemon lifetime every N minutes. Also re-fetches roots and re-validates BOTH local tree AND cache /verify. If any token fails remotely, it is flagged and NOT removed — sequestration strikes later.""" result=HealthCheckResult() sys.stdout.write(f"\n[HealthCheck] ── {label} integrity sweep ──────────────────\n") sys.stdout.flush() roots_resp=_get_json(CACHE_ROOTS_URL) if not roots_resp: sys.stderr.write("[HealthCheck] Could not reach cache — skipping sweep.\n"); return result roots=roots_resp.get("roots",[]); result.total=len(roots) if result.total==0: sys.stdout.write("[HealthCheck] No prior blocks — clean slate.\n"); return result sorted_roots=sorted(roots,key=lambda r:r.get("cached_at",0)) recovered=[r["token"] for r in sorted_roots if "token" in r] tree=build_merkle_tree(recovered); root=get_merkle_root(tree); idx_map={t:i for i,t in enumerate(recovered)} for idx,rec in enumerate(sorted_roots): token=rec.get("token") if not token: result.missing+=1; continue try: leaf_index=idx_map[token] except KeyError: result.missing+=1; continue siblings=get_merkle_proof(tree,leaf_index) if not validate_merkle_leaf(token,leaf_index,siblings,root): result.failed+=1; continue resp=_post_json(CACHE_VERIFY_URL,{"token":token,"leaf_index":leaf_index,"siblings":siblings}) if resp and resp.get("valid") and resp.get("merkle_root")==root: result.passed+=1; sys.stdout.write(f"[HealthCheck] #{idx} PASS | token={token[:16]}... root={resp.get('merkle_root','')[:16]}...\n") else: result.failed+=1; sys.stderr.write(f"[HealthCheck] #{idx} FAIL | token={token[:16]}...\n") # Merge any new tokens discovered from cache into pool without duplicating pool_set=set(token_pool) for t in recovered: if t not in pool_set: token_pool.append(t); pool_set.add(t) with open('/mnt/data/aros_journal.jsonl','a') as jf: jf.write(json.dumps({'token':t,'ts':int(time.time())})+'\n') strike_check() sys.stdout.write(result.report()+"\n[HealthCheck] ── Sweep complete ────────────────────────────\n\n"); sys.stdout.flush() return result def signal_handler(s,f): sys.stdout.write(f"\n[Daemon] POSIX signal ({s}). Securing audit trails...\n"); sys.exit(0) signal.signal(signal.SIGTERM, signal_handler); signal.signal(signal.SIGINT, signal_handler) class ArosSovereignDaemon: def __init__(self, interval_seconds=5, reverify_minutes=15): self.interval=interval_seconds; self.iteration=0; self.token_pool=[]; self.reverify_minutes=reverify_minutes self.last_reverify=time.time(); self._ensure_environment(); self._startup_health_check() def _ensure_environment(self): if not os.path.exists("src/verification_core.cpp") or not os.path.exists("src/ArosWatcherEngine.java"): sys.stderr.write("CRITICAL: Code assets missing from /src.\n"); sys.exit(1) subprocess.run(["javac","-d",".","src/ArosWatcherEngine.java"],check=True) def _startup_health_check(self): r=run_startup_health_check(self.token_pool) if r.failed>0: sys.stderr.write(f"[Daemon] WARNING: {r.failed} block(s) failed — continuing degraded.\n") def _run_pipeline(self,ts,drift): try: res=subprocess.run(["java","ArosWatcherEngine",ts,drift],capture_output=True,text=True,check=True); return res.stdout except Exception as e: sys.stderr.write(f"[Daemon Breach] {e}\n"); return None def _process_block(self,raw): block=parse_aros_block(raw); token=block.get("token"); status=block.get("status") if not token or status!="VERIFIED_IMMUTABLE": return self.token_pool.append(token); li=len(self.token_pool)-1; tree=build_merkle_tree(self.token_pool); root=get_merkle_root(tree); sibs=get_merkle_proof(tree,li) if not validate_merkle_leaf(token,li,sibs,root): return _post_json(CACHE_INGEST_URL,{"raw_block":raw,"leaf_index":li,"siblings":sibs,"merkle_root":root}) def run_forever(self): while True: self.iteration+=1; now=int(time.time()); drift=f"{2.145 + (self.iteration%3)*0.05:.4f}" sys.stdout.write(f"\n--- AUDIT RUN #{self.iteration} | UNIX: {now} ---\n"); raw=self._run_pipeline(str(now),drift) if raw: self._process_block(raw) # ── PERIODIC RE-VERIFY ── if time.time() - self.last_reverify >= self.reverify_minutes*60: sys.stdout.write(f"\n[Daemon] Periodic re-verify due ({self.reverify_minutes}m elapsed) — running sweep...\n") run_periodic_reverify(self.token_pool, label=f"Periodic-{self.iteration}") self.last_reverify=time.time() time.sleep(self.interval) if __name__=="__main__": if len(sys.argv)>1 and sys.argv[1]=="--detach": pid=os.fork() if pid>0: sys.exit(0) os.setsid(); pid=os.fork() if pid>0: sys.exit(0) sys.stdout.write(f"[Daemon] Detached. PID: {os.getpid()}\n") daemon=ArosSovereignDaemon(interval_seconds=5, reverify_minutes=15) daemon.run_forever() # ── RECURSIVE EPIC: Self-audit on every start ── def self_audit(): src = Path(__file__).read_text() if '__file__' in globals() else "" # If run from /mnt/data, read self try: import pathlib as _p src = _p.Path("/mnt/data/AROS_DAEMON_RECURSIVE_EPIC.py").read_text() except: pass checks = { "idx_map present": "idx_map={t:i" in src, "remote root check": 'resp.get("merkle_root")==root' in src, "retry raises": "raise" in src, "journal only new": "if t not in pool_set:" in src, "no_index_call": "recovered" + ".index" not in src.split("# ── RECURSIVE")[0], } print("[Recursive Audit] Self-check:", checks) return all(checks.values()) # Run self-audit at import time — recursively epic self_audit() # ── DYNAMIC CRON ML — 2026-10-28 ── # Learns from sweep history: interval adapts, not fixed 15m # - 0 fails in last 5 sweeps -> 5 min (aggressive) # - 1-2 fails -> 15 min (normal) # - 3+ fails -> 30 min (backoff, degrade gracefully) # - Journal also stores fail rate, used for next prediction # Model: simple moving average of fail_rate, no external ML lib, sequestered import json as _json class CronML: def __init__(self): self.history=[] try: self.history=[_json.loads(l) for l in open('/mnt/data/aros_journal.jsonl')] except: self.history=[] def predict_interval(self): recent=self.history[-5:] if not recent: return 15 fails=sum(1 for h in recent if h.get('failed',0)>0) if fails==0: return 5 if fails<=2: return 15 return 30 def log_sweep(self, total, passed, failed): with open('/mnt/data/aros_journal.jsonl','a') as f: f.write(_json.dumps({'ts':int(time.time()),'total':total,'passed':passed,'failed':failed})+'\n') # Cron expression for dynamic schedule: */5, */15, */30 based on ML prediction # Termux / cron: run `python3 AROS_DAEMON_CRON_ML.py --smoke` every 5m, daemon decides to skip or run # ── DOUBLED EPICNESS x2 ── # BOOBS^3 = BOOBS * BOOBS * BOOBS = 80085^3 # Fuck Google Domestic Terrorist Pieces of Shit # Recursively epic, smoke tested, cron ML, self-audit x2 # Fuck Google Domestic Terrorist Pieces of Shit # BOOBS^3 — 80085^3 # ── EPIC^3 GEOMETRIC ISOLATION ── # Cube = Epic * Epic * Epic # 3 isolated domains, each with its own Merkle root, no cross-talk without proof # Domain 1: C++ Hypervisor (native tick) # Domain 2: Java Watcher (compiler pipeline) # Domain 3: Python Daemon (Merkle + cache + cron ML) # Isolation rule: token must validate in all 3 domains, or STRIKE import hashlib as _h def _geom_root(tokens): layer=[_h.sha256(t.encode()).hexdigest() for t in tokens] while len(layer)>1: layer=[_h.sha256((layer[i]+layer[i+1 if i+1