Vahter: gated text-repeat ML feature behind ML_REPEAT_COUNT_ENABLED

src/VahterBanBot/Bot.fs

@@ -683,16 +683,30 @@ type BotService(
     // Private members — ML / LLM auto-verdict
     // -----------------------------------------------------------------------
 
+    /// Repeat-count lookup gated by feature flag. Skipped when the flag is
+    /// off so we don't pay the DB round-trip on the hot inference path; the
+    /// trained pipeline doesn't reference repeatCountF in that case anyway.
+    /// Window matches MlTrainInterval so the inference distribution lines up
+    /// with the training-time text_repeat_counts CTE in DB.MlData.
+    member private _.GetMlRepeatCount(text: string) = task {
+        if not botConfig.Value.MlRepeatCountEnabled || isNull text then
+            return 0
+        else
+            let since = utcNow() - botConfig.Value.MlTrainInterval
+            return! db.GetTextRepeatCount(text, since)
+    }
+
     /// Fast text-only ML check used to short-circuit before Azure OCR.
     /// Returns Some score when ML alone is confident the message is spam
     /// (records the ML score so callers don't double-record). Returns None
     /// for null text, low ML score, warning band, or old-user immunity —
     /// in which case the caller should run Azure OCR for any cache misses
     /// and then call GetAutoVerdict for the full verdict.
-    member private _.PreOcrMlCheck(msg: TgMessage, usrMsgCount: int) = task {
+    member private this.PreOcrMlCheck(msg: TgMessage, usrMsgCount: int) = task {
         if isNull msg.Text then return None
         else
-        match ml.Predict(msg.Text, usrMsgCount, msg.Entities) with
+        let! repeatCount = this.GetMlRepeatCount(msg.Text)
+        match ml.Predict(msg.Text, usrMsgCount, msg.Entities, repeatCount) with
         | None -> return None
         | Some prediction ->
             // Old-user immunity: defer to GetAutoVerdict so the
@@ -707,8 +721,9 @@ type BotService(
     }
 
     /// Runs ML prediction + optional LLM triage, returns verdict.
-    member private _.GetAutoVerdict(msg: TgMessage, usrMsgCount: int) = task {
-        match ml.Predict(msg.Text, usrMsgCount, msg.Entities) with
+    member private this.GetAutoVerdict(msg: TgMessage, usrMsgCount: int) = task {
+        let! repeatCount = this.GetMlRepeatCount(msg.Text)
+        match ml.Predict(msg.Text, usrMsgCount, msg.Entities, repeatCount) with
         | None -> return None
         | Some prediction ->
             do! db.RecordMlScoredMessage(msg.ChatId, msg.MessageId, float prediction.Score, prediction.Score >= botConfig.Value.MlSpamThreshold)

src/VahterBanBot/DB.fs

@@ -24,6 +24,7 @@ type SpamOrHamDb =
       spam: bool
       less_than_n_messages: bool
       custom_emoji_count: int
+      text_repeat_count: int
       created_at: DateTime }
 
 type DbService(connString: string, timeProvider: TimeProvider) =
@@ -586,6 +587,7 @@ WITH final_messages AS (
         (data->>'userId')::BIGINT              AS user_id,
         data->>'text'                          AS text,
         data->'rawMessage'->'entities'         AS entities,
+        msg_text_md5,
         created_at
     FROM event
     WHERE event_type IN ('MessageReceived', 'MessageEdited')
@@ -599,6 +601,16 @@ user_msg_counts AS (
     FROM final_messages
     GROUP BY user_id
 ),
+text_repeat_counts AS (
+    -- How many times the same text appears in the training window.
+    -- Repeated spam blasts (campaigns) get a high count; one-off chatter
+    -- gets 1. Symmetric with the GetTextRepeatCount inference query so
+    -- training and inference see the same feature distribution.
+    SELECT msg_text_md5, COUNT(*)::INT AS text_repeat_count
+    FROM final_messages
+    WHERE msg_text_md5 IS NOT NULL
+    GROUP BY msg_text_md5
+),
 verdicts AS (
     -- All verdict-bearing events, unified across message and moderation streams
     SELECT
@@ -630,18 +642,43 @@ SELECT m.text,
        COALESCE(u.less_than_n_messages, TRUE)                              AS less_than_n_messages,
        (SELECT COUNT(*) FROM jsonb_array_elements(m.entities) ent
         WHERE ent->>'type' = 'custom_emoji')::INT                          AS custom_emoji_count,
+       COALESCE(rc.text_repeat_count, 1)                                   AS text_repeat_count,
        MAX(m.created_at)                                                   AS created_at
 FROM final_messages m
 LEFT JOIN last_verdict v ON v.chat_id = m.chat_id AND v.message_id = m.message_id
 LEFT JOIN user_msg_counts u ON u.user_id = m.user_id
-GROUP BY m.text, v.is_spam, u.less_than_n_messages, m.entities
+LEFT JOIN text_repeat_counts rc ON rc.msg_text_md5 = m.msg_text_md5
+GROUP BY m.text, v.is_spam, u.less_than_n_messages, m.entities, rc.text_repeat_count
 ORDER BY MAX(m.created_at);
 """
 
             let! data = conn.QueryAsync<SpamOrHamDb>(sql, {| criticalDate = criticalDate; criticalMsgCount = criticalMsgCount |})
             return Array.ofSeq data
         }
 
+    /// Counts how many MessageReceived events with the same text exist
+    /// in the time window [since, NOW]. Used at inference time to populate
+    /// the repeat-count ML feature symmetrically with how training computes
+    /// it from text_repeat_counts CTE in MlData.
+    /// Backed by idx_event_msg_text_md5 (V35).
+    member _.GetTextRepeatCount(text: string, since: DateTime) : Task<int> =
+        task {
+            use conn = new NpgsqlConnection(connString)
+
+            //language=postgresql
+            let sql =
+                """
+SELECT COUNT(*)::INT
+FROM event
+WHERE event_type = 'MessageReceived'
+  AND msg_text_md5 = md5(@text)
+  AND created_at >= @since;
+"""
+
+            let! count = conn.ExecuteScalarAsync<int>(sql, {| text = text; since = since |})
+            return count
+        }
+
     /// Saves a trained ML model to the database (singleton row, upsert).
     member _.SaveTrainedModel(modelStream: Stream) : Task =
         task {

src/VahterBanBot/ML.fs

@@ -23,6 +23,11 @@ type SpamOrHam =
       spam: bool
       lessThanNMessagesF: single
       moreThanNEmojisF: single
+      /// log1p(text_repeat_count). Only consumed by the pipeline when
+      /// MlRepeatCountEnabled was true at training time; otherwise the
+      /// trained pipeline doesn't reference this column and the value
+      /// is ignored.
+      repeatCountF: single
       weight: single
       createdAt: DateTime }
 
@@ -105,7 +110,10 @@ type MachineLearning(
                   createdAt = x.created_at
                   weight = w
                   moreThanNEmojisF = if x.custom_emoji_count > botConf.Value.MlCustomEmojiThreshold then 1.0f else 0.0f
-                  lessThanNMessagesF = if x.less_than_n_messages then 1.0f else 0.0f }
+                  lessThanNMessagesF = if x.less_than_n_messages then 1.0f else 0.0f
+                  // log1p compresses the long tail (max ~200 → ~5.3) so SDCA
+                  // doesn't over-weight a single very-repeated campaign.
+                  repeatCountF = single (Math.Log(1.0 + float x.text_repeat_count)) }
             )
             |> fun x ->
                 if botConf.Value.MlTrainRandomSortData then
@@ -117,10 +125,17 @@ type MachineLearning(
         let trainingData = trainTestSplit.TrainSet
         let testData = trainTestSplit.TestSet
 
+        let featureColumns =
+            [|
+                "TextFeaturized"
+                "lessThanNMessagesF"
+                "moreThanNEmojisF"
+                if botConf.Value.MlRepeatCountEnabled then "repeatCountF"
+            |]
         let featurePipeline =
             mlContext.Transforms.Text
                 .FeaturizeText(outputColumnName = "TextFeaturized", inputColumnName = "text")
-                .Append(mlContext.Transforms.Concatenate(outputColumnName = "Features", inputColumnNames = [|"TextFeaturized"; "lessThanNMessagesF"; "moreThanNEmojisF"|]))
+                .Append(mlContext.Transforms.Concatenate(outputColumnName = "Features", inputColumnNames = featureColumns))
 
         let dataProcessPipeline =
             let options = SdcaLogisticRegressionBinaryTrainer.Options(
@@ -179,7 +194,7 @@ type MachineLearning(
     // if ML is ready (either disabled or model is trained)
     member _.IsReady = not botConf.Value.MlEnabled || predictionEngine.IsSome
 
-    member _.Predict(text: string, userMsgCount: int, entities: MessageEntity array) =
+    member _.Predict(text: string, userMsgCount: int, entities: MessageEntity array, repeatCount: int) =
         try
             match predictionEngine with
             | Some predictionEngine ->
@@ -189,12 +204,13 @@ type MachineLearning(
                     |> Option.defaultValue [||]
                     |> Seq.filter (fun x -> x.Type = MessageEntityType.CustomEmoji)
                     |> Seq.length
-                
+
                 predictionEngine.Predict
                     { text = text
                       spam = false
                       lessThanNMessagesF = if userMsgCount < botConf.Value.MlTrainCriticalMsgCount then 1.0f else 0.0f
                       moreThanNEmojisF = if emojiCount > botConf.Value.MlCustomEmojiThreshold then 1.0f else 0.0f
+                      repeatCountF = single (Math.Log(1.0 + float repeatCount))
                       weight = 1.0f
                       createdAt = timeProvider.GetUtcNow().UtcDateTime }
                 |> Some

src/VahterBanBot/Program.fs

@@ -108,6 +108,7 @@ let buildBotConf () =
       MlCustomEmojiThreshold = getSettingOr "ML_CUSTOM_EMOJI_THRESHOLD" "20" |> int
       MlStopWordsInChats = getSettingOr "ML_STOP_WORDS_IN_CHATS" "{}" |> fromJson
       MlWeightDecayK = getSettingOr "ML_WEIGHT_DECAY_K" "0" |> float
+      MlRepeatCountEnabled = getSettingOr "ML_REPEAT_COUNT_ENABLED" "false" |> bool.Parse
       MlOldUserMsgCount = getSettingOr "ML_OLD_USER_MSG_COUNT" "50" |> int
       // Reaction spam detection
       ReactionSpamEnabled = getSettingOr "REACTION_SPAM_ENABLED" "false" |> bool.Parse

src/VahterBanBot/Types.fs

@@ -252,6 +252,11 @@ type BotConfiguration =
       MlStopWordsInChats: Dictionary<int64, string list>
       /// Time-decay weight parameter: w(t) = exp(-k * age_in_days). 0 = no decay (all weights 1.0).
       MlWeightDecayK: float
+      /// Feature flag: include text-repeat count as an ML feature.
+      /// Affects training only — inference always populates the field correctly,
+      /// so toggling without retrain leaves predictions consistent with the
+      /// currently loaded model's pipeline.
+      MlRepeatCountEnabled: bool
       /// Users with >= this many unique messages are immune from ML/LLM triage.
       MlOldUserMsgCount: int
       // Reaction spam detection

src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql

@@ -0,0 +1,13 @@
+-- Partial index on event.msg_text_md5 for MessageReceived events.
+-- Powers the per-message repeat-count lookup used by the ML inference path
+-- (DbService.GetTextRepeatCount) and by the training-set SQL's
+-- text_repeat_counts CTE. Without this index those become sequential scans
+-- over the full event log (~700k+ rows).
+--
+-- Partial WHERE matches the access pattern: we only ever count
+-- MessageReceived rows with a non-null hash. msg_text_md5 is NULL for
+-- events whose data has no text field (most non-MessageReceived events).
+
+CREATE INDEX IF NOT EXISTS idx_event_msg_text_md5
+    ON event(msg_text_md5)
+    WHERE event_type = 'MessageReceived' AND msg_text_md5 IS NOT NULL;

tests/VahterBanBot.Tests/ContainerTestBase.fs

@@ -618,9 +618,25 @@ type MlEnabledVahterTestContainers() =
 /// Variant that DELIBERATELY skips fixture preload to exercise the production training pipeline
 /// end-to-end. Used by MLTrainingPipelineTests as a smoke test that training still produces a
 /// usable model (the most important property of the bot — autonomous spam detection).
+///
+/// Also enables ML_REPEAT_COUNT_ENABLED=true so the smoke test exercises the
+/// repeat-count feature path through training AND inference. The non-FF path
+/// is the production default and is implicitly covered by every other ML test.
 type MlTrainingFromScratchTestContainers() =
     inherit VahterTestContainers(mlEnabled = true)
 
+    override this.SeedDatabase(connString: string) =
+        let baseSeed = base.SeedDatabase(connString)
+        task {
+            do! baseSeed
+            use conn = new NpgsqlConnection(connString)
+            do! conn.OpenAsync()
+            do! conn.ExecuteAsync(
+                    "INSERT INTO bot_setting(key,value,type,feature_group) \
+                     VALUES('ML_REPEAT_COUNT_ENABLED','true','FEATURE_FLAG','ML')")
+                :> Task
+        }
+
     override this.AfterStart() =
         task {
             // Same /ready wait, but never extract bytes — we don't want a throwaway fresh-train

tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs

@@ -12,6 +12,12 @@ open BotTestInfra
 /// deliberately starts with no model in the DB — forcing the bot's prod
 /// MachineLearning.StartAsync to train end-to-end.
 ///
+/// The fixture also enables ML_REPEAT_COUNT_ENABLED=true so this test
+/// exercises the repeat-count feature path through training (conditional
+/// Concatenate of repeatCountF) and inference (DB.GetTextRepeatCount lookup
+/// gated by the FF in Bot.GetMlRepeatCount). Production default is FF=off,
+/// covered implicitly by every other ML test (which uses the pinned model).
+///
 /// The training pipeline is the most important property of this bot (it's what
 /// kills 90% of spam autonomously), so we assert it can:
 ///   1. Train successfully from seed data within a reasonable timeout