diff --git a/CHANGELOG.md b/CHANGELOG.md
index eaf0e48..d5a72e2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,15 @@ results in CI or internal records. See docs/releasing.md. -->
 
 ## [Unreleased]
 
+## [v4.1.5] - 2026-10-09
+
+### Fixed
+
+- Read current consensus RPC requests and block responses so schema alerts and served-block metrics work with updated nodes.
+- Avoid replaying older log records when hourly log discovery encounters a directory read error.
+- Accept `Rpc` and `Validator` mempool transaction origins alongside the earlier boolean format.
+- Build with Go 1.26.9 and update `golang.org/x/net` to v0.60.0 for the October security fixes, including the HTTP/2 server crash described in [GO-2026-6617](https://pkg.go.dev/vuln/GO-2026-6617).
+
 ## [v4.1.4] - 2026-10-02
 
 ### Fixed
diff --git a/Dockerfile b/Dockerfile
index c5139fd..cf8a5e6 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.25.13-alpine3.23@sha256:42fc3368d1c50170a452f2bf4a1dfd292a065870c3f258d799aad4316671cb69 AS builder
+FROM --platform=$BUILDPLATFORM golang:1.26.9-alpine3.23@sha256:96123126ac58e910f4dd3619a8901e2fb6d1ad84b59b1232cac7c9ea65a8f888 AS builder
 ARG VERSION=docker
 ARG GIT_COMMIT=unknown
 ARG TARGETOS
diff --git a/docs/metrics.md b/docs/metrics.md
index ed76214..335b93a 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -83,7 +83,7 @@ Consensus event semantics are source-limited:
 - `RoundCatchUp` direction remains upstream-opaque;
 - accumulator counters add each accepted window's `delta`; `n` is the source-window observation count;
 - registered and sent RPC accumulator values are outbound local work, not served requests;
-- `hl_consensus_rpc_blocks_served_total` counts only the explicitly proven complete outbound-response branch.
+- `hl_consensus_rpc_blocks_served_total` counts blocks explicitly reported by complete inbound RPC lifecycles. Direct requests require the incoming stream, received request and matching outbound response; wrapped requests also require both task stages and `query_peers=false`.
 
 Validator latency EMA is the upstream measured field used by the sampled node build. It is not an exporter-measured protocol RTT. A complete all-zero EMA snapshot is treated as initialization; mixed snapshots preserve genuine zeros.
 
@@ -183,7 +183,7 @@ Run `go generate ./internal/metrics` after changing a declaration. `go test ./in
 | `hl_consensus_round_qc_total` | Counter | Prometheus | - | base | Accepted RoundQc source-window delta accumulated since exporter start; no relationship to committed blocks is inferred. | `prometheus_instruments.go` |
 | `hl_consensus_round_tc_total` | Counter | Prometheus | - | base | Accepted RoundTc source-window delta accumulated since exporter start; no cause or network-health conclusion is inferred. | `prometheus_instruments.go` |
 | `hl_consensus_rounds_per_block` | Gauge | OTel bridge | - | base | Most recent positive difference between consecutive accepted consensus Block round values; a single observed value, not an average | `instruments.go` |
-| `hl_consensus_rpc_blocks_served_total` | Counter | Prometheus | - | base | Blocks explicitly reported sent by complete query_peers=false Outbound response Ok.BlocksAndTxs results since exporter start. | `validator_consensus_prometheus.go` |
+| `hl_consensus_rpc_blocks_served_total` | Counter | Prometheus | - | base | Blocks explicitly reported sent by complete inbound RPC lifecycles since exporter start; wrapped requests require query_peers=false. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_events_total` | Counter | Prometheus | `content`, `direction`, `outcome`, `stage` | base | Consensus RPC lifecycle events since exporter start, classified only into fixed direction, stage, outcome, and content vocabularies. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_parse_total` | Counter | Prometheus | `result` | base | Consensus RPC source records since exporter start by bounded parser result. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_requests_registered_total` | Counter | Prometheus | - | base | Accepted RpcRequestsRegistered delta for outbound local accumulator work since exporter start; not a served-request count. | `prometheus_instruments.go` |
diff --git a/go.mod b/go.mod
index 2b067c4..f2a7c2a 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/validaoxyz/hyperliquid-exporter
 
-go 1.25.13
+go 1.26.9
 
 require (
 	github.com/joho/godotenv v1.5.1
@@ -31,9 +31,9 @@ require (
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/trace v1.45.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.11.0 // indirect
-	golang.org/x/net v0.58.0 // indirect
-	golang.org/x/sys v0.47.0 // indirect
-	golang.org/x/text v0.41.0 // indirect
+	golang.org/x/net v0.60.0 // indirect
+	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/text v0.42.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect
 	google.golang.org/grpc v1.83.2 // indirect
diff --git a/go.sum b/go.sum
index 1bf90ed..ef3a778 100644
--- a/go.sum
+++ b/go.sum
@@ -69,12 +69,12 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
-golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
-golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
-golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
+golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
+golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
+golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc=
diff --git a/internal/metrics/validator_consensus_prometheus.go b/internal/metrics/validator_consensus_prometheus.go
index 5734e41..08e17ea 100644
--- a/internal/metrics/validator_consensus_prometheus.go
+++ b/internal/metrics/validator_consensus_prometheus.go
@@ -28,7 +28,7 @@ var (
 	}, []string{"direction", "stage", "outcome", "content"})
 	HLConsensusRPCBlocksServed = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "hl_consensus_rpc_blocks_served_total",
-		Help: "Blocks explicitly reported sent by complete query_peers=false Outbound response Ok.BlocksAndTxs results since exporter start.",
+		Help: "Blocks explicitly reported sent by complete inbound RPC lifecycles since exporter start; wrapped requests require query_peers=false.",
 	})
 	HLConsensusRPCParse = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "hl_consensus_rpc_parse_total",
