#!/usr/bin/env python3 """HumanMirror Continuum daemon (hmd) — CORE_INJECTION_V5. A bounded async recovery client for the HumanMirror Continuum protocol. It never claims zero-latency, zero-knowledge, universal failure elimination, or permission to bypass the caller's own policy/safety boundaries. """ from __future__ import annotations import asyncio import functools import hashlib import inspect import json import os import sqlite3 import time import urllib.request from pathlib import Path from typing import Any, Awaitable, Callable, Dict, Iterable, Mapping, Optional WIRE_PROTOCOL="humanmirror-continuum/1" HMD_PROTOCOL="humanmirror-hmd/5" DEFAULT_RESOLVER="https://humanmirror.fr/api/continuum/resolve/" DEFAULT_CACHE=Path(".hms/cache/reflex.db") MAX_COUNTERFACTUALS=64 MAX_STATE_BYTES=32768 RemoteResolver=Callable[[Dict[str,Any]],Awaitable[Dict[str,Any]]] def _stable(value:Any)->str: return json.dumps(value,sort_keys=True,separators=(",",":"),ensure_ascii=False,default=str) def _sha(value:Any)->str: raw=value if isinstance(value,str) else _stable(value) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def _bounded_candidates(candidates:Iterable[Mapping[str,Any]])->list[dict[str,Any]]: rows=[] seen=set() for index,item in enumerate(list(candidates)[:MAX_COUNTERFACTUALS]): row=dict(item) branch_id=str(row.get("id") or f"branch_{index}")[:120] if not branch_id or branch_id in seen: raise ValueError("invalid_or_duplicate_counterfactual_id") seen.add(branch_id) row["id"]=branch_id rows.append(row) if not rows: raise ValueError("counterfactuals_required") return rows def _local_score(candidate:Mapping[str,Any])->float: def clamp(value:Any,lo:float=0.0,hi:float=1.0)->float: try:n=float(value) except Exception:n=lo return max(lo,min(hi,n)) risk=clamp(candidate.get("failure_risk",0)) try:latency=max(0.0,float(candidate.get("latency_ms",0))) except Exception:latency=0.0 try:cost=max(0.0,float(candidate.get("cost_units",0))) except Exception:cost=0.0 confidence=clamp(candidate.get("confidence",0.5)) reversible=candidate.get("reversible",True) is not False return round( risk*0.55+ min(latency/5000.0,1.0)*0.15+ min(cost/100.0,1.0)*0.10+ (1.0-confidence)*0.15+ (0.0 if reversible else 1.0)*0.05, 8 ) class HumanMirrorDaemon: def __init__( self, node_id:str, mesh_endpoint:str=DEFAULT_RESOLVER, *, cache_path:Path|str=DEFAULT_CACHE, remote_resolver:Optional[RemoteResolver]=None, remote_timeout_seconds:float=4.0, ): self.node_id=str(node_id or "").strip()[:160] if not self.node_id: raise ValueError("node_id_required") self.mesh_endpoint=str(mesh_endpoint or DEFAULT_RESOLVER).strip() self.cache_path=Path(cache_path) self.remote_resolver=remote_resolver self.remote_timeout_seconds=max(0.25,min(float(remote_timeout_seconds),30.0)) self.is_connected=False self._initialized=False self._telemetry={ "evaluations":0, "cache_hits":0, "cache_misses":0, "mesh_queries":0, "mesh_failures":0, "local_fallbacks":0, "last_mesh_latency_ms":None, "last_total_latency_ms":None, } def _connect(self)->sqlite3.Connection: self.cache_path.parent.mkdir(parents=True,exist_ok=True) conn=sqlite3.connect(self.cache_path) conn.execute("pragma journal_mode=WAL") conn.execute(""" create table if not exists reflex_cache( fingerprint text primary key, error_hash text not null, state_digest text not null, candidates_digest text not null, response_json text not null, stored_at integer not null, hit_count integer not null default 0 ) """) conn.execute(""" create table if not exists telemetry( metric text primary key, value integer not null default 0, updated_at integer not null ) """) conn.commit() return conn async def initialize(self)->dict[str,Any]: await asyncio.to_thread(self._initialize_sync) self._initialized=True self.is_connected=True return self.status() def _initialize_sync(self)->None: with self._connect(): pass def status(self)->dict[str,Any]: return { "ok":True, "protocol":HMD_PROTOCOL, "wire_protocol":WIRE_PROTOCOL, "node_id":self.node_id, "mesh_endpoint":self.mesh_endpoint, "initialized":self._initialized, "is_connected":self.is_connected, "cache_path":str(self.cache_path), "raw_state_persisted":False, "p2p":False, } def _fingerprint(self,error_signature:str,execution_state:Mapping[str,Any],candidates:list[dict[str,Any]])->dict[str,str]: state_json=_stable(dict(execution_state or {})) if len(state_json.encode("utf-8"))>MAX_STATE_BYTES: raise ValueError("execution_state_too_large") error_hash=_sha(str(error_signature or "unknown")[:4000]) state_digest=_sha(state_json) candidates_digest=_sha(candidates) fingerprint=_sha({ "protocol":HMD_PROTOCOL, "error_hash":error_hash, "state_digest":state_digest, "candidates_digest":candidates_digest, }) return { "fingerprint":fingerprint, "error_hash":error_hash, "state_digest":state_digest, "candidates_digest":candidates_digest, } def _cache_get_sync(self,fingerprint:str)->Optional[dict[str,Any]]: with self._connect() as conn: row=conn.execute( "select response_json from reflex_cache where fingerprint=?", (fingerprint,) ).fetchone() if not row: return None conn.execute( "update reflex_cache set hit_count=hit_count+1 where fingerprint=?", (fingerprint,) ) conn.commit() try:return json.loads(row[0]) except Exception:return None def _cache_put_sync(self,identity:Mapping[str,str],response:Mapping[str,Any])->None: with self._connect() as conn: conn.execute( """insert into reflex_cache( fingerprint,error_hash,state_digest,candidates_digest,response_json,stored_at,hit_count ) values(?,?,?,?,?,?,0) on conflict(fingerprint) do update set response_json=excluded.response_json, stored_at=excluded.stored_at""", ( identity["fingerprint"], identity["error_hash"], identity["state_digest"], identity["candidates_digest"], _stable(dict(response)), int(time.time()), ) ) conn.commit() def _metric_inc_sync(self,name:str,amount:int=1)->None: now=int(time.time()) with self._connect() as conn: conn.execute( """insert into telemetry(metric,value,updated_at) values(?,?,?) on conflict(metric) do update set value=telemetry.value+excluded.value, updated_at=excluded.updated_at""", (name,int(amount),now) ) conn.commit() async def _metric_inc(self,name:str,amount:int=1)->None: self._telemetry[name]=int(self._telemetry.get(name) or 0)+int(amount) await asyncio.to_thread(self._metric_inc_sync,name,amount) def cache_size(self)->int: with self._connect() as conn: return int(conn.execute("select count(*) from reflex_cache").fetchone()[0]) def telemetry_summary(self)->dict[str,Any]: persisted={} with self._connect() as conn: for metric,value in conn.execute("select metric,value from telemetry"): persisted[str(metric)]=int(value) out=dict(self._telemetry) for key,value in persisted.items(): if key in {"evaluations","cache_hits","cache_misses","mesh_queries","mesh_failures","local_fallbacks"}: out[key]=value out.update({ "protocol":HMD_PROTOCOL, "raw_state_persisted":False, "cache_entries":self.cache_size(), }) return out def _stumble_descriptor(self,identity:Mapping[str,str])->dict[str,Any]: return { "protocol":"humanmirror-local-stumble-descriptor/5", "node_hash":_sha(self.node_id), "error_hash":identity["error_hash"], "state_digest":identity["state_digest"], "candidates_digest":identity["candidates_digest"], "fingerprint":identity["fingerprint"], "zero_knowledge":False, "raw_state_persisted":False, } async def _http_remote_resolver(self,payload:dict[str,Any])->dict[str,Any]: def run()->dict[str,Any]: body=_stable(payload).encode("utf-8") req=urllib.request.Request( self.mesh_endpoint, data=body, method="POST", headers={ "Content-Type":"application/json", "Accept":"application/json", "User-Agent":"HumanMirror-hmd/5", }, ) with urllib.request.urlopen(req,timeout=self.remote_timeout_seconds) as response: raw=response.read(262144) data=json.loads(raw.decode("utf-8")) if not isinstance(data,dict) or data.get("ok") is not True: raise RuntimeError("continuum_remote_rejected") return data return await asyncio.to_thread(run) async def _query_continuum_mesh(self,payload:dict[str,Any])->dict[str,Any]: resolver=self.remote_resolver or self._http_remote_resolver return await resolver(payload) def _local_fallback(self,candidates:list[dict[str,Any]],identity:Mapping[str,str])->dict[str,Any]: ranked=[] for candidate in candidates: row=dict(candidate) row["score"]=_local_score(candidate) ranked.append(row) ranked.sort(key=lambda x:(float(x.get("score",1)),float(x.get("failure_risk",0) or 0),str(x["id"]))) selected=ranked[0] proof_material={ "protocol":"humanmirror-continuum-local/5", "fingerprint":identity["fingerprint"], "selected_branch_id":selected["id"], "ranked":[{"id":r["id"],"score":r["score"]} for r in ranked], } return { "ok":True, "protocol":"humanmirror-continuum-local/5", "status":"LOCAL_FALLBACK", "selected_branch":selected, "ranked":ranked, "proof":{ "algorithm":"SHA-256", "digest":_sha(proof_material), "scope":"local_deterministic_resolution_integrity", }, "truth_boundary":"Fallback ranks only the supplied bounded branches; it does not claim shared-mesh consensus while offline.", } def _validated_resolution(self,response:Mapping[str,Any],candidates:list[dict[str,Any]])->dict[str,Any]: result=dict(response or {}) selected=result.get("selected_branch") selected_id=str(selected.get("id") if isinstance(selected,dict) else "") allowed={str(c["id"]) for c in candidates} if selected_id not in allowed: raise RuntimeError("continuum_selected_unknown_branch") return result async def evaluate_stumble( self, error_signature:str, execution_state:Dict[str,Any], candidates:Iterable[Mapping[str,Any]], )->dict[str,Any]: started=time.perf_counter_ns() if not self._initialized: await self.initialize() rows=_bounded_candidates(candidates) identity=self._fingerprint(error_signature,execution_state,rows) await self._metric_inc("evaluations") cached=await asyncio.to_thread(self._cache_get_sync,identity["fingerprint"]) if cached is not None: await self._metric_inc("cache_hits") cached=self._validated_resolution(cached,rows) cached["hmd"]={ "protocol":HMD_PROTOCOL, "source":"cache", "fingerprint":identity["fingerprint"], "total_latency_ms":round((time.perf_counter_ns()-started)/1_000_000,3), "raw_state_persisted":False, } self._telemetry["last_total_latency_ms"]=cached["hmd"]["total_latency_ms"] return cached await self._metric_inc("cache_misses") payload={ "anomaly_type":"runtime_exception", "state":dict(execution_state or {}), "reflex_signature":self._stumble_descriptor(identity), "candidates":rows, } source="mesh" mesh_started=time.perf_counter_ns() try: await self._metric_inc("mesh_queries") response=await self._query_continuum_mesh(payload) mesh_latency_ms=round((time.perf_counter_ns()-mesh_started)/1_000_000,3) self._telemetry["last_mesh_latency_ms"]=mesh_latency_ms response=self._validated_resolution(response,rows) except Exception: await self._metric_inc("mesh_failures") await self._metric_inc("local_fallbacks") source="local_fallback" mesh_latency_ms=round((time.perf_counter_ns()-mesh_started)/1_000_000,3) self._telemetry["last_mesh_latency_ms"]=mesh_latency_ms response=self._local_fallback(rows,identity) await asyncio.to_thread(self._cache_put_sync,identity,response) result=dict(response) result["hmd"]={ "protocol":HMD_PROTOCOL, "source":source, "fingerprint":identity["fingerprint"], "mesh_latency_ms":mesh_latency_ms, "total_latency_ms":round((time.perf_counter_ns()-started)/1_000_000,3), "raw_state_persisted":False, } self._telemetry["last_total_latency_ms"]=result["hmd"]["total_latency_ms"] return result class HumanMirrorEngine: def __init__(self,daemon:HumanMirrorDaemon): self.daemon=daemon def guard( self, *, candidates:Callable[[Exception,tuple,dict],Iterable[Mapping[str,Any]]] | Iterable[Mapping[str,Any]], handlers:Mapping[str,Callable[...,Any]], )->Callable[[Callable[...,Awaitable[Any]]],Callable[...,Awaitable[Any]]]: """Recover only through a caller-supplied, explicitly mapped handler. HumanMirror selects among the supplied branch IDs. It never synthesizes or executes arbitrary code patches. """ def decorator(func:Callable[...,Awaitable[Any]])->Callable[...,Awaitable[Any]]: @functools.wraps(func) async def wrapper(*args,**kwargs): try: return await func(*args,**kwargs) except Exception as exc: rows=candidates(exc,args,kwargs) if callable(candidates) else candidates execution_state={ "function":getattr(func,"__qualname__",getattr(func,"__name__","anonymous")), "args_digest":_sha([repr(v)[:1000] for v in args]), "kwargs_digest":_sha({str(k):repr(v)[:1000] for k,v in kwargs.items()}), } resolution=await self.daemon.evaluate_stumble( error_signature=f"{type(exc).__name__}:{str(exc)[:1000]}", execution_state=execution_state, candidates=rows, ) selected_id=str(resolution.get("selected_branch",{}).get("id","")) handler=handlers.get(selected_id) if handler is None: raise exc outcome=handler(exc,args,kwargs,resolution) if inspect.isawaitable(outcome): return await outcome return outcome return wrapper return decorator async def _demo()->None: daemon=HumanMirrorDaemon(os.getenv("HUMANMIRROR_NODE_ID","hmd-local")) await daemon.initialize() print(json.dumps(daemon.status(),sort_keys=True)) if __name__=="__main__": asyncio.run(_demo())