diff --git a/src/VahterBanBot/Bot.fs b/src/VahterBanBot/Bot.fs index ca90bcc..2562bfa 100644 --- a/src/VahterBanBot/Bot.fs +++ b/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) diff --git a/src/VahterBanBot/DB.fs b/src/VahterBanBot/DB.fs index 01f4e7a..3cb5514 100644 --- a/src/VahterBanBot/DB.fs +++ b/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,11 +642,13 @@ 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); """ @@ -642,6 +656,29 @@ ORDER BY MAX(m.created_at); 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 = + 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(sql, {| text = text; since = since |}) + return count + } + /// Saves a trained ML model to the database (singleton row, upsert). member _.SaveTrainedModel(modelStream: Stream) : Task = task { diff --git a/src/VahterBanBot/ML.fs b/src/VahterBanBot/ML.fs index aaf2683..ae9f057 100644 --- a/src/VahterBanBot/ML.fs +++ b/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 diff --git a/src/VahterBanBot/Program.fs b/src/VahterBanBot/Program.fs index 2000882..56bdb17 100644 --- a/src/VahterBanBot/Program.fs +++ b/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 diff --git a/src/VahterBanBot/Types.fs b/src/VahterBanBot/Types.fs index b001059..45c15d6 100644 --- a/src/VahterBanBot/Types.fs +++ b/src/VahterBanBot/Types.fs @@ -252,6 +252,11 @@ type BotConfiguration = MlStopWordsInChats: Dictionary /// 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 diff --git a/src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql b/src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql new file mode 100644 index 0000000..cb2d78e --- /dev/null +++ b/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; diff --git a/tests/VahterBanBot.Tests/ContainerTestBase.fs b/tests/VahterBanBot.Tests/ContainerTestBase.fs index 17b6f85..266680c 100644 --- a/tests/VahterBanBot.Tests/ContainerTestBase.fs +++ b/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 diff --git a/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs b/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs index a5f3dd5..f1166b1 100644 --- a/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs +++ b/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