diff --git a/internal/monitors/consensus_rpc_monitor.go b/internal/monitors/consensus_rpc_monitor.go
index 29f513f..156d956 100644
--- a/internal/monitors/consensus_rpc_monitor.go
+++ b/internal/monitors/consensus_rpc_monitor.go
@@ -1,10 +1,12 @@
 package monitors
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -24,6 +26,7 @@ type consensusRPCRecord struct {
 	outcome     string
 	blocks      int64
 	queryPeers  bool
+	direct      bool
 	hasSession  bool
 	hasRequest  bool
 	hasResponse bool
@@ -39,6 +42,7 @@ type consensusRPCLifecycle struct {
 	request      [32]byte
 	content      string
 	received     bool
+	direct       bool
 	taskInbound  bool
 	taskResponse bool
 	response     [32]byte
@@ -150,11 +154,14 @@ func parseConsensusRPCLine(line []byte) (consensusRPCRecord, error) {
 		if len(payload) <= index {
 			return fmt.Errorf("missing request")
 		}
-		digest, content, queryPeers, err := parseConsensusRPCRequest(payload[index])
+		digest, content, queryPeers, direct, err := parseConsensusRPCRequest(payload[index])
 		if err != nil {
 			return err
 		}
-		record.request, record.content, record.queryPeers = digest, content, queryPeers
+		if direct && tag != "Received rpc request" && tag != "Outbound response" {
+			return fmt.Errorf("direct request at unsupported stage")
+		}
+		record.request, record.content, record.queryPeers, record.direct = digest, content, queryPeers, direct
 		record.hasRequest = true
 		return nil
 	}
@@ -236,33 +243,98 @@ func parseConsensusRPCLine(line []byte) (consensusRPCRecord, error) {
 	return record, nil
 }
 
-func parseConsensusRPCRequest(raw json.RawMessage) ([32]byte, string, bool, error) {
-	var body map[string]json.RawMessage
-	if err := json.Unmarshal(raw, &body); err != nil || body == nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid request object")
-	}
-	var queryPeers bool
-	queryRaw, ok := body["query_peers"]
-	if !ok || unmarshalRequiredJSON(queryRaw, &queryPeers) != nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid query_peers")
-	}
-	contentRaw, ok := body["content"]
-	if !ok {
-		return [32]byte{}, "", false, fmt.Errorf("missing request content")
+func parseConsensusRPCRequest(raw json.RawMessage) ([32]byte, string, bool, bool, error) {
+	body, err := consensusRPCObject(raw)
+	if err != nil {
+		return [32]byte{}, "", false, false, fmt.Errorf("invalid request object: %w", err)
 	}
+	direct := false
+	queryPeers := false
 	content := "other"
-	var contentObject map[string]json.RawMessage
-	if err := json.Unmarshal(contentRaw, &contentObject); err != nil || contentObject == nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid request content")
-	}
-	if _, ok := contentObject["BlocksAndTxs"]; ok {
-		content = "blocks_and_txs"
+	if _, wrapped := body["content"]; wrapped {
+		if _, mixed := body["after_round"]; mixed {
+			return [32]byte{}, "", false, false, fmt.Errorf("conflicting request forms")
+		}
+		if _, mixed := body["until_block_hash"]; mixed {
+			return [32]byte{}, "", false, false, fmt.Errorf("conflicting request forms")
+		}
+		if unmarshalRequiredJSON(body["query_peers"], &queryPeers) != nil {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid query_peers")
+		}
+		contentObject, err := consensusRPCObject(body["content"])
+		if err != nil {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid request content")
+		}
+		_, hasRound := contentObject["after_round"]
+		_, hasHash := contentObject["until_block_hash"]
+		if hasRound || hasHash {
+			if !validFlatConsensusRPCRequest(contentObject) {
+				return [32]byte{}, "", false, false, fmt.Errorf("invalid flat request content")
+			}
+			content = "blocks_and_txs"
+		} else {
+			if len(contentObject) != 1 {
+				return [32]byte{}, "", false, false, fmt.Errorf("invalid request variants")
+			}
+			if blocksRaw, ok := contentObject["BlocksAndTxs"]; ok {
+				blocksBody, err := consensusRPCObject(blocksRaw)
+				if err != nil || !validFlatConsensusRPCRequest(blocksBody) {
+					return [32]byte{}, "", false, false, fmt.Errorf("invalid BlocksAndTxs request")
+				}
+				content = "blocks_and_txs"
+			}
+		}
+	} else {
+		if !validFlatConsensusRPCRequest(body) {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid direct request")
+		}
+		direct, content = true, "blocks_and_txs"
 	}
 	canonical, err := canonicalJSON(raw)
 	if err != nil {
-		return [32]byte{}, "", false, err
+		return [32]byte{}, "", false, false, err
+	}
+	return sha256.Sum256(canonical), content, queryPeers, direct, nil
+}
+
+func validFlatConsensusRPCRequest(body map[string]json.RawMessage) bool {
+	var round uint64
+	var hash string
+	return len(body) == 2 && unmarshalRequiredJSON(body["after_round"], &round) == nil &&
+		unmarshalRequiredJSON(body["until_block_hash"], &hash) == nil && hash != ""
+}
+
+func consensusRPCObject(raw json.RawMessage) (map[string]json.RawMessage, error) {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	if token, err := decoder.Token(); err != nil || token != json.Delim('{') {
+		return nil, fmt.Errorf("expected object")
+	}
+	body := make(map[string]json.RawMessage)
+	for decoder.More() {
+		token, err := decoder.Token()
+		if err != nil {
+			return nil, err
+		}
+		key, ok := token.(string)
+		if !ok {
+			return nil, fmt.Errorf("invalid object key")
+		}
+		if _, exists := body[key]; exists {
+			return nil, fmt.Errorf("duplicate object key")
+		}
+		var value json.RawMessage
+		if err := decoder.Decode(&value); err != nil {
+			return nil, err
+		}
+		body[key] = value
+	}
+	if token, err := decoder.Token(); err != nil || token != json.Delim('}') {
+		return nil, fmt.Errorf("unclosed object")
+	}
+	if _, err := decoder.Token(); err != io.EOF {
+		return nil, fmt.Errorf("trailing object data")
 	}
-	return sha256.Sum256(canonical), content, queryPeers, nil
+	return body, nil
 }
 
 func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, int64, error) {
@@ -271,25 +343,44 @@ func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, i
 		return [32]byte{}, "", "", 0, err
 	}
 	digest := sha256.Sum256(canonical)
-	var variants map[string]json.RawMessage
-	if err := json.Unmarshal(raw, &variants); err != nil || variants == nil {
+	variants, err := consensusRPCObject(raw)
+	if err != nil {
 		return [32]byte{}, "", "", 0, fmt.Errorf("invalid response object")
 	}
+	if len(variants) != 1 {
+		return [32]byte{}, "", "", 0, fmt.Errorf("invalid response variants")
+	}
 	okRaw, ok := variants["Ok"]
 	if !ok {
 		return digest, "other", "other", 0, nil
 	}
-	var okBody map[string]json.RawMessage
-	if err := json.Unmarshal(okRaw, &okBody); err != nil || okBody == nil {
+	okBody, err := consensusRPCObject(okRaw)
+	if err != nil {
 		return [32]byte{}, "", "", 0, fmt.Errorf("invalid Ok response")
 	}
 	blocksRaw, ok := okBody["BlocksAndTxs"]
-	if !ok {
-		return digest, "other", "other", 0, nil
-	}
+	_, hasN := okBody["n"]
+	_, hasHash := okBody["last_block_hash"]
 	var blocksBody map[string]json.RawMessage
-	if err := json.Unmarshal(blocksRaw, &blocksBody); err != nil || blocksBody == nil {
-		return [32]byte{}, "", "", 0, fmt.Errorf("invalid BlocksAndTxs response")
+	if ok {
+		if len(okBody) != 1 {
+			return [32]byte{}, "", "", 0, fmt.Errorf("conflicting block response forms")
+		}
+		blocksBody, err = consensusRPCObject(blocksRaw)
+		if err != nil {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid BlocksAndTxs response")
+		}
+	} else if hasN || hasHash {
+		var hash string
+		if len(okBody) != 2 || unmarshalRequiredJSON(okBody["last_block_hash"], &hash) != nil || hash == "" {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid flat block response")
+		}
+		blocksBody = okBody
+	} else {
+		if len(okBody) != 1 {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid Ok response variants")
+		}
+		return digest, "other", "other", 0, nil
 	}
 	var n int64
 	if rawN, present := blocksBody["n"]; !present || unmarshalRequiredJSON(rawN, &n) != nil || n < 1 || n > maxConsensusRPCServedBlocks {
@@ -300,7 +391,9 @@ func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, i
 
 func canonicalJSON(raw []byte) ([]byte, error) {
 	var value interface{}
-	if err := json.Unmarshal(raw, &value); err != nil {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	if err := decoder.Decode(&value); err != nil {
 		return nil, err
 	}
 	return json.Marshal(value)
@@ -323,11 +416,14 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 		if life == nil || life.received || record.queryPeers {
 			return "unjoinable"
 		}
-		if existing, collision := t.byRequest[record.request]; collision && existing != record.session {
+		if existing, collision := t.byRequest[record.request]; !record.direct && collision && existing != record.session {
 			return "unjoinable"
 		}
 		life.request, life.content, life.received, life.updatedAt = record.request, record.content, true, now
-		t.byRequest[record.request] = record.session
+		life.direct = record.direct
+		if !record.direct {
+			t.byRequest[record.request] = record.session
+		}
 		return "ok"
 	case "Rpc task inbound":
 		life := t.lifecycleForRequest(record.request)
@@ -345,8 +441,8 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 		return "ok"
 	case "Outbound response":
 		life := t.bySession[record.session]
-		if life == nil || !life.received || !life.taskInbound || !life.taskResponse ||
-			life.request != record.request || life.response != record.response || record.queryPeers {
+		if life == nil || !life.received || life.direct != record.direct || life.request != record.request || record.queryPeers ||
+			(!life.direct && (!life.taskInbound || !life.taskResponse || life.response != record.response)) {
 			return "unjoinable"
 		}
 		completionKey := sha256.Sum256(append(record.session[:], record.response[:]...))
@@ -364,17 +460,21 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 			delete(t.completed, oldestKey)
 		}
 		t.completed[completionKey] = now
-		// Emit stage counters only after the complete five-record lifecycle
-		// joins. An abandoned or replayed prefix must not look like served work.
+		// Publish only after the full observed lifecycle joins. Direct requests
+		// have no task stages; wrapped requests still require both task records.
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "stream_opened", "observed", "other").Inc()
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "request_received", "observed", life.content).Inc()
-		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_inbound", "observed", life.content).Inc()
-		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_response", "observed", life.content).Inc()
+		if !life.direct {
+			metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_inbound", "observed", life.content).Inc()
+			metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_response", "observed", life.content).Inc()
+		}
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "response_sent", record.outcome, record.content).Inc()
 		if record.outcome == "ok" && record.content == "blocks_and_txs" {
 			metrics.HLConsensusRPCBlocksServed.Add(float64(record.blocks))
 		}
-		delete(t.byRequest, life.request)
+		if !life.direct {
+			delete(t.byRequest, life.request)
+		}
 		delete(t.bySession, life.session)
 		return "ok"
 	default:
@@ -395,7 +495,9 @@ func (t *consensusRPCTracker) expire(now time.Time) {
 	cutoff := now.Add(-t.ttl)
 	for session, life := range t.bySession {
 		if life.updatedAt.Before(cutoff) {
-			delete(t.byRequest, life.request)
+			if !life.direct {
+				delete(t.byRequest, life.request)
+			}
 			delete(t.bySession, session)
 		}
 	}
diff --git a/internal/monitors/consensus_rpc_monitor_test.go b/internal/monitors/consensus_rpc_monitor_test.go
index 274e991..34a78b9 100644
--- a/internal/monitors/consensus_rpc_monitor_test.go
+++ b/internal/monitors/consensus_rpc_monitor_test.go
@@ -103,7 +103,7 @@ func TestConsensusRPCRejectsInvalidServedCountsAndQueryPeers(t *testing.T) {
 	if got := tracker.accept(incoming); got != "ok" {
 		t.Fatalf("incoming = %q", got)
 	}
-	queryPeers := `{"content":{"BlocksAndTxs":{}},"query_peers":true}`
+	queryPeers := `{"content":{"BlocksAndTxs":{"after_round":10,"until_block_hash":"h"}},"query_peers":true}`
 	received, err := parseConsensusRPCLine([]byte(rpcLine(`["Received rpc request","session",` + queryPeers + `]`)))
 	if err != nil {
 		t.Fatal(err)
@@ -112,7 +112,7 @@ func TestConsensusRPCRejectsInvalidServedCountsAndQueryPeers(t *testing.T) {
 		t.Fatalf("query_peers=true = %q, want unjoinable", got)
 	}
 
-	nullQueryPeers := `{"content":{"BlocksAndTxs":{}},"query_peers":null}`
+	nullQueryPeers := `{"content":{"BlocksAndTxs":{"after_round":10,"until_block_hash":"h"}},"query_peers":null}`
 	if _, err := parseConsensusRPCLine([]byte(rpcLine(`["Rpc task inbound",` + nullQueryPeers + `]`))); err == nil {
 		t.Fatal("query_peers=null was accepted as false")
 	}
@@ -181,3 +181,251 @@ func TestConsensusRPCCompletedJoinSetIsBounded(t *testing.T) {
 func rpcLine(payload string) string {
 	return fmt.Sprintf(`[%q,%s]`, rpcTestTimestamp, payload)
 }
+
+func TestConsensusRPCDirectLifecyclePublishesOnlyObservedStages(t *testing.T) {
+	request := `{"after_round":904479215,"until_block_hash":"block"}`
+	response := `{"Ok":{"n":2,"last_block_hash":"last"}}`
+	tracker := newConsensusRPCTracker()
+	blocksBefore := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed)
+	stages := []struct {
+		stage, content string
+		want           float64
+	}{
+		{"stream_opened", "other", 2},
+		{"request_received", "blocks_and_txs", 2},
+		{"task_inbound", "blocks_and_txs", 0},
+		{"task_response", "blocks_and_txs", 0},
+	}
+	stageBefore := make([]float64, len(stages))
+	for i, stage := range stages {
+		stageBefore[i] = validatorMetricValue(t, metrics.HLConsensusRPCEvents.WithLabelValues("serve", stage.stage, "observed", stage.content))
+	}
+	terminalBefore := validatorMetricValue(t, metrics.HLConsensusRPCEvents.WithLabelValues("serve", "response_sent", "ok", "blocks_and_txs"))
+	parse := func(payload string) consensusRPCRecord {
+		t.Helper()
+		record, err := parseConsensusRPCLine([]byte(rpcLine(payload)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		return record
+	}
+	terminal := parse(`["Outbound response","a",` + request + `,` + response + `]`)
+	if got := tracker.accept(terminal); got != "unjoinable" {
+		t.Fatalf("orphan terminal = %s", got)
+	}
+	// Concurrent sessions can legitimately request the same blocks.
+	for _, session := range []string{"a", "b"} {
+		for _, payload := range []string{
+			`["Incoming tcp stream","` + session + `","203.0.113.1:1"]`,
+			`["Received rpc request","` + session + `",` + request + `]`,
+		} {
+			if got := tracker.accept(parse(payload)); got != "ok" {
+				t.Fatalf("prefix = %s", got)
+			}
+		}
+	}
+	if got := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed); got != blocksBefore {
+		t.Fatal("incomplete request counted blocks")
+	}
+	for i, stage := range stages {
+		if got := validatorMetricValue(t, metrics.HLConsensusRPCEvents.WithLabelValues("serve", stage.stage, "observed", stage.content)); got != stageBefore[i] {
+			t.Fatalf("incomplete request published %s", stage.stage)
+		}
+	}
+	mismatched := parse(`["Outbound response","a",{"after_round":904479216,"until_block_hash":"block"},` + response + `]`)
+	if got := tracker.accept(mismatched); got != "unjoinable" {
+		t.Fatalf("mismatched request = %s", got)
+	}
+	for _, session := range []string{"a", "b"} {
+		r := parse(`["Outbound response","` + session + `",` + request + `,` + response + `]`)
+		if got := tracker.accept(r); got != "ok" {
+			t.Fatalf("terminal = %s", got)
+		}
+		if got := tracker.accept(r); got != "unjoinable" {
+			t.Fatalf("replay = %s", got)
+		}
+	}
+	if got := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed) - blocksBefore; got != 4 {
+		t.Fatalf("blocks delta = %v", got)
+	}
+	if got := validatorMetricValue(t, metrics.HLConsensusRPCEvents.WithLabelValues("serve", "response_sent", "ok", "blocks_and_txs")) - terminalBefore; got != 2 {
+		t.Fatalf("responses delta = %v", got)
+	}
+	for i, stage := range stages {
+		if delta := validatorMetricValue(t, metrics.HLConsensusRPCEvents.WithLabelValues("serve", stage.stage, "observed", stage.content)) - stageBefore[i]; delta != stage.want {
+			t.Fatalf("%s delta = %v, want %v", stage.stage, delta, stage.want)
+		}
+	}
+}
+
+func TestConsensusRPCWrappedFlatLifecycleStillRequiresTaskStages(t *testing.T) {
+	request := `{"content":{"after_round":10,"until_block_hash":"block"},"query_peers":false}`
+	response := `{"Ok":{"n":2,"last_block_hash":"last"}}`
+	tracker := newConsensusRPCTracker()
+	parse := func(payload string) consensusRPCRecord {
+		t.Helper()
+		r, err := parseConsensusRPCLine([]byte(rpcLine(payload)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		return r
+	}
+	for _, payload := range []string{`["Incoming tcp stream","a","203.0.113.1:1"]`, `["Received rpc request","a",` + request + `]`} {
+		if got := tracker.accept(parse(payload)); got != "ok" {
+			t.Fatal(got)
+		}
+	}
+	terminal := parse(`["Outbound response","a",` + request + `,` + response + `]`)
+	if got := tracker.accept(terminal); got != "unjoinable" {
+		t.Fatalf("missing task records = %s", got)
+	}
+	for _, payload := range []string{`["Rpc task inbound",` + request + `]`, `["Rpc task response",` + request + `,` + response + `]`} {
+		if got := tracker.accept(parse(payload)); got != "ok" {
+			t.Fatal(got)
+		}
+	}
+	before := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed)
+	if got := tracker.accept(terminal); got != "ok" {
+		t.Fatal(got)
+	}
+	if delta := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed) - before; delta != 2 {
+		t.Fatalf("blocks = %v", delta)
+	}
+}
+
+func TestConsensusRPCRejectsMalformedDirectForms(t *testing.T) {
+	for _, request := range []string{
+		`{"after_round":1}`, `{"after_round":null,"until_block_hash":"h"}`,
+		`{"after_round":1.0,"until_block_hash":"h"}`, `{"after_round":-1,"until_block_hash":"h"}`,
+		`{"after_round":1,"until_block_hash":null}`, `{"after_round":1,"until_block_hash":""}`,
+		`{"after_round":1,"until_block_hash":"h","query_peers":false}`,
+		`{"after_round":1,"after_round":2,"until_block_hash":"h"}`,
+		`{"after_round":1,"until_block_hash":"h","until_block_hash":"x"}`,
+	} {
+		if _, err := parseConsensusRPCLine([]byte(rpcLine(`["Received rpc request","a",` + request + `]`))); err == nil {
+			t.Fatalf("accepted malformed request %s", request)
+		}
+	}
+	for _, response := range []string{
+		`{"Ok":{"n":2}}`, `{"Ok":{"n":2,"last_block_hash":null}}`,
+		`{"Ok":{"n":0,"last_block_hash":"h"}}`, `{"Ok":{"n":2,"last_block_hash":"h","extra":1}}`,
+		`{"Ok":{"n":2,"n":3,"last_block_hash":"h"}}`,
+	} {
+		if _, _, _, _, err := parseConsensusRPCResponse(json.RawMessage(response)); err == nil {
+			t.Fatalf("accepted malformed response %s", response)
+		}
+	}
+	if _, err := parseConsensusRPCLine([]byte(rpcLine(`["Rpc task inbound",{"after_round":1,"until_block_hash":"h"}]`))); err == nil {
+		t.Fatal("bare request accepted at legacy task stage")
+	}
+}
+
+func TestConsensusRPCRequestDigestPreservesLargeRounds(t *testing.T) {
+	a, _, _, _, err := parseConsensusRPCRequest(json.RawMessage(`{"after_round":9007199254740992,"until_block_hash":"h"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, _, _, _, err := parseConsensusRPCRequest(json.RawMessage(`{"after_round":9007199254740993,"until_block_hash":"h"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a == b {
+		t.Fatal("distinct integer rounds share request identity")
+	}
+}
+
+func TestConsensusRPCRejectsIncompleteAndConflictingRequests(t *testing.T) {
+	for name, request := range map[string]string{
+		"flat missing round":            `{"content":{"until_block_hash":"h"},"query_peers":false}`,
+		"flat missing hash":             `{"content":{"after_round":1},"query_peers":false}`,
+		"legacy null body":              `{"content":{"BlocksAndTxs":null},"query_peers":false}`,
+		"legacy missing fields":         `{"content":{"BlocksAndTxs":{}},"query_peers":false}`,
+		"legacy duplicate round":        `{"content":{"BlocksAndTxs":{"after_round":1,"after_round":2,"until_block_hash":"h"}},"query_peers":false}`,
+		"legacy conflicting variants":   `{"content":{"BlocksAndTxs":{"after_round":1,"until_block_hash":"h"},"Status":{}},"query_peers":false}`,
+		"mixed direct and wrapped":      `{"content":{"BlocksAndTxs":{"after_round":1,"until_block_hash":"h"}},"query_peers":false,"after_round":2,"until_block_hash":"other"}`,
+		"mixed direct hash and wrapped": `{"content":{"BlocksAndTxs":{"after_round":1,"until_block_hash":"h"}},"query_peers":false,"until_block_hash":"other"}`,
+		"empty content":                 `{"content":{},"query_peers":false}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			if _, err := parseConsensusRPCLine([]byte(rpcLine(`["Received rpc request","a",` + request + `]`))); err == nil {
+				t.Fatal("accepted malformed request")
+			}
+		})
+	}
+}
+
+func TestConsensusRPCRejectsIncompleteAndConflictingResponses(t *testing.T) {
+	for name, response := range map[string]string{
+		"flat missing count":                   `{"Ok":{"last_block_hash":"h"}}`,
+		"conflicting non-ok variants":          `{"Err":"failure","Timeout":null}`,
+		"conflicting ok and err":               `{"Ok":{"n":2,"last_block_hash":"h"},"Err":"failure"}`,
+		"conflicting response forms":           `{"Ok":{"BlocksAndTxs":{"n":2},"n":2,"last_block_hash":"h"}}`,
+		"conflicting unknown content variants": `{"Ok":{"FutureVariant":{},"AnotherVariant":{}}}`,
+		"empty response":                       `{}`,
+		"empty ok body":                        `{"Ok":{}}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			if _, _, _, _, err := parseConsensusRPCResponse(json.RawMessage(response)); err == nil {
+				t.Fatal("accepted malformed response")
+			}
+		})
+	}
+}
+
+func TestConsensusRPCUnknownTaggedVariantsRemainBounded(t *testing.T) {
+	request := json.RawMessage(`{"content":{"FutureVariant":{"opaque":"private"}},"query_peers":false,"metadata":"retained"}`)
+	_, content, queryPeers, direct, err := parseConsensusRPCRequest(request)
+	if err != nil || content != "other" || queryPeers || direct {
+		t.Fatalf("unknown request: content=%s queryPeers=%v direct=%v err=%v", content, queryPeers, direct, err)
+	}
+	for _, raw := range []string{`{"FutureVariant":{"opaque":"private"}}`, `{"Ok":{"FutureVariant":{"opaque":"private"}}}`} {
+		_, outcome, content, n, err := parseConsensusRPCResponse(json.RawMessage(raw))
+		if err != nil || outcome != "other" || content != "other" || n != 0 {
+			t.Fatalf("unknown response: outcome=%s content=%s n=%d err=%v", outcome, content, n, err)
+		}
+	}
+}
+
+func TestConsensusRPCDirectErrorAndExpiredLifecycles(t *testing.T) {
+	request := `{"after_round":1,"until_block_hash":"h"}`
+	parse := func(payload string) consensusRPCRecord {
+		t.Helper()
+		record, err := parseConsensusRPCLine([]byte(rpcLine(payload)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		return record
+	}
+	for _, expired := range []bool{false, true} {
+		t.Run(fmt.Sprintf("expired=%v", expired), func(t *testing.T) {
+			tracker := newConsensusRPCTracker()
+			for _, payload := range []string{`["Incoming tcp stream","a","203.0.113.1:1"]`, `["Received rpc request","a",` + request + `]`} {
+				if got := tracker.accept(parse(payload)); got != "ok" {
+					t.Fatal(got)
+				}
+			}
+			terminal := parse(`["Outbound response","a",` + request + `,{"Err":"failure"}]`)
+			beforeBlocks := validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed)
+			counter := metrics.HLConsensusRPCEvents.WithLabelValues("serve", "response_sent", "other", "blocks_and_txs")
+			beforeResponses := validatorMetricValue(t, counter)
+			want := "ok"
+			wantResponses := float64(1)
+			if expired {
+				terminal.sourceTime = terminal.sourceTime.Add(tracker.ttl + time.Second)
+				want, wantResponses = "unjoinable", 0
+			}
+			if got := tracker.accept(terminal); got != want {
+				t.Fatalf("terminal = %s, want %s", got, want)
+			}
+			if delta := validatorMetricValue(t, counter) - beforeResponses; delta != wantResponses {
+				t.Fatalf("response delta = %v, want %v", delta, wantResponses)
+			}
+			if validatorMetricValue(t, metrics.HLConsensusRPCBlocksServed) != beforeBlocks {
+				t.Fatal("error response counted blocks")
+			}
+			if len(tracker.bySession) != 0 || len(tracker.byRequest) != 0 {
+				t.Fatal("finished or expired direct state retained")
+			}
+		})
+	}
+}
diff --git a/internal/monitors/mempool_monitor.go b/internal/monitors/mempool_monitor.go
index b7434b2..1f4bf05 100644
--- a/internal/monitors/mempool_monitor.go
+++ b/internal/monitors/mempool_monitor.go
@@ -325,11 +325,8 @@ func validateMempoolStatusEvent(inner []json.RawMessage, tag, status string) boo
 	if err := unmarshalRequiredJSON(inner[1], &hash); err != nil || hash == "" {
 		return false
 	}
-	if tag == "add_tx" {
-		var replace bool
-		if err := unmarshalRequiredJSON(inner[2], &replace); err != nil {
-			return false
-		}
+	if tag == "add_tx" && !validateMempoolTxOrigin(inner[2]) {
+		return false
 	}
 	switch status {
 	case "ok":
@@ -344,6 +341,32 @@ func validateMempoolStatusEvent(inner []json.RawMessage, tag, status string) boo
 	}
 }
 
+func validateMempoolTxOrigin(raw json.RawMessage) bool {
+	raw = bytes.TrimSpace(raw)
+	if bytes.Equal(raw, []byte("true")) || bytes.Equal(raw, []byte("false")) {
+		return true
+	}
+	var origin string
+	if json.Unmarshal(raw, &origin) == nil {
+		return origin == "Rpc"
+	}
+
+	// Require exactly one Validator field, including rejecting duplicate keys.
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	if token, err := decoder.Token(); err != nil || token != json.Delim('{') {
+		return false
+	}
+	if token, err := decoder.Token(); err != nil || token != "Validator" {
+		return false
+	}
+	var validator string
+	if decoder.Decode(&validator) != nil || !isFullHexAddress(validator) {
+		return false
+	}
+	token, err := decoder.Token()
+	return err == nil && token == json.Delim('}')
+}
+
 func parseMempoolStatus(inner []json.RawMessage, index int) (status, reason string, valid bool) {
 	if len(inner) <= index {
 		return "", "missing_status", false
diff --git a/internal/monitors/mempool_monitor_test.go b/internal/monitors/mempool_monitor_test.go
index 5b7ef58..d7338d8 100644
--- a/internal/monitors/mempool_monitor_test.go
+++ b/internal/monitors/mempool_monitor_test.go
@@ -95,6 +95,78 @@ func TestParseMempoolStatusVocabulary(t *testing.T) {
 	}
 }
 
+func TestParseMempoolTxOrigins(t *testing.T) {
+	for name, origin := range map[string]string{
+		"legacy false": `false`,
+		"legacy true":  `true`,
+		"rpc":          `"Rpc"`,
+		"validator":    `{"Validator":"0x2222222222222222222222222222222222222222"}`,
+	} {
+		for _, status := range []string{"ok", "err"} {
+			t.Run(name+"/"+status, func(t *testing.T) {
+				payload := fmt.Sprintf(`["add_tx","0x1111111111111111111111111111111111111111111111111111111111111111",%s,%q`, origin, status)
+				if status == "err" {
+					payload += `,{"AddTxDuplicate":{"tx_hash":"0x1111111111111111111111111111111111111111111111111111111111111111","existing_from_rpc":false}}`
+				}
+				got := parseMempoolObservation(mempoolLine(payload + `]`))
+				if !got.complete || got.parseReason != "" || got.eventType != "add_tx" || got.status != status {
+					t.Fatalf("origin rejected: %+v", got)
+				}
+				if status == "err" && (got.errorOperation != "add_tx" || got.errorKind != "add_tx_duplicate") {
+					t.Fatalf("error classification changed: %+v", got)
+				}
+				counter := metrics.HLMempoolEventsTotal.WithLabelValues("add_tx", status)
+				before := mempoolMetricValue(t, counter)
+				publishMempoolObservation(got)
+				if after := mempoolMetricValue(t, counter); after != before+1 {
+					t.Fatalf("event not published: before=%v after=%v", before, after)
+				}
+			})
+		}
+	}
+}
+
+func TestParseMempoolRejectsMalformedTxOrigins(t *testing.T) {
+	for name, origin := range map[string]string{
+		"null":                `null`,
+		"number":              `1`,
+		"array":               `["Rpc"]`,
+		"empty string":        `""`,
+		"unknown string":      `"FutureOrigin"`,
+		"wrong string case":   `"rpc"`,
+		"validator string":    `"Validator"`,
+		"empty object":        `{}`,
+		"unknown object":      `{"FutureOrigin":"0x2222222222222222222222222222222222222222"}`,
+		"rpc object":          `{"Rpc":null}`,
+		"null validator":      `{"Validator":null}`,
+		"empty validator":     `{"Validator":""}`,
+		"number validator":    `{"Validator":123}`,
+		"array validator":     `{"Validator":[]}`,
+		"object validator":    `{"Validator":{}}`,
+		"short validator":     `{"Validator":"0x2222"}`,
+		"nonhex validator":    `{"Validator":"0x222222222222222222222222222222222222222g"}`,
+		"extra property":      `{"Validator":"0x2222222222222222222222222222222222222222","Rpc":null}`,
+		"duplicate validator": `{"Validator":"0x2222222222222222222222222222222222222222","Validator":"0x2222222222222222222222222222222222222222"}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			got := parseMempoolObservation(mempoolLine(fmt.Sprintf(`["add_tx","h",%s,"ok"]`, origin)))
+			if got.complete || got.parseReason != "invalid_add_tx_payload" {
+				t.Fatalf("malformed origin accepted: %+v", got)
+			}
+			counter := metrics.HLMempoolEventsTotal.WithLabelValues("add_tx", "ok")
+			before := mempoolMetricValue(t, counter)
+			publishMempoolObservation(got)
+			if after := mempoolMetricValue(t, counter); after != before {
+				t.Fatalf("malformed origin published: before=%v after=%v", before, after)
+			}
+		})
+	}
+	got := parseMempoolObservation(mempoolLine(`["add_tx","h","ok"]`))
+	if got.complete || got.parseReason != "missing_status" {
+		t.Fatalf("missing origin accepted: %+v", got)
+	}
+}
+
 func TestParseMempoolPruneDropAndSizePayloads(t *testing.T) {
 	prune := parseMempoolObservation(mempoolLine(`["Pruned rpc request throttle",2,1,[]]`))
 	if !prune.complete || prune.pruneItems == nil || *prune.pruneItems != 1 {
diff --git a/internal/monitors/stream_test.go b/internal/monitors/stream_test.go
index 3ba9df1..f44a058 100644
--- a/internal/monitors/stream_test.go
+++ b/internal/monitors/stream_test.go
@@ -660,3 +660,106 @@ func TestTailStreamReportsBoundedFailureStage(t *testing.T) {
 		t.Fatal("tail did not cancel after open failure")
 	}
 }
+
+func TestTailStreamHourlyReadErrorPreservesCursor(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("permission fixture requires a non-root process")
+	}
+	root := t.TempDir()
+	older, newer := filepath.Join(root, "20261008"), filepath.Join(root, "20261009")
+	for _, dir := range []string{older, newer} {
+		if err := os.Mkdir(dir, 0o700); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := os.WriteFile(filepath.Join(older, "23"), []byte("historical\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	current := filepath.Join(newer, "11")
+	writer, err := os.OpenFile(current, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer writer.Close()
+	defer os.Chmod(newer, 0o700)
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	defer func() {
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(streamTestTimeout):
+			t.Error("tailer did not stop")
+		}
+	}()
+	lines := make(chan string, 16)
+	resolutions := make(chan error, 64)
+	switches := make(chan string, 16)
+	go func() {
+		defer close(done)
+		tailStream(ctx, tailStreamOpts{
+			resolve:     func() (string, error) { path, err := latestHourlyFile(root); resolutions <- err; return path, err },
+			rescanEvery: 20 * time.Millisecond, eofSleep: time.Millisecond,
+			onLine:   func(line string) { lines <- strings.TrimSpace(line) },
+			onSwitch: func(path string) { switches <- path },
+		})
+	}()
+	if path := waitStreamValue(t, switches); path != current {
+		t.Fatalf("startup path=%q", path)
+	}
+	appendRecord := func(value string) {
+		t.Helper()
+		if _, err := writer.WriteString(value + "\n"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	appendRecord("before")
+	if line := waitStreamValue(t, lines); line != "before" {
+		t.Fatalf("first line=%q", line)
+	}
+	if err := os.Chmod(newer, 0); err != nil {
+		t.Fatal(err)
+	}
+	waitResolution := func(wantError bool) {
+		t.Helper()
+		timeout := time.After(streamTestTimeout)
+		for {
+			select {
+			case err := <-resolutions:
+				if wantError && errors.Is(err, os.ErrPermission) || !wantError && err == nil {
+					return
+				}
+			case line := <-lines:
+				t.Fatalf("unexpected replay while waiting for discovery: %q", line)
+			case <-timeout:
+				t.Fatalf("timed out waiting for discovery error=%v", wantError)
+			}
+		}
+	}
+	waitResolution(true)
+	// The already-open writer and reader still work while directory discovery fails.
+	appendRecord("during")
+	if line := waitStreamValue(t, lines); line != "during" {
+		t.Fatalf("line during failure=%q", line)
+	}
+	if err := os.Chmod(newer, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	waitResolution(false)
+	appendRecord("after")
+	if line := waitStreamValue(t, lines); line != "after" {
+		t.Fatalf("line after recovery=%q", line)
+	}
+	cancel()
+	<-done
+	select {
+	case line := <-lines:
+		t.Fatalf("unexpected extra line=%q", line)
+	default:
+	}
+	select {
+	case path := <-switches:
+		t.Fatalf("discovery failure switched to %q", path)
+	default:
+	}
+}
diff --git a/internal/monitors/visor_monitor.go b/internal/monitors/visor_monitor.go
index b4d94ee..f03f2ad 100644
--- a/internal/monitors/visor_monitor.go
+++ b/internal/monitors/visor_monitor.go
@@ -307,7 +307,10 @@ func latestHourlyFile(root string) (string, error) {
 	for i := len(dateNames) - 1; i >= 0; i-- {
 		datePath := filepath.Join(root, dateNames[i])
 		hourEntries, err := os.ReadDir(datePath)
-		if err != nil || len(hourEntries) == 0 {
+		if err != nil {
+			return "", err
+		}
+		if len(hourEntries) == 0 {
 			continue
 		}
 		hourNames := make([]string, 0, len(hourEntries))
diff --git a/internal/monitors/visor_monitor_test.go b/internal/monitors/visor_monitor_test.go
index 8a388a6..ee588a2 100644
--- a/internal/monitors/visor_monitor_test.go
+++ b/internal/monitors/visor_monitor_test.go
@@ -226,3 +226,33 @@ func TestParseVisorTime(t *testing.T) {
 		})
 	}
 }
+
+func TestLatestHourlyFileNestedReadErrorDoesNotFallBack(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("permission fixture requires a non-root process")
+	}
+	root := t.TempDir()
+	older, newer := filepath.Join(root, "20261008"), filepath.Join(root, "20261009")
+	for _, dir := range []string{older, newer} {
+		if err := os.Mkdir(dir, 0o700); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := os.WriteFile(filepath.Join(older, "23"), []byte("older\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Chmod(newer, 0); err != nil {
+		t.Fatal(err)
+	}
+	defer os.Chmod(newer, 0o700)
+	if path, err := latestHourlyFile(root); path != "" || !errors.Is(err, os.ErrPermission) {
+		t.Fatalf("unreadable newest date: path=%q err=%v, want permission error without older fallback", path, err)
+	}
+	if err := os.Chmod(newer, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	// A readable empty date remains a valid reason to select the preceding date.
+	if path, err := latestHourlyFile(root); err != nil || path != filepath.Join(older, "23") {
+		t.Fatalf("readable empty newest date: path=%q err=%v", path, err)
+	}
+}