From 5e88956a10b1a014f78c1899abbe6390084594b1 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 22:09:21 +0000
Subject: [PATCH 01/26] Carry coarse label-source provenance into Under the
 Hood reports
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

TweetSafetyLabelEvent already has SafetyLabelSource (BotMaker vs Tool),
but the daily job dropped it and the report had no field to show whether
a label was applied manually. Persist a coarse category only — no
rule_id or actor_ldap — and emit it on each post-label row.

Fixes xai-org/x-algorithm#39

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../scalding/UnderTheHoodCommon.scala         |  2 +-
 .../scalding/UthDailyPostsBackfillJob.scala   | 54 +++++------
 .../scalding/UthDailyPostsJob.scala           | 94 +++++++++++++------
 under-the-hood/scalding/UthLabelSource.scala  | 29 ++++++
 .../scalding/UthUserMonthMhPublisherJob.scala | 19 ++--
 .../columns/underTheHoodReport.User.strato    |  1 +
 under-the-hood/thrift/uth_serving.thrift      |  6 ++
 7 files changed, 139 insertions(+), 66 deletions(-)
 create mode 100644 under-the-hood/scalding/UthLabelSource.scala

diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala
index 8460e6b3..c0d33a6a 100644
--- a/under-the-hood/scalding/UnderTheHoodCommon.scala
+++ b/under-the-hood/scalding/UnderTheHoodCommon.scala
@@ -101,7 +101,7 @@ object UnderTheHoodCommon {
     reducers: Int
   ): TypedPipe[UthDailyPostLabel] =
     applyReducers(
-      rows.groupBy(r => (r.userId, r.authoredYyyymmdd, r.label)),
+      rows.groupBy(r => (r.userId, r.authoredYyyymmdd, r.label, r.source)),
       reducers
     ).reduce { (a, b) =>
       if (a.asOfYyyymmdd.getOrElse(Int.MinValue) >= b.asOfYyyymmdd.getOrElse(Int.MinValue)) a
diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
index 6739b139..e8950a33 100644
--- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
@@ -100,7 +100,7 @@ class UthDailyPostsBackfillApp {
         observationMs,
         config)
         .map {
-          case (userId, authoredDay, label, asOfDay, carried, removed) =>
+          case (userId, authoredDay, label, source, asOfDay, carried, removed) =>
             val age = calendarDaysBetween(authoredDay, asOfDay)
             UthDailyPostLabel(
               userId = Some(userId),
@@ -111,7 +111,8 @@ class UthDailyPostsBackfillApp {
               asOfYyyymmdd = Some(asOfDay),
               observationAgeDays = Some(age),
               isFinal = Some(age >= config.postObservationDays),
-              postObservationDays = Some(config.postObservationDays)
+              postObservationDays = Some(config.postObservationDays),
+              source = source
             )
         }
 
@@ -204,10 +205,11 @@ object UthDailyPostsBackfillApp {
           if (expiresStr == null || expiresStr.isEmpty) Long.MaxValue else expiresStr.toLong
         if (expires <= rangeLookbackStartMs) None
         else if (createdStr == null || createdStr.isEmpty)
-          Some((tweetId, (labelName, Long.MinValue, true, expires)))
+          Some((tweetId, (labelName, Long.MinValue, true, expires, None)))
         else {
           val created = createdStr.toLong
-          if (created < rangeEndMs) Some((tweetId, (labelName, created, true, expires)))
+          if (created < rangeEndMs)
+            Some((tweetId, (labelName, created, true, expires, None)))
           else None
         }
     }
@@ -240,7 +242,7 @@ object UthDailyPostsBackfillApp {
     rangeEndMs: Long,
     observationMs: Long,
     config: UthDailyPostsConfig
-  ): TypedPipe[(Long, Int, String, Int, Long, Long)] = {
+  ): TypedPipe[(Long, Int, String, Option[String], Int, Long, Long)] = {
     val postByTweetId = applyReducers(
       posts.map {
         case (tweetId, userId, day, logicalId, createdMs) =>
@@ -261,7 +263,10 @@ object UthDailyPostsBackfillApp {
       else scopedRows.join(postByTweetId)
 
     val inHorizon = joined.flatMap {
-      case (_, ((label, eventMs, isApply, expiresMs), (logicalId, userId, day, createdMs))) =>
+      case (
+            _,
+            ((label, eventMs, isApply, expiresMs, source), (logicalId, userId, day, createdMs))
+          ) =>
         asOfDayStartsFor(createdMs, rangeStartMs, rangeEndMs, observationMs).flatMap { dStartMs =>
           val dayEndMs = dStartMs + DayMs
           val deadline = math.min(createdMs + observationMs, dayEndMs)
@@ -270,43 +275,36 @@ object UthDailyPostsBackfillApp {
           else
             UthDailyPostsApp
               .actionInHorizon(eventMs, isApply, expiresMs, createdMs, deadline)
-              .map {
-                case (everApply, ts, lastApply, exp) =>
-                  (
-                    (logicalId, userId, day, label, yyyymmdd(dStartMs), createdMs),
-                    (everApply, ts, lastApply, exp)
-                  )
+              .map { _ =>
+                (
+                  (logicalId, userId, day, label, yyyymmdd(dStartMs), createdMs),
+                  UthDailyPostsApp.actionAggFromEvent(isApply, eventMs, expiresMs, source)
+                )
               }
         }
     }
 
-    val reduced = applyReducers(inHorizon.group, config.reducers).reduce { (a, b) =>
-      val everApply = a._1 || b._1
-      val last =
-        if (a._2 != b._2) {
-          if (a._2 > b._2) (a._2, a._3, a._4) else (b._2, b._3, b._4)
-        } else if (a._3 || b._3) {
-          val exp = if (a._3) a._4 else b._4
-          (a._2, true, exp)
-        } else (a._2, false, a._4)
-      (everApply, last._1, last._2, last._3)
-    }.toTypedPipe
+    val reduced = applyReducers(inHorizon.group, config.reducers)
+      .reduce(UthDailyPostsApp.mergeActionAgg)
+      .toTypedPipe
 
     applyReducers(
       reduced.collect {
-        case ((_, userId, day, label, asOfDay, createdMs), (everApply, _, lastApply, lastExpires))
-            if everApply =>
+        case (
+              (_, userId, day, label, asOfDay, createdMs),
+              (everApply, _, lastApply, lastExpires, _, source)
+            ) if everApply =>
           val deadline = math.min(createdMs + observationMs, yyyymmddToMs(asOfDay) + DayMs)
           val removed =
             if (UthDailyPostsApp
                 .removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L
             else 0L
-          ((userId, day, label, asOfDay), (1L, removed))
+          ((userId, day, label, source, asOfDay), (1L, removed))
       }.group,
       config.reducers
     ).sum.toTypedPipe.map {
-      case ((userId, day, label, asOfDay), (carried, removed)) =>
-        (userId, day, label, asOfDay, carried, removed)
+      case ((userId, day, label, source, asOfDay), (carried, removed)) =>
+        (userId, day, label, source, asOfDay, carried, removed)
     }
   }
 }
diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index dbafd530..b45d241e 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -107,7 +107,7 @@ class UthDailyPostsApp {
         observationMs,
         config)
         .map {
-          case (userId, authoredDay, label, carried, removed) =>
+          case (userId, authoredDay, label, source, carried, removed) =>
             val age = calendarDaysBetween(authoredDay, asOfDay)
             UthDailyPostLabel(
               userId = Some(userId),
@@ -118,7 +118,8 @@ class UthDailyPostsApp {
               asOfYyyymmdd = Some(asOfDay),
               observationAgeDays = Some(age),
               isFinal = Some(age >= config.postObservationDays),
-              postObservationDays = Some(config.postObservationDays)
+              postObservationDays = Some(config.postObservationDays),
+              source = source
             )
         }
 
@@ -175,7 +176,10 @@ class UthDailyPostsApp {
 object UthDailyPostsApp {
   import UnderTheHoodCommon._
 
-  type LabelRow = (Long, (String, Long, Boolean, Long))
+  // tweetId -> (label, eventMs, isApply, expiresMs, coarseSource)
+  type LabelRow = (Long, (String, Long, Boolean, Long, Option[String]))
+  // everApply, lastTs, lastIsApply, lastExpires, lastApplyTs, lastApplySource
+  type ActionAgg = (Boolean, Long, Boolean, Long, Long, Option[String])
 
   val NsfwAdminStampedLabel = "NSFW_ADMIN_STAMPED"
   val TweetFlagColumns: Set[String] = Set("nsfwAdmin")
@@ -268,7 +272,12 @@ object UthDailyPostsApp {
         val expiresMs =
           if (isApply) event.label.flatMap(_.expiresAtMsec).getOrElse(Long.MaxValue)
           else Long.MaxValue
-        Some((event.tweetId, (name, eventMs, isApply, expiresMs)))
+        Some(
+          (
+            event.tweetId,
+            (name, eventMs, isApply, expiresMs, UthLabelSource.fromEventLabel(event.label))
+          )
+        )
       } else None
     }
 
@@ -285,8 +294,10 @@ object UthDailyPostsApp {
       else
         f.createdAtMsec match {
           case Some(created) =>
-            if (created < dayEndMs) Some((f.tweetId, (name, created, true, expires))) else None
-          case None => Some((f.tweetId, (name, Long.MinValue, true, expires)))
+            if (created < dayEndMs)
+              Some((f.tweetId, (name, created, true, expires, None)))
+            else None
+          case None => Some((f.tweetId, (name, Long.MinValue, true, expires, None)))
         }
     }
 
@@ -301,10 +312,12 @@ object UthDailyPostsApp {
           if (expiresStr == null || expiresStr.isEmpty) Long.MaxValue else expiresStr.toLong
         if (expires <= lookbackStartMs) None
         else if (createdStr == null || createdStr.isEmpty)
-          Some((tweetId, (labelName, Long.MinValue, true, expires)))
+          Some((tweetId, (labelName, Long.MinValue, true, expires, None)))
         else {
           val created = createdStr.toLong
-          if (created < dayEndMs) Some((tweetId, (labelName, created, true, expires))) else None
+          if (created < dayEndMs)
+            Some((tweetId, (labelName, created, true, expires, None)))
+          else None
         }
     }
 
@@ -322,6 +335,33 @@ object UthDailyPostsApp {
     else Some((isApply, eventMs, isApply, expiresMs))
   }
 
+  private[under_the_hood] def actionAggFromEvent(
+    isApply: Boolean,
+    eventMs: Long,
+    expiresMs: Long,
+    source: Option[String]
+  ): ActionAgg = {
+    val applyTs = if (isApply) eventMs else Long.MinValue
+    val applySrc = if (isApply) source else None
+    (isApply, eventMs, isApply, expiresMs, applyTs, applySrc)
+  }
+
+  private[under_the_hood] def mergeActionAgg(a: ActionAgg, b: ActionAgg): ActionAgg = {
+    val everApply = a._1 || b._1
+    val last =
+      if (a._2 != b._2) {
+        if (a._2 > b._2) (a._2, a._3, a._4) else (b._2, b._3, b._4)
+      } else if (a._3 || b._3) {
+        val exp = if (a._3) a._4 else b._4
+        (a._2, true, exp)
+      } else (a._2, false, a._4)
+    val (applyTs, applySrc) =
+      if (a._5 != b._5) {
+        if (a._5 > b._5) (a._5, a._6) else (b._5, b._6)
+      } else (a._5, a._6.orElse(b._6))
+    (everApply, last._1, last._2, last._3, applyTs, applySrc)
+  }
+
   private[under_the_hood] def removedAfterLastAction(
     lastIsApply: Boolean,
     lastExpiresMs: Long,
@@ -340,7 +380,7 @@ object UthDailyPostsApp {
     dayEndMs: Long,
     observationMs: Long,
     config: UthDailyPostsConfig
-  ): TypedPipe[(Long, Int, String, Long, Long)] = {
+  ): TypedPipe[(Long, Int, String, Option[String], Long, Long)] = {
     val postByTweetId = posts.map {
       case (tweetId, userId, day, logicalId, createdMs) =>
         (tweetId, (logicalId, userId, day, createdMs))
@@ -358,39 +398,35 @@ object UthDailyPostsApp {
       else scopedRows.join(postByTweetId)
 
     val inHorizon = joined.flatMap {
-      case (_, ((label, eventMs, isApply, expiresMs), (logicalId, userId, day, createdMs))) =>
+      case (_, ((label, eventMs, isApply, expiresMs, source), (logicalId, userId, day, createdMs))) =>
         val deadline = math.min(createdMs + observationMs, dayEndMs)
-        actionInHorizon(eventMs, isApply, expiresMs, createdMs, deadline).map {
-          case (everApply, ts, lastApply, exp) =>
-            ((logicalId, userId, day, label, createdMs), (everApply, ts, lastApply, exp))
+        actionInHorizon(eventMs, isApply, expiresMs, createdMs, deadline).map { _ =>
+          (
+            (logicalId, userId, day, label, createdMs),
+            actionAggFromEvent(isApply, eventMs, expiresMs, source)
+          )
         }
     }
 
-    val reduced = applyReducers(inHorizon.group, config.reducers).reduce { (a, b) =>
-      val everApply = a._1 || b._1
-      val last =
-        if (a._2 != b._2) {
-          if (a._2 > b._2) (a._2, a._3, a._4) else (b._2, b._3, b._4)
-        } else if (a._3 || b._3) {
-          val exp = if (a._3) a._4 else b._4
-          (a._2, true, exp)
-        } else (a._2, false, a._4)
-      (everApply, last._1, last._2, last._3)
-    }.toTypedPipe
+    val reduced = applyReducers(inHorizon.group, config.reducers)
+      .reduce(mergeActionAgg)
+      .toTypedPipe
 
     applyReducers(
       reduced.collect {
-        case ((_, userId, day, label, createdMs), (everApply, _, lastApply, lastExpires))
-            if everApply =>
+        case (
+              (_, userId, day, label, createdMs),
+              (everApply, _, lastApply, lastExpires, _, source)
+            ) if everApply =>
           val deadline = math.min(createdMs + observationMs, dayEndMs)
           val removed =
             if (removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L else 0L
-          ((userId, day, label), (1L, removed))
+          ((userId, day, label, source), (1L, removed))
       }.group,
       config.reducers
     ).sum.toTypedPipe.map {
-      case ((userId, day, label), (carried, removed)) =>
-        (userId, day, label, carried, removed)
+      case ((userId, day, label, source), (carried, removed)) =>
+        (userId, day, label, source, carried, removed)
     }
   }
 }
diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
new file mode 100644
index 00000000..66ba59ea
--- /dev/null
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -0,0 +1,29 @@
+package com.twitter.visibility.under_the_hood
+
+import com.twitter.spam.rtf.thriftscala.SafetyLabel
+import com.twitter.spam.rtf.thriftscala.SafetyLabelSource
+
+object UthLabelSource {
+
+  val Automated = "automated"
+  val Manual = "manual"
+  val Other = "other"
+
+  def fromEventLabel(label: Option[SafetyLabel]): Option[String] =
+    label.flatMap(_.safetyLabelSource).map(coarseCategory)
+
+  /**
+   * Maps the per-event SafetyLabelSource union to a public category.
+   * Does not emit rule_id, actor_ldap, or agent_tool.
+   *
+   * BotMakerAction = automated systems; ToolAction = manual/tool apply.
+   * Any other set variant (including LLM annotations if present on the IDL)
+   * is "other" so unknown union members stay compile-safe.
+   */
+  private[under_the_hood] def coarseCategory(source: SafetyLabelSource): String =
+    source match {
+      case SafetyLabelSource.BotMakerAction(_) => Automated
+      case SafetyLabelSource.ToolAction(_) => Manual
+      case _ => Other
+    }
+}
diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
index 31ed6037..e88abdde 100644
--- a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
+++ b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
@@ -165,7 +165,10 @@ object UthUserMonthMhPublisherApp {
           label <- row.label
           carried <- row.carried
           removed <- row.removed
-        } yield ((userId, monthBucket(day)), (label, dayOfMonth(day), carried, removed))
+        } yield (
+          (userId, monthBucket(day)),
+          (label, row.source, dayOfMonth(day), carried, removed)
+        )
       },
       reducers
     )
@@ -202,24 +205,24 @@ object UthUserMonthMhPublisherApp {
 
           val postLabelAgg = labelsOpt
             .getOrElse(Nil)
-            .groupBy { case (label, _, _, _) => label }
+            .groupBy { case (label, source, _, _, _) => (label, source) }
             .map {
-              case (label, rows) =>
+              case ((label, source), rows) =>
                 val days = rows
-                  .groupBy { case (_, day, _, _) => day }
+                  .groupBy { case (_, _, day, _, _) => day }
                   .map {
                     case (day, dayRows) =>
                       val best = dayRows.maxBy {
-                        case (_, _, carried, removed) => (carried, removed)
+                        case (_, _, _, carried, removed) => (carried, removed)
                       }
-                      UthDayCarriedRemoved(Some(day), Some(best._3), Some(best._4))
+                      UthDayCarriedRemoved(Some(day), Some(best._4), Some(best._5))
                   }
                   .toList
                   .sortBy(_.dayOfMonth.getOrElse(0))
-                UthPostLabelAggregate(Some(label), Some(days))
+                UthPostLabelAggregate(Some(label), Some(days), source)
             }
             .toList
-            .sortBy(_.label.getOrElse(""))
+            .sortBy(a => (a.label.getOrElse(""), a.source.getOrElse("")))
 
           val accountLabelAgg = accountOpt
             .getOrElse(Nil)
diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index 6be87004..cb84c425 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -194,6 +194,7 @@ def buildReportJson(
           label = underTheHoodLabels.postLabelName(raw),
           about = underTheHoodLabels.postLabelAbout(raw),
           effect = underTheHoodLabels.postLabelEffect(raw),
+          source = agg.source.filter { s => s.nonEmpty }.getOrElse("unavailable"),
           posts = posts,
           totalPostsInMonth = postCount,
           percentageOfPosts = formatPercentage(posts, postCount),
diff --git a/under-the-hood/thrift/uth_serving.thrift b/under-the-hood/thrift/uth_serving.thrift
index 299c551f..9887ddfd 100644
--- a/under-the-hood/thrift/uth_serving.thrift
+++ b/under-the-hood/thrift/uth_serving.thrift
@@ -53,6 +53,9 @@ enum UthFollowerClass {
 struct UthPostLabelAggregate {
   1: optional string label (personalDataType = 'TweetSafetyLabels')
   2: optional list<UthDayCarriedRemoved> days
+  // Coarse SafetyLabelSource category: automated | manual | other.
+  // Unset when the daily event/snapshot did not carry a source.
+  3: optional string source
 }(persisted = 'true', hasPersonalData = 'true')
 
 struct UthAccountLabelAggregate {
@@ -129,6 +132,9 @@ struct UthDailyPostLabel {
   7: optional i32 observationAgeDays
   8: optional bool isFinal
   9: optional i32 postObservationDays
+  // Coarse SafetyLabelSource category: automated | manual | other.
+  // Unset for snapshot/flag rows and events with no source variant.
+  10: optional string source
 }(persisted = 'true', hasPersonalData = 'true')
 
 struct UthDailyAccountLabel {

From 816315c6ed15afe0402273fc4fff401edf9eb906 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 22:57:52 +0000
Subject: [PATCH 02/26] Tighten UTH reportJson source to
 automated|manual|llm|unknown

Replace other/unavailable with unknown. reportJson clamps to the four
tokens. GrokAnnotationAction is not on the published spam.rtf IDL the
jobs compile against, so it is not matched and folds to unknown.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala    | 17 ++++++++++-------
 .../columns/underTheHoodReport.User.strato      | 10 +++++++++-
 under-the-hood/thrift/uth_serving.thrift        |  4 ++--
 3 files changed, 21 insertions(+), 10 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 66ba59ea..25d57453 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -7,23 +7,26 @@ object UthLabelSource {
 
   val Automated = "automated"
   val Manual = "manual"
-  val Other = "other"
+  val Llm = "llm"
+  val Unknown = "unknown"
 
   def fromEventLabel(label: Option[SafetyLabel]): Option[String] =
     label.flatMap(_.safetyLabelSource).map(coarseCategory)
 
   /**
-   * Maps the per-event SafetyLabelSource union to a public category.
-   * Does not emit rule_id, actor_ldap, or agent_tool.
+   * Public reportJson tokens: automated | manual | llm | unknown.
+   * Does not emit rule_id, actor_ldap, agent_tool, or VF-client type names.
    *
-   * BotMakerAction = automated systems; ToolAction = manual/tool apply.
-   * Any other set variant (including LLM annotations if present on the IDL)
-   * is "other" so unknown union members stay compile-safe.
+   * The jobs compile against spam.rtf SafetyLabelSource. Published in-repo
+   * usage of that IDL only names BotMakerAction and ToolAction, so those
+   * map to automated/manual. GrokAnnotationAction exists on unpublished
+   * xai_x_thrift (VF client), not this IDL — it is not matched here and
+   * folds to unknown rather than inventing a case that may not compile.
    */
   private[under_the_hood] def coarseCategory(source: SafetyLabelSource): String =
     source match {
       case SafetyLabelSource.BotMakerAction(_) => Automated
       case SafetyLabelSource.ToolAction(_) => Manual
-      case _ => Other
+      case _ => Unknown
     }
 }
diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index cb84c425..e7fa8eae 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -155,6 +155,14 @@ def sumDayCounts(days: Option[Seq[UthDayCount]]): Long =
     acc + d.count.getOrElse(0L)
   }
 
+def postLabelSource(source: Option[String]): String =
+  source match {
+    case Some("automated") => "automated"
+    case Some("manual") => "manual"
+    case Some("llm") => "llm"
+    case _ => "unknown"
+  }
+
 def formatPercentage(numerator: Long, denominator: Long): String = {
   if (denominator <= 0L) {
     "0%"
@@ -194,7 +202,7 @@ def buildReportJson(
           label = underTheHoodLabels.postLabelName(raw),
           about = underTheHoodLabels.postLabelAbout(raw),
           effect = underTheHoodLabels.postLabelEffect(raw),
-          source = agg.source.filter { s => s.nonEmpty }.getOrElse("unavailable"),
+          source = postLabelSource(agg.source),
           posts = posts,
           totalPostsInMonth = postCount,
           percentageOfPosts = formatPercentage(posts, postCount),
diff --git a/under-the-hood/thrift/uth_serving.thrift b/under-the-hood/thrift/uth_serving.thrift
index 9887ddfd..4dafe7c1 100644
--- a/under-the-hood/thrift/uth_serving.thrift
+++ b/under-the-hood/thrift/uth_serving.thrift
@@ -53,7 +53,7 @@ enum UthFollowerClass {
 struct UthPostLabelAggregate {
   1: optional string label (personalDataType = 'TweetSafetyLabels')
   2: optional list<UthDayCarriedRemoved> days
-  // Coarse SafetyLabelSource category: automated | manual | other.
+  // Coarse SafetyLabelSource category: automated | manual | llm | unknown.
   // Unset when the daily event/snapshot did not carry a source.
   3: optional string source
 }(persisted = 'true', hasPersonalData = 'true')
@@ -132,7 +132,7 @@ struct UthDailyPostLabel {
   7: optional i32 observationAgeDays
   8: optional bool isFinal
   9: optional i32 postObservationDays
-  // Coarse SafetyLabelSource category: automated | manual | other.
+  // Coarse SafetyLabelSource category: automated | manual | llm | unknown.
   // Unset for snapshot/flag rows and events with no source variant.
   10: optional string source
 }(persisted = 'true', hasPersonalData = 'true')

From e43ac6f6c39bee082bec5d19a80f0e5973e91049 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 22:59:50 +0000
Subject: [PATCH 03/26] Collapse unset UTH sources and map Grok to llm without
 a hard case

Persist only automated|manual|llm so snapshot/flag/unmapped rows share
one unset key and do not split into duplicate unknown report rows.
Detect a GrokAnnotation* union via Product prefix so llm works when
that variant exists without naming a case that may be absent from
spam.rtf.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../scalding/UnderTheHoodCommon.scala         |  3 +-
 .../scalding/UthDailyPostsBackfillJob.scala   |  2 +-
 .../scalding/UthDailyPostsJob.scala           |  2 +-
 under-the-hood/scalding/UthLabelSource.scala  | 36 +++++++++++++------
 .../scalding/UthUserMonthMhPublisherJob.scala |  2 +-
 5 files changed, 31 insertions(+), 14 deletions(-)

diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala
index c0d33a6a..e8d6f6ab 100644
--- a/under-the-hood/scalding/UnderTheHoodCommon.scala
+++ b/under-the-hood/scalding/UnderTheHoodCommon.scala
@@ -101,7 +101,8 @@ object UnderTheHoodCommon {
     reducers: Int
   ): TypedPipe[UthDailyPostLabel] =
     applyReducers(
-      rows.groupBy(r => (r.userId, r.authoredYyyymmdd, r.label, r.source)),
+      rows.groupBy(r =>
+        (r.userId, r.authoredYyyymmdd, r.label, UthLabelSource.persistToken(r.source))),
       reducers
     ).reduce { (a, b) =>
       if (a.asOfYyyymmdd.getOrElse(Int.MinValue) >= b.asOfYyyymmdd.getOrElse(Int.MinValue)) a
diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
index e8950a33..5dda0534 100644
--- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
@@ -112,7 +112,7 @@ class UthDailyPostsBackfillApp {
               observationAgeDays = Some(age),
               isFinal = Some(age >= config.postObservationDays),
               postObservationDays = Some(config.postObservationDays),
-              source = source
+              source = UthLabelSource.persistToken(source)
             )
         }
 
diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index b45d241e..e53a9acb 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -119,7 +119,7 @@ class UthDailyPostsApp {
               observationAgeDays = Some(age),
               isFinal = Some(age >= config.postObservationDays),
               postObservationDays = Some(config.postObservationDays),
-              source = source
+              source = UthLabelSource.persistToken(source)
             )
         }
 
diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 25d57453..9f506805 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -11,22 +11,38 @@ object UthLabelSource {
   val Unknown = "unknown"
 
   def fromEventLabel(label: Option[SafetyLabel]): Option[String] =
-    label.flatMap(_.safetyLabelSource).map(coarseCategory)
+    persistToken(label.flatMap(_.safetyLabelSource).flatMap(coarseCategory))
+
+  /** Persist only automated | manual | llm. Unset/unmapped stay empty. */
+  def persistToken(source: Option[String]): Option[String] =
+    source match {
+      case Some(Automated) => Some(Automated)
+      case Some(Manual) => Some(Manual)
+      case Some(Llm) => Some(Llm)
+      case _ => None
+    }
 
   /**
    * Public reportJson tokens: automated | manual | llm | unknown.
    * Does not emit rule_id, actor_ldap, agent_tool, or VF-client type names.
    *
-   * The jobs compile against spam.rtf SafetyLabelSource. Published in-repo
-   * usage of that IDL only names BotMakerAction and ToolAction, so those
-   * map to automated/manual. GrokAnnotationAction exists on unpublished
-   * xai_x_thrift (VF client), not this IDL — it is not matched here and
-   * folds to unknown rather than inventing a case that may not compile.
+   * Named cases are only those present on published spam.rtf usage
+   * (BotMakerAction, ToolAction). A Grok/LLM union member is detected by
+   * Product prefix so this compiles if that case is absent from the IDL.
    */
-  private[under_the_hood] def coarseCategory(source: SafetyLabelSource): String =
+  private[under_the_hood] def coarseCategory(source: SafetyLabelSource): Option[String] =
     source match {
-      case SafetyLabelSource.BotMakerAction(_) => Automated
-      case SafetyLabelSource.ToolAction(_) => Manual
-      case _ => Unknown
+      case SafetyLabelSource.BotMakerAction(_) => Some(Automated)
+      case SafetyLabelSource.ToolAction(_) => Some(Manual)
+      case other if isLlmVariant(other) => Some(Llm)
+      case _ => None
+    }
+
+  private def isLlmVariant(source: SafetyLabelSource): Boolean = {
+    val name = source match {
+      case p: Product => p.productPrefix
+      case _ => ""
     }
+    name == "GrokAnnotationAction" || name.startsWith("GrokAnnotation")
+  }
 }
diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
index e88abdde..b9c58f90 100644
--- a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
+++ b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
@@ -167,7 +167,7 @@ object UthUserMonthMhPublisherApp {
           removed <- row.removed
         } yield (
           (userId, monthBucket(day)),
-          (label, row.source, dayOfMonth(day), carried, removed)
+          (label, UthLabelSource.persistToken(row.source), dayOfMonth(day), carried, removed)
         )
       },
       reducers

From a0178ea979768a511e68bf5c6c882cf81350e3cd Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 22:59:54 +0000
Subject: [PATCH 04/26] Note that unset UTH source becomes reportJson unknown

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/thrift/uth_serving.thrift | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/under-the-hood/thrift/uth_serving.thrift b/under-the-hood/thrift/uth_serving.thrift
index 4dafe7c1..f3a88410 100644
--- a/under-the-hood/thrift/uth_serving.thrift
+++ b/under-the-hood/thrift/uth_serving.thrift
@@ -53,8 +53,7 @@ enum UthFollowerClass {
 struct UthPostLabelAggregate {
   1: optional string label (personalDataType = 'TweetSafetyLabels')
   2: optional list<UthDayCarriedRemoved> days
-  // Coarse SafetyLabelSource category: automated | manual | llm | unknown.
-  // Unset when the daily event/snapshot did not carry a source.
+  // Persisted token: automated | manual | llm. Unset → reportJson unknown.
   3: optional string source
 }(persisted = 'true', hasPersonalData = 'true')
 
@@ -132,8 +131,7 @@ struct UthDailyPostLabel {
   7: optional i32 observationAgeDays
   8: optional bool isFinal
   9: optional i32 postObservationDays
-  // Coarse SafetyLabelSource category: automated | manual | llm | unknown.
-  // Unset for snapshot/flag rows and events with no source variant.
+  // Persisted token: automated | manual | llm. Unset → reportJson unknown.
   10: optional string source
 }(persisted = 'true', hasPersonalData = 'true')
 

From f0dd4fd6ec22d51920622901e057665c67a48875 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:03:46 +0000
Subject: [PATCH 05/26] Coalesce leftover UTH source tokens with unset

Treat stored unknown/other/unavailable as unset so earlier daily rows
do not split aggregates. Also match GrokAnnotation on the union class
name when the value is not a Product.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 9f506805..762bbf55 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -19,6 +19,7 @@ object UthLabelSource {
       case Some(Automated) => Some(Automated)
       case Some(Manual) => Some(Manual)
       case Some(Llm) => Some(Llm)
+      case Some(Unknown) | Some("other") | Some("unavailable") => None
       case _ => None
     }
 
@@ -39,10 +40,14 @@ object UthLabelSource {
     }
 
   private def isLlmVariant(source: SafetyLabelSource): Boolean = {
-    val name = source match {
+    val productName = source match {
       case p: Product => p.productPrefix
       case _ => ""
     }
-    name == "GrokAnnotationAction" || name.startsWith("GrokAnnotation")
+    val typeName = source.getClass.getSimpleName
+    isGrokName(productName) || isGrokName(typeName)
   }
+
+  private def isGrokName(name: String): Boolean =
+    name == "GrokAnnotationAction" || name.contains("GrokAnnotation")
 }

From 4185ebf192be6b5d73e2145f11ed5a9ef13e4c00 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:05:02 +0000
Subject: [PATCH 06/26] Rewrite latest-as-of UTH rows to a persistable source
 token

Group-by already coalesced leftover tokens; the kept row still carried
the raw string. Normalize after reduce so month assembly always sees
automated|manual|llm or unset, including single-row groups.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UnderTheHoodCommon.scala | 1 +
 1 file changed, 1 insertion(+)

diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala
index e8d6f6ab..91a51ab9 100644
--- a/under-the-hood/scalding/UnderTheHoodCommon.scala
+++ b/under-the-hood/scalding/UnderTheHoodCommon.scala
@@ -108,6 +108,7 @@ object UnderTheHoodCommon {
       if (a.asOfYyyymmdd.getOrElse(Int.MinValue) >= b.asOfYyyymmdd.getOrElse(Int.MinValue)) a
       else b
     }.values
+      .map(r => r.copy(source = UthLabelSource.persistToken(r.source)))
 
   def parseUserIds(args: Args): Set[Long] = {
     val raw = args.list("userIds").flatMap(_.split(",")).map(_.trim).filter(_.nonEmpty)

From 50dddaf04b7976df7b1166c1882ae4c9a1229d75 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:06:35 +0000
Subject: [PATCH 07/26] Clamp UTH source before daily label counts

Count keys used the raw last-apply source. Leftover tokens that persist
to unset could split a label, then collapse at write and drop counts.
Normalize in actionAgg and the count key so one persistable token is
counted once.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthDailyPostsBackfillJob.scala | 2 +-
 under-the-hood/scalding/UthDailyPostsJob.scala         | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
index 5dda0534..d9f8803d 100644
--- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
@@ -299,7 +299,7 @@ object UthDailyPostsBackfillApp {
             if (UthDailyPostsApp
                 .removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L
             else 0L
-          ((userId, day, label, source, asOfDay), (1L, removed))
+          ((userId, day, label, UthLabelSource.persistToken(source), asOfDay), (1L, removed))
       }.group,
       config.reducers
     ).sum.toTypedPipe.map {
diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index e53a9acb..0912213c 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -342,7 +342,7 @@ object UthDailyPostsApp {
     source: Option[String]
   ): ActionAgg = {
     val applyTs = if (isApply) eventMs else Long.MinValue
-    val applySrc = if (isApply) source else None
+    val applySrc = if (isApply) UthLabelSource.persistToken(source) else None
     (isApply, eventMs, isApply, expiresMs, applyTs, applySrc)
   }
 
@@ -421,7 +421,7 @@ object UthDailyPostsApp {
           val deadline = math.min(createdMs + observationMs, dayEndMs)
           val removed =
             if (removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L else 0L
-          ((userId, day, label, source), (1L, removed))
+          ((userId, day, label, UthLabelSource.persistToken(source)), (1L, removed))
       }.group,
       config.reducers
     ).sum.toTypedPipe.map {

From d19e74b2a0d799bd75d34b18972beed42f867cde Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:07:57 +0000
Subject: [PATCH 08/26] Normalize UTH source tokens before persist and report

Trim and lowercase so Automated / ' llm ' do not miss the allowlist,
split aggregates, or render as unknown.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala                 | 2 +-
 under-the-hood/strato/columns/underTheHoodReport.User.strato | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 762bbf55..f8af4181 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -15,7 +15,7 @@ object UthLabelSource {
 
   /** Persist only automated | manual | llm. Unset/unmapped stay empty. */
   def persistToken(source: Option[String]): Option[String] =
-    source match {
+    source.map(_.trim.toLowerCase) match {
       case Some(Automated) => Some(Automated)
       case Some(Manual) => Some(Manual)
       case Some(Llm) => Some(Llm)
diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index e7fa8eae..185ab8d6 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -156,7 +156,7 @@ def sumDayCounts(days: Option[Seq[UthDayCount]]): Long =
   }
 
 def postLabelSource(source: Option[String]): String =
-  source match {
+  source.map { s => s.trim.toLowerCase } match {
     case Some("automated") => "automated"
     case Some("manual") => "manual"
     case Some("llm") => "llm"

From bafda6056b9b0a63ea1033598c8bcafaa8a693c6 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:09:19 +0000
Subject: [PATCH 09/26] Avoid Scrooge .copy when rewriting latest-as-of source

UthDailyPostLabel is constructed with named args everywhere else in
this lane. Reconstruct the row the same way so source rewrite does not
depend on a case-class copy method the IDL compiler may not emit.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UnderTheHoodCommon.scala | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala
index 91a51ab9..e638766f 100644
--- a/under-the-hood/scalding/UnderTheHoodCommon.scala
+++ b/under-the-hood/scalding/UnderTheHoodCommon.scala
@@ -108,7 +108,20 @@ object UnderTheHoodCommon {
       if (a.asOfYyyymmdd.getOrElse(Int.MinValue) >= b.asOfYyyymmdd.getOrElse(Int.MinValue)) a
       else b
     }.values
-      .map(r => r.copy(source = UthLabelSource.persistToken(r.source)))
+      .map { r =>
+        UthDailyPostLabel(
+          userId = r.userId,
+          authoredYyyymmdd = r.authoredYyyymmdd,
+          label = r.label,
+          carried = r.carried,
+          removed = r.removed,
+          asOfYyyymmdd = r.asOfYyyymmdd,
+          observationAgeDays = r.observationAgeDays,
+          isFinal = r.isFinal,
+          postObservationDays = r.postObservationDays,
+          source = UthLabelSource.persistToken(r.source)
+        )
+      }
 
   def parseUserIds(args: Args): Set[Long] = {
     val raw = args.list("userIds").flatMap(_.split(",")).map(_.trim).filter(_.nonEmpty)

From b29d20364034963b56c6573b088ef13a16011e35 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:10:52 +0000
Subject: [PATCH 10/26] Write month post-label source with named Scrooge fields

Positional UthPostLabelAggregate(label, days, source) depends on field
order. Use named args and persistToken so the MH row matches the daily
rewrite path.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthUserMonthMhPublisherJob.scala | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
index b9c58f90..6cf6c2fc 100644
--- a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
+++ b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
@@ -219,7 +219,11 @@ object UthUserMonthMhPublisherApp {
                   }
                   .toList
                   .sortBy(_.dayOfMonth.getOrElse(0))
-                UthPostLabelAggregate(Some(label), Some(days), source)
+                UthPostLabelAggregate(
+                  label = Some(label),
+                  days = Some(days),
+                  source = UthLabelSource.persistToken(source)
+                )
             }
             .toList
             .sortBy(a => (a.label.getOrElse(""), a.source.getOrElse("")))

From 6d0d4b718f0863665222b9eb5781a2f2a5f99565 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:13:01 +0000
Subject: [PATCH 11/26] Dedupe UTH reportJson post labels that clamp to the
 same source

Leftover month rows stored as unset vs unknown/other collapse to the
same report token. Keep one row per (label, source), taking the larger
post count so the same posts are not shown twice.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../columns/underTheHoodReport.User.strato     | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index 185ab8d6..28184d9e 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -209,6 +209,24 @@ def buildReportJson(
         }
       }
     }
+    .groupBy { row => (row.label, row.source) }
+    .flatMap { g =>
+      val rows = g._2
+      rows.headOption.map { head =>
+        val posts = rows.foldLeft(0L) { (acc, r) =>
+          if (r.posts > acc) { r.posts } else { acc }
+        }
+        {
+          label = head.label,
+          about = head.about,
+          effect = head.effect,
+          source = head.source,
+          posts = posts,
+          totalPostsInMonth = postCount,
+          percentageOfPosts = formatPercentage(posts, postCount),
+        }
+      }
+    }
 
   val accountLabels = userMonth.accountLabelAgg
     .getOrElse(Seq.empty)

From 58445ef45d64effdbd2f5194db1a6a1a4c6ba187 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:13:53 +0000
Subject: [PATCH 12/26] Sort UTH reportJson post labels by label and source

groupBy after source dedupe left row order hash-dependent. Sort so
the same month aggregate always emits a stable reportJson list.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/strato/columns/underTheHoodReport.User.strato | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index 28184d9e..a9066fce 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -227,6 +227,8 @@ def buildReportJson(
         }
       }
     }
+    .toList
+    .sortBy { row => (row.label, row.source) }
 
   val accountLabels = userMonth.accountLabelAgg
     .getOrElse(Seq.empty)

From 418b20c20d30b1b42a9c2b7f2987905b6c02a132 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:15:26 +0000
Subject: [PATCH 13/26] Tighten Grok union name match for UTH llm source

Match the last simple class segment with startsWith(GrokAnnotation)
instead of contains(), so an unrelated name cannot be classified llm.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index f8af4181..154469de 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -29,7 +29,7 @@ object UthLabelSource {
    *
    * Named cases are only those present on published spam.rtf usage
    * (BotMakerAction, ToolAction). A Grok/LLM union member is detected by
-   * Product prefix so this compiles if that case is absent from the IDL.
+   * simple class name so this compiles if that case is absent from the IDL.
    */
   private[under_the_hood] def coarseCategory(source: SafetyLabelSource): Option[String] =
     source match {
@@ -48,6 +48,8 @@ object UthLabelSource {
     isGrokName(productName) || isGrokName(typeName)
   }
 
-  private def isGrokName(name: String): Boolean =
-    name == "GrokAnnotationAction" || name.contains("GrokAnnotation")
+  private def isGrokName(name: String): Boolean = {
+    val simple = name.split('.').last.split('$').last
+    simple == "GrokAnnotationAction" || simple.startsWith("GrokAnnotation")
+  }
 }

From e3b92eb94cc6d0e6158b19fad561f7f7cf739758 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:16:56 +0000
Subject: [PATCH 14/26] Leave tweet-flag UTH rows with an explicit unset source

Flag inputs have no SafetyLabelSource. Set source = None at both daily
and backfill writes so a future thrift default cannot invent provenance.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthDailyPostsBackfillJob.scala | 3 ++-
 under-the-hood/scalding/UthDailyPostsJob.scala         | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
index d9f8803d..8358064f 100644
--- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
@@ -136,7 +136,8 @@ class UthDailyPostsBackfillApp {
               asOfYyyymmdd = Some(authoredDay),
               observationAgeDays = Some(0),
               isFinal = Some(true),
-              postObservationDays = Some(config.postObservationDays)
+              postObservationDays = Some(config.postObservationDays),
+              source = None
             )
         }
 
diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index 0912213c..04d1a239 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -142,7 +142,8 @@ class UthDailyPostsApp {
               asOfYyyymmdd = Some(asOfDay),
               observationAgeDays = Some(0),
               isFinal = Some(true),
-              postObservationDays = Some(config.postObservationDays)
+              postObservationDays = Some(config.postObservationDays),
+              source = None
             )
         }
 

From 4bdb34ed95238e1a3b581ff8bf8523a7e858c20a Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:18:39 +0000
Subject: [PATCH 15/26] Merge leftover UTH post-label days instead of taking
 max posts

max(posts) dropped disjoint leftover days after source clamp. Union
days and keep the larger carried/removed per day, matching month
assembly, then sum.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../columns/underTheHoodReport.User.strato    | 57 ++++++++++---------
 1 file changed, 31 insertions(+), 26 deletions(-)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index a9066fce..b294638a 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -155,7 +155,15 @@ def sumDayCounts(days: Option[Seq[UthDayCount]]): Long =
     acc + d.count.getOrElse(0L)
   }
 
-def postLabelSource(source: Option[String]): String =
+def bestDayRow(a: UthDayCarriedRemoved, b: UthDayCarriedRemoved): UthDayCarriedRemoved = {
+  val ac = a.carried.getOrElse(0L)
+  val ar = a.removed.getOrElse(0L)
+  val bc = b.carried.getOrElse(0L)
+  val br = b.removed.getOrElse(0L)
+  if (ac > bc || (ac == bc && ar >= br)) { a } else { b }
+}
+
+def postLabelSource(source: Option[String]): String = {
   source.map { s => s.trim.toLowerCase } match {
     case Some("automated") => "automated"
     case Some("manual") => "manual"
@@ -197,35 +205,32 @@ def buildReportJson(
     .getOrElse(Seq.empty)
     .flatMap { agg =>
       agg.label.filter(underTheHoodLabels.isPostLabel).map { raw =>
-        val posts = sumCarried(agg.days)
-        {
-          label = underTheHoodLabels.postLabelName(raw),
-          about = underTheHoodLabels.postLabelAbout(raw),
-          effect = underTheHoodLabels.postLabelEffect(raw),
-          source = postLabelSource(agg.source),
-          posts = posts,
-          totalPostsInMonth = postCount,
-          percentageOfPosts = formatPercentage(posts, postCount),
-        }
+        (underTheHoodLabels.postLabelName(raw), postLabelSource(agg.source), agg)
       }
     }
-    .groupBy { row => (row.label, row.source) }
+    .groupBy { t => (t._1, t._2) }
     .flatMap { g =>
-      val rows = g._2
-      rows.headOption.map { head =>
-        val posts = rows.foldLeft(0L) { (acc, r) =>
-          if (r.posts > acc) { r.posts } else { acc }
-        }
-        {
-          label = head.label,
-          about = head.about,
-          effect = head.effect,
-          source = head.source,
-          posts = posts,
-          totalPostsInMonth = postCount,
-          percentageOfPosts = formatPercentage(posts, postCount),
+      val name = g._1._1
+      val source = g._1._2
+      val days = g._2
+        .flatMap { t => t._3.days.getOrElse(Seq.empty) }
+        .groupBy { d => d.dayOfMonth.getOrElse(0) }
+        .flatMap { dg =>
+          dg._2.headOption.map { first =>
+            dg._2.foldLeft(first) { (best, d) => bestDayRow(best, d) }
+          }
         }
-      }
+        .toList
+      val posts = sumCarried(Some(days))
+      Some({
+        label = name,
+        about = underTheHoodLabels.postLabelAbout(name),
+        effect = underTheHoodLabels.postLabelEffect(name),
+        source = source,
+        posts = posts,
+        totalPostsInMonth = postCount,
+        percentageOfPosts = formatPercentage(posts, postCount),
+      })
     }
     .toList
     .sortBy { row => (row.label, row.source) }

From 90a0711fb5d45b95e9ab94f992bdfd447ef5f4c9 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:20:02 +0000
Subject: [PATCH 16/26] Avoid Map.flatMap when merging UTH leftover source rows

groupBy returns a Map; flatMap there expects key/value pairs and can
drop or fail to type the report rows. Convert toList first.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/strato/columns/underTheHoodReport.User.strato | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index b294638a..a2ed96fb 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -209,18 +209,19 @@ def buildReportJson(
       }
     }
     .groupBy { t => (t._1, t._2) }
+    .toList
     .flatMap { g =>
       val name = g._1._1
       val source = g._1._2
       val days = g._2
         .flatMap { t => t._3.days.getOrElse(Seq.empty) }
         .groupBy { d => d.dayOfMonth.getOrElse(0) }
+        .toList
         .flatMap { dg =>
           dg._2.headOption.map { first =>
             dg._2.foldLeft(first) { (best, d) => bestDayRow(best, d) }
           }
         }
-        .toList
       val posts = sumCarried(Some(days))
       Some({
         label = name,

From 5d842d57f35bd52cfc03cadf387bce6a79a5ceaa Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:21:38 +0000
Subject: [PATCH 17/26] Fix Grok class-name split: $ is regex end-of-string

String.split('$') does not split inner-class names. Take the segment
after the last '.' or '$' and also check getName so binary names like
SafetyLabelSource$GrokAnnotationAction still map to llm.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 154469de..023da58d 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -44,12 +44,19 @@ object UthLabelSource {
       case p: Product => p.productPrefix
       case _ => ""
     }
-    val typeName = source.getClass.getSimpleName
-    isGrokName(productName) || isGrokName(typeName)
+    val cls = source.getClass
+    isGrokName(productName) || isGrokName(cls.getSimpleName) || isGrokName(cls.getName)
   }
 
   private def isGrokName(name: String): Boolean = {
-    val simple = name.split('.').last.split('$').last
+    val afterDot = name.lastIndexOf('.') match {
+      case -1 => name
+      case i => name.substring(i + 1)
+    }
+    val simple = afterDot.lastIndexOf('$') match {
+      case -1 => afterDot
+      case i => afterDot.substring(i + 1)
+    }
     simple == "GrokAnnotationAction" || simple.startsWith("GrokAnnotation")
   }
 }

From fb5ece60bb30995f7b953bdc4093100665587ff8 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:23:04 +0000
Subject: [PATCH 18/26] Pattern-match UTH leftover source groups instead of
 _1._1

Nested tuple accessors can silently pick the wrong slot if the group
shape changes. Destructure (name, source) and day rows by case.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../strato/columns/underTheHoodReport.User.strato    | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index a2ed96fb..2476501b 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -210,16 +210,14 @@ def buildReportJson(
     }
     .groupBy { t => (t._1, t._2) }
     .toList
-    .flatMap { g =>
-      val name = g._1._1
-      val source = g._1._2
-      val days = g._2
+    .flatMap { case ((name, source), rows) =>
+      val days = rows
         .flatMap { t => t._3.days.getOrElse(Seq.empty) }
         .groupBy { d => d.dayOfMonth.getOrElse(0) }
         .toList
-        .flatMap { dg =>
-          dg._2.headOption.map { first =>
-            dg._2.foldLeft(first) { (best, d) => bestDayRow(best, d) }
+        .flatMap { case (_, dayRows) =>
+          dayRows.headOption.map { first =>
+            dayRows.foldLeft(first) { (best, d) => bestDayRow(best, d) }
           }
         }
       val posts = sumCarried(Some(days))

From 1a097b1eaffbce99fe6a230567c251d06a70d100 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:25:11 +0000
Subject: [PATCH 19/26] Close postLabelSource and look up about/effect by raw
 id

postLabelSource was missing its function closer, so formatPercentage
was nested inside it. Post-label about/effect must use the raw label
id, matching the account path; display name stays on label only.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../columns/underTheHoodReport.User.strato    | 25 +++++++++++--------
 1 file changed, 14 insertions(+), 11 deletions(-)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index 2476501b..bd40aa26 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -170,6 +170,7 @@ def postLabelSource(source: Option[String]): String = {
     case Some("llm") => "llm"
     case _ => "unknown"
   }
+}
 
 def formatPercentage(numerator: Long, denominator: Long): String = {
   if (denominator <= 0L) {
@@ -205,14 +206,14 @@ def buildReportJson(
     .getOrElse(Seq.empty)
     .flatMap { agg =>
       agg.label.filter(underTheHoodLabels.isPostLabel).map { raw =>
-        (underTheHoodLabels.postLabelName(raw), postLabelSource(agg.source), agg)
+        (underTheHoodLabels.postLabelName(raw), postLabelSource(agg.source), raw, agg)
       }
     }
     .groupBy { t => (t._1, t._2) }
     .toList
     .flatMap { case ((name, source), rows) =>
       val days = rows
-        .flatMap { t => t._3.days.getOrElse(Seq.empty) }
+        .flatMap { case (_, _, _, agg) => agg.days.getOrElse(Seq.empty) }
         .groupBy { d => d.dayOfMonth.getOrElse(0) }
         .toList
         .flatMap { case (_, dayRows) =>
@@ -221,15 +222,17 @@ def buildReportJson(
           }
         }
       val posts = sumCarried(Some(days))
-      Some({
-        label = name,
-        about = underTheHoodLabels.postLabelAbout(name),
-        effect = underTheHoodLabels.postLabelEffect(name),
-        source = source,
-        posts = posts,
-        totalPostsInMonth = postCount,
-        percentageOfPosts = formatPercentage(posts, postCount),
-      })
+      rows.headOption.map { case (_, _, raw, _) =>
+        {
+          label = name,
+          about = underTheHoodLabels.postLabelAbout(raw),
+          effect = underTheHoodLabels.postLabelEffect(raw),
+          source = source,
+          posts = posts,
+          totalPostsInMonth = postCount,
+          percentageOfPosts = formatPercentage(posts, postCount),
+        }
+      }
     }
     .toList
     .sortBy { row => (row.label, row.source) }

From 089a90993531961eb20fba1051b55abc36874e2e Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:34:26 +0000
Subject: [PATCH 20/26] Drop older asOf UTH rows when source is rewritten

latestAsOf grouped by persistable source, so an earlier unset row
could survive next to a later sourced rewrite of the same user/day/label.
Keep every source from the latest asOf only.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../scalding/UnderTheHoodCommon.scala         | 52 ++++++++++++-------
 1 file changed, 33 insertions(+), 19 deletions(-)

diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala
index e638766f..4e2ea228 100644
--- a/under-the-hood/scalding/UnderTheHoodCommon.scala
+++ b/under-the-hood/scalding/UnderTheHoodCommon.scala
@@ -96,32 +96,46 @@ object UnderTheHoodCommon {
   def finalThroughDay(completeThroughDay: Int, asOfDay: Int, postObservationDays: Int): Int =
     math.min(completeThroughDay, addCalendarDays(asOfDay, -postObservationDays))
 
+  // Latest asOf is a full rewrite of (user, authored day, label). Keep every
+  // persistable source from that asOf; drop older asOf rows so a later sourced
+  // rewrite cannot sit beside an earlier unset row.
   def latestAsOfPostLabelRows(
     rows: TypedPipe[UthDailyPostLabel],
     reducers: Int
   ): TypedPipe[UthDailyPostLabel] =
     applyReducers(
-      rows.groupBy(r =>
-        (r.userId, r.authoredYyyymmdd, r.label, UthLabelSource.persistToken(r.source))),
+      rows.groupBy(r => (r.userId, r.authoredYyyymmdd, r.label)),
       reducers
-    ).reduce { (a, b) =>
-      if (a.asOfYyyymmdd.getOrElse(Int.MinValue) >= b.asOfYyyymmdd.getOrElse(Int.MinValue)) a
-      else b
-    }.values
-      .map { r =>
-        UthDailyPostLabel(
-          userId = r.userId,
-          authoredYyyymmdd = r.authoredYyyymmdd,
-          label = r.label,
-          carried = r.carried,
-          removed = r.removed,
-          asOfYyyymmdd = r.asOfYyyymmdd,
-          observationAgeDays = r.observationAgeDays,
-          isFinal = r.isFinal,
-          postObservationDays = r.postObservationDays,
-          source = UthLabelSource.persistToken(r.source)
-        )
+    ).mapValueStream { values =>
+      val all = values.toSeq
+      if (all.isEmpty) Iterator.empty
+      else {
+        val latest = all.map(_.asOfYyyymmdd.getOrElse(Int.MinValue)).max
+        all
+          .filter(_.asOfYyyymmdd.getOrElse(Int.MinValue) == latest)
+          .groupBy(r => UthLabelSource.persistToken(r.source))
+          .values
+          .iterator
+          .map { sameSource =>
+            val r = sameSource.maxBy { x =>
+              (x.carried.getOrElse(0L), x.removed.getOrElse(0L))
+            }
+            UthDailyPostLabel(
+              userId = r.userId,
+              authoredYyyymmdd = r.authoredYyyymmdd,
+              label = r.label,
+              carried = r.carried,
+              removed = r.removed,
+              asOfYyyymmdd = r.asOfYyyymmdd,
+              observationAgeDays = r.observationAgeDays,
+              isFinal = r.isFinal,
+              postObservationDays = r.postObservationDays,
+              source = UthLabelSource.persistToken(r.source)
+            )
+          }
       }
+    }.toTypedPipe
+      .map { case (_, row) => row }
 
   def parseUserIds(args: Args): Set[Long] = {
     val raw = args.list("userIds").flatMap(_.split(",")).map(_.trim).filter(_.nonEmpty)

From 988af0c4d5bdcb66e53b187993363339375fe89f Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:39:12 +0000
Subject: [PATCH 21/26] Skip trailing synthetic $ when matching UTH llm source

lastIndexOf('$') on GrokAnnotationAction$ (Scala module suffix) left
an empty segment and missed llm. Take the last non-empty . / $ piece.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala | 21 +++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index 023da58d..e4448912 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -49,14 +49,21 @@ object UthLabelSource {
   }
 
   private def isGrokName(name: String): Boolean = {
-    val afterDot = name.lastIndexOf('.') match {
-      case -1 => name
-      case i => name.substring(i + 1)
+    val simple = lastNonEmptySegment(name)
+    simple == "GrokAnnotationAction" || simple.startsWith("GrokAnnotation")
+  }
+
+  // Last `.` / `$` segment, skipping a trailing synthetic `$` (Scala module suffix).
+  // Do not String.split("$"): `$` is regex end-of-string.
+  private def lastNonEmptySegment(name: String): String = {
+    var end = name.length
+    while (end > 0 && (name.charAt(end - 1) == '.' || name.charAt(end - 1) == '$')) {
+      end -= 1
     }
-    val simple = afterDot.lastIndexOf('$') match {
-      case -1 => afterDot
-      case i => afterDot.substring(i + 1)
+    if (end == 0) ""
+    else {
+      val start = name.lastIndexOf('.', end - 1).max(name.lastIndexOf('$', end - 1))
+      name.substring(start + 1, end)
     }
-    simple == "GrokAnnotationAction" || simple.startsWith("GrokAnnotation")
   }
 }

From 04252948c4bd6c85e17548255b9df2c695bf7b7d Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:40:24 +0000
Subject: [PATCH 22/26] Keep event source when an unset snapshot apply is later

mergeActionAgg took the later apply's source even when that row was
an unset snapshot duplicate, wiping BotMaker/Tool/Grok provenance.
Last apply still wins the timestamp; fall back to the other persistable
source when the later row is unset.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthDailyPostsJob.scala | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index 04d1a239..2f921a9e 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -358,7 +358,10 @@ object UthDailyPostsApp {
       } else (a._2, false, a._4)
     val (applyTs, applySrc) =
       if (a._5 != b._5) {
-        if (a._5 > b._5) (a._5, a._6) else (b._5, b._6)
+        val (later, earlier) = if (a._5 > b._5) (a, b) else (b, a)
+        // Last apply wins the timestamp. If that row is an unset snapshot
+        // duplicate, keep the other apply's persistable source.
+        (later._5, later._6.orElse(earlier._6))
       } else (a._5, a._6.orElse(b._6))
     (everApply, last._1, last._2, last._3, applyTs, applySrc)
   }

From 6a8944e8a1e1f69517a84020fed0cec7fc69f12e Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:55:56 +0000
Subject: [PATCH 23/26] Pattern-match UTH post-label leftover group keys

groupBy still used t._1 / t._2 after raw was added as a third
tuple field. Name the display name and source in the key, same as
the leftover flatMap.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/strato/columns/underTheHoodReport.User.strato | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato
index bd40aa26..65338764 100644
--- a/under-the-hood/strato/columns/underTheHoodReport.User.strato
+++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato
@@ -209,7 +209,7 @@ def buildReportJson(
         (underTheHoodLabels.postLabelName(raw), postLabelSource(agg.source), raw, agg)
       }
     }
-    .groupBy { t => (t._1, t._2) }
+    .groupBy { case (name, source, _, _) => (name, source) }
     .toList
     .flatMap { case ((name, source), rows) =>
       val days = rows

From 8db548be425af0dffe8d2795abe93a32317960b5 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:57:17 +0000
Subject: [PATCH 24/26] Skip getSimpleName when detecting UTH llm source

Class.getSimpleName can throw InternalError on Scala $ names and
would fail the daily job for an unmapped SafetyLabelSource. getName
plus the last non-empty segment already yields the simple class name.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthLabelSource.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/under-the-hood/scalding/UthLabelSource.scala b/under-the-hood/scalding/UthLabelSource.scala
index e4448912..fde73a84 100644
--- a/under-the-hood/scalding/UthLabelSource.scala
+++ b/under-the-hood/scalding/UthLabelSource.scala
@@ -44,8 +44,8 @@ object UthLabelSource {
       case p: Product => p.productPrefix
       case _ => ""
     }
-    val cls = source.getClass
-    isGrokName(productName) || isGrokName(cls.getSimpleName) || isGrokName(cls.getName)
+    // getName + lastNonEmptySegment; skip getSimpleName (can throw on Scala $ names).
+    isGrokName(productName) || isGrokName(source.getClass.getName)
   }
 
   private def isGrokName(name: String): Boolean = {

From d98fe358d9cc420ce32b3f0386bbd9fbbafaa877 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Fri, 14 Aug 2026 23:59:59 +0000
Subject: [PATCH 25/26] Pattern-match leftover UTH month day counts

Same-day leftover rows still used best._4 / best._5 after source
was added to the tuple. Take carried/removed by pattern match.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 under-the-hood/scalding/UthUserMonthMhPublisherJob.scala | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
index 6cf6c2fc..830c7263 100644
--- a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
+++ b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala
@@ -212,10 +212,10 @@ object UthUserMonthMhPublisherApp {
                   .groupBy { case (_, _, day, _, _) => day }
                   .map {
                     case (day, dayRows) =>
-                      val best = dayRows.maxBy {
-                        case (_, _, _, carried, removed) => (carried, removed)
-                      }
-                      UthDayCarriedRemoved(Some(day), Some(best._4), Some(best._5))
+                      val (carried, removed) = dayRows
+                        .map { case (_, _, _, c, r) => (c, r) }
+                        .max
+                      UthDayCarriedRemoved(Some(day), Some(carried), Some(removed))
                   }
                   .toList
                   .sortBy(_.dayOfMonth.getOrElse(0))

From 7b6ca476b45f58646cfb309253d475ed8f14590e Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Sat, 15 Aug 2026 00:06:05 +0000
Subject: [PATCH 26/26] Only inherit source from an unset snapshot gap-fill

mergeActionAgg fell back to an earlier persistable source for any
later unset apply, so a later unmapped event inherited BotMaker/Tool
provenance. Thread isSnapshot and fall back only for snapshot rows.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>
---
 .../scalding/UthDailyPostsBackfillJob.scala   | 10 +--
 .../scalding/UthDailyPostsJob.scala           | 63 ++++++++++++-------
 2 files changed, 46 insertions(+), 27 deletions(-)

diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
index 8358064f..0e908595 100644
--- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala
@@ -206,11 +206,11 @@ object UthDailyPostsBackfillApp {
           if (expiresStr == null || expiresStr.isEmpty) Long.MaxValue else expiresStr.toLong
         if (expires <= rangeLookbackStartMs) None
         else if (createdStr == null || createdStr.isEmpty)
-          Some((tweetId, (labelName, Long.MinValue, true, expires, None)))
+          Some((tweetId, (labelName, Long.MinValue, true, expires, None, true)))
         else {
           val created = createdStr.toLong
           if (created < rangeEndMs)
-            Some((tweetId, (labelName, created, true, expires, None)))
+            Some((tweetId, (labelName, created, true, expires, None, true)))
           else None
         }
     }
@@ -266,7 +266,7 @@ object UthDailyPostsBackfillApp {
     val inHorizon = joined.flatMap {
       case (
             _,
-            ((label, eventMs, isApply, expiresMs, source), (logicalId, userId, day, createdMs))
+            ((label, eventMs, isApply, expiresMs, source, isSnapshot), (logicalId, userId, day, createdMs))
           ) =>
         asOfDayStartsFor(createdMs, rangeStartMs, rangeEndMs, observationMs).flatMap { dStartMs =>
           val dayEndMs = dStartMs + DayMs
@@ -279,7 +279,7 @@ object UthDailyPostsBackfillApp {
               .map { _ =>
                 (
                   (logicalId, userId, day, label, yyyymmdd(dStartMs), createdMs),
-                  UthDailyPostsApp.actionAggFromEvent(isApply, eventMs, expiresMs, source)
+                  UthDailyPostsApp.actionAggFromEvent(isApply, eventMs, expiresMs, source, isSnapshot)
                 )
               }
         }
@@ -293,7 +293,7 @@ object UthDailyPostsBackfillApp {
       reduced.collect {
         case (
               (_, userId, day, label, asOfDay, createdMs),
-              (everApply, _, lastApply, lastExpires, _, source)
+              (everApply, _, lastApply, lastExpires, _, source, _)
             ) if everApply =>
           val deadline = math.min(createdMs + observationMs, yyyymmddToMs(asOfDay) + DayMs)
           val removed =
diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala
index 2f921a9e..796c5d10 100644
--- a/under-the-hood/scalding/UthDailyPostsJob.scala
+++ b/under-the-hood/scalding/UthDailyPostsJob.scala
@@ -177,10 +177,10 @@ class UthDailyPostsApp {
 object UthDailyPostsApp {
   import UnderTheHoodCommon._
 
-  // tweetId -> (label, eventMs, isApply, expiresMs, coarseSource)
-  type LabelRow = (Long, (String, Long, Boolean, Long, Option[String]))
-  // everApply, lastTs, lastIsApply, lastExpires, lastApplyTs, lastApplySource
-  type ActionAgg = (Boolean, Long, Boolean, Long, Long, Option[String])
+  // tweetId -> (label, eventMs, isApply, expiresMs, coarseSource, isSnapshot)
+  type LabelRow = (Long, (String, Long, Boolean, Long, Option[String], Boolean))
+  // everApply, lastTs, lastIsApply, lastExpires, lastApplyTs, lastApplySource, lastApplyIsSnapshot
+  type ActionAgg = (Boolean, Long, Boolean, Long, Long, Option[String], Boolean)
 
   val NsfwAdminStampedLabel = "NSFW_ADMIN_STAMPED"
   val TweetFlagColumns: Set[String] = Set("nsfwAdmin")
@@ -276,7 +276,7 @@ object UthDailyPostsApp {
         Some(
           (
             event.tweetId,
-            (name, eventMs, isApply, expiresMs, UthLabelSource.fromEventLabel(event.label))
+            (name, eventMs, isApply, expiresMs, UthLabelSource.fromEventLabel(event.label), false)
           )
         )
       } else None
@@ -296,9 +296,9 @@ object UthDailyPostsApp {
         f.createdAtMsec match {
           case Some(created) =>
             if (created < dayEndMs)
-              Some((f.tweetId, (name, created, true, expires, None)))
+              Some((f.tweetId, (name, created, true, expires, None, true)))
             else None
-          case None => Some((f.tweetId, (name, Long.MinValue, true, expires, None)))
+          case None => Some((f.tweetId, (name, Long.MinValue, true, expires, None, true)))
         }
     }
 
@@ -313,11 +313,11 @@ object UthDailyPostsApp {
           if (expiresStr == null || expiresStr.isEmpty) Long.MaxValue else expiresStr.toLong
         if (expires <= lookbackStartMs) None
         else if (createdStr == null || createdStr.isEmpty)
-          Some((tweetId, (labelName, Long.MinValue, true, expires, None)))
+          Some((tweetId, (labelName, Long.MinValue, true, expires, None, true)))
         else {
           val created = createdStr.toLong
           if (created < dayEndMs)
-            Some((tweetId, (labelName, created, true, expires, None)))
+            Some((tweetId, (labelName, created, true, expires, None, true)))
           else None
         }
     }
@@ -340,11 +340,12 @@ object UthDailyPostsApp {
     isApply: Boolean,
     eventMs: Long,
     expiresMs: Long,
-    source: Option[String]
+    source: Option[String],
+    isSnapshot: Boolean
   ): ActionAgg = {
     val applyTs = if (isApply) eventMs else Long.MinValue
     val applySrc = if (isApply) UthLabelSource.persistToken(source) else None
-    (isApply, eventMs, isApply, expiresMs, applyTs, applySrc)
+    (isApply, eventMs, isApply, expiresMs, applyTs, applySrc, isApply && isSnapshot)
   }
 
   private[under_the_hood] def mergeActionAgg(a: ActionAgg, b: ActionAgg): ActionAgg = {
@@ -356,14 +357,29 @@ object UthDailyPostsApp {
         val exp = if (a._3) a._4 else b._4
         (a._2, true, exp)
       } else (a._2, false, a._4)
-    val (applyTs, applySrc) =
-      if (a._5 != b._5) {
-        val (later, earlier) = if (a._5 > b._5) (a, b) else (b, a)
-        // Last apply wins the timestamp. If that row is an unset snapshot
-        // duplicate, keep the other apply's persistable source.
-        (later._5, later._6.orElse(earlier._6))
-      } else (a._5, a._6.orElse(b._6))
-    (everApply, last._1, last._2, last._3, applyTs, applySrc)
+    val (aTs, aSrc, aSnap) = (a._5, a._6, a._7)
+    val (bTs, bSrc, bSnap) = (b._5, b._6, b._7)
+    val (applyTs, applySrc, applySnap) =
+      if (aTs != bTs) {
+        val (laterTs, laterSrc, laterSnap, earlierSrc) =
+          if (aTs > bTs) (aTs, aSrc, aSnap, bSrc) else (bTs, bSrc, bSnap, aSrc)
+        // Last apply wins. Fall back to the earlier persistable source only
+        // when the later row is an unset snapshot gap-fill, not a later event.
+        val src =
+          if (laterSrc.nonEmpty) laterSrc
+          else if (laterSnap) earlierSrc.orElse(laterSrc)
+          else laterSrc
+        (laterTs, src, laterSnap && src.isEmpty)
+      } else {
+        val src =
+          (aSnap, bSnap) match {
+            case (false, true) => aSrc.orElse(bSrc)
+            case (true, false) => bSrc.orElse(aSrc)
+            case _ => aSrc.orElse(bSrc)
+          }
+        (aTs, src, src.isEmpty && aSnap && bSnap)
+      }
+    (everApply, last._1, last._2, last._3, applyTs, applySrc, applySnap)
   }
 
   private[under_the_hood] def removedAfterLastAction(
@@ -402,12 +418,15 @@ object UthDailyPostsApp {
       else scopedRows.join(postByTweetId)
 
     val inHorizon = joined.flatMap {
-      case (_, ((label, eventMs, isApply, expiresMs, source), (logicalId, userId, day, createdMs))) =>
+      case (
+            _,
+            ((label, eventMs, isApply, expiresMs, source, isSnapshot), (logicalId, userId, day, createdMs))
+          ) =>
         val deadline = math.min(createdMs + observationMs, dayEndMs)
         actionInHorizon(eventMs, isApply, expiresMs, createdMs, deadline).map { _ =>
           (
             (logicalId, userId, day, label, createdMs),
-            actionAggFromEvent(isApply, eventMs, expiresMs, source)
+            actionAggFromEvent(isApply, eventMs, expiresMs, source, isSnapshot)
           )
         }
     }
@@ -420,7 +439,7 @@ object UthDailyPostsApp {
       reduced.collect {
         case (
               (_, userId, day, label, createdMs),
-              (everApply, _, lastApply, lastExpires, _, source)
+              (everApply, _, lastApply, lastExpires, _, source, _)
             ) if everApply =>
           val deadline = math.min(createdMs + observationMs, dayEndMs)
           val removed =