feat: yennefer proxy, wqflop monitor, NodeStateDO enhancements

Igor Holt · Igor Holt · commit 6749f883d777 · 2026-06-27T16:28:35.000Z
- Add src/yennefer-proxy.ts for Yennefer orchestration bridging
- Add scripts/wqflop_monitor.py for quantum workload monitoring
- Enhance NodeStateDO.ts, index.ts, types.ts for extended state management
- Update mcp-config.json and package.json
- Expand test coverage in health.test.ts
- Align wrangler.toml
diff --git a/mcp-config.json b/mcp-config.json
@@ -38,6 +38,23 @@
       "health": "https://gc-mcp.iholt.workers.dev/health",
       "verified": true,
       "notes": "Development/fallback endpoint, fully functional"
+    },
+    {
+      "name": "payments-mcp",
+      "transport": "stdio",
+      "status": "installed",
+      "command": "node",
+      "args": ["/home/diamondnode/.payments-mcp/bundle.js"],
+      "verified": false,
+      "notes": "Coinbase Agentic Wallet + x402 payments MCP (v1.0.5 installed). Electron-based; fails headless (no DISPLAY/libasound). Configure Coinbase creds for embedded/agentic wallets on Base. Complements direct EOA flows."
+    },
+    {
+      "name": "coinbase-cdp-docs",
+      "url": "https://docs.cdp.coinbase.com/mcp",
+      "transport": "streamable-http",
+      "status": "available",
+      "verified": false,
+      "notes": "Official Coinbase Developer Platform Documentation MCP (read-only). Server: 'Coinbase Developer Documentation' v1.0.0. Provides search/retrieval over public docs (search tools + resources). Initialize succeeds with tools+resources capabilities. Use for AgentKit, cdp-sdk, Trade API, onchain actions, Base/EVM guidance. Complements (does not replace) diamondnode-qflop-base-wallet-mcp + direct ethers in Yennefer/aerodrome-lp/qflop-backfill for our 25 funded real EOA wallets (BACKFILL_MASTER_MNEMONIC) + owner PRIVATE_KEY doing real Aerodrome volatile LP adds/swaps on Base 8453 (QFLOP_SIM_MODE=0). For live CDP-managed wallet control: set CDP_API_KEY_NAME + CDP_API_KEY_PRIVATE_KEY then use @coinbase/agentkit + cdp-sdk (getMcpTools) or payments-mcp with keys. No keys present; no auto-install performed."
     }
   ],
   "protocol": {
diff --git a/package.json b/package.json
@@ -13,23 +13,7 @@
     "test": "vitest run",
     "gen-identity": "node scripts/gen-identity.mjs"
   },
-  "devDependencies": {
-    "@cloudflare/workers-types": "^4.20240524.0",
-    "typescript": "^5.5.0",
-    "vitest": "^1.6.0",
-    "wrangler": "^4.93.0"
-  },
   "dependencies": {
-    "@appsignal/javascript": "^1.6.1",
-    "@opentelemetry/api": "^1.9.1",
-    "@opentelemetry/auto-instrumentations-node": "^0.75.0",
-    "@opentelemetry/exporter-metrics-otlp-proto": "^0.217.0",
-    "@opentelemetry/exporter-trace-otlp-proto": "^0.217.0",
-    "@opentelemetry/resources": "^2.7.1",
-    "@opentelemetry/sdk-metrics": "^2.7.1",
-    "@opentelemetry/sdk-node": "^0.217.0",
-    "@opentelemetry/sdk-trace-node": "^2.7.1",
-    "@opentelemetry/semantic-conventions": "^1.40.0",
-    "botid": "^1.5.11"
+    "@appsignal/javascript": "^1.6.1"
   }
 }
diff --git a/scripts/wqflop_monitor.py b/scripts/wqflop_monitor.py
@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""wQFLOP engine liveness monitor — polls dexpaprika, posts heartbeat EVTs to DN.
+Cron: */15 * * * *   (via crontab)
+On high-severity alerts (volume > $100): also posts to news.genesisconductor.io.
+"""
+import json, sys, os, datetime, subprocess
+import requests
+
+DN_INGEST  = "https://dn.genesisconductor.io/api/rpsi/ingest"
+NEWS_BASE  = "https://news.genesisconductor.io"
+DEXPAPRIKA = "https://api.dexpaprika.com/networks/base/pools/0x4abc6d796cd036b6f1e433a97f9784a00f90c53e"
+POOL_ID    = "base-aerodrome-weth-wqflop-0x4abc6d796cd036b6f1e433a97f9784a00f90c53e"
+ENV_FILE   = os.path.expanduser("~/.env")
+
+ALERT_VOL_USD = 100.0
+UA = "Mozilla/5.0 (compatible; DiamondNode-wQFLOP-Monitor/1.0)"
+
+def load_env():
+    env = {}
+    try:
+        with open(ENV_FILE) as f:
+            for line in f:
+                line = line.strip()
+                if '=' in line and not line.startswith('#'):
+                    k, _, v = line.partition('=')
+                    env[k.strip()] = v.strip().strip('"').strip("'")
+    except Exception:
+        pass
+    return env
+
+def main():
+    env = load_env()
+    ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+    pool = None
+    try:
+        pool = requests.get(DEXPAPRIKA, headers={"User-Agent": UA}, timeout=10).json()
+    except Exception as e:
+        print(f"dexpaprika error: {e}", file=sys.stderr)
+
+    tvl_usd = vol_24h_usd = 0.0
+    buys_24h = txns_24h = 0
+    last_price_usd = 0.0
+    severity = "low"
+    alert_msgs = []
+
+    if pool and isinstance(pool, dict):
+        for r in pool.get("token_reserves", []):
+            tvl_usd += float(r.get("reserve_usd") or 0)
+        s = pool.get("24h", {})
+        vol_24h_usd   = float(s.get("volume_usd", 0) or 0)
+        buys_24h      = int(s.get("buys", 0) or 0)
+        txns_24h      = int(s.get("txns", 0) or 0)
+        last_price_usd = float(pool.get("last_price_usd") or 0)
+        if vol_24h_usd > ALERT_VOL_USD:
+            severity = "high"
+            alert_msgs.append(f"volume spike: ${vol_24h_usd:.4f} > ${ALERT_VOL_USD}")
+
+    hb = f"evt-wqflop-hb-{ts[:16].replace(':','-').replace('T','-')}"
+    evts = [{
+        "evt_id": hb,
+        "record_type": "monitoring_hook",
+        "timestamp": ts,
+        "payload": {
+            "target_asset_id": POOL_ID,
+            "monitor_type": "heartbeat",
+            "pool_state": {
+                "tvl_usd": round(tvl_usd, 6),
+                "volume_24h_usd": round(vol_24h_usd, 6),
+                "buys_24h": buys_24h,
+                "txns_24h": txns_24h,
+                "last_price_usd": round(last_price_usd, 4),
+                "data_source": "dexpaprika"
+            },
+            "severity": severity,
+            "alerts": alert_msgs,
+            "engine_alive": True,
+            "crystal_score": {"passed": True, "meets_target": True,
+                              "value": 1.0, "target": 1.0, "metric": "heartbeat_ok"}
+        }
+    }]
+
+    if severity == "high":
+        evts.append({
+            "evt_id": hb + "-qubo",
+            "record_type": "qubo_modeling_request",
+            "timestamp": ts,
+            "payload": {
+                "trigger": "volume_spike", "asset_id": POOL_ID,
+                "volume_usd": round(vol_24h_usd, 4),
+                "modeling_objective": "Re-evaluate allocation on volume spike",
+                "constraints": {"max_allocation_usd": 50, "requires_maru_guard": True},
+                "crystal_score": {"passed": True, "meets_target": True, "value": 1.0, "target": 1.0}
+            }
+        })
+
+    dn_ok = False
+    try:
+        resp = requests.post(DN_INGEST, json=evts, timeout=10)
+        dn_ok = resp.json().get("ok", False)
+    except Exception as e:
+        print(f"DN error: {e}", file=sys.stderr)
+
+    # Post to news on high-severity alert
+    news_ok = False
+    if severity == "high":
+        secret = env.get("NEWS_PUBLISH_SECRET", "")
+        if secret:
+            date_str = ts[:10]
+            slug = f"wqflop-alert-{date_str}-vol-spike"
+            md = f"""# wQFLOP Alert — Volume Spike Detected
+
+**Time**: {ts}
+**Pool**: wQFLOP/WETH · Aerodrome · Base
+**Severity**: HIGH
+
+## Alert Details
+
+{chr(10).join(f'- {a}' for a in alert_msgs)}
+
+## Pool State
+
+| Metric | Value |
+|---|---|
+| TVL USD | ${tvl_usd:.4f} |
+| 24h Volume | ${vol_24h_usd:.4f} |
+| 24h Txns | {txns_24h} |
+| 24h Buys | {buys_24h} |
+| Last Price | ${last_price_usd:.4f} WETH/wQFLOP |
+
+## Actions Triggered
+
+- QUBO modeling request queued (maru guard required)
+- monitoring_hook EVT posted to DN pipeline
+- News alert published
+
+*Automated alert from wQFLOP engine monitor (*/15 cron)*
+"""
+            try:
+                nr = requests.post(f"{NEWS_BASE}/api/publish",
+                    json={"title": f"wQFLOP Alert — {date_str}", "md": md, "slug": slug},
+                    headers={"Authorization": f"Bearer {secret}"}, timeout=15)
+                news_ok = nr.json().get("ok", False)
+            except Exception as e:
+                print(f"news error: {e}", file=sys.stderr)
+
+    out = {
+        "ts": ts, "tvl_usd": round(tvl_usd, 6),
+        "vol_24h_usd": round(vol_24h_usd, 6), "txns_24h": txns_24h,
+        "severity": severity, "alerts": alert_msgs,
+        "dn_ok": dn_ok, "news_ok": news_ok
+    }
+    print(json.dumps(out))
+
+if __name__ == "__main__":
+    main()
diff --git a/src/NodeStateDO.ts b/src/NodeStateDO.ts
@@ -1,8 +1,104 @@
-import { DurableObjectState } from "cloudflare:workers";
-
 export class NodeStateDO {
-  constructor(state: DurableObjectState) {}
-  async fetch(request: Request) {
-    return new Response("NodeStateDO is working", { status: 200 });
+  private sql: any;
+
+  constructor(state: DurableObjectState) {
+    this.sql = (state.storage as any).sql;
+    this.sql.exec(`
+      CREATE TABLE IF NOT EXISTS rpsi_evts (
+        id          INTEGER PRIMARY KEY AUTOINCREMENT,
+        evt_id      TEXT NOT NULL UNIQUE,
+        record_type TEXT NOT NULL,
+        ts          TEXT NOT NULL,
+        payload     TEXT NOT NULL,
+        received_at TEXT NOT NULL
+      )
+    `);
+  }
+
+  async fetch(request: Request): Promise<Response> {
+    const url   = new URL(request.url);
+    const path  = url.pathname;
+    const method = request.method;
+
+    if (method === "OPTIONS") {
+      return new Response(null, {
+        status: 204,
+        headers: {
+          "Access-Control-Allow-Origin": "*",
+          "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
+          "Access-Control-Allow-Headers": "Content-Type, Authorization",
+        },
+      });
+    }
+
+    // POST /ingest — single EVT or array of EVTs
+    if (path === "/ingest" && method === "POST") {
+      const body = await request.json() as Record<string, unknown> | Record<string, unknown>[];
+      const evts = Array.isArray(body) ? body : [body];
+      const received_at = new Date().toISOString();
+      const inserted: string[] = [];
+
+      for (const evt of evts) {
+        const evt_id      = (evt.evt_id as string)      ?? crypto.randomUUID();
+        const record_type = (evt.record_type as string) ?? "unknown";
+        const ts          = (evt.timestamp as string)   ?? received_at;
+        this.sql.exec(
+          `INSERT OR IGNORE INTO rpsi_evts (evt_id, record_type, ts, payload, received_at)
+           VALUES (?, ?, ?, ?, ?)`,
+          evt_id, record_type, ts, JSON.stringify(evt), received_at,
+        );
+        inserted.push(evt_id);
+      }
+      return Response.json({ ok: true, inserted, received_at }, {
+        headers: { "Access-Control-Allow-Origin": "*" },
+      });
+    }
+
+    // GET /feed?n=20 — latest N EVTs, newest first
+    if (path === "/feed" && method === "GET") {
+      const n = Math.min(parseInt(url.searchParams.get("n") ?? "20"), 200);
+      const rows = [...this.sql.exec(
+        `SELECT * FROM rpsi_evts ORDER BY id DESC LIMIT ?`, n,
+      )];
+      return Response.json({
+        count: rows.length,
+        evts: rows.map((r: any) => ({ ...r, payload: JSON.parse(r.payload) })),
+      }, { headers: { "Access-Control-Allow-Origin": "*" } });
+    }
+
+    // GET /status — latest EVT per record_type (pipeline snapshot)
+    if (path === "/status" && method === "GET") {
+      const rows = [...this.sql.exec(`
+        SELECT * FROM rpsi_evts
+        WHERE id IN (SELECT MAX(id) FROM rpsi_evts GROUP BY record_type)
+        ORDER BY record_type
+      `)];
+      const status: Record<string, unknown> = {};
+      for (const r of rows as any[]) {
+        status[r.record_type] = {
+          evt_id: r.evt_id,
+          ts: r.ts,
+          received_at: r.received_at,
+          payload: JSON.parse(r.payload),
+        };
+      }
+      return Response.json({ status, ts: new Date().toISOString(), count: rows.length }, {
+        headers: { "Access-Control-Allow-Origin": "*" },
+      });
+    }
+
+    // GET /purge?keep=500 — housekeeping
+    if (path === "/purge" && method === "GET") {
+      const keep = parseInt(url.searchParams.get("keep") ?? "500");
+      this.sql.exec(
+        `DELETE FROM rpsi_evts WHERE id <= (SELECT id FROM rpsi_evts ORDER BY id DESC LIMIT 1 OFFSET ?)`,
+        keep,
+      );
+      return Response.json({ ok: true, kept: keep }, {
+        headers: { "Access-Control-Allow-Origin": "*" },
+      });
+    }
+
+    return new Response("Not Found", { status: 404, headers: { "Access-Control-Allow-Origin": "*" } });
   }
-}
+}
diff --git a/src/index.ts b/src/index.ts
@@ -6,6 +6,7 @@ import { makeEvent, signEvent, signRadixClaim } from "./identity.js";
 import { appendAudit } from "./audit.js";
 import { handleSystemStatus } from "./cortex.js";
 import { handleSEORoutes } from "./seo-routes.js";
+import { handleYenneferProxy } from "./yennefer-proxy.js";
 import { YENNEFER_DASHBOARD_HTML } from "./yennefer-dashboard.js";
 import { NodeStateDO } from "./NodeStateDO.js";
 
@@ -62,9 +63,13 @@ export default {
           response = seoResponse;
         } else if (pathname === "/healthz" || pathname === "/health") {
           response = await handleHealth(request, env, ctx);
+        } else if (pathname === "/api/health" && method === "GET") {
+          response = await handleHealth(request, env, ctx);
         } else if (pathname === "/api/system-status" && method === "GET") {
           // Live cortex feed aggregation — powers Yennefer Cortex dashboard
           response = await handleSystemStatus(request, env, ctx);
+        } else if (pathname.startsWith("/api/yennefer/")) {
+          response = await handleYenneferProxy(request, env);
         } else if (pathname === "/audit/replay") {
           response = await handleAuditReplay(request);
         } else if (pathname === "/.well-known/diamond-node.json") {
@@ -137,6 +142,17 @@ export default {
               headers: { "Access-Control-Allow-Origin": "*" },
             });
           }
+        } else if (pathname.startsWith("/api/rpsi/")) {
+          // Route RPSI EVT ingestion and feed queries to NodeStateDO
+          const doId = env.NodeStateDO.idFromName("rpsi-pipeline");
+          const stub = env.NodeStateDO.get(doId);
+          const doUrl = new URL(request.url);
+          doUrl.pathname = pathname.replace("/api/rpsi", "");
+          response = await stub.fetch(new Request(doUrl.toString(), {
+            method: request.method,
+            headers: request.headers,
+            body: request.method !== "GET" && request.method !== "HEAD" ? request.body : undefined,
+          }));
         }
       }
 
diff --git a/src/types.ts b/src/types.ts
@@ -4,6 +4,8 @@ export interface Env {
   DIAMOND_NODE_ED25519_PUB: string;  // base64-encoded SPKI public key
   DIAMOND_VAULT_AUDIT_URL: string;   // upstream audit endpoint (optional)
   APPSIGNAL_KEY?: string;            // AppSignal API key (optional, for monitoring)
+  YENNEFER_API_URL?: string;          // Yennefer orchestrator API/tunnel origin
+  NodeStateDO: DurableObjectNamespace;  // RPSI EVT storage
 
   // Vars — set in wrangler.toml [vars]
   NODE_VERSION: string;
diff --git a/src/yennefer-proxy.ts b/src/yennefer-proxy.ts
diff --git a/test/health.test.ts b/test/health.test.ts
diff --git a/wrangler.toml b/wrangler.toml
