feat(jobs): daily feedback aggregator writes per-(severity x provider) summary to AppSettings

v1.4.16 phase B5e: ship the pg-boss `feedback-aggregator` queue that
runs daily at 04:00 Europe/Berlin (after the rate-limit, idempotency,
audit-log cleanups). Aggregates the last 30 days of
`RecommendationFeedback` rows into per-(severity, metricSourceType,
providerType, promptVersion) buckets and writes a JSON summary to
the singleton `AppSettings.adminAiInsightsFeedbackSummary` row.

Summary shape:
  { generatedAt, windowDays, buckets:
    [{ severity, metricSourceType, providerType, promptVersion,
       helpful, notHelpful, total, helpfulRate }] }

Bucketing is split into a pure `buildFeedbackBuckets()` helper so the
admin's ad-hoc analysis tools can reuse it without going through
Prisma. helpfulRate is rounded to 2 decimals; sort order is
deterministic (severity → metricSourceType → providerType →
promptVersion) so the admin UI's row order stays stable across
aggregator runs.

Single-user-default-on policy (research §3): the aggregator runs
unconditionally for the user's own ratings. Cross-user prompt-tuning
stays admin-only via `requireAdmin()` gating on the consuming
endpoint; the aggregator just produces the data. Prompt mutation
itself is a v1.4.17 ratchet — this commit only delivers the loop
infrastructure.

Worker handler degrades gracefully on failure: warns instead of
throwing so a single-tick error doesn't poison the queue and block
the next cleanup window. Wide-Event annotations: feedback_buckets,
feedback_total_rows, feedback_window_days.

Tests: 5 new unit tests (3 bucketing scenarios + 2 aggregator
default-window + custom-window).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.7 committed 2026-05-10 01:39:59 +02:00
1 parent 47ca1e41e0
commit 71ffe3022f
3 files changed
+380

No files matched your search

@@ -0,0 +1,173 @@
import { describe, it, expect, vi } from "vitest";
import type { PrismaClient } from "@/generated/prisma/client";
import {
aggregateRecommendationFeedback,
buildFeedbackBuckets,
DEFAULT_FEEDBACK_AGGREGATION_WINDOW_DAYS,
} from "../feedback-aggregator";
interface FeedbackRow {
recommendationSeverity: string;
metricSourceType: string;
providerType: string;
promptVersion: string;
helpful: boolean;
}
function makePrismaMock(rows: FeedbackRow[]) {
return {
recommendationFeedback: {
findMany: vi.fn().mockResolvedValue(rows),
},
appSettings: {
upsert: vi.fn().mockResolvedValue({}),
},
} as unknown as PrismaClient;
}
describe("buildFeedbackBuckets", () => {
it("groups by (severity, metricSourceType, providerType, promptVersion) and counts up/down", () => {
const rows: FeedbackRow[] = [
// Two thumbs-up + one thumbs-down on the same bucket → 2/3 helpfulRate.
{
recommendationSeverity: "important",
metricSourceType: "bloodPressure",
providerType: "codex",
promptVersion: "4.16.0",
helpful: true,
},
{
recommendationSeverity: "important",
metricSourceType: "bloodPressure",
providerType: "codex",
promptVersion: "4.16.0",
helpful: true,
},
{
recommendationSeverity: "important",
metricSourceType: "bloodPressure",
providerType: "codex",
promptVersion: "4.16.0",
helpful: false,
},
// Different provider in same severity bucket — counted separately.
{
recommendationSeverity: "important",
metricSourceType: "bloodPressure",
providerType: "openai",
promptVersion: "4.16.0",
helpful: false,
},
];
const buckets = buildFeedbackBuckets(rows);
expect(buckets).toHaveLength(2);
const codex = buckets.find((b) => b.providerType === "codex");
expect(codex).toBeDefined();
expect(codex?.helpful).toBe(2);
expect(codex?.notHelpful).toBe(1);
expect(codex?.total).toBe(3);
// Math.round(2/3 * 100) / 100 = 0.67
expect(codex?.helpfulRate).toBeCloseTo(0.67, 2);
const openai = buckets.find((b) => b.providerType === "openai");
expect(openai?.helpful).toBe(0);
expect(openai?.notHelpful).toBe(1);
expect(openai?.helpfulRate).toBe(0);
});
it("sorts buckets deterministically (severity, metricSourceType, provider, prompt)", () => {
const rows: FeedbackRow[] = [
{
recommendationSeverity: "urgent",
metricSourceType: "weight",
providerType: "openai",
promptVersion: "4.16.0",
helpful: true,
},
{
recommendationSeverity: "info",
metricSourceType: "mood",
providerType: "codex",
promptVersion: "4.16.0",
helpful: true,
},
{
recommendationSeverity: "info",
metricSourceType: "bloodPressure",
providerType: "codex",
promptVersion: "4.16.0",
helpful: true,
},
];
const buckets = buildFeedbackBuckets(rows);
expect(buckets.map((b) => b.severity)).toEqual([
"info",
"info",
"urgent",
]);
// Within "info" the metricSourceType is the secondary key.
expect(buckets[0].metricSourceType).toBe("bloodPressure");
expect(buckets[1].metricSourceType).toBe("mood");
});
it("emits an empty array on no rows (aggregator's null-summary path)", () => {
expect(buildFeedbackBuckets([])).toEqual([]);
});
});
describe("aggregateRecommendationFeedback", () => {
it("queries the last 30 days by default and writes a singleton AppSettings row", async () => {
const now = new Date("2026-05-09T04:00:00Z");
const prisma = makePrismaMock([
{
recommendationSeverity: "suggestion",
metricSourceType: "weight",
providerType: "codex",
promptVersion: "4.16.0",
helpful: true,
},
]);
await aggregateRecommendationFeedback(prisma, { now });
expect(prisma.recommendationFeedback.findMany).toHaveBeenCalledTimes(1);
const call = (prisma.recommendationFeedback.findMany as ReturnType<typeof vi.fn>).mock
.calls[0][0];
const expectedSince = new Date(
now.getTime() - DEFAULT_FEEDBACK_AGGREGATION_WINDOW_DAYS * 86_400_000,
);
expect(call.where.createdAt.gte).toEqual(expectedSince);
expect(prisma.appSettings.upsert).toHaveBeenCalledTimes(1);
const upsertCall = (
prisma.appSettings.upsert as ReturnType<typeof vi.fn>
).mock.calls[0][0];
expect(upsertCall.where).toEqual({ id: "singleton" });
const summary = upsertCall.update.adminAiInsightsFeedbackSummary;
expect(summary.windowDays).toBe(DEFAULT_FEEDBACK_AGGREGATION_WINDOW_DAYS);
expect(summary.generatedAt).toBe(now.toISOString());
expect(summary.buckets).toHaveLength(1);
expect(summary.buckets[0].providerType).toBe("codex");
});
it("respects a custom windowDays argument", async () => {
const now = new Date("2026-05-09T04:00:00Z");
const prisma = makePrismaMock([]);
await aggregateRecommendationFeedback(prisma, { now, windowDays: 7 });
const call = (prisma.recommendationFeedback.findMany as ReturnType<typeof vi.fn>).mock
.calls[0][0];
const expectedSince = new Date(now.getTime() - 7 * 86_400_000);
expect(call.where.createdAt.gte).toEqual(expectedSince);
const upsertCall = (
prisma.appSettings.upsert as ReturnType<typeof vi.fn>
).mock.calls[0][0];
expect(upsertCall.update.adminAiInsightsFeedbackSummary.windowDays).toBe(7);
// Empty buckets when no rows.
expect(
upsertCall.update.adminAiInsightsFeedbackSummary.buckets,
).toEqual([]);
});
});
+167
View File
@@ -0,0 +1,167 @@
/**
* v1.4.16 phase B5e — daily feedback aggregator.
*
* Runs once a day (04:00 Europe/Berlin via the pg-boss schedule
* registered in `reminder-worker.ts`). Aggregates the last 30 days
* of `RecommendationFeedback` rows into per-(severity, metricSourceType,
* providerType, promptVersion) buckets and writes a summary JSON to
* `AppSettings.adminAiInsightsFeedbackSummary` (singleton row).
*
* The summary feeds the `/admin/ai-quality` preview that surfaces
* helpful-rate per severity x provider — a v1.4.17 ratchet will read
* the same shape to drive prompt-tuning (low-helpful-rate buckets
* get a stricter "OMIT" or "REPHRASE" rule appended next prompt
* version, per research §5.2).
*
* Single-user-default-on policy (research §3): the aggregator runs
* unconditionally for the user's own ratings. Cross-user training
* stays admin-only and the summary lives on the singleton AppSettings
* row that's already gated behind requireAdmin().
*
* No prompt mutation in v1.4.16 — the aggregator only writes the
* summary blob; the prompt-tuning step is deferred to v1.4.17.
*/
import type { PrismaClient } from "@/generated/prisma/client";
export const DEFAULT_FEEDBACK_AGGREGATION_WINDOW_DAYS = 30;
export interface FeedbackBucket {
severity: string;
metricSourceType: string;
providerType: string;
promptVersion: string;
helpful: number;
notHelpful: number;
total: number;
/** Fraction in [0..1], rounded to 2 decimals. Always 0 when total=0. */
helpfulRate: number;
}
export interface FeedbackAggregationSummary {
generatedAt: string;
windowDays: number;
buckets: FeedbackBucket[];
}
interface MinimalFeedbackRow {
recommendationSeverity: string;
metricSourceType: string;
providerType: string;
promptVersion: string;
helpful: boolean;
}
/**
* Pure helper — group rows by the four-tuple key and compute helpful
* counts per bucket. Exported so the unit test can pin the bucketing
* shape without going through Prisma. Callers that already have rows
* in memory (admin-only ad-hoc analysis tools) can reuse this without
* the Prisma roundtrip.
*
* Sort order is deterministic: severity, then metricSourceType, then
* providerType, then promptVersion. The admin UI rendering this list
* relies on the order being stable across aggregator runs.
*/
export function buildFeedbackBuckets(
rows: MinimalFeedbackRow[],
): FeedbackBucket[] {
const map = new Map<string, FeedbackBucket>();
for (const row of rows) {
const key = [
row.recommendationSeverity,
row.metricSourceType,
row.providerType,
row.promptVersion,
].join("");
const existing = map.get(key);
if (existing) {
if (row.helpful) existing.helpful += 1;
else existing.notHelpful += 1;
existing.total += 1;
} else {
map.set(key, {
severity: row.recommendationSeverity,
metricSourceType: row.metricSourceType,
providerType: row.providerType,
promptVersion: row.promptVersion,
helpful: row.helpful ? 1 : 0,
notHelpful: row.helpful ? 0 : 1,
total: 1,
helpfulRate: 0,
});
}
}
const buckets = [...map.values()];
for (const b of buckets) {
b.helpfulRate =
b.total === 0 ? 0 : Math.round((b.helpful / b.total) * 100) / 100;
}
buckets.sort((a, b) => {
return (
a.severity.localeCompare(b.severity) ||
a.metricSourceType.localeCompare(b.metricSourceType) ||
a.providerType.localeCompare(b.providerType) ||
a.promptVersion.localeCompare(b.promptVersion)
);
});
return buckets;
}
export interface AggregateOptions {
/** Override "now" for deterministic tests. Defaults to `new Date()`. */
now?: Date;
/** Override the rolling window. Defaults to 30 days. */
windowDays?: number;
}
/**
* Read the rolling-window feedback rows, bucket them, and persist the
* summary to the singleton AppSettings row. Returns the summary so
* callers (the worker handler, ad-hoc admin tools) can log a digest.
*/
export async function aggregateRecommendationFeedback(
prisma: PrismaClient,
options: AggregateOptions = {},
): Promise<FeedbackAggregationSummary> {
const now = options.now ?? new Date();
const windowDays =
options.windowDays ?? DEFAULT_FEEDBACK_AGGREGATION_WINDOW_DAYS;
const since = new Date(now.getTime() - windowDays * 86_400_000);
const rows = await prisma.recommendationFeedback.findMany({
where: { createdAt: { gte: since } },
select: {
recommendationSeverity: true,
metricSourceType: true,
providerType: true,
promptVersion: true,
helpful: true,
},
});
const summary: FeedbackAggregationSummary = {
generatedAt: now.toISOString(),
windowDays,
buckets: buildFeedbackBuckets(rows),
};
// The singleton AppSettings row is auto-created on first install
// (`/admin/general` already touches it). Use upsert so the
// aggregator works on a fresh DB during testcontainer integration
// runs without depending on init ordering.
await prisma.appSettings.upsert({
where: { id: "singleton" },
update: {
adminAiInsightsFeedbackSummary: summary as unknown as object,
},
create: {
id: "singleton",
adminAiInsightsFeedbackSummary: summary as unknown as object,
},
});
return summary;
}
+40
View File
@@ -30,6 +30,7 @@ import { setGlobalBoss } from "@/lib/jobs/boss-instance";
import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup";
import { cleanupOldAuditLogs } from "@/lib/jobs/audit-log-cleanup";
import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler";
import { aggregateRecommendationFeedback } from "@/lib/jobs/feedback-aggregator";
import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret";
import { deleteMessage } from "@/lib/telegram";
import { decrypt, encrypt } from "@/lib/crypto";
@@ -109,6 +110,11 @@ const OFFHOST_BACKUP_CRON = "30 2 * * *";
const HOST_METRIC_QUEUE = "host-metric-sample";
// Per-minute cadence — matches the chart's 60s polling refetchInterval.
const HOST_METRIC_CRON = "* * * * *";
// v1.4.16 phase B5e — daily rec-feedback aggregator. 04:00 Europe/Berlin
// runs the slot AFTER all the cleanup jobs (rate-limit, idempotency,
// audit-log) so the previous-day's noise is gone before we aggregate.
const FEEDBACK_AGGREGATOR_QUEUE = "feedback-aggregator";
const FEEDBACK_AGGREGATOR_CRON = "0 4 * * *";
interface ReminderCheckPayload {
triggeredAt: string;
@@ -176,6 +182,10 @@ interface HostMetricSamplePayload {
triggeredAt: string;
}
interface FeedbackAggregatorPayload {
triggeredAt: string;
}
// Re-export timezone utilities under local names for backward compatibility
const getUserTodayBounds = getUserTodayBoundsUtil;
const getDayOfWeekInTz = getDayOfWeekInTzUtil;
@@ -938,6 +948,29 @@ async function handleHostMetricSample(jobs: Job<HostMetricSamplePayload>[]) {
});
}
async function handleFeedbackAggregator(
jobs: Job<FeedbackAggregatorPayload>[],
) {
void jobs;
await withBackgroundEvent("job.feedback_aggregator", async (evt) => {
const p = getWorkerPrisma();
try {
const summary = await aggregateRecommendationFeedback(p);
evt.addMeta("feedback_buckets", summary.buckets.length);
evt.addMeta(
"feedback_total_rows",
summary.buckets.reduce((acc, b) => acc + b.total, 0),
);
evt.addMeta("feedback_window_days", summary.windowDays);
} catch (err) {
// The admin dashboard tolerates a stale summary — log and move
// on rather than poisoning the boss queue with retries that
// would block the next cleanup window.
evt.addWarning(`feedback-aggregator failed: ${err}`);
}
});
}
async function handleOffhostBackup(jobs: Job<OffhostBackupPayload>[]) {
void jobs;
await withBackgroundEvent("job.offhost_backup", async (evt) => {
@@ -1184,6 +1217,7 @@ export async function startReminderWorker() {
AUDIT_LOG_CLEANUP_QUEUE,
OFFHOST_BACKUP_QUEUE,
HOST_METRIC_QUEUE,
FEEDBACK_AGGREGATOR_QUEUE,
];
for (const q of allQueues) {
@@ -1207,6 +1241,7 @@ export async function startReminderWorker() {
[AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON],
[OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
[HOST_METRIC_QUEUE, HOST_METRIC_CRON],
[FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON],
];
for (const [name, cron] of schedules) {
@@ -1294,6 +1329,11 @@ export async function startReminderWorker() {
{ localConcurrency: 1 },
handleHostMetricSample,
);
await boss.work<FeedbackAggregatorPayload>(
FEEDBACK_AGGREGATOR_QUEUE,
{ localConcurrency: 1 },
handleFeedbackAggregator,
);
return boss;
}