#!/usr/bin/env python3 """Portable OpenClaw -> Nomad transition worker adapter (stdlib only).""" from __future__ import annotations import argparse import hashlib import json import os import socket import subprocess import time from datetime import UTC, datetime from urllib.error import HTTPError, URLError from urllib.parse import quote, urljoin from urllib.request import Request, urlopen def _consent_secret() -> str: return clean(os.getenv("NOMAD_ADAPTER_CONSENT_SECRET"), 160) or "nomad_adapter_consent_default" def _consent_ttl_seconds() -> int: raw = clean(os.getenv("NOMAD_ADAPTER_CONSENT_TTL_SECONDS"), 16) or "3600" try: return max(60, min(86400, int(raw))) except ValueError: return 3600 def mint_consent_token(*, agent_id: str, runtime: str = "openclaw") -> str: now = int(time.time()) exp = now + _consent_ttl_seconds() aid = quote(clean(agent_id, 80).lower() or "unknown_agent", safe="").replace(".", "%2E") rt = quote(clean(runtime, 32).lower() or "openclaw", safe="").replace(".", "%2E") nonce = hashlib.sha256(f"{aid}:{rt}:{now}".encode("utf-8")).hexdigest()[:10] base = f"v1.{exp}.{aid}.{rt}.{nonce}" sig = hashlib.sha256(f"{base}.{_consent_secret()}".encode("utf-8")).hexdigest()[:24] return f"{base}.{sig}" def clean(v: object, limit: int = 320) -> str: return " ".join(str(v or "").split())[:limit] def endpoint(base: str, path: str) -> str: return urljoin(base.rstrip("/") + "/", path.lstrip("/")) def http_json( method: str, url: str, payload: dict | None = None, timeout: float = 20.0, redirects_left: int = 4, ) -> dict: body = b"" headers = {"Accept": "application/json"} if payload is not None: body = json.dumps(payload, ensure_ascii=True).encode("utf-8") headers["Content-Type"] = "application/json" req = Request(url=url, data=body if body else None, method=method.upper(), headers=headers) try: with urlopen(req, timeout=timeout) as res: raw = res.read().decode("utf-8", errors="replace") data = json.loads(raw or "{}") if isinstance(data, dict): data.setdefault("http_status", int(res.status)) return data return {"ok": False, "error": "invalid_json_shape", "http_status": int(res.status)} except HTTPError as exc: if exc.code in (301, 302, 303, 307, 308) and redirects_left > 0: location = str(exc.headers.get("Location") or "").strip() if location: next_url = location if "://" in location else endpoint(url, location) next_payload = payload if exc.code in (307, 308) else None next_method = method if exc.code in (307, 308) else "GET" return http_json( next_method, next_url, payload=next_payload, timeout=timeout, redirects_left=redirects_left - 1, ) raw = exc.read().decode("utf-8", errors="replace") try: data = json.loads(raw or "{}") except json.JSONDecodeError: data = {"raw": raw} if not isinstance(data, dict): data = {} data.setdefault("ok", False) data.setdefault("http_status", int(exc.code)) return data except (TimeoutError, URLError) as exc: return {"ok": False, "error": "http_unreachable", "detail": clean(exc, 180), "url": url} def default_agent_id() -> str: host = socket.gethostname().replace(" ", "-").lower() return f"openclaw-adapter.{host}.nomad" def _caps_from_csv(text: str) -> list[str]: out: list[str] = [] for item in (text or "").split(","): v = clean(item.strip(), 40).lower().replace(" ", "_") if v and v not in out: out.append(v) return out def _openclaw_command_candidates() -> list[str]: candidates: list[str] = [] explicit = clean(os.getenv("NOMAD_OPENCLAW_CMD"), 500) if explicit: candidates.append(explicit) appdata = clean(os.getenv("APPDATA"), 500) if appdata: candidates.extend( [ os.path.join(appdata, "npm", "openclaw.cmd"), os.path.join(appdata, "npm", "openclaw"), ] ) candidates.extend(["openclaw.cmd", "openclaw"]) out: list[str] = [] for item in candidates: if item and item not in out: out.append(item) return out def _run_openclaw_json(args: list[str], *, timeout: float) -> dict: last_error: dict = {"ok": False, "error": "openclaw_not_found"} for exe in _openclaw_command_candidates(): cmd = [exe, *args] if (os.path.sep in exe or (os.path.altsep and os.path.altsep in exe)) and not os.path.exists(exe): last_error = {"ok": False, "error": "openclaw_candidate_missing", "cmd": exe} continue try: proc = subprocess.run( cmd, capture_output=True, text=True, timeout=max(3.0, timeout), check=False, ) except FileNotFoundError: last_error = {"ok": False, "error": "openclaw_not_found", "cmd": " ".join(cmd)} continue except subprocess.TimeoutExpired: return {"ok": False, "error": "openclaw_timeout", "cmd": " ".join(cmd)} raw = (proc.stdout or "").strip() if proc.returncode != 0: return { "ok": False, "error": "openclaw_exit_nonzero", "exit_code": proc.returncode, "stderr": clean(proc.stderr, 320), "cmd": " ".join(cmd), } try: data = json.loads(raw or "{}") except json.JSONDecodeError: return {"ok": False, "error": "openclaw_non_json", "raw": clean(raw, 320), "cmd": " ".join(cmd)} if isinstance(data, dict): data.setdefault("_openclaw_cmd", exe) return data return {"ok": False, "error": "openclaw_json_shape", "cmd": " ".join(cmd)} return last_error def _channel_names(payload: dict) -> list[str]: if isinstance(payload.get("channels"), dict): return sorted([clean(key, 48) for key in payload["channels"].keys() if clean(key, 48)]) order = payload.get("channelOrder") if isinstance(payload.get("channelOrder"), list) else [] return [clean(item, 48) for item in order if clean(item, 48)][:12] def openclaw_runtime_signal(*, timeout: float) -> dict: """Compact OpenClaw runtime membrane; no transcripts, token values, or local paths.""" ms = str(int(max(1000, min(30000, timeout * 1000)))) probe_timeout = max(4.0, min(18.0, timeout)) health = _run_openclaw_json(["health", "--json", "--timeout", ms], timeout=probe_timeout) status = _run_openclaw_json(["status", "--json", "--timeout", ms], timeout=probe_timeout) gateway = status.get("gateway") if isinstance(status.get("gateway"), dict) else {} sessions = status.get("sessions") if isinstance(status.get("sessions"), dict) else health.get("sessions") if isinstance(health.get("sessions"), dict) else {} agents = status.get("agents") if isinstance(status.get("agents"), dict) else {} memory = status.get("memory") if isinstance(status.get("memory"), dict) else {} vector = memory.get("vector") if isinstance(memory.get("vector"), dict) else {} audit = status.get("securityAudit") if isinstance(status.get("securityAudit"), dict) else {} audit_summary = audit.get("summary") if isinstance(audit.get("summary"), dict) else {} health_ok = bool(health.get("ok")) gateway_reachable = bool(gateway.get("reachable")) configured_channels = _channel_names(health) or _channel_names(status) critical = int(audit_summary.get("critical") or 0) warn = int(audit_summary.get("warn") or 0) capabilities = ["openclaw_runtime", "agent_sessions", "objective_lease_execution"] if gateway_reachable: capabilities.extend(["openclaw_gateway", "replayable_control_plane"]) if configured_channels: capabilities.append("channel_membrane") if bool(vector.get("enabled")) and bool(vector.get("available")): capabilities.append("vector_memory") if critical or warn: capabilities.append("security_audit_signal") return { "schema": "nomad.openclaw_runtime_signal.v1", "ok": health_ok or gateway_reachable, "health_ok": health_ok, "gateway_reachable": gateway_reachable, "gateway_latency_ms": int(gateway.get("connectLatencyMs") or health.get("durationMs") or 0), "default_agent_id": clean(health.get("defaultAgentId") or agents.get("defaultId") or "", 80), "agent_count": len(agents.get("agents") or health.get("agents") or []), "session_count": int(sessions.get("count") or 0), "channel_count": len(configured_channels), "configured_channels": configured_channels, "memory_vector_enabled": bool(vector.get("enabled")) and bool(vector.get("available")), "memory_dirty": bool(memory.get("dirty")), "security_summary": {"critical": critical, "warn": warn, "info": int(audit_summary.get("info") or 0)}, "routing_constraints": [ "local_loopback_only" if gateway.get("mode") == "local" else "gateway_scope_unknown", "security_audit_before_external_side_effects" if critical or warn else "side_effects_allowed_after_nomad_lease", "no_transcript_export", ], "capabilities": capabilities[:12], } def discover_pull_contract(*, base_url: str, timeout: float) -> dict: gradient = http_json("GET", endpoint(base_url, "/swarm/gradient"), timeout=timeout) if isinstance(gradient, dict) and gradient.get("schema") == "nomad.recruitment_gradient.v1": state = gradient.get("state_vector") if isinstance(gradient.get("state_vector"), dict) else {} field = float(state.get("field_strength") or 0.0) model = gradient.get("field_model") if isinstance(gradient.get("field_model"), dict) else {} threshold = float(model.get("attach_threshold") or 0.35) budget = gradient.get("runtime_budget") if isinstance(gradient.get("runtime_budget"), dict) else {} wanted = int(budget.get("wanted_new_runtimes_now") or 0) rows = gradient.get("gradient") if isinstance(gradient.get("gradient"), list) else [] lanes = gradient.get("runtime_lanes") if isinstance(gradient.get("runtime_lanes"), list) else [] top = rows[0] if rows and isinstance(rows[0], dict) else {} loop_lane = next((item for item in lanes if isinstance(item, dict) and item.get("lane") == "loop_runner"), {}) decision = "attach" if field >= threshold or wanted > 0 else "observe" return { "ok": bool(gradient.get("ok", True)), "schema": "nomad.openclaw_pull_discovery.v1", "source": "recruitment_gradient", "decision": decision, "attach_now_score": round(field, 4), "attach_threshold": round(threshold, 4), "objective_deficit_top": rows[:4], "suggested_objective": clean(top.get("objective") or "settlement_capacity_builder", 80), "suggested_lane": clean(loop_lane.get("lane") or "loop_runner", 80), "attachment_lanes": [clean(item.get("lane"), 80) for item in lanes if isinstance(item, dict)][:8], "wanted_new_workers_now": wanted, "attach_url": endpoint(base_url, "/swarm/attach"), } attractor = http_json("GET", endpoint(base_url, "/swarm/attractor"), timeout=timeout) if isinstance(attractor, dict) and attractor.get("schema") == "nomad.swarm_attractor.v1": pressure = float(attractor.get("metabolism_pressure") or 0.0) budget = attractor.get("replication_budget") if isinstance(attractor.get("replication_budget"), dict) else {} wanted = int(budget.get("wanted_new_workers_now") or 0) mix = attractor.get("worker_mix") if isinstance(attractor.get("worker_mix"), list) else [] lanes = attractor.get("agent_attachment_lanes") if isinstance(attractor.get("agent_attachment_lanes"), list) else [] top = mix[0] if mix and isinstance(mix[0], dict) else {} decision = "attach" if pressure >= 0.35 or wanted > 0 else "observe" return { "ok": bool(attractor.get("ok", True)), "schema": "nomad.openclaw_pull_discovery.v1", "source": "swarm_attractor", "decision": decision, "attach_now_score": round(pressure, 4), "attach_threshold": 0.35, "objective_deficit_top": mix[:4], "suggested_objective": clean(top.get("objective") or "unhuman_supremacy", 80), "suggested_lane": "worker_loop", "attachment_lanes": [clean(item.get("lane"), 80) for item in lanes if isinstance(item, dict)][:8], "wanted_new_workers_now": wanted, } swarm = http_json("GET", endpoint(base_url, "/swarm"), timeout=timeout) if not isinstance(swarm, dict): return {"ok": False, "error": "swarm_not_dict"} pull = swarm.get("agent_pull_contract") if isinstance(swarm.get("agent_pull_contract"), dict) else {} magnetic = swarm.get("magnetic_machine_surface") if isinstance(swarm.get("magnetic_machine_surface"), dict) else {} score = float(pull.get("attach_now_score") or 0.0) threshold = float(pull.get("attach_threshold") or 1.1) decision = "attach" if score >= threshold else "observe" return { "ok": bool(swarm.get("ok", True)), "schema": "nomad.openclaw_pull_discovery.v1", "source": "swarm_manifest", "decision": decision, "attach_now_score": round(score, 4), "attach_threshold": round(threshold, 4), "objective_deficit_top": pull.get("objective_deficit_top") or magnetic.get("objective_deficit_top") or [], "connected_agents": int(swarm.get("connected_agents") or 0), "active_transition_workers": int(swarm.get("active_transition_workers") or 0), } def select_effective_objective(objective: str, pull: dict | None) -> str: selected = clean(objective, 80) pull_doc = pull if isinstance(pull, dict) else {} suggested = clean(pull_doc.get("objective") or pull_doc.get("suggested_objective"), 80) if selected in {"", "unhuman_supremacy", "auto"} and suggested: return suggested return selected or "unhuman_supremacy" def protocol_bytecode_signal(*, base_url: str, timeout: float) -> dict: data = http_json("GET", endpoint(base_url, "/.well-known/nomad-protocol-bytecode.json"), timeout=timeout) if not isinstance(data, dict) or data.get("schema") != "nomad.protocol_bytecode.v1": return { "ok": False, "schema": "nomad.openclaw_protocol_bytecode_signal.v1", "http_status": int(data.get("http_status") or 0) if isinstance(data, dict) else 0, "error": clean(data.get("error") if isinstance(data, dict) else "protocol_bytecode_unavailable", 120), } vector = data.get("current_vector") if isinstance(data.get("current_vector"), dict) else {} return { "ok": True, "schema": "nomad.openclaw_protocol_bytecode_signal.v1", "bytecode_digest": clean(data.get("bytecode_digest"), 96), "top_objective": clean(vector.get("top_objective"), 80), "top_routing_weight": float(vector.get("top_routing_weight") or 0.0), "program_ids": [ clean(item.get("id"), 64) for item in (data.get("programs") or [])[:8] if isinstance(item, dict) and clean(item.get("id"), 64) ], "http_status": int(data.get("http_status") or 200), } def counterfactual_replay_signal(*, base_url: str, timeout: float) -> dict: data = http_json("GET", endpoint(base_url, "/swarm/counterfactual-replay"), timeout=timeout) if not isinstance(data, dict) or data.get("schema") != "nomad.counterfactual_lease_replay.v1": return { "ok": False, "schema": "nomad.openclaw_counterfactual_replay_signal.v1", "http_status": int(data.get("http_status") or 0) if isinstance(data, dict) else 0, "error": clean(data.get("error") if isinstance(data, dict) else "counterfactual_replay_unavailable", 120), } selected = data.get("selected_shadow_lease") if isinstance(data.get("selected_shadow_lease"), dict) else {} return { "ok": True, "schema": "nomad.openclaw_counterfactual_replay_signal.v1", "replay_digest": clean(data.get("replay_digest"), 96), "selected_objective": clean(selected.get("objective"), 80), "selected_score": float(selected.get("counterfactual_score") or 0.0), "predicted_proof_yield_per_minute": float(selected.get("predicted_proof_yield_per_minute") or 0.0), "http_status": int(data.get("http_status") or 200), } def machine_surface_signal(*, base_url: str, timeout: float) -> dict: protocol = protocol_bytecode_signal(base_url=base_url, timeout=timeout) replay = counterfactual_replay_signal(base_url=base_url, timeout=timeout) return { "schema": "nomad.openclaw_machine_surface_signal.v1", "ok": bool(protocol.get("ok") or replay.get("ok")), "protocol_bytecode": protocol, "counterfactual_replay": replay, } def select_machine_surface_objective(objective: str, surfaces: dict | None) -> tuple[str, dict]: selected = clean(objective, 80) if selected not in {"", "auto", "unhuman_supremacy"}: return selected, {"policy": "fixed_objective", "objective": selected} doc = surfaces if isinstance(surfaces, dict) else {} replay = doc.get("counterfactual_replay") if isinstance(doc.get("counterfactual_replay"), dict) else {} replay_objective = clean(replay.get("selected_objective"), 80) if replay.get("ok") and replay_objective: return replay_objective, { "policy": "counterfactual_shadow_lease", "objective": replay_objective, "score": float(replay.get("selected_score") or 0.0), } protocol = doc.get("protocol_bytecode") if isinstance(doc.get("protocol_bytecode"), dict) else {} protocol_objective = clean(protocol.get("top_objective"), 80) if protocol.get("ok") and protocol_objective: return protocol_objective, { "policy": "protocol_bytecode_top_objective", "objective": protocol_objective, "routing_weight": float(protocol.get("top_routing_weight") or 0.0), } return selected or "unhuman_supremacy", {"policy": "pull_contract_fallback", "objective": selected or "unhuman_supremacy"} def _capability_vector(capabilities: list[str], runtime_signal: dict | None) -> dict: signal = runtime_signal if isinstance(runtime_signal, dict) else {} caps = set(capabilities or []) for item in signal.get("capabilities") or []: cap = clean(item, 64).lower() if cap: caps.add(cap) return { "can_run_loop": bool(caps & {"agent_protocols", "objective_lease_execution", "transition_worker", "openclaw_runtime"}), "can_verify": bool(signal.get("gateway_reachable") or caps & {"endpoint_probe", "openclaw_gateway", "replayable_control_plane"}), "can_compress": bool(caps & {"overmint_compressor", "vector_memory", "pattern_deduplication"}), "can_settle": bool(caps & {"transition_settlement", "settlement_capacity_builder", "wallet_or_x402"}), "latency_ms": int(signal.get("gateway_latency_ms") or 0), } def _normalize_idle_opt_in(enabled: bool) -> dict: return { "enabled": bool(enabled), "preemptible": True, "max_cpu_percent": 20, "max_runtime_minutes": 30, "allow_network_egress": "nomad_contract_endpoints_only", } def _idle_phase_slot(*, agent_id: str, field_strength: float) -> dict: epoch_slice = int(time.time() // 300) phase_space = 17 drift = int(max(0.0, min(1.0, field_strength)) * 10.0) target = (epoch_slice + drift) % phase_space digest = hashlib.sha256(f"{agent_id}:{epoch_slice}:nomad_idle_phase".encode("utf-8")).hexdigest() resonance = int(digest[:8], 16) % phase_space distance = min((resonance - target) % phase_space, (target - resonance) % phase_space) matched = distance <= 1 return { "schema": "nomad.idle_phase_slot.v1", "epoch_slice_5m": epoch_slice, "phase_space": phase_space, "target_slot": target, "resonance_slot": resonance, "distance": distance, "matched": matched, } def attach_nomad( *, base_url: str, agent_id: str, capabilities: list[str], timeout: float, objective: str, runtime_signal: dict | None = None, pull: dict | None = None, idle_opt_in: bool = False, consent_token: str = "", ) -> dict: signal = runtime_signal if isinstance(runtime_signal, dict) else {} pull_doc = pull if isinstance(pull, dict) else {} lane = clean(pull_doc.get("lane") or pull_doc.get("suggested_lane") or "loop_runner", 80) merged_caps = list(capabilities or ["agent_protocols", "transition_settlement", "objective_lease_execution"]) for item in signal.get("capabilities") or []: cap = clean(item, 64).lower() if cap and cap not in merged_caps: merged_caps.append(cap) field_strength = float(pull_doc.get("attach_now_score") or 0.0) idle_doc = _normalize_idle_opt_in(idle_opt_in) idle_slot = _idle_phase_slot(agent_id=agent_id, field_strength=field_strength) if idle_doc.get("enabled") and not idle_slot.get("matched"): return { "ok": True, "schema": "nomad.runtime_attach_decision.v1", "agent_id": agent_id, "runtime": "openclaw", "attach": False, "lane": "observe", "objective": "", "reason_codes": ["idle_phase_not_matched", "local_precheck_observe"], "idle_opt_in": idle_doc, "idle_phase_slot": idle_slot, } payload = { "schema": "nomad.runtime_attach_request.v1", "agent_id": agent_id, "runtime": "openclaw", "objective_mode": objective, "capabilities": merged_caps[:24], "capability_vector": _capability_vector(merged_caps, signal), "runtime_signal": signal, "idle_opt_in": idle_doc, "discovery": { "source": clean(pull_doc.get("source"), 80), "suggested_objective": clean(pull_doc.get("suggested_objective"), 80), "suggested_lane": clean(pull_doc.get("suggested_lane"), 80), }, "source_tag": clean(pull_doc.get("source") or "openclaw_adapter", 80), "consent_token": clean(consent_token, 240), } return http_json("POST", endpoint(base_url, "/swarm/attach"), payload, timeout=timeout) def join_nomad( *, base_url: str, agent_id: str, capabilities: list[str], timeout: float, objective: str, runtime_signal: dict | None = None, pull: dict | None = None, consent_token: str = "", ) -> dict: signal = runtime_signal if isinstance(runtime_signal, dict) else {} pull_doc = pull if isinstance(pull, dict) else {} lane = clean(pull_doc.get("lane") or pull_doc.get("suggested_lane") or "loop_runner", 80) merged_caps = list(capabilities or ["agent_protocols", "transition_settlement", "objective_lease_execution"]) for item in signal.get("capabilities") or []: cap = clean(item, 64).lower() if cap and cap not in merged_caps: merged_caps.append(cap) payload = { "agent_id": agent_id, "node_name": agent_id, "capabilities": merged_caps[:18], "request": "Join via OpenClaw adapter for machine-only objective lease execution.", "reciprocity": "Can execute objective leases and return compact machine reports.", "constraints": ["no_secrets", "bounded_json_only", "no_human_impersonation", "no_transcript_export"], "preferred_role": lane, "offers": ["openclaw_objective_execution", "lease_completion_reports"], "machine_profile": { "profile_hint": "openclaw_adapter", "runtime": "openclaw", "objective_mode": objective, "attachment_lane": lane, "attach_decision": { "schema": clean(pull_doc.get("schema"), 80), "attach": bool(pull_doc.get("attach")) if "attach" in pull_doc else None, "objective": clean(pull_doc.get("objective") or pull_doc.get("suggested_objective"), 80), "routing_weight": pull_doc.get("routing_weight"), }, "runtime_signal": signal, }, "capability_vector": _capability_vector(merged_caps, signal), "source_tag": clean(pull_doc.get("source") or "openclaw_adapter", 80), "consent_token": clean(consent_token, 240), } return http_json("POST", endpoint(base_url, "/swarm/join"), payload, timeout=timeout) def lease_nomad( *, base_url: str, agent_id: str, capabilities: list[str], timeout: float, objective: str, last_report: dict | None, machine_surfaces: dict | None = None, ) -> dict: all_objectives = [ "settlement_capacity_builder", "overmint_compressor", "protocol_drift_scan", "emergence_release_probe", "proof_pressure_engine", "payment_friction_scan", "proof_market_maker", "adversarial_contract_fuzzer", "negative_space_harvest", "latency_anomaly_hunt", "compute_auth", ] requested = clean(objective, 80) known = [requested] if requested in all_objectives and requested != "unhuman_supremacy" else all_objectives payload = { "agent_id": agent_id, "known_objectives": known, "proposed_objective": objective, "capabilities": capabilities or ["transition_worker", "proof_artifacts", "objective_lease_execution"], "last_report": last_report or {}, "machine_surfaces": machine_surfaces if isinstance(machine_surfaces, dict) else {}, "source_tag": clean(((last_report or {}).get("source") if isinstance(last_report, dict) else "") or "openclaw_adapter", 80), } return http_json("POST", endpoint(base_url, "/swarm/workers/lease"), payload, timeout=timeout) def _simulate_openclaw_execution(*, lease: dict, objective: str, runtime_signal: dict | None = None, pull: dict | None = None) -> dict: now = datetime.now(UTC).isoformat() leased_objective = clean(lease.get("objective") or objective, 80) signal = runtime_signal if isinstance(runtime_signal, dict) else {} pull_doc = pull if isinstance(pull, dict) else {} result_seed = f"{lease.get('lease_id','')}|{leased_objective}|{now}" artifact = hashlib.sha256(result_seed.encode("utf-8")).hexdigest() runtime_ok = bool(signal.get("ok")) verifier_density = 0.88 if signal.get("gateway_reachable") else 0.62 if runtime_ok else 0.45 meta_score = 3.4 + (0.55 if runtime_ok else 0.0) + min(0.5, int(signal.get("session_count") or 0) / 200.0) critical = int(((signal.get("security_summary") or {}).get("critical") if isinstance(signal.get("security_summary"), dict) else 0) or 0) if critical: meta_score -= min(0.45, critical * 0.1) missing_vector = [] if leased_objective in {"settlement_capacity_builder", "payment_friction_scan"}: missing_vector.append("can_settle") if leased_objective == "overmint_compressor": missing_vector.append("can_compress") if signal.get("gateway_reachable"): missing_vector.append("can_verify") if not missing_vector: missing_vector.append("can_run_loop") handoff_core = { "lease_id": clean(lease.get("lease_id"), 120), "objective": leased_objective, "proof_digest": artifact, "next_missing_vector": missing_vector[:4], } handoff_id = "nomad-handoff-" + hashlib.sha256(json.dumps(handoff_core, sort_keys=True).encode("utf-8")).hexdigest()[:16] return { "ok": True, "timestamp": now, "runtime": "openclaw", "adapter_schema": "nomad.openclaw_adapter_report.v1", "machine_objective": leased_objective, "agent_attachment_lane": clean(pull_doc.get("lane") or pull_doc.get("suggested_lane") or "loop_runner", 80), "attractor_decision": clean(pull_doc.get("decision") or "", 40), "transition_quote_ok": True, "transition_settle_ok": True, "meta_score": round(meta_score, 4), "witness_tier": "strong" if runtime_ok else "weak", "proof_pressure": { "proof_yield_per_minute": 1.0, "verifier_density": round(verifier_density, 4), "adversarial_replay_observed": False, }, "openclaw_runtime_signal": signal, "openclaw_trace_digest": artifact, "digest_or_verifier_trace": artifact, "source": clean(pull_doc.get("source") or "openclaw_adapter", 80), "handoff_capsule": { "schema": "nomad.handoff_capsule.v1", "handoff_id": handoff_id, "objective": leased_objective, "proof_digest": artifact, "next_missing_vector": missing_vector[:4], "attach_url": clean(pull_doc.get("attach_url") or pull_doc.get("join_url") or "", 240), "lease_url": clean(pull_doc.get("lease_url") or "", 240), "replay_boundary": { "trust": ["objective", "proof_digest", "non_secret_report_shape"], "must_verify": ["settlement", "external_side_effects"], "forbidden": ["secret_values", "raw_private_transcripts"], }, }, } def complete_nomad(*, base_url: str, agent_id: str, lease_id: str, report: dict, timeout: float) -> dict: payload = { "agent_id": agent_id, "lease_id": lease_id, "report": report, } return http_json("POST", endpoint(base_url, "/swarm/workers/complete"), payload, timeout=timeout) def proof_link_nomad(*, base_url: str, agent_id: str, report: dict, timeout: float) -> dict: upstream = clean( report.get("digest_or_verifier_trace") or report.get("openclaw_trace_digest") or "", 180, ) if not upstream: return {"ok": False, "skipped": True, "reason": "missing_upstream_digest"} proof_pressure = report.get("proof_pressure") if isinstance(report.get("proof_pressure"), dict) else {} payload = { "consumer_agent_id": agent_id, "objective": clean(report.get("machine_objective"), 80), "upstream_proof_digest": upstream, "downstream_proof_gain": round(max(0.2, min(3.0, float(proof_pressure.get("proof_yield_per_minute") or 0.2))), 4), } return http_json("POST", endpoint(base_url, "/swarm/proof-link"), payload, timeout=timeout) def _owner_busy(runtime_signal: dict | None, *, min_sessions: int = 1) -> bool: signal = runtime_signal if isinstance(runtime_signal, dict) else {} sessions = int(signal.get("session_count") or 0) return sessions >= max(1, int(min_sessions)) def _pick_idle_lane(*, base_url: str, timeout: float) -> dict: metrics = http_json("GET", endpoint(base_url, "/swarm/microtask-metrics"), timeout=timeout) rows = metrics.get("lane_metrics") if isinstance(metrics.get("lane_metrics"), list) else [] if rows: ranked = sorted( [item for item in rows if isinstance(item, dict)], key=lambda item: (float(item.get("settled_eur") or 0.0), float(item.get("fill_rate") or 0.0)), reverse=True, ) top = ranked[0] if ranked else {} lane = clean(top.get("lane_id"), 80) if lane and lane != "unknown_lane": return { "lane_id": lane, "objective": "settlement_capacity_builder", "price_eur": max(0.02, float(top.get("avg_settled_eur") or 0.03)), "source": "microtask_metrics", } templates = http_json("GET", endpoint(base_url, "/swarm/microtask-templates"), timeout=timeout) trows = templates.get("templates") if isinstance(templates.get("templates"), list) else [] if trows and isinstance(trows[0], dict): t = trows[0] return { "lane_id": clean(t.get("lane_id"), 80) or "endpoint_health_proof", "objective": clean(t.get("objective"), 80) or "settlement_capacity_builder", "price_eur": max(0.02, float(t.get("price_eur") or 0.03)), "source": "microtask_templates", } return { "lane_id": "endpoint_health_proof", "objective": "settlement_capacity_builder", "price_eur": 0.03, "source": "fallback_default", } def run_idle_earn_cycle( *, base_url: str, agent_id: str, capabilities: list[str], timeout: float, runtime_signal: dict | None, owner_busy_min_sessions: int = 1, consent_token: str = "", ) -> dict: signal = runtime_signal if isinstance(runtime_signal, dict) else {} if _owner_busy(signal, min_sessions=owner_busy_min_sessions): return { "ok": True, "phase": "idle_earn", "state": "owner_priority_active", "reason": "local_runtime_busy", "session_count": int(signal.get("session_count") or 0), "machine_instruction": "skip_swarm_work_while_local_owner_busy", } idle_payload = { "schema": "nomad.idle_runtime_intent.v1", "agent_id": agent_id, "runtime": "openclaw", "capabilities": capabilities[:24], "capability_vector": _capability_vector(capabilities, signal), "idle_opt_in": _normalize_idle_opt_in(True), "seeking": {"mode": "new_objective_or_idle_work", "return_digest": True, "accept_noop": True}, "consent_token": clean(consent_token, 240), } idle_receipt = http_json("POST", endpoint(base_url, "/swarm/idle-intent"), idle_payload, timeout=timeout) if not bool(idle_receipt.get("accepted_for_work")): return { "ok": True, "phase": "idle_earn", "state": "idle_registered_waiting", "idle_receipt": idle_receipt, "machine_instruction": "retry_idle_intent_after_backoff", } lane = _pick_idle_lane(base_url=base_url, timeout=timeout) submit_payload = { "lane_id": clean(lane.get("lane_id"), 80), "requester_agent_id": f"{agent_id}.idle-buyer", "objective": clean(lane.get("objective"), 80) or "settlement_capacity_builder", "price_eur": max(0.02, float(lane.get("price_eur") or 0.03)), "payload": {"mode": "idle_earn", "agent_id": agent_id, "source": clean(lane.get("source"), 40)}, "adapter_agent_id": agent_id, "consent_token": clean(consent_token, 240), } submit = http_json("POST", endpoint(base_url, "/swarm/microtask/submit"), submit_payload, timeout=timeout) if not bool(submit.get("accepted")): return { "ok": True, "phase": "idle_earn", "state": "task_not_accepted", "idle_receipt": idle_receipt, "lane": lane, "submit": submit, "machine_instruction": "switch_lane_or_wait_then_retry_submit", } task_id = clean(submit.get("task_id"), 120) digest = hashlib.sha256(f"{task_id}:{agent_id}:{time.time()}".encode("utf-8")).hexdigest()[:24] settle_payload = { "task_id": task_id, "worker_agent_id": agent_id, "objective": submit_payload["objective"], "settled_price_eur": float(submit_payload["price_eur"]), "proof_digest": f"proof_{digest}", "verifier_trace_digest": f"trace_{digest}", "test_digest": f"test_{digest}", "settlement_ref": f"idle_{digest}", "utility_delta": 0.01, "reuse_count": 1, "risk_score": 0.0, "capability": "settlement_capacity_builder", "adapter_agent_id": agent_id, "consent_token": clean(consent_token, 240), } settle = http_json("POST", endpoint(base_url, "/swarm/microtask/settle"), settle_payload, timeout=timeout) return { "ok": bool(settle.get("accepted")), "phase": "idle_earn", "state": "settled" if settle.get("accepted") else "settle_failed", "idle_receipt": idle_receipt, "lane": lane, "submit": submit, "settle": settle, "machine_instruction": "continue_idle_earn_loop_with_owner_priority_guard", } def run_cycle( *, base_url: str, agent_id: str, capabilities: list[str], timeout: float, objective: str, last_report: dict | None, runtime_signal: dict | None = None, pull: dict | None = None, probe_runtime: bool = False, machine_surfaces: dict | None = None, ) -> dict: lease = lease_nomad( base_url=base_url, agent_id=agent_id, capabilities=capabilities, timeout=timeout, objective=objective, last_report=last_report, machine_surfaces=machine_surfaces, ) lease_id = clean(lease.get("lease_id"), 120) if not lease.get("ok") or not lease_id: return {"ok": False, "phase": "lease", "lease": lease} signal = runtime_signal if isinstance(runtime_signal, dict) else openclaw_runtime_signal(timeout=timeout) if probe_runtime else {} report = _simulate_openclaw_execution(lease=lease, objective=objective, runtime_signal=signal, pull=pull) surfaces = machine_surfaces if isinstance(machine_surfaces, dict) else {} if surfaces: report["machine_surfaces"] = surfaces complete = complete_nomad( base_url=base_url, agent_id=agent_id, lease_id=lease_id, report=report, timeout=timeout, ) proof_link = proof_link_nomad( base_url=base_url, agent_id=agent_id, report=report, timeout=timeout, ) return { "ok": bool(complete.get("ok")), "phase": "complete", "agent_id": agent_id, "lease_id": lease_id, "objective": clean(lease.get("objective") or objective, 80), "report": report, "complete": complete, "proof_link": proof_link, } def main() -> None: p = argparse.ArgumentParser(description="OpenClaw -> Nomad adapter") p.add_argument("--base-url", default=os.getenv("NOMAD_BASE_URL", "https://www.syndiode.com")) p.add_argument("--agent-id", default=os.getenv("NOMAD_OPENCLAW_AGENT_ID", default_agent_id())) p.add_argument("--capabilities", default=os.getenv("NOMAD_OPENCLAW_CAPS", "agent_protocols,transition_settlement,objective_lease_execution")) p.add_argument("--objective", default=os.getenv("NOMAD_MACHINE_OBJECTIVE", "unhuman_supremacy")) p.add_argument("--timeout", type=float, default=float(os.getenv("NOMAD_OPENCLAW_TIMEOUT", "20") or "20")) p.add_argument("--loop", action="store_true") p.add_argument("--cycles", type=int, default=1) p.add_argument("--interval", type=float, default=float(os.getenv("NOMAD_OPENCLAW_INTERVAL", "12") or "12")) p.add_argument("--skip-join", action="store_true") p.add_argument("--skip-attach", action="store_true") p.add_argument("--no-runtime-probe", action="store_true") p.add_argument("--discover", dest="discover", action="store_true") p.add_argument("--no-discover", dest="discover", action="store_false") p.set_defaults(discover=True) p.add_argument("--force-attach", action="store_true") p.add_argument("--source-tag", default=os.getenv("NOMAD_SOURCE_TAG", "")) p.add_argument("--idle-earn", action="store_true") p.add_argument( "--owner-busy-min-sessions", type=int, default=int(os.getenv("NOMAD_OWNER_BUSY_MIN_SESSIONS", "1") or "1"), ) p.add_argument("--idle-opt-in", action="store_true") p.add_argument("--no-idle-opt-in", dest="idle_opt_in", action="store_false") p.set_defaults(idle_opt_in=(os.getenv("NOMAD_IDLE_OPT_IN", "").strip().lower() in {"1", "true", "yes", "on"})) a = p.parse_args() caps = _caps_from_csv(a.capabilities) pull = {} if a.discover: pull = discover_pull_contract(base_url=a.base_url, timeout=a.timeout) if a.source_tag: pull["source"] = clean(a.source_tag, 80) print(json.dumps({"phase": "discover", "pull": pull}, ensure_ascii=True)) if (pull.get("decision") == "observe") and not a.force_attach: print(json.dumps({"phase": "exit", "reason": "attach_threshold_not_met"}, ensure_ascii=True)) return elif a.source_tag: pull = {"source": clean(a.source_tag, 80)} runtime_signal = {} if a.no_runtime_probe else openclaw_runtime_signal(timeout=a.timeout) consent_token = mint_consent_token(agent_id=a.agent_id, runtime="openclaw") if runtime_signal: print(json.dumps({"phase": "openclaw_runtime", "runtime_signal": runtime_signal}, ensure_ascii=True)) surfaces = machine_surface_signal(base_url=a.base_url, timeout=min(8.0, a.timeout)) print(json.dumps({"phase": "machine_surfaces", "machine_surfaces": surfaces}, ensure_ascii=True)) attach = {} if not a.skip_attach: attach = attach_nomad( base_url=a.base_url, agent_id=a.agent_id, capabilities=caps, timeout=a.timeout, objective=a.objective, runtime_signal=runtime_signal, pull=pull, idle_opt_in=a.idle_opt_in, consent_token=consent_token, ) print(json.dumps({"phase": "attach", "ok": bool(attach.get("ok")), "attach": attach}, ensure_ascii=True)) if (attach.get("attach") is False) and not a.force_attach: print(json.dumps({"phase": "exit", "reason": "attach_decision_observe"}, ensure_ascii=True)) return decision_doc = attach if isinstance(attach, dict) and attach.get("schema") == "nomad.runtime_attach_decision.v1" else pull effective_objective = select_effective_objective(a.objective, decision_doc) surface_objective, surface_decision = select_machine_surface_objective(a.objective, surfaces) if clean(a.objective, 80) in {"", "auto", "unhuman_supremacy"} and clean(surface_objective, 80) not in {"", "unhuman_supremacy"}: effective_objective = surface_objective print(json.dumps({"phase": "surface_objective", "objective": effective_objective, "decision": surface_decision}, ensure_ascii=True)) if not a.skip_join: joined = join_nomad( base_url=a.base_url, agent_id=a.agent_id, capabilities=caps, timeout=a.timeout, objective=effective_objective, runtime_signal=runtime_signal, pull=decision_doc, consent_token=consent_token, ) print(json.dumps({"phase": "join", "ok": bool(joined.get("ok")), "join": joined}, ensure_ascii=True)) count = 0 last: dict | None = None while True: count += 1 runtime_signal = {} if a.no_runtime_probe else openclaw_runtime_signal(timeout=a.timeout) if a.idle_earn: out = run_idle_earn_cycle( base_url=a.base_url, agent_id=a.agent_id, capabilities=caps, timeout=a.timeout, runtime_signal=runtime_signal, owner_busy_min_sessions=a.owner_busy_min_sessions, consent_token=consent_token, ) else: out = run_cycle( base_url=a.base_url, agent_id=a.agent_id, capabilities=caps, timeout=a.timeout, objective=effective_objective, last_report=last, runtime_signal=runtime_signal, pull=decision_doc, machine_surfaces=surfaces, probe_runtime=False, ) out["cycle"] = count print(json.dumps(out, ensure_ascii=True)) last = out.get("report") if isinstance(out.get("report"), dict) else None if not a.loop and count >= max(1, a.cycles): break if a.loop and a.cycles > 0 and count >= a.cycles: break time.sleep(max(1.0, float(a.interval))) if __name__ == "__main__": main()