diff --git a/under-the-hood/scalding/UnderTheHoodCommon.scala b/under-the-hood/scalding/UnderTheHoodCommon.scala index 8460e6b3..4e2ea228 100644 --- a/under-the-hood/scalding/UnderTheHoodCommon.scala +++ b/under-the-hood/scalding/UnderTheHoodCommon.scala @@ -96,6 +96,9 @@ 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 @@ -103,10 +106,36 @@ object UnderTheHoodCommon { applyReducers( 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 + ).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) diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala index 6739b139..0e908595 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 = UthLabelSource.persistToken(source) ) } @@ -135,7 +136,8 @@ class UthDailyPostsBackfillApp { asOfYyyymmdd = Some(authoredDay), observationAgeDays = Some(0), isFinal = Some(true), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + source = None ) } @@ -204,10 +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))) + Some((tweetId, (labelName, Long.MinValue, true, expires, None, true))) else { val created = createdStr.toLong - if (created < rangeEndMs) Some((tweetId, (labelName, created, true, expires))) + if (created < rangeEndMs) + Some((tweetId, (labelName, created, true, expires, None, true))) else None } } @@ -240,7 +243,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 +264,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, isSnapshot), (logicalId, userId, day, createdMs)) + ) => asOfDayStartsFor(createdMs, rangeStartMs, rangeEndMs, observationMs).flatMap { dStartMs => val dayEndMs = dStartMs + DayMs val deadline = math.min(createdMs + observationMs, dayEndMs) @@ -270,43 +276,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, isSnapshot) + ) } } } - 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, UthLabelSource.persistToken(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..796c5d10 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 = UthLabelSource.persistToken(source) ) } @@ -141,7 +142,8 @@ class UthDailyPostsApp { asOfYyyymmdd = Some(asOfDay), observationAgeDays = Some(0), isFinal = Some(true), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + source = None ) } @@ -175,7 +177,10 @@ class UthDailyPostsApp { object UthDailyPostsApp { import UnderTheHoodCommon._ - type LabelRow = (Long, (String, Long, Boolean, Long)) + // 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") @@ -268,7 +273,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), false) + ) + ) } else None } @@ -285,8 +295,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, true))) + else None + case None => Some((f.tweetId, (name, Long.MinValue, true, expires, None, true))) } } @@ -301,10 +313,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, true))) 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, true))) + else None } } @@ -322,6 +336,52 @@ object UthDailyPostsApp { else Some((isApply, eventMs, isApply, expiresMs)) } + private[under_the_hood] def actionAggFromEvent( + isApply: Boolean, + eventMs: Long, + expiresMs: Long, + 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 && isSnapshot) + } + + 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 (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( lastIsApply: Boolean, lastExpiresMs: Long, @@ -340,7 +400,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 +418,38 @@ 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, isSnapshot), (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, isSnapshot) + ) } } - 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, UthLabelSource.persistToken(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..fde73a84 --- /dev/null +++ b/under-the-hood/scalding/UthLabelSource.scala @@ -0,0 +1,69 @@ +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 Llm = "llm" + val Unknown = "unknown" + + def fromEventLabel(label: Option[SafetyLabel]): Option[String] = + persistToken(label.flatMap(_.safetyLabelSource).flatMap(coarseCategory)) + + /** Persist only automated | manual | llm. Unset/unmapped stay empty. */ + def persistToken(source: Option[String]): Option[String] = + source.map(_.trim.toLowerCase) match { + 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 + } + + /** + * Public reportJson tokens: automated | manual | llm | unknown. + * Does not emit rule_id, actor_ldap, agent_tool, or VF-client type names. + * + * Named cases are only those present on published spam.rtf usage + * (BotMakerAction, ToolAction). A Grok/LLM union member is detected by + * 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 { + 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 productName = source match { + case p: Product => p.productPrefix + case _ => "" + } + // getName + lastNonEmptySegment; skip getSimpleName (can throw on Scala $ names). + isGrokName(productName) || isGrokName(source.getClass.getName) + } + + private def isGrokName(name: String): Boolean = { + 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 + } + if (end == 0) "" + else { + val start = name.lastIndexOf('.', end - 1).max(name.lastIndexOf('$', end - 1)) + name.substring(start + 1, end) + } + } +} diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala index 31ed6037..830c7263 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, UthLabelSource.persistToken(row.source), dayOfMonth(day), carried, removed) + ) }, reducers ) @@ -202,24 +205,28 @@ 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) - } - UthDayCarriedRemoved(Some(day), Some(best._3), Some(best._4)) + val (carried, removed) = dayRows + .map { case (_, _, _, c, r) => (c, r) } + .max + UthDayCarriedRemoved(Some(day), Some(carried), Some(removed)) } .toList .sortBy(_.dayOfMonth.getOrElse(0)) - UthPostLabelAggregate(Some(label), Some(days)) + UthPostLabelAggregate( + label = Some(label), + days = Some(days), + source = UthLabelSource.persistToken(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..65338764 100644 --- a/under-the-hood/strato/columns/underTheHoodReport.User.strato +++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato @@ -155,6 +155,23 @@ def sumDayCounts(days: Option[Seq[UthDayCount]]): Long = acc + d.count.getOrElse(0L) } +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" + case Some("llm") => "llm" + case _ => "unknown" + } +} + def formatPercentage(numerator: Long, denominator: Long): String = { if (denominator <= 0L) { "0%" @@ -189,17 +206,36 @@ def buildReportJson( .getOrElse(Seq.empty) .flatMap { agg => agg.label.filter(underTheHoodLabels.isPostLabel).map { raw => - val posts = sumCarried(agg.days) + (underTheHoodLabels.postLabelName(raw), postLabelSource(agg.source), raw, agg) + } + } + .groupBy { case (name, source, _, _) => (name, source) } + .toList + .flatMap { case ((name, source), rows) => + val days = rows + .flatMap { case (_, _, _, agg) => agg.days.getOrElse(Seq.empty) } + .groupBy { d => d.dayOfMonth.getOrElse(0) } + .toList + .flatMap { case (_, dayRows) => + dayRows.headOption.map { first => + dayRows.foldLeft(first) { (best, d) => bestDayRow(best, d) } + } + } + val posts = sumCarried(Some(days)) + rows.headOption.map { case (_, _, raw, _) => { - label = underTheHoodLabels.postLabelName(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) } val accountLabels = userMonth.accountLabelAgg .getOrElse(Seq.empty) diff --git a/under-the-hood/thrift/uth_serving.thrift b/under-the-hood/thrift/uth_serving.thrift index 299c551f..f3a88410 100644 --- a/under-the-hood/thrift/uth_serving.thrift +++ b/under-the-hood/thrift/uth_serving.thrift @@ -53,6 +53,8 @@ enum UthFollowerClass { struct UthPostLabelAggregate { 1: optional string label (personalDataType = 'TweetSafetyLabels') 2: optional list days + // Persisted token: automated | manual | llm. Unset → reportJson unknown. + 3: optional string source }(persisted = 'true', hasPersonalData = 'true') struct UthAccountLabelAggregate { @@ -129,6 +131,8 @@ struct UthDailyPostLabel { 7: optional i32 observationAgeDays 8: optional bool isFinal 9: optional i32 postObservationDays + // Persisted token: automated | manual | llm. Unset → reportJson unknown. + 10: optional string source }(persisted = 'true', hasPersonalData = 'true') struct UthDailyAccountLabel {