fix(view): reverse path flags describe the published path; raw revTr loop exposed as revtr_loop_cleaned

src/hermes/sql/queries/create_events_enriched.sql

@@ -427,16 +427,54 @@ SELECT
     reverse_geolocated_hop_count AS geolocated_hop_count,
     SAFE_DIVIDE(reverse_geolocated_hop_count, reverse_total_hop_count)
       AS geolocation_coverage,
-    -- Reverse flags describe the reverse path AS MEASURED (A B A loop; A ... A
-    -- gap over hops with no ASN), computed by step 04 BEFORE its cleanup drops
-    -- ambiguous hop_type=4 hops and truncates at the first AS re-entry. They
-    -- cannot be recomputed from reverse_hops: a loop never survives cleanup.
-    -- The raw path is not stored, so partitions written before step 04 was
-    -- fixed cannot be corrected; they are exposed as NULL (unknown) rather
-    -- than with the old definition. The date is the first partition written by
-    -- the fixed step 04 (a sentinel until that has run).
-    IF(partition_date >= DATE '9999-12-31', reverse_loop, NULL) AS loop_detected,
-    IF(partition_date >= DATE '9999-12-31', reverse_unresponsive_within_AS, NULL) AS unresponsive_within_as,
+    -- Same definitions as the forward path, over the PUBLISHED reverse hops, so
+    -- both directions' flags describe the path shown in as_path / hops. Step 04
+    -- drops ambiguous hop_type=4 excursions and truncates the revTr at the first
+    -- AS re-entry, so a published reverse path never has a loop: loop_detected is
+    -- FALSE by construction (NULL when there are no hops). Whether the RAW revTr
+    -- had a loop is revtr_loop_cleaned below.
+    IF(
+      ARRAY_LENGTH(reverse_hops) = 0,
+      NULL,
+      (
+        SELECT COUNT(*) > COUNT(DISTINCT asn)
+        FROM (
+          SELECT h.asn, LAG(h.asn) OVER (ORDER BY o) AS prev_asn
+          FROM UNNEST(reverse_hops) AS h WITH OFFSET o
+          WHERE h.asn IS NOT NULL AND h.asn != -1
+        )
+        WHERE prev_asn IS NULL OR asn != prev_asn
+      )
+    ) AS loop_detected,
+    -- TRUE when an AS reappears after one or more hops with no ASN (A ... A):
+    -- silent hops or hops that replied but could not be mapped, so part of
+    -- that AS's internal path is unknown. Same definition as step 04 and the forward path.
+    IF(
+      ARRAY_LENGTH(reverse_hops) = 0,
+      NULL,
+      COALESCE((
+        SELECT LOGICAL_OR(asn = prev_asn AND o - prev_o > 1)
+        FROM (
+          SELECT
+            h.asn,
+            o,
+            LAG(h.asn) OVER (ORDER BY o) AS prev_asn,
+            LAG(o) OVER (ORDER BY o) AS prev_o
+          FROM UNNEST(reverse_hops) AS h WITH OFFSET o
+          WHERE h.asn IS NOT NULL AND h.asn != -1
+        )
+      ), FALSE)
+    ) AS unresponsive_within_as,
+    -- TRUE when the reverse traceroute AS MEASURED contained an AS loop (A B A)
+    -- that step 04 then removed (dropped an ambiguous hop_type=4 hop) or cut
+    -- (truncated the path at the first AS re-entry, so the published path is
+    -- shorter than what was measured). ~18% of reverse paths on 2026-09-30 and
+    -- 2026-10-01 -- a property of raw revTr output, not of the published path.
+    -- Step 04 computes reverse_loop on the raw path; partitions before
+    -- 2026-09-30 (the first written by the fixed step 04, image 08d689b+) used a
+    -- wrong definition (~75% TRUE) and the raw path is not stored, so they are
+    -- NULL rather than wrong.
+    IF(partition_date >= DATE '2026-09-30', reverse_loop, NULL) AS revtr_loop_cleaned,
     ARRAY(SELECT h.asn FROM UNNEST(reverse_hops) h WHERE h.asn IS NOT NULL)
       AS as_path,
     ARRAY(SELECT h.country_code FROM UNNEST(reverse_hops) h

tests/test_events_enriched_view.py

@@ -165,6 +165,7 @@ def test_view_exposes_every_column_the_pipeline_writes():
 # what they claim to hold from the hop arrays. See
 # test_path_flags_are_derived_from_the_hops_not_the_legacy_columns.
 SUPERSEDED_BY_HOPS = [
+    "reverse_unresponsive_within_AS",
     "forward_unresponse_within_AS",
     "forward_loop",
 ]
@@ -201,28 +202,34 @@ def test_path_flags_are_derived_from_the_hops_not_the_legacy_columns():
     # a loop is an AS that reappears after a DIFFERENT AS (A B A), and
     # "unresponsive within an AS" is an AS that reappears after hops with no
     # ASN (A ... A), whether they were silent or replied but went unmapped.
-    assert sql.count("LAG(h.asn) OVER (ORDER BY o)") == 2
-    assert sql.count("o - prev_o > 1") == 1
+    assert sql.count("LAG(h.asn) OVER (ORDER BY o)") == 4
+    assert sql.count("o - prev_o > 1") == 2
 
 
-def test_reverse_flags_describe_the_measured_path_and_are_null_before_the_fix():
-    """Reverse flags describe the reverse path AS MEASURED, before step 04's
-    cleanup truncates it at the first AS re-entry, so they cannot be recomputed
-    from the published hops (a loop never survives the cleanup). The view reads
-    step 04's columns, and only for partitions written by the fixed step 04:
-    the raw path is not stored, so older values cannot be corrected and are
-    exposed as NULL rather than as the old, wrong definition.
-    """
+def test_both_directions_flag_the_published_path_with_one_definition():
+    """loop_detected / unresponsive_within_as describe the path shown beside them
+    (as_path, hops) in BOTH directions. The reverse flags used to read step 04's
+    raw-revTr columns, so ~18% of reverse paths showed loop_detected = TRUE next to
+    an as_path with no loop (step 04 truncates at the first AS re-entry)."""
     sql = _view_sql()
+    for hops in ("forward_hops", "reverse_hops"):
+        assert f"FROM UNNEST({hops}) AS h WITH OFFSET o" in sql, hops
+        assert sql.count(f"ARRAY_LENGTH({hops}) = 0") == 2, hops
+    assert "reverse_unresponsive_within_AS" not in sql
+    assert not re.search(r"\breverse_loop, NULL\) AS loop_detected", sql)
 
+
+def test_raw_revtr_loop_is_published_under_its_own_name():
+    """Step 04's reverse_loop describes the RAW revTr: it is exposed as
+    revtr_loop_cleaned, NULL before the first partition with the fixed
+    definition (the raw path is not stored, so older values cannot be fixed)."""
+    sql = _view_sql()
     gate = f"partition_date >= DATE '{REVERSE_PATH_FLAGS_FROM}'"
-    assert f"IF({gate}, reverse_loop, NULL) AS loop_detected" in sql
-    assert f"IF({gate}, reverse_unresponsive_within_AS, NULL) AS unresponsive_within_as" in sql
+    assert f"IF({gate}, reverse_loop, NULL) AS revtr_loop_cleaned" in sql
 
 
-# First partition_date written by the fixed step 04. A sentinel until the
-# fixed image has processed its first date; then set to that date.
-REVERSE_PATH_FLAGS_FROM = "9999-12-31"
+# First partition_date written by the fixed step 04 (image 08d689b and later).
+REVERSE_PATH_FLAGS_FROM = "2026-09-30"
 
 
 def test_step_04_writes_the_same_flag_definitions():