From 0d4c3a6c25ab89bde0c1a7028731486eb925fb21 Mon Sep 17 00:00:00 2001
From: Florent Tapponnier <contact@mobula.io>
Date: Thu, 9 Jul 2026 14:04:30 +0200
Subject: [PATCH] head-lag: parallel Processed sub for solana
 (onUnconfirmedEventsCreated deprecated by codex, API-key path already killed)

---
 .../cmd/script/head_lag_monitor.go            | 49 +++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go
index 51038d68..dc9b04d9 100644
--- a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go
+++ b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go
@@ -510,6 +510,40 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 			return fmt.Errorf("subscribe to %s failed: %w", pool.Name, err)
 		}
 
+		// onUnconfirmedEventsCreated is deprecated (Codex changelog 2026-04-20)
+		// and already hard-killed for API-key auth since ~07-06. Subscribe the
+		// replacement (onEventsCreated + commitmentLevel Processed) in parallel:
+		// logs tell us which one actually delivers on the JWT path, and we keep
+		// data flowing whichever side Codex cuts next. Gauge takes last write,
+		// so double delivery is harmless.
+		if pool.ChainName == "solana" {
+			procMsg := map[string]interface{}{
+				"type": "subscribe",
+				"id":   fmt.Sprintf("headlag_proc_%d", i),
+				"payload": map[string]interface{}{
+					"query": `subscription OnPoolEvents($id: String!, $cl: [EventCommitmentLevel!]) {
+						onEventsCreated(id: $id, commitmentLevel: $cl) {
+							address
+							networkId
+							events {
+								blockNumber
+								timestamp
+								transactionHash
+								eventType
+							}
+						}
+					}`,
+					"variables": map[string]interface{}{
+						"id": fmt.Sprintf("%s:%d", pool.Address, pool.NetworkID),
+						"cl": []string{"Processed"},
+					},
+				},
+			}
+			if err := conn.WriteJSON(procMsg); err != nil {
+				return fmt.Errorf("processed subscribe to %s failed: %w", pool.Name, err)
+			}
+		}
+
 		time.Sleep(100 * time.Millisecond) // Small delay between subscriptions
 	}
 
@@ -532,6 +566,7 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 	// connection (the read loop errors out and the outer loop redials).
 	var lastEventMu sync.Mutex
 	lastEventByChain := make(map[string]time.Time)
+	firstEventBySub := make(map[string]bool)
 	for _, pool := range headLagPools {
 		lastEventByChain[pool.ChainName] = time.Now()
 	}
@@ -625,6 +660,13 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 			// Codex killed the Solana sub exactly this way. Force a full
 			// reconnect + resubscribe instead.
 			if wsMsg.Type == "complete" || wsMsg.Type == "error" {
+				// The experimental Processed sub failing must not tear down the
+				// legacy flow: log it and keep the connection alive.
+				if strings.HasPrefix(wsMsg.ID, "headlag_proc_") {
+					payloadStr, _ := json.Marshal(wsMsg.Payload)
+					log.Printf("[HEAD-LAG][CODEX] ⚠️ Processed sub %q terminated (type=%s payload=%s) — legacy sub continues", wsMsg.ID, wsMsg.Type, string(payloadStr))
+					continue
+				}
 				return fmt.Errorf("subscription %q terminated by server (type=%s)", wsMsg.ID, wsMsg.Type)
 			}
 
@@ -633,6 +675,13 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 				continue
 			}
 
+			// One-shot log per subscription id: tells us whether the legacy
+			// unconfirmed sub, the Processed sub, or both deliver on this tier.
+			if !firstEventBySub[wsMsg.ID] {
+				firstEventBySub[wsMsg.ID] = true
+				log.Printf("[HEAD-LAG][CODEX] 📬 First event on subscription %q", wsMsg.ID)
+			}
+
 			// Parse event data
 			payloadBytes, _ := json.Marshal(wsMsg.Payload)
 			var eventData CodexEventData