From f7a7652f45f76e9b4ec12709227b58fb6cfc9a45 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Tue, 29 Sep 2026 23:58:15 +0100 Subject: [PATCH 1/2] Add historical moderation quality aggregation --- src/VahterBanBot/Cleanup.fs | 7 +- src/VahterBanBot/ModerationQuality.fs | 155 ++++++++++++++++++ src/VahterBanBot/ModerationQualityHistory.fs | 96 +++++++++++ src/VahterBanBot/Program.fs | 21 +++ src/VahterBanBot/VahterBanBot.fsproj | 2 + src/VahterBanBot/docs/MODERATION-QUALITY.md | 71 ++++++++ .../V54__moderation_quality_history.sql | 20 +++ .../ModerationQualityTests.fs | 129 +++++++++++++++ .../VahterBanBot.Tests.fsproj | 1 + 9 files changed, 501 insertions(+), 1 deletion(-) create mode 100644 src/VahterBanBot/ModerationQuality.fs create mode 100644 src/VahterBanBot/ModerationQualityHistory.fs create mode 100644 src/VahterBanBot/docs/MODERATION-QUALITY.md create mode 100644 src/vahter-bot/migrations/V54__moderation_quality_history.sql create mode 100644 tests/VahterBanBot.Tests/ModerationQualityTests.fs diff --git a/src/VahterBanBot/Cleanup.fs b/src/VahterBanBot/Cleanup.fs index efc33439..c176bcfa 100644 --- a/src/VahterBanBot/Cleanup.fs +++ b/src/VahterBanBot/Cleanup.fs @@ -19,7 +19,8 @@ type CleanupService( tg: ITelegramApi, botConf: IOptions, ml: MachineLearning, - db: DbService + db: DbService, + quality: ModerationQualityHistory ) = let podId = getEnvOr "POD_NAME" Environment.MachineName let mutable cts: CancellationTokenSource = null @@ -121,6 +122,10 @@ type CleanupService( try // Check scheduled jobs do! tryRunJob "daily_cleanup" (TimeSpan.FromHours botConf.Value.CleanupScheduledHour) runCleanup + do! tryRunJob "moderation_quality_daily" (TimeSpan.FromHours botConf.Value.CleanupScheduledHour) (fun () -> task { + let! _ = quality.Run(None, ct) + () + }) do! tryRunJob "daily_stats" (TimeSpan.FromHours botConf.Value.StatsScheduledHour) runStats if botConf.Value.MlEnabled && botConf.Value.MlRetrainScheduledEnabled then diff --git a/src/VahterBanBot/ModerationQuality.fs b/src/VahterBanBot/ModerationQuality.fs new file mode 100644 index 00000000..ac9baff7 --- /dev/null +++ b/src/VahterBanBot/ModerationQuality.fs @@ -0,0 +1,155 @@ +namespace VahterBanBot + +open System +open BotInfra +open VahterBanBot.Types + +[] +type QualityOutcome = + | TruePositive | FalsePositive | FalseNegative | TrueNegative | Abstained | Unresolved | Excluded + member this.Code = + match this with + | QualityOutcome.TruePositive -> "tp" + | QualityOutcome.FalsePositive -> "fp" + | QualityOutcome.FalseNegative -> "fn" + | QualityOutcome.TrueNegative -> "tn" + | QualityOutcome.Abstained -> "abstained" + | QualityOutcome.Unresolved -> "unresolved" + | QualityOutcome.Excluded -> "excluded" + +type QualityContribution = + { Cohort: string + System: string + LlmModel: string option + Outcome: QualityOutcome } + +type QualityEvaluation = { EvaluatedAt: DateTime; Contributions: QualityContribution list } + +type QualityUnban = { At: DateTime; Id: int64; ChatId: int64; MessageId: int64 } + +module ModerationQuality = + let private key (row: RawEvent) = row.created_at, row.id + + let unbans (store: EventStore) (rows: RawEvent list) = + let mutable target = None + [ for row in List.sortBy key rows do + match store.Deserialize row with + | UserBanned e -> + target <- + match e.chatId, e.messageId, e.bannedBy with + | Some chat, Some msg, _ -> Some (chat, msg) + | _, _, Some (BannedByVahter b) -> Some (b.chatId, int64 b.messageId) + | _, _, Some (BannedByAI b) -> Some (b.chatId, int64 b.messageId) + | _ -> None + | UserUnbanned e -> + match e.actor, e.unbannedBy, target with + | Some (Actor.User _), _, Some (chat, msg) + | None, Some _, Some (chat, msg) when chat <> 0L && msg <> 0L -> + yield { At = row.created_at; Id = row.id; ChatId = chat; MessageId = msg } + | _ -> () + | _ -> () ] + + let private classify prediction truth = + match prediction, truth with + | true, Some SpamClassification.Spam -> QualityOutcome.TruePositive + | true, Some SpamClassification.Ham -> QualityOutcome.FalsePositive + | false, Some SpamClassification.Spam -> QualityOutcome.FalseNegative + | false, Some SpamClassification.Ham -> QualityOutcome.TrueNegative + | _ -> QualityOutcome.Unresolved + + let evaluate (store: EventStore) (message: RawEvent list) (moderation: RawEvent list) + (detection: RawEvent list) (corrections: QualityUnban list) = + let scores = detection |> List.filter (fun r -> r.event_type = "MlScoredMessage") |> List.sortBy key + let labels = + [ for row in message do + match store.Deserialize row with + | MessageMarkedHam _ -> yield key row, SpamClassification.Ham + | MessageMarkedSpam _ -> yield key row, SpamClassification.Spam + | _ -> () + for row in moderation do + match store.Deserialize row with + | VahterActed e -> + match e.actionType with + | PotentialKill | ManualBan -> yield key row, SpamClassification.Spam + | PotentialNotSpam | DetectedNotSpam -> yield key row, SpamClassification.Ham + | _ -> () + | _ -> () + for c in corrections do yield (c.At, c.Id), SpamClassification.Ham ] + let human = labels |> List.sortBy fst |> List.tryLast + let beforeReview row = human |> Option.forall (fun (at, _) -> key row < at) + let scoreValue row = + match store.Deserialize row with + | MlScoredMessage e -> e.score + | _ -> invalidArg (nameof row) "Expected ML score" + let changedSince at = + let content limit = + message + |> List.filter (fun r -> key r <= limit && (r.event_type = "MessageReceived" || r.event_type = "MessageEdited")) + |> List.sortBy key |> List.tryLast + |> Option.bind (fun r -> + match store.Deserialize r with + | MessageReceived e -> e.text + | MessageEdited e -> e.text + | _ -> None) + human |> Option.exists (fun (reviewAt, _) -> content reviewAt <> content at) + match List.tryHead scores, scores |> List.filter beforeReview |> List.tryLast with + | None, _ -> None + | Some first, latest -> + let llm = + detection |> List.filter (fun r -> beforeReview r && (r.event_type = "LlmClassified" || r.event_type = "LlmVerdictCacheHit")) + |> List.sortBy key |> List.tryLast + |> Option.bind (fun r -> + match store.Deserialize r with + | LlmClassified e -> Some (r, e.verdict, e.modelName) + | LlmVerdictCacheHit e -> Some (r, e.verdict, e.modelName) + | _ -> None) + let deletion = + moderation |> List.filter (fun r -> r.event_type = "BotAutoDeleted" && beforeReview r) + |> List.sortBy key |> List.tryLast + |> Option.map (fun r -> + match store.Deserialize r with + | BotAutoDeleted e -> e.reason + | _ -> invalidArg (nameof r) "Expected automatic deletion") + let automated, excluded = + match deletion with + | Some (MlSpam _ | LlmSpam _) -> true, false + | Some _ -> false, true + | None -> false, false + let llmTruth verdict = + match verdict with + | "SPAM" | "KILL" -> Some SpamClassification.Spam + | "NOT_SPAM" -> Some SpamClassification.Ham + | _ -> None + let truth = + match human with + | Some (_, label) -> Some label + | None when automated -> Some SpamClassification.Spam + | None -> + match latest, llm with + | Some ml, Some (r, verdict, _) when key r >= key ml -> llmTruth verdict + | _ -> Some SpamClassification.Ham + let all system prediction = + { Cohort = "all_scored"; System = system; LlmModel = None + Outcome = + match latest with + | None -> QualityOutcome.Excluded + | Some ml when excluded || changedSince (key ml) -> QualityOutcome.Excluded + | Some _ -> classify prediction truth } + let contributions = + [ yield all "pipeline" automated + yield all "ml" (latest |> Option.exists (fun r -> scoreValue r > 0.0)) + match llm with + | Some (r, verdict, model) -> + match scores |> List.filter (fun s -> key s < key r) |> List.tryLast with + | Some ml when scoreValue ml >= -0.5 && scoreValue ml < 1.5 -> + let reference = human |> Option.map snd |> Option.orElseWith (fun () -> llmTruth verdict) + let result prediction = + match changedSince (key r), verdict with + | true, _ -> QualityOutcome.Excluded + | false, "SKIP" -> QualityOutcome.Abstained + | _ -> classify prediction reference + yield { Cohort = "llm_triaged"; System = "ml"; LlmModel = model; Outcome = result (scoreValue ml > 0.0) } + yield { Cohort = "llm_triaged"; System = "llm"; LlmModel = model; Outcome = result (verdict = "SPAM" || verdict = "KILL") } + | _ -> () + | None -> () ] + Some { EvaluatedAt = first.created_at; Contributions = contributions } diff --git a/src/VahterBanBot/ModerationQualityHistory.fs b/src/VahterBanBot/ModerationQualityHistory.fs new file mode 100644 index 00000000..de8bfba8 --- /dev/null +++ b/src/VahterBanBot/ModerationQualityHistory.fs @@ -0,0 +1,96 @@ +namespace VahterBanBot + +open System +open System.Data +open System.Threading +open BotInfra +open Dapper +open Npgsql +open VahterBanBot.Types +open VahterBanBot.Utils + +type ModerationQualityHistory(connectionString: string, clock: TimeProvider) = + let store = EventStore(connectionString, "event", eventJsonOpts, "event_snapshot") + let setting (conn: NpgsqlConnection) name fallback = task { + let! value = conn.QuerySingleOrDefaultAsync("SELECT value FROM bot_setting WHERE key = @name", {| name = name |}) + return match Int32.TryParse value with true, n when n > 0 -> n | _ -> fallback + } + + let rebuild (conn: NpgsqlConnection) (day: DateTime) (ct: CancellationToken) = task { + use! tx = conn.BeginTransactionAsync(IsolationLevel.RepeatableRead, ct) + let! candidates = conn.QueryAsync(CommandDefinition(""" + SELECT DISTINCT stream_id FROM event + WHERE event_type = 'MlScoredMessage' AND created_at >= @day AND created_at < @finish + """, {| day = day; finish = day.AddDays 1.0 |}, tx, cancellationToken = ct)) + let counts = Collections.Generic.Dictionary() + let increment cohort system model code = + let key = cohort, system, model, code + let found, count = counts.TryGetValue key + counts[key] <- (if found then count else 0L) + 1L + for batch in candidates |> Seq.chunkBySize 250 do + ct.ThrowIfCancellationRequested() + let suffixes = batch |> Array.map (fun s -> s.Substring("detection:".Length)) + let ids = [ for s in suffixes do for prefix in ["message:"; "moderation:"; "detection:"] do yield prefix + s ] + let! rows = store.ReadRawEventsForStreams(conn, tx, ids) + let users = + [ for s in suffixes do + for r in rows["message:" + s] do + match store.Deserialize r with + | MessageReceived e -> yield $"user:{e.userId}" + | _ -> () ] |> List.distinct + let! bans = conn.QueryAsync(CommandDefinition(""" + SELECT id, stream_id, stream_version, event_type, data::text, created_at FROM event + WHERE stream_id = ANY(@users) AND event_type IN ('UserBanned', 'UserUnbanned') + """, {| users = List.toArray users |}, tx, cancellationToken = ct)) + let unbans = bans |> Seq.groupBy (fun r -> r.stream_id) |> Seq.collect (fun (_, rs) -> ModerationQuality.unbans store (Seq.toList rs)) |> Seq.toList + for s in suffixes do + let parts = s.Split ':' + let corrections = unbans |> List.filter (fun c -> c.ChatId = Int64.Parse parts[0] && c.MessageId = Int64.Parse parts[1]) + match ModerationQuality.evaluate store rows["message:" + s] rows["moderation:" + s] rows["detection:" + s] corrections with + | Some evaluation when evaluation.EvaluatedAt.Date = day -> + for c in evaluation.Contributions do increment c.Cohort c.System c.LlmModel c.Outcome.Code + | _ -> () + for system in ["pipeline"; "ml"] do + let key = "all_scored", system, None, "tp" + if not (counts.ContainsKey key) then counts[key] <- 0L + let! _ = conn.ExecuteAsync(CommandDefinition("DELETE FROM moderation_quality_daily WHERE day = @day", {| day = day |}, tx, cancellationToken = ct)) + for ((cohort, system, model), values) in counts |> Seq.groupBy (fun kv -> let c, s, m, _ = kv.Key in c, s, m) do + let value code = values |> Seq.sumBy (fun kv -> let _, _, _, c = kv.Key in if c = code then kv.Value else 0L) + let! _ = conn.ExecuteAsync(CommandDefinition(""" + INSERT INTO moderation_quality_daily (day, cohort, system, llm_model, tp, tn, fp, fn, abstained, unresolved, excluded) + VALUES (@day, @cohort, @system, @model, @tp, @tn, @fp, @fn, @abstained, @unresolved, @excluded) + """, {| day = day; cohort = cohort; system = system; model = Option.toObj model; + tp = value "tp"; tn = value "tn"; fp = value "fp"; fn = value "fn"; + abstained = value "abstained"; unresolved = value "unresolved"; excluded = value "excluded" |}, tx, cancellationToken = ct)) + () + do! tx.CommitAsync ct + } + + member _.Run(day: DateTime option, ct: CancellationToken) = task { + let connection = NpgsqlConnectionStringBuilder(connectionString) + connection.Pooling <- false + use conn = new NpgsqlConnection(connection.ConnectionString) + do! conn.OpenAsync ct + let! settling = setting conn "QUALITY_SETTLING_DAYS" 14 + let cutoff = clock.GetUtcNow().UtcDateTime.Date.AddDays(-float settling) + if day |> Option.exists (fun d -> d.Kind <> DateTimeKind.Utc || d <> d.Date || d >= cutoff) then + invalidArg (nameof day) "Day must be a mature UTC date" + let! acquired = conn.QuerySingleAsync("SELECT pg_try_advisory_lock(734829105)") + if not acquired then invalidOp "A quality history job is already running" + let! limit = setting conn "QUALITY_BACKFILL_DAYS" 7 + let! days = task { + match day with + | Some d -> return [d] + | None -> + let! result = conn.QueryAsync(CommandDefinition(""" + SELECT d AT TIME ZONE 'UTC' FROM generate_series( + (SELECT min(created_at) AT TIME ZONE 'UTC' FROM event WHERE event_type = 'MlScoredMessage')::date::timestamp, + (@cutoff AT TIME ZONE 'UTC') - interval '1 day', interval '1 day') d + WHERE NOT EXISTS (SELECT 1 FROM moderation_quality_daily q WHERE q.day = d::date AND q.cohort = 'all_scored' AND q.system = 'pipeline') + ORDER BY d LIMIT @limit + """, {| cutoff = cutoff; limit = min limit 31 |}, cancellationToken = ct)) + return Seq.toList result + } + for d in days do do! rebuild conn d ct + return days.Length + } diff --git a/src/VahterBanBot/Program.fs b/src/VahterBanBot/Program.fs index fb976a55..26b9c601 100644 --- a/src/VahterBanBot/Program.fs +++ b/src/VahterBanBot/Program.fs @@ -222,6 +222,8 @@ WebhookHost.configureSharedServices webhookCfg builder %builder.Services .AddSingleton>(botConfOptions) + .AddSingleton(fun sp -> + ModerationQualityHistory(connString, sp.GetRequiredService())) .AddSingleton(fun sp -> DbService(connString, sp.GetRequiredService())) .AddSingleton(fun _ -> OcrCacheRepository(connString) :> IOcrCache) @@ -318,6 +320,25 @@ SettingsDump.mapConfigDumpEndpoint SettingsDump.toJson eventJsonOpts live) app +%app.MapPost("/quality-history", Func>(fun ctx -> task { + if not (WebhookHost.validateApiKey webhookCfg.SecretToken ctx) then + return Results.Text("Access Denied", statusCode = 401) + else + let raw = string ctx.Request.Query["day"] + let parsed = DateTime.TryParseExact(raw, "yyyy-MM-dd", Globalization.CultureInfo.InvariantCulture, + Globalization.DateTimeStyles.AssumeUniversal ||| Globalization.DateTimeStyles.AdjustToUniversal) + match raw, parsed with + | "", _ | _, (true, _) -> + try + let day = if raw = "" then None else Some (snd parsed) + let! count = ctx.RequestServices.GetRequiredService().Run(day, ctx.RequestAborted) + return Results.Ok {| completedDays = count |} + with + | :? ArgumentException as ex -> return Results.BadRequest ex.Message + | :? InvalidOperationException as ex -> return Results.Conflict ex.Message + | _ -> return Results.BadRequest "Expected day=yyyy-MM-dd, or omit day to backfill missing days" +})) + // Backfill/repair of the snapshot_* read models and aggregate snapshots from the event log. // Idempotent; run manually. `?scope=users` limits it to user streams (skips the message pass). %app.MapPost("/rebuild-snapshots", Func>(fun ctx -> diff --git a/src/VahterBanBot/VahterBanBot.fsproj b/src/VahterBanBot/VahterBanBot.fsproj index 09aa4adc..6b3f21c2 100644 --- a/src/VahterBanBot/VahterBanBot.fsproj +++ b/src/VahterBanBot/VahterBanBot.fsproj @@ -15,6 +15,8 @@ + + diff --git a/src/VahterBanBot/docs/MODERATION-QUALITY.md b/src/VahterBanBot/docs/MODERATION-QUALITY.md new file mode 100644 index 00000000..c4ba6cf4 --- /dev/null +++ b/src/VahterBanBot/docs/MODERATION-QUALITY.md @@ -0,0 +1,71 @@ +# Historical moderation quality + +`moderation_quality_daily` stores correction-based estimates, not independently audited ground truth. Uncorrected decisions are assumed correct; missed spam that nobody reports remains invisible. + +## Populations + +| Cohort | System | Meaning | +| --- | --- | --- | +| `all_scored` | `pipeline` | Actual ML + LLM automatic deletion, once per ML-scored message | +| `all_scored` | `ml` | Counterfactual ML alone, SPAM when score > 0, HAM otherwise | +| `llm_triaged` | `ml` | Counterfactual ML on the same messages as the paired LLM row | +| `llm_triaged` | `llm` | Recorded LLM verdict, grouped by recorded deployment name | + +Filter both cohort and system before summing. The populations overlap: adding systems or cohorts double-counts messages. In the paired cohort, `llm_model` identifies the partner LLM deployment for both systems; NULL means the historical event has no model attribution. ML model versions are not recorded. + +The paired cohort requires an LLM event and its preceding ML score in `[-0.5, 1.5)`; 1.5 is the direct-ML spam boundary. It includes cached LLM verdicts. SKIP contributes `abstained` to both paired rows, keeping the precision/recall populations identical. + +The UTC day comes from the first ML score. Edits do not create a second message. The latest score and LLM verdict before the latest human correction are evaluated. A text change between evaluation and correction excludes that comparison. + +Latest decisive human correction wins across message labels, reviewer actions, and human unbans. An unban corrects only the message targeted by that user's preceding ban. Without a correction, automatic deletion implies SPAM, otherwise the current LLM verdict supplies the reference, otherwise HAM is assumed. + +Whole-pipeline rows measure automatic deletions, so human-caught spam without an automatic deletion is FN. Unreviewed SKIP is unresolved; reviewed SKIP can contribute to pipeline counts. Non-ML/LLM deletion reasons are excluded from whole-pipeline comparisons. Messages without an ML score, including deterministic prefilters and older uninstrumented history, are outside these populations. + +## Scheduling and backfill + +The existing scheduler runs the job at `CLEANUP_SCHEDULED_HOUR_UTC`. `QUALITY_SETTLING_DAYS` defaults to 14 full UTC days after a day ends; the latest eligible day is therefore today minus 15 days. + +The job fills missing eligible days from the first recorded ML score, oldest first. `QUALITY_BACKFILL_DAYS` defaults to 7 and is capped at 31 per invocation; once caught up, one day becomes eligible each day. These settings are read from `bot_setting` on each run. + +Authenticated `POST /quality-history` processes another bounded batch. Repeat until `completedDays` is zero to backfill without waiting for the daily schedule. `POST /quality-history?day=YYYY-MM-DD` replaces one eligible day, including corrections that arrived after its previous calculation. + +No settling period guarantees that corrections are final. A correction older than the settling window requires rebuilding its original scoring day. Increasing the window does not remove already-computed rows; dashboard eligibility should use the same window. + +Each day uses one repeatable-read transaction and atomically replaces its rows. Completed days survive interruption, empty days get zero rows, and retries cannot multiply counts. A database session advisory lock serializes scheduled and manual runs across pods; the dedicated unpooled connection releases it on disposal. + +The job reads indexed event ranges and batches message streams. Dashboard reads use the aggregate table's day-leading unique index; no event replay occurs in a dashboard query. The existing event-type/time and stream indexes support backfill. + +## Dashboard SQL + +Use half-open UTC dates, `$1` inclusive and `$2` exclusive. Parameters are bound by the dashboard/query client. Return NULL for empty denominators. + +```sql +SELECT day, + sum(tp) AS tp, sum(tn) AS tn, sum(fp) AS fp, sum(fn) AS fn, + 100.0 * sum(tp) / nullif(sum(tp + fp), 0) AS precision_pct, + 100.0 * sum(tp) / nullif(sum(tp + fn), 0) AS recall_pct, + sum(abstained) AS abstained, sum(unresolved) AS unresolved, + sum(excluded) AS excluded +FROM moderation_quality_daily +WHERE day >= $1::date AND day < $2::date + AND cohort = 'all_scored' AND system = 'pipeline' +GROUP BY day +ORDER BY day; +``` + +For an interval, remove `day` from SELECT and remove GROUP BY/ORDER BY. Sum counts first; do not average daily percentages or multiply ML and LLM precision. + +```sql +SELECT system, llm_model, + sum(tp) AS tp, sum(tn) AS tn, sum(fp) AS fp, sum(fn) AS fn, + 100.0 * sum(tp) / nullif(sum(tp + fp), 0) AS precision_pct, + 100.0 * sum(tp) / nullif(sum(tp + fn), 0) AS recall_pct, + sum(abstained) AS abstained, sum(unresolved) AS unresolved, + sum(excluded) AS excluded +FROM moderation_quality_daily +WHERE day >= $1::date AND day < $2::date AND cohort = 'llm_triaged' +GROUP BY system, llm_model +ORDER BY llm_model, system; +``` + +For the latest 30 settled days at the default window, bind `$1 = CURRENT_DATE - 44` and `$2 = CURRENT_DATE - 14` with the database session in UTC. Show the covered date range and missing days alongside the metrics while backfill is catching up. diff --git a/src/vahter-bot/migrations/V54__moderation_quality_history.sql b/src/vahter-bot/migrations/V54__moderation_quality_history.sql new file mode 100644 index 00000000..54215b10 --- /dev/null +++ b/src/vahter-bot/migrations/V54__moderation_quality_history.sql @@ -0,0 +1,20 @@ +CREATE TABLE moderation_quality_daily ( + day date NOT NULL, + cohort text NOT NULL CHECK (cohort IN ('all_scored', 'llm_triaged')), + system text NOT NULL CHECK (system IN ('pipeline', 'ml', 'llm')), + llm_model text, + tp bigint NOT NULL DEFAULT 0, + tn bigint NOT NULL DEFAULT 0, + fp bigint NOT NULL DEFAULT 0, + fn bigint NOT NULL DEFAULT 0, + abstained bigint NOT NULL DEFAULT 0, + unresolved bigint NOT NULL DEFAULT 0, + excluded bigint NOT NULL DEFAULT 0, + computed_at timestamptz NOT NULL DEFAULT now(), + UNIQUE NULLS NOT DISTINCT (day, cohort, system, llm_model) +); +GRANT SELECT, INSERT, UPDATE, DELETE ON moderation_quality_daily TO vahter_bot_ban_service; +INSERT INTO scheduled_job (job_name) VALUES ('moderation_quality_daily'); +INSERT INTO bot_setting (key, value, type, feature_group, description) VALUES + ('QUALITY_SETTLING_DAYS', '14', 'FREE_FORM', 'quality', 'Full UTC days after day end before quality aggregation'), + ('QUALITY_BACKFILL_DAYS', '7', 'FREE_FORM', 'quality', 'Maximum missing historical days processed per daily run'); diff --git a/tests/VahterBanBot.Tests/ModerationQualityTests.fs b/tests/VahterBanBot.Tests/ModerationQualityTests.fs new file mode 100644 index 00000000..b8746b88 --- /dev/null +++ b/tests/VahterBanBot.Tests/ModerationQualityTests.fs @@ -0,0 +1,129 @@ +module VahterBanBot.Tests.ModerationQualityTests + +open System +open System.Threading +open Dapper +open Npgsql +open VahterBanBot +open VahterBanBot.Types +open VahterBanBot.Tests.ContainerTestBase +open Xunit + +[] +type QualityRow = + { cohort: string; system: string; llm_model: string | null + tp: int64; tn: int64; fp: int64; fn: int64; abstained: int64; unresolved: int64; excluded: int64 } + +type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = + let day = DateTime(2001, 2, 3, 0, 0, 0, DateTimeKind.Utc) + let chat = -900010L + let job = ModerationQualityHistory(fixture.DbConnectionString, TimeProvider.System) + let add id score verdict model = task { + do! fixture.InsertRawEvent($"message:{chat}:{id}", 1, + MessageReceived {| chatId = chat; messageId = id; userId = id; text = Some "example"; rawMessage = "{}" |}, day.AddMinutes 1.0) + do! fixture.InsertRawEvent($"detection:{chat}:{id}", 1, + MlScoredMessage {| chatId = chat; messageId = id; score = score; isSpam = score > 0.0 |}, day.AddMinutes 2.0) + do! fixture.InsertRawEvent($"detection:{chat}:{id}", 2, + LlmClassified {| chatId = chat; messageId = id; verdict = verdict; reason = None; modelName = model + promptTokens = 1; completionTokens = 1; latencyMs = 1; promptHash = None |}, day.AddMinutes 3.0) + if verdict = "SPAM" then + do! fixture.InsertRawEvent($"moderation:{chat}:{id}", 1, + BotAutoDeleted {| chatId = chat; messageId = id; userId = id + reason = LlmSpam {| score = score; modelName = defaultArg model "unknown"; reason = None; cacheScope = None |} |}, day.AddMinutes 4.0) + } + let rows () = task { + use conn = new NpgsqlConnection(fixture.DbConnectionString) + let! result = conn.QueryAsync("SELECT * FROM moderation_quality_daily WHERE day = @day", {| day = day |}) + return Seq.toList result + } + + [] + member _.``Historical counts compare identical messages and replace corrected days`` () = task { + do! add 91001L 0.7 "SPAM" (Some "sol") + do! add 91002L -0.2 "SPAM" (Some "sol") + do! add 91003L 0.0 "NOT_SPAM" (Some "sol") + do! add 91004L 0.4 "NOT_SPAM" (Some "sol") + do! add 91005L 0.5 "SKIP" (Some "sol") + do! add 91006L 0.7 "SPAM" (Some "mini") + do! fixture.InsertRawEvent($"message:{chat}:91006", 2, + MessageMarkedHam {| chatId = chat; messageId = 91006L; text = "example"; markedBy = Some 1L |}, day.AddDays 2.0) + let! completed = job.Run(Some day, CancellationToken.None) + Assert.Equal(1, completed) + let! first = rows () + let row cohort system model = first |> List.find (fun r -> r.cohort = cohort && r.system = system && r.llm_model = model) + let ml = row "llm_triaged" "ml" "sol" + Assert.Equal((1L, 1L, 1L, 1L, 1L), (ml.tp, ml.tn, ml.fp, ml.fn, ml.abstained)) + let llm = row "llm_triaged" "llm" "sol" + Assert.Equal((2L, 2L, 0L, 0L, 1L), (llm.tp, llm.tn, llm.fp, llm.fn, llm.abstained)) + Assert.Equal(1L, (row "llm_triaged" "llm" "mini").fp) + let pipeline = row "all_scored" "pipeline" null + Assert.Equal((2L, 2L, 1L, 0L, 1L), (pipeline.tp, pipeline.tn, pipeline.fp, pipeline.fn, pipeline.unresolved)) + do! fixture.InsertRawEvent($"message:{chat}:91006", 3, + MessageMarkedSpam {| chatId = chat; messageId = 91006L; markedBy = Some 1L |}, day.AddDays 20.0) + let! _ = job.Run(Some day, CancellationToken.None) + let! second = rows () + let mini = second |> List.find (fun r -> r.cohort = "llm_triaged" && r.system = "llm" && r.llm_model = "mini") + Assert.Equal((1L, 0L), (mini.tp, mini.fp)) + let! _ = job.Run(Some day, CancellationToken.None) + let! third = rows () + Assert.Equal(List.sortBy (fun r -> r.cohort, r.system, r.llm_model) second, + List.sortBy (fun r -> r.cohort, r.system, r.llm_model) third) + } + + [] + member _.``Unban corrects its ban target and empty days are persisted`` () = task { + let targetDay = day.AddDays 1.0 + let id = 92001L + do! fixture.InsertRawEvent($"message:{chat}:{id}", 1, + MessageReceived {| chatId = chat; messageId = id; userId = id; text = Some "example"; rawMessage = "{}" |}, targetDay) + do! fixture.InsertRawEvent($"detection:{chat}:{id}", 1, + MlScoredMessage {| chatId = chat; messageId = id; score = 3.0; isSpam = true |}, targetDay.AddMinutes 1.0) + do! fixture.InsertRawEvent($"moderation:{chat}:{id}", 1, + BotAutoDeleted {| chatId = chat; messageId = id; userId = id; reason = MlSpam {| score = 3.0 |} |}, targetDay.AddMinutes 2.0) + do! fixture.InsertRawEvent($"user:{id}", 1, + UserBanned {| userId = id; bannedBy = None; actor = Some Actor.ML; chatId = Some chat; messageId = Some id + messageText = None; bannedAt = targetDay.AddMinutes 2.0 |}, targetDay.AddMinutes 2.0) + do! fixture.InsertRawEvent($"user:{id}", 2, + UserUnbanned {| userId = id; unbannedBy = Some 1L; actor = Some (Actor.User {| userId = 1L; username = None |}) |}, targetDay.AddDays 3.0) + let! _ = job.Run(Some targetDay, CancellationToken.None) + use conn = new NpgsqlConnection(fixture.DbConnectionString) + let! fp = conn.QuerySingleAsync("SELECT fp FROM moderation_quality_daily WHERE day = @day AND system = 'pipeline'", {| day = targetDay |}) + Assert.Equal(1L, fp) + let! _ = job.Run(Some (day.AddDays 2.0), CancellationToken.None) + let! n = conn.QuerySingleAsync("SELECT count(*) FROM moderation_quality_daily WHERE day = @day AND tp + tn + fp + fn = 0", {| day = day.AddDays 2.0 |}) + Assert.Equal(2L, n) + } + + [] + member _.``Unsettled dates are rejected`` () = task { + let! _ = Assert.ThrowsAsync(fun () -> job.Run(Some DateTime.UtcNow.Date, CancellationToken.None)) + () + } + + [] + member _.``Backfill resumes missing UTC days and respects the maturity boundary`` () = task { + let start = DateTime(1999, 2, 1, 0, 0, 0, DateTimeKind.Utc) + let clock = { new TimeProvider() with override _.GetUtcNow() = DateTimeOffset(start.AddDays 18.0) } + let history = ModerationQualityHistory(fixture.DbConnectionString, clock) + do! fixture.InsertRawEvent($"detection:{chat}:93001", 1, + MlScoredMessage {| chatId = chat; messageId = 93001L; score = -2.0; isSpam = false |}, start.AddHours 1.0) + let! count = history.Run(None, CancellationToken.None) + Assert.Equal(4, count) + let! resumed = history.Run(None, CancellationToken.None) + Assert.Equal(0, resumed) + let! _ = Assert.ThrowsAsync(fun () -> history.Run(Some (start.AddDays 4.0), CancellationToken.None)) + let! lastMature = history.Run(Some (start.AddDays 3.0), CancellationToken.None) + Assert.Equal(1, lastMature) + } + + [] + member _.``History endpoint validates dates and requires authentication`` () = task { + use payload = new System.Net.Http.StringContent("") + use! invalid = fixture.BotHttp.PostAsync("/quality-history?day=invalid", payload) + Assert.Equal(System.Net.HttpStatusCode.BadRequest, invalid.StatusCode) + use! immature = fixture.BotHttp.PostAsync("/quality-history?day=9999-01-01", payload) + Assert.Equal(System.Net.HttpStatusCode.BadRequest, immature.StatusCode) + use client = new System.Net.Http.HttpClient(BaseAddress = fixture.BotHttp.BaseAddress) + use! unauthorized = client.PostAsync("/quality-history", payload) + Assert.Equal(System.Net.HttpStatusCode.Unauthorized, unauthorized.StatusCode) + } diff --git a/tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj b/tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj index d62264ed..521ae6d4 100644 --- a/tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj +++ b/tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj @@ -39,6 +39,7 @@ + From 264417ceff03768b9904b996ee5d2ef3210c02a6 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Wed, 30 Sep 2026 08:07:14 +0100 Subject: [PATCH 2/2] Refresh rolling quality days and repair older event changes --- src/VahterBanBot/Cleanup.fs | 2 +- src/VahterBanBot/ModerationQualityHistory.fs | 89 +++++++++-- src/VahterBanBot/Program.fs | 24 +-- src/VahterBanBot/docs/MODERATION-QUALITY.md | 18 +-- .../V54__moderation_quality_history.sql | 25 ++- .../ModerationQualityTests.fs | 145 ++++++++++++++++-- 6 files changed, 256 insertions(+), 47 deletions(-) diff --git a/src/VahterBanBot/Cleanup.fs b/src/VahterBanBot/Cleanup.fs index c176bcfa..4967de7f 100644 --- a/src/VahterBanBot/Cleanup.fs +++ b/src/VahterBanBot/Cleanup.fs @@ -123,7 +123,7 @@ type CleanupService( // Check scheduled jobs do! tryRunJob "daily_cleanup" (TimeSpan.FromHours botConf.Value.CleanupScheduledHour) runCleanup do! tryRunJob "moderation_quality_daily" (TimeSpan.FromHours botConf.Value.CleanupScheduledHour) (fun () -> task { - let! _ = quality.Run(None, ct) + let! _ = quality.Run(QualityHistoryMode.Daily, ct) () }) do! tryRunJob "daily_stats" (TimeSpan.FromHours botConf.Value.StatsScheduledHour) runStats diff --git a/src/VahterBanBot/ModerationQualityHistory.fs b/src/VahterBanBot/ModerationQualityHistory.fs index de8bfba8..66a33046 100644 --- a/src/VahterBanBot/ModerationQualityHistory.fs +++ b/src/VahterBanBot/ModerationQualityHistory.fs @@ -2,6 +2,9 @@ namespace VahterBanBot open System open System.Data +open System.Diagnostics +open Microsoft.Extensions.Logging +open Microsoft.Extensions.Logging.Abstractions open System.Threading open BotInfra open Dapper @@ -9,7 +12,14 @@ open Npgsql open VahterBanBot.Types open VahterBanBot.Utils -type ModerationQualityHistory(connectionString: string, clock: TimeProvider) = +[] +type QualityHistoryMode = Daily | Backfill | Rebuild of DateTime + +[] +type private QualityPending = { event_id: int64; stream_id: string } + +type ModerationQualityHistory(connectionString: string, clock: TimeProvider, ?logger: ILogger) = + let logger = defaultArg logger NullLogger.Instance let store = EventStore(connectionString, "event", eventJsonOpts, "event_snapshot") let setting (conn: NpgsqlConnection) name fallback = task { let! value = conn.QuerySingleOrDefaultAsync("SELECT value FROM bot_setting WHERE key = @name", {| name = name |}) @@ -63,25 +73,78 @@ type ModerationQualityHistory(connectionString: string, clock: TimeProvider) = tp = value "tp"; tn = value "tn"; fp = value "fp"; fn = value "fn"; abstained = value "abstained"; unresolved = value "unresolved"; excluded = value "excluded" |}, tx, cancellationToken = ct)) () + let! _ = conn.ExecuteAsync(CommandDefinition("DELETE FROM moderation_quality_dirty_day WHERE day = @day", {| day = day |}, tx, cancellationToken = ct)) + do! tx.CommitAsync ct + } + + let collectDirtyDays (conn: NpgsqlConnection) (ct: CancellationToken) = task { + use! tx = conn.BeginTransactionAsync(ct) + let! pending = conn.QueryAsync(CommandDefinition( + "SELECT event_id, stream_id FROM moderation_quality_pending", transaction = tx, cancellationToken = ct)) + let pending = Seq.toArray pending + let days = Collections.Generic.HashSet() + for batch in Array.chunkBySize 500 pending do + let eventIds = batch |> Array.map (fun e -> e.event_id) |> Set.ofArray + let users = batch |> Array.filter (fun e -> e.stream_id.StartsWith("user:")) |> Array.map (fun e -> e.stream_id) |> Array.distinct + let! bans = conn.QueryAsync(CommandDefinition(""" + SELECT id, stream_id, stream_version, event_type, data::text, created_at FROM event + WHERE stream_id = ANY(@users) AND event_type IN ('UserBanned', 'UserUnbanned') + """, {| users = users |}, tx, cancellationToken = ct)) + let targets = + [| for e in batch do + if e.stream_id.StartsWith("detection:") then yield e.stream_id + for _, rows in bans |> Seq.groupBy (fun r -> r.stream_id) do + for unban in ModerationQuality.unbans store (Seq.toList rows) do + if eventIds.Contains unban.Id then yield $"detection:{unban.ChatId}:{unban.MessageId}" |] + |> Array.distinct + let! affected = conn.QueryAsync(CommandDefinition(""" + SELECT DISTINCT date_trunc('day', first.created_at, 'UTC') FROM unnest(@targets::text[]) AS target + CROSS JOIN LATERAL ( + SELECT created_at FROM event WHERE stream_id = target AND event_type = 'MlScoredMessage' + ORDER BY created_at, id LIMIT 1 + ) first + """, {| targets = targets |}, tx, cancellationToken = ct)) + for day in affected do %days.Add day + if days.Count > 0 then + let! _ = conn.ExecuteAsync(CommandDefinition(""" + INSERT INTO moderation_quality_dirty_day (day) + SELECT (d AT TIME ZONE 'UTC')::date FROM unnest(@days::timestamptz[]) AS d + ON CONFLICT DO NOTHING + """, {| days = Seq.toArray days |}, tx, cancellationToken = ct)) + () + if pending.Length > 0 then + let! _ = conn.ExecuteAsync(CommandDefinition( + "DELETE FROM moderation_quality_pending WHERE event_id = ANY(@ids)", + {| ids = pending |> Array.map (fun e -> e.event_id) |}, tx, cancellationToken = ct)) + () do! tx.CommitAsync ct } - member _.Run(day: DateTime option, ct: CancellationToken) = task { + member _.Run(mode: QualityHistoryMode, ct: CancellationToken) = task { + let elapsed = Stopwatch.StartNew() let connection = NpgsqlConnectionStringBuilder(connectionString) connection.Pooling <- false use conn = new NpgsqlConnection(connection.ConnectionString) do! conn.OpenAsync ct - let! settling = setting conn "QUALITY_SETTLING_DAYS" 14 - let cutoff = clock.GetUtcNow().UtcDateTime.Date.AddDays(-float settling) - if day |> Option.exists (fun d -> d.Kind <> DateTimeKind.Utc || d <> d.Date || d >= cutoff) then - invalidArg (nameof day) "Day must be a mature UTC date" + let today = clock.GetUtcNow().UtcDateTime.Date + let cutoff = today.AddDays -14.0 + match mode with + | QualityHistoryMode.Rebuild day when day.Kind <> DateTimeKind.Utc || day <> day.Date || day > today -> + invalidArg (nameof mode) "Day must be a UTC date no later than today" + | _ -> () let! acquired = conn.QuerySingleAsync("SELECT pg_try_advisory_lock(734829105)") if not acquired then invalidOp "A quality history job is already running" - let! limit = setting conn "QUALITY_BACKFILL_DAYS" 7 let! days = task { - match day with - | Some d -> return [d] - | None -> + match mode with + | QualityHistoryMode.Rebuild day -> return [day] + | QualityHistoryMode.Daily -> + do! collectDirtyDays conn ct + let! dirty = conn.QueryAsync(CommandDefinition( + "SELECT day::timestamp AT TIME ZONE 'UTC' FROM moderation_quality_dirty_day WHERE day <= @today", + {| today = today |}, cancellationToken = ct)) + return [ yield! dirty; for offset in -14..0 do yield today.AddDays(float offset) ] |> List.distinct |> List.sort + | QualityHistoryMode.Backfill -> + let! limit = setting conn "QUALITY_BACKFILL_DAYS" 7 let! result = conn.QueryAsync(CommandDefinition(""" SELECT d AT TIME ZONE 'UTC' FROM generate_series( (SELECT min(created_at) AT TIME ZONE 'UTC' FROM event WHERE event_type = 'MlScoredMessage')::date::timestamp, @@ -91,6 +154,10 @@ type ModerationQualityHistory(connectionString: string, clock: TimeProvider) = """, {| cutoff = cutoff; limit = min limit 31 |}, cancellationToken = ct)) return Seq.toList result } - for d in days do do! rebuild conn d ct + for day in days do do! rebuild conn day ct + let name = match mode with QualityHistoryMode.Daily -> "daily" | QualityHistoryMode.Backfill -> "backfill" | QualityHistoryMode.Rebuild _ -> "rebuild" + logger.LogInformation("Moderation quality {Mode} completed {CompletedDays} days in {DurationSeconds} seconds", name, days.Length, elapsed.Elapsed.TotalSeconds) + if mode = QualityHistoryMode.Daily && elapsed.Elapsed.TotalSeconds > 30.0 then + logger.LogWarning("Moderation quality daily refresh exceeded 30-second budget: {DurationSeconds} seconds", elapsed.Elapsed.TotalSeconds) return days.Length } diff --git a/src/VahterBanBot/Program.fs b/src/VahterBanBot/Program.fs index 26b9c601..33a4693e 100644 --- a/src/VahterBanBot/Program.fs +++ b/src/VahterBanBot/Program.fs @@ -223,7 +223,7 @@ WebhookHost.configureSharedServices webhookCfg builder %builder.Services .AddSingleton>(botConfOptions) .AddSingleton(fun sp -> - ModerationQualityHistory(connString, sp.GetRequiredService())) + ModerationQualityHistory(connString, sp.GetRequiredService(), sp.GetRequiredService>())) .AddSingleton(fun sp -> DbService(connString, sp.GetRequiredService())) .AddSingleton(fun _ -> OcrCacheRepository(connString) :> IOcrCache) @@ -320,23 +320,29 @@ SettingsDump.mapConfigDumpEndpoint SettingsDump.toJson eventJsonOpts live) app +let parseQualityHistoryMode (ctx: HttpContext) = + let raw = string ctx.Request.Query["day"] + let parsed = DateTime.TryParseExact(raw, "yyyy-MM-dd", Globalization.CultureInfo.InvariantCulture, + Globalization.DateTimeStyles.AssumeUniversal ||| Globalization.DateTimeStyles.AdjustToUniversal) + match string ctx.Request.Query["mode"], raw, parsed with + | ("" | "daily"), "", _ -> Some QualityHistoryMode.Daily + | "backfill", "", _ -> Some QualityHistoryMode.Backfill + | ("" | "rebuild"), _, (true, day) -> Some (QualityHistoryMode.Rebuild day) + | _ -> None + %app.MapPost("/quality-history", Func>(fun ctx -> task { if not (WebhookHost.validateApiKey webhookCfg.SecretToken ctx) then return Results.Text("Access Denied", statusCode = 401) else - let raw = string ctx.Request.Query["day"] - let parsed = DateTime.TryParseExact(raw, "yyyy-MM-dd", Globalization.CultureInfo.InvariantCulture, - Globalization.DateTimeStyles.AssumeUniversal ||| Globalization.DateTimeStyles.AdjustToUniversal) - match raw, parsed with - | "", _ | _, (true, _) -> + match parseQualityHistoryMode ctx with + | Some mode -> try - let day = if raw = "" then None else Some (snd parsed) - let! count = ctx.RequestServices.GetRequiredService().Run(day, ctx.RequestAborted) + let! count = ctx.RequestServices.GetRequiredService().Run(mode, ctx.RequestAborted) return Results.Ok {| completedDays = count |} with | :? ArgumentException as ex -> return Results.BadRequest ex.Message | :? InvalidOperationException as ex -> return Results.Conflict ex.Message - | _ -> return Results.BadRequest "Expected day=yyyy-MM-dd, or omit day to backfill missing days" + | None -> return Results.BadRequest "Use mode=daily, mode=backfill, or day=yyyy-MM-dd" })) // Backfill/repair of the snapshot_* read models and aggregate snapshots from the event log. diff --git a/src/VahterBanBot/docs/MODERATION-QUALITY.md b/src/VahterBanBot/docs/MODERATION-QUALITY.md index c4ba6cf4..b78b6a90 100644 --- a/src/VahterBanBot/docs/MODERATION-QUALITY.md +++ b/src/VahterBanBot/docs/MODERATION-QUALITY.md @@ -23,24 +23,24 @@ Whole-pipeline rows measure automatic deletions, so human-caught spam without an ## Scheduling and backfill -The existing scheduler runs the job at `CLEANUP_SCHEDULED_HOUR_UTC`. `QUALITY_SETTLING_DAYS` defaults to 14 full UTC days after a day ends; the latest eligible day is therefore today minus 15 days. +The existing scheduler runs at `CLEANUP_SCHEDULED_HOUR_UTC`. Every daily run replaces today and the previous 14 UTC days, plus older affected days. Today is partial as of `computed_at`; recent days are provisional because reviewers can still correct decisions. -The job fills missing eligible days from the first recorded ML score, oldest first. `QUALITY_BACKFILL_DAYS` defaults to 7 and is capped at 31 per invocation; once caught up, one day becomes eligible each day. These settings are read from `bot_setting` on each run. +An event-table insert trigger atomically enqueues relevant message, scoring, moderation, and unban events. The job resolves their original scoring days, persists them in `moderation_quality_dirty_day`, and acknowledges only the pending event IDs it read, in one transaction. Late commits remain pending regardless of event-ID order. -Authenticated `POST /quality-history` processes another bounded batch. Repeat until `completedDays` is zero to backfill without waiting for the daily schedule. `POST /quality-history?day=YYYY-MM-DD` replaces one eligible day, including corrections that arrived after its previous calculation. +Each day uses a repeatable-read transaction to replace its rows and remove its dirty marker together. New events during rebuilding remain pending for the next run. Failed or interrupted rebuilds leave dirty days available for retry; older edits and corrections are covered without a maximum age. -No settling period guarantees that corrections are final. A correction older than the settling window requires rebuilding its original scoring day. Increasing the window does not remove already-computed rows; dashboard eligibility should use the same window. +Authenticated `POST /quality-history` (or `?mode=daily`) runs a refresh. `POST /quality-history?day=YYYY-MM-DD` replaces one day up to and including today. A session advisory lock serializes scheduled, backfill, and manual jobs across pods. -Each day uses one repeatable-read transaction and atomically replaces its rows. Completed days survive interruption, empty days get zero rows, and retries cannot multiply counts. A database session advisory lock serializes scheduled and manual runs across pods; the dedicated unpooled connection releases it on disposal. +One-time backfill is explicit: `POST /quality-history?mode=backfill` fills missing days before the rolling window, oldest first. `QUALITY_BACKFILL_DAYS` defaults to 7 and is capped at 31 per invocation. Repeat until `completedDays` is zero; completed and empty days are persisted, so retries resume safely. Daily refresh does not run this backfill. -The job reads indexed event ranges and batches message streams. Dashboard reads use the aggregate table's day-leading unique index; no event replay occurs in a dashboard query. The existing event-type/time and stream indexes support backfill. +Structured completion logs include mode, completed-day count, and elapsed seconds. Daily refresh warns when it exceeds the 30-second performance budget; it continues to finish its work. Dashboard queries read the day-indexed aggregate table and never replay events. ## Dashboard SQL Use half-open UTC dates, `$1` inclusive and `$2` exclusive. Parameters are bound by the dashboard/query client. Return NULL for empty denominators. ```sql -SELECT day, +SELECT day, day >= CURRENT_DATE - 14 AS provisional, max(computed_at) AS computed_at, sum(tp) AS tp, sum(tn) AS tn, sum(fp) AS fp, sum(fn) AS fn, 100.0 * sum(tp) / nullif(sum(tp + fp), 0) AS precision_pct, 100.0 * sum(tp) / nullif(sum(tp + fn), 0) AS recall_pct, @@ -53,7 +53,7 @@ GROUP BY day ORDER BY day; ``` -For an interval, remove `day` from SELECT and remove GROUP BY/ORDER BY. Sum counts first; do not average daily percentages or multiply ML and LLM precision. +For an interval, remove `day`, replace the provisional expression with `bool_or(day >= CURRENT_DATE - 14)`, and remove GROUP BY/ORDER BY. Sum counts first; do not average daily percentages or multiply ML and LLM precision. ```sql SELECT system, llm_model, @@ -68,4 +68,4 @@ GROUP BY system, llm_model ORDER BY llm_model, system; ``` -For the latest 30 settled days at the default window, bind `$1 = CURRENT_DATE - 44` and `$2 = CURRENT_DATE - 14` with the database session in UTC. Show the covered date range and missing days alongside the metrics while backfill is catching up. +For 30 days including today, bind `$1 = CURRENT_DATE - 29` and `$2 = CURRENT_DATE + 1` with the database session in UTC. For 30 complete days, use `$1 = CURRENT_DATE - 30` and `$2 = CURRENT_DATE`. Show coverage, missing days, and `computed_at`; even older days can change after edits or corrections. diff --git a/src/vahter-bot/migrations/V54__moderation_quality_history.sql b/src/vahter-bot/migrations/V54__moderation_quality_history.sql index 54215b10..bcd6c515 100644 --- a/src/vahter-bot/migrations/V54__moderation_quality_history.sql +++ b/src/vahter-bot/migrations/V54__moderation_quality_history.sql @@ -16,5 +16,26 @@ CREATE TABLE moderation_quality_daily ( GRANT SELECT, INSERT, UPDATE, DELETE ON moderation_quality_daily TO vahter_bot_ban_service; INSERT INTO scheduled_job (job_name) VALUES ('moderation_quality_daily'); INSERT INTO bot_setting (key, value, type, feature_group, description) VALUES - ('QUALITY_SETTLING_DAYS', '14', 'FREE_FORM', 'quality', 'Full UTC days after day end before quality aggregation'), - ('QUALITY_BACKFILL_DAYS', '7', 'FREE_FORM', 'quality', 'Maximum missing historical days processed per daily run'); + ('QUALITY_BACKFILL_DAYS', '7', 'FREE_FORM', 'quality', 'Maximum older missing days processed per explicit backfill invocation'); + +CREATE TABLE moderation_quality_pending ( + event_id bigint PRIMARY KEY, + stream_id text NOT NULL +); +CREATE TABLE moderation_quality_dirty_day (day date PRIMARY KEY); +GRANT SELECT, INSERT, UPDATE, DELETE ON moderation_quality_pending, moderation_quality_dirty_day + TO vahter_bot_ban_service; + +CREATE FUNCTION moderation_quality_enqueue() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + INSERT INTO moderation_quality_pending (event_id, stream_id) + VALUES (NEW.id, CASE WHEN NEW.event_type = 'UserUnbanned' THEN NEW.stream_id + ELSE 'detection:' || substring(NEW.stream_id FROM position(':' IN NEW.stream_id) + 1) END); + RETURN NULL; +END $$; +CREATE TRIGGER moderation_quality_enqueue AFTER INSERT ON event + FOR EACH ROW WHEN (NEW.event_type IN ( + 'MessageReceived', 'MessageEdited', 'MessageMarkedHam', 'MessageMarkedSpam', + 'MlScoredMessage', 'LlmClassified', 'LlmVerdictCacheHit', + 'VahterActed', 'BotAutoDeleted', 'UserUnbanned')) + EXECUTE FUNCTION moderation_quality_enqueue(); diff --git a/tests/VahterBanBot.Tests/ModerationQualityTests.fs b/tests/VahterBanBot.Tests/ModerationQualityTests.fs index b8746b88..e3392ff0 100644 --- a/tests/VahterBanBot.Tests/ModerationQualityTests.fs +++ b/tests/VahterBanBot.Tests/ModerationQualityTests.fs @@ -1,6 +1,7 @@ module VahterBanBot.Tests.ModerationQualityTests open System +open System.Text.Json open System.Threading open Dapper open Npgsql @@ -37,6 +38,20 @@ type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = return Seq.toList result } + let historyAt today = + let clock = { new TimeProvider() with override _.GetUtcNow() = DateTimeOffset(today) } + ModerationQualityHistory(fixture.DbConnectionString, clock) + let scoredAt at id score = task { + do! fixture.InsertRawEvent($"message:{chat}:{id}", 1, + MessageReceived {| chatId = chat; messageId = id; userId = id; text = Some "example"; rawMessage = "{}" |}, at) + do! fixture.InsertRawEvent($"detection:{chat}:{id}", 1, + MlScoredMessage {| chatId = chat; messageId = id; score = score; isSpam = score > 0.0 |}, at.AddMinutes 1.0) + } + let mlAt at = task { + use conn = new NpgsqlConnection(fixture.DbConnectionString) + return! conn.QuerySingleAsync("SELECT * FROM moderation_quality_daily WHERE day = @day AND cohort = 'all_scored' AND system = 'ml'", {| day = at |}) + } + [] member _.``Historical counts compare identical messages and replace corrected days`` () = task { do! add 91001L 0.7 "SPAM" (Some "sol") @@ -47,7 +62,7 @@ type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = do! add 91006L 0.7 "SPAM" (Some "mini") do! fixture.InsertRawEvent($"message:{chat}:91006", 2, MessageMarkedHam {| chatId = chat; messageId = 91006L; text = "example"; markedBy = Some 1L |}, day.AddDays 2.0) - let! completed = job.Run(Some day, CancellationToken.None) + let! completed = job.Run(QualityHistoryMode.Rebuild day, CancellationToken.None) Assert.Equal(1, completed) let! first = rows () let row cohort system model = first |> List.find (fun r -> r.cohort = cohort && r.system = system && r.llm_model = model) @@ -60,11 +75,11 @@ type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = Assert.Equal((2L, 2L, 1L, 0L, 1L), (pipeline.tp, pipeline.tn, pipeline.fp, pipeline.fn, pipeline.unresolved)) do! fixture.InsertRawEvent($"message:{chat}:91006", 3, MessageMarkedSpam {| chatId = chat; messageId = 91006L; markedBy = Some 1L |}, day.AddDays 20.0) - let! _ = job.Run(Some day, CancellationToken.None) + let! _ = job.Run(QualityHistoryMode.Rebuild day, CancellationToken.None) let! second = rows () let mini = second |> List.find (fun r -> r.cohort = "llm_triaged" && r.system = "llm" && r.llm_model = "mini") Assert.Equal((1L, 0L), (mini.tp, mini.fp)) - let! _ = job.Run(Some day, CancellationToken.None) + let! _ = job.Run(QualityHistoryMode.Rebuild day, CancellationToken.None) let! third = rows () Assert.Equal(List.sortBy (fun r -> r.cohort, r.system, r.llm_model) second, List.sortBy (fun r -> r.cohort, r.system, r.llm_model) third) @@ -83,37 +98,41 @@ type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = do! fixture.InsertRawEvent($"user:{id}", 1, UserBanned {| userId = id; bannedBy = None; actor = Some Actor.ML; chatId = Some chat; messageId = Some id messageText = None; bannedAt = targetDay.AddMinutes 2.0 |}, targetDay.AddMinutes 2.0) + let! _ = job.Run(QualityHistoryMode.Rebuild targetDay, CancellationToken.None) do! fixture.InsertRawEvent($"user:{id}", 2, UserUnbanned {| userId = id; unbannedBy = Some 1L; actor = Some (Actor.User {| userId = 1L; username = None |}) |}, targetDay.AddDays 3.0) - let! _ = job.Run(Some targetDay, CancellationToken.None) + let! _ = job.Run(QualityHistoryMode.Daily, CancellationToken.None) use conn = new NpgsqlConnection(fixture.DbConnectionString) let! fp = conn.QuerySingleAsync("SELECT fp FROM moderation_quality_daily WHERE day = @day AND system = 'pipeline'", {| day = targetDay |}) Assert.Equal(1L, fp) - let! _ = job.Run(Some (day.AddDays 2.0), CancellationToken.None) + let! _ = job.Run(QualityHistoryMode.Rebuild (day.AddDays 2.0), CancellationToken.None) let! n = conn.QuerySingleAsync("SELECT count(*) FROM moderation_quality_daily WHERE day = @day AND tp + tn + fp + fn = 0", {| day = day.AddDays 2.0 |}) Assert.Equal(2L, n) } [] - member _.``Unsettled dates are rejected`` () = task { - let! _ = Assert.ThrowsAsync(fun () -> job.Run(Some DateTime.UtcNow.Date, CancellationToken.None)) + member _.``Future dates are rejected`` () = task { + let! _ = Assert.ThrowsAsync(fun () -> job.Run(QualityHistoryMode.Rebuild (DateTime.UtcNow.Date.AddDays 1.0), CancellationToken.None)) () } [] - member _.``Backfill resumes missing UTC days and respects the maturity boundary`` () = task { + member _.``Backfill resumes missing UTC days and leaves the rolling window to daily refresh`` () = task { let start = DateTime(1999, 2, 1, 0, 0, 0, DateTimeKind.Utc) let clock = { new TimeProvider() with override _.GetUtcNow() = DateTimeOffset(start.AddDays 18.0) } let history = ModerationQualityHistory(fixture.DbConnectionString, clock) do! fixture.InsertRawEvent($"detection:{chat}:93001", 1, MlScoredMessage {| chatId = chat; messageId = 93001L; score = -2.0; isSpam = false |}, start.AddHours 1.0) - let! count = history.Run(None, CancellationToken.None) + let! count = history.Run(QualityHistoryMode.Backfill, CancellationToken.None) Assert.Equal(4, count) - let! resumed = history.Run(None, CancellationToken.None) + let! resumed = history.Run(QualityHistoryMode.Backfill, CancellationToken.None) Assert.Equal(0, resumed) - let! _ = Assert.ThrowsAsync(fun () -> history.Run(Some (start.AddDays 4.0), CancellationToken.None)) - let! lastMature = history.Run(Some (start.AddDays 3.0), CancellationToken.None) - Assert.Equal(1, lastMature) + use conn = new NpgsqlConnection(fixture.DbConnectionString) + let! recentBefore = conn.QuerySingleAsync("SELECT count(*) FROM moderation_quality_daily WHERE day >= @first AND day <= @last", {| first = start.AddDays 4.0; last = start.AddDays 18.0 |}) + Assert.Equal(0L, recentBefore) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! recentAfter = conn.QuerySingleAsync("SELECT count(DISTINCT day) FROM moderation_quality_daily WHERE day >= @first AND day <= @last", {| first = start.AddDays 4.0; last = start.AddDays 18.0 |}) + Assert.Equal(15L, recentAfter) } [] @@ -121,9 +140,105 @@ type ModerationQualityTests(fixture: MlDisabledVahterTestContainers) = use payload = new System.Net.Http.StringContent("") use! invalid = fixture.BotHttp.PostAsync("/quality-history?day=invalid", payload) Assert.Equal(System.Net.HttpStatusCode.BadRequest, invalid.StatusCode) - use! immature = fixture.BotHttp.PostAsync("/quality-history?day=9999-01-01", payload) - Assert.Equal(System.Net.HttpStatusCode.BadRequest, immature.StatusCode) + use! future = fixture.BotHttp.PostAsync("/quality-history?day=9999-01-01", payload) + Assert.Equal(System.Net.HttpStatusCode.BadRequest, future.StatusCode) use client = new System.Net.Http.HttpClient(BaseAddress = fixture.BotHttp.BaseAddress) use! unauthorized = client.PostAsync("/quality-history", payload) Assert.Equal(System.Net.HttpStatusCode.Unauthorized, unauthorized.StatusCode) } + + [] + member _.``Daily refresh replaces today's provisional counts`` () = task { + let today = DateTime(2010, 1, 15, 0, 0, 0, DateTimeKind.Utc) + let history = historyAt today + let id = 94001L + do! scoredAt today id 0.7 + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! before = mlAt today + Assert.Equal((0L, 1L), (before.tp, before.fp)) + do! fixture.InsertRawEvent($"message:{chat}:{id}", 2, + MessageMarkedSpam {| chatId = chat; messageId = id; markedBy = Some 1L |}, today.AddHours 1.0) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! after = mlAt today + Assert.Equal((1L, 0L), (after.tp, after.fp)) + } + + [] + member _.``An old edited message rebuilds its original day without double counting today`` () = task { + let today = DateTime(2011, 8, 1, 0, 0, 0, DateTimeKind.Utc) + let original = today.AddDays -149.0 + let history = historyAt today + let id = 95001L + do! scoredAt original id -2.0 + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! before = mlAt original + Assert.Equal(1L, before.tn) + do! fixture.InsertRawEvent($"message:{chat}:{id}", 2, + MessageEdited {| chatId = chat; messageId = id; userId = id; text = Some "edited"; rawMessage = "{}" |}, today) + do! fixture.InsertRawEvent($"detection:{chat}:{id}", 2, + MlScoredMessage {| chatId = chat; messageId = id; score = 0.7; isSpam = true |}, today.AddMinutes 1.0) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! after = mlAt original + Assert.Equal((0L, 1L), (after.tn, after.fp)) + let! recent = mlAt today + Assert.Equal(0L, recent.tp + recent.tn + recent.fp + recent.fn) + } + + [] + member _.``A lower event id committed after a refresh is processed and rollback leaves no pending work`` () = task { + let today = DateTime(2012, 8, 1, 0, 0, 0, DateTimeKind.Utc) + let original = today.AddDays -50.0 + let history = historyAt today + let id = 96001L + do! scoredAt original id 2.0 + do! fixture.InsertRawEvent($"moderation:{chat}:{id}", 1, + BotAutoDeleted {| chatId = chat; messageId = id; userId = id; reason = MlSpam {| score = 2.0 |} |}, original.AddMinutes 2.0) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + use writer = new NpgsqlConnection(fixture.DbConnectionString) + do! writer.OpenAsync() + use! tx = writer.BeginTransactionAsync() + let ham = MessageMarkedHam {| chatId = chat; messageId = id; text = "example"; markedBy = Some 1L |} + let! lowerId = writer.QuerySingleAsync("INSERT INTO event(stream_id,stream_version,data,created_at) VALUES (@stream,2,@data::jsonb,@at) RETURNING id", + {| stream = $"message:{chat}:{id}"; data = JsonSerializer.Serialize(ham, eventJsonOpts); at = today |}, tx) + do! scoredAt today 96002L -2.0 + use reader = new NpgsqlConnection(fixture.DbConnectionString) + let! higherId = reader.QuerySingleAsync("SELECT max(id) FROM event") + Assert.True(higherId > lowerId) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! before = mlAt original + Assert.Equal(1L, before.tp) + do! tx.CommitAsync() + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! after = mlAt original + Assert.Equal((0L, 1L), (after.tp, after.fp)) + use! rolledBack = writer.BeginTransactionAsync() + let spam = MessageMarkedSpam {| chatId = chat; messageId = id; markedBy = Some 1L |} + let! rolledId = writer.QuerySingleAsync("INSERT INTO event(stream_id,stream_version,data,created_at) VALUES (@stream,3,@data::jsonb,@at) RETURNING id", + {| stream = $"message:{chat}:{id}"; data = JsonSerializer.Serialize(spam, eventJsonOpts); at = today.AddHours 1.0 |}, rolledBack) + do! rolledBack.RollbackAsync() + let! pending = reader.QuerySingleAsync("SELECT count(*) FROM moderation_quality_pending WHERE event_id = @id", {| id = rolledId |}) + Assert.Equal(0L, pending) + } + + [] + member _.``A failed old-day rebuild preserves its dirty marker for retry`` () = task { + let today = DateTime(2013, 8, 1, 0, 0, 0, DateTimeKind.Utc) + let original = today.AddDays -40.0 + let history = historyAt today + let id = 97001L + do! scoredAt original id -2.0 + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + use conn = new NpgsqlConnection(fixture.DbConnectionString) + let malformed = JsonSerializer.Serialize {| Case = "MessageEdited"; chatId = chat; messageId = id; userId = id; text = [|"invalid"|]; rawMessage = "{}" |} + let! brokenId = conn.QuerySingleAsync("INSERT INTO event(stream_id,stream_version,data,created_at) VALUES (@stream,2,@data::jsonb,@at) RETURNING id", + {| stream = $"message:{chat}:{id}"; data = malformed; at = today |}) + let! _ = Assert.ThrowsAnyAsync(fun () -> history.Run(QualityHistoryMode.Daily, CancellationToken.None)) + let! dirty = conn.QuerySingleAsync("SELECT count(*) FROM moderation_quality_dirty_day WHERE day = @day", {| day = original |}) + Assert.Equal(1L, dirty) + let! unchanged = mlAt original + Assert.Equal(1L, unchanged.tn) + let! _ = conn.ExecuteAsync("DELETE FROM event WHERE id = @id", {| id = brokenId |}) + let! _ = history.Run(QualityHistoryMode.Daily, CancellationToken.None) + let! remaining = conn.QuerySingleAsync("SELECT count(*) FROM moderation_quality_dirty_day WHERE day = @day", {| day = original |}) + Assert.Equal(0L, remaining) + }