feat: implement onboarding ledger event handling and related tests

package.json

@@ -6,6 +6,7 @@
   "scripts": {
     "dev": "tsx watch src/index.ts",
     "build": "tsc -p tsconfig.json",
+    "test:onboarding": "tsx scripts/onboarding-ledger.test.ts",
     "start": "node dist/index.js",
     "typecheck": "tsc -p tsconfig.json --noEmit",
     "workflows:smoke": "tsx src/workflows/smoke-test.ts",

scripts/onboarding-ledger.test.ts

@@ -0,0 +1,74 @@
+import assert from "node:assert/strict";
+import {
+  completedAtFromOnboardingPayload,
+  isValidOnboardingLedgerEvent,
+  normalizeOnboardingEventType,
+} from "../src/events/onboarding-ledger.js";
+
+const now = "2026-06-28T00:00:00.000Z";
+
+assert.equal(normalizeOnboardingEventType("user_onboarding_completed"), "user.onboarding.completed");
+
+assert.equal(
+  isValidOnboardingLedgerEvent({
+    type: "onboarding.completed",
+    payload: {},
+  }),
+  true,
+  "explicit completion event should satisfy onboarding status",
+);
+
+assert.equal(
+  isValidOnboardingLedgerEvent({
+    type: "user.onboarding.completed",
+    payload: {},
+  }),
+  true,
+  "user completion event should satisfy onboarding status",
+);
+
+assert.equal(
+  isValidOnboardingLedgerEvent({
+    type: "profile.onboarding.completed",
+    payload: {},
+  }),
+  true,
+  "profile completion event should satisfy onboarding status",
+);
+
+assert.equal(
+  isValidOnboardingLedgerEvent({
+    type: "onboarding.snapshot.saved",
+    payload: { onboarding: { current_step: 2 } },
+  }),
+  false,
+  "intermediate snapshots must not bypass onboarding",
+);
+
+assert.equal(
+  isValidOnboardingLedgerEvent({
+    type: "onboarding.snapshot.saved",
+    payload: { onboarding: { completed_at: now } },
+  }),
+  true,
+  "completion snapshots should satisfy onboarding status",
+);
+
+assert.equal(
+  completedAtFromOnboardingPayload({
+    preferences: { onboarding: { completed_at: now } },
+  }),
+  now,
+  "completion timestamp should be extracted from preferences snapshot",
+);
+
+assert.equal(
+  completedAtFromOnboardingPayload({
+    completedAt: "not-a-date",
+  })?.endsWith("Z"),
+  true,
+  "invalid completion timestamps should normalize to a valid ISO timestamp",
+);
+
+console.log("onboarding-ledger tests passed");
+process.exit(0);

src/events/onboarding-ledger.ts

@@ -0,0 +1,192 @@
+import { and, desc, eq, inArray } from "drizzle-orm";
+import { db } from "../db/client.js";
+import { growEvents, type GrowEventRow } from "../db/schema.js";
+import { asRecord } from "./envelope.js";
+import {
+  markGrowEventFailed,
+  markGrowEventProcessed,
+  markGrowEventProcessing,
+  recordGrowEvent,
+} from "./record-grow-event.js";
+import { ensureOnboardingBaselineQscoreForCompletedAt } from "./onboarding-qscore.js";
+import {
+  onboardingCompletedAtFromEvent,
+  runCuratorOnboardingLoopSafely,
+} from "../v1/curator/curator-onboarding-loop.js";
+
+export const ONBOARDING_LEDGER_EVENT_TYPES = [
+  "onboarding.snapshot.saved",
+  "onboarding.completed",
+  "user.onboarding.completed",
+  "profile.onboarding.completed",
+] as const;
+
+export type OnboardingLedgerEventType = (typeof ONBOARDING_LEDGER_EVENT_TYPES)[number];
+
+export const ONBOARDING_LEDGER_QUERY_TYPES = [
+  ...ONBOARDING_LEDGER_EVENT_TYPES,
+  "onboarding_snapshot_saved",
+  "onboarding_completed",
+  "user_onboarding_completed",
+  "profile_onboarding_completed",
+] as const;
+
+export type OnboardingStatusEvent = {
+  id: string;
+  type: string;
+  occurredAt: Date;
+  processingStatus: string;
+};
+
+const COMPLETION_EVENT_TYPES = new Set<string>([
+  "onboarding.completed",
+  "user.onboarding.completed",
+  "profile.onboarding.completed",
+]);
+
+export function normalizeOnboardingEventType(type: string) {
+  return type.toLowerCase().replaceAll("_", ".");
+}
+
+export function completedAtFromOnboardingPayload(payload: Record<string, unknown> | null | undefined) {
+  const data = payload ?? {};
+  const preferences = asRecord(data.preferences);
+  const onboarding = asRecord(data.onboarding ?? preferences.onboarding);
+  const candidate =
+    data.completedAt ??
+    data.completed_at ??
+    data.onboardingCompletedAt ??
+    data.onboarding_completed_at ??
+    onboarding.completed_at ??
+    onboarding.completedAt ??
+    asRecord(preferences.onboarding).completed_at ??
+    asRecord(preferences.onboarding).completedAt;
+
+  if (candidate instanceof Date) {
+    return Number.isNaN(candidate.getTime()) ? undefined : candidate.toISOString();
+  }
+  if (typeof candidate !== "string" || !candidate.trim()) return undefined;
+  const parsed = new Date(candidate);
+  return Number.isNaN(parsed.getTime()) ? new Date().toISOString() : parsed.toISOString();
+}
+
+export function isValidOnboardingLedgerEvent(event: Pick<GrowEventRow, "type" | "payload">) {
+  const normalizedType = normalizeOnboardingEventType(event.type);
+  if (COMPLETION_EVENT_TYPES.has(normalizedType)) return true;
+
+  // Snapshots are status-valid only when they are completion snapshots. Plain
+  // intermediate step saves must not let a new seeker bypass onboarding.
+  if (normalizedType === "onboarding.snapshot.saved") {
+    return Boolean(completedAtFromOnboardingPayload(event.payload));
+  }
+
+  return false;
+}
+
+export async function getLatestValidOnboardingLedgerEvent(userId: string): Promise<OnboardingStatusEvent | null> {
+  const rows = await db
+    .select({
+      id: growEvents.id,
+      type: growEvents.type,
+      payload: growEvents.payload,
+      occurredAt: growEvents.occurredAt,
+      processingStatus: growEvents.processingStatus,
+    })
+    .from(growEvents)
+    .where(
+      and(
+        eq(growEvents.userId, userId),
+        inArray(growEvents.type, [...ONBOARDING_LEDGER_QUERY_TYPES]),
+      ),
+    )
+    .orderBy(desc(growEvents.occurredAt))
+    .limit(25);
+
+  const row = rows.find((event) => isValidOnboardingLedgerEvent(event));
+  if (!row) return null;
+  const { payload: _payload, ...statusEvent } = row;
+  return statusEvent;
+}
+
+export async function ensureOnboardingBaselineQscoreFromLedger(userId: string) {
+  const event = await getLatestValidOnboardingLedgerEvent(userId);
+  if (!event) return false;
+  return ensureOnboardingBaselineQscoreForCompletedAt(userId, event.occurredAt);
+}
+
+function onboardingContextFromInput(context?: Record<string, unknown>) {
+  const input = context ?? {};
+  const preferences = asRecord(input.preferences);
+  const onboarding = asRecord(input.onboarding ?? preferences.onboarding);
+  return {
+    ...input,
+    onboarding,
+    preferences: Object.keys(preferences).length ? preferences : { onboarding },
+  };
+}
+
+export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow) {
+  if (!event.userId || !isValidOnboardingLedgerEvent(event)) {
+    return {
+      qscoreBaselineSeeded: false,
+      curatorOnboarding: { status: "skipped" as const, reason: event.userId ? "not_onboarding_completion" : "missing_user_id" },
+    };
+  }
+
+  const completedAt =
+    onboardingCompletedAtFromEvent(event) ??
+    completedAtFromOnboardingPayload(event.payload) ??
+    event.occurredAt.toISOString();
+
+  const qscoreBaselineSeeded = await ensureOnboardingBaselineQscoreForCompletedAt(event.userId, completedAt);
+  const curatorOnboarding = await runCuratorOnboardingLoopSafely({
+    userId: event.userId,
+    completedAt,
+    sourceEventId: event.id,
+    source: event.source,
+    context: onboardingContextFromInput(event.payload),
+  });
+
+  return { qscoreBaselineSeeded, curatorOnboarding };
+}
+
+export async function recordAndProcessOnboardingCompletion(input: {
+  userId: string;
+  completedAt: string | Date;
+  source?: string;
+  context?: Record<string, unknown>;
+}) {
+  const completedAt =
+    input.completedAt instanceof Date
+      ? input.completedAt.toISOString()
+      : input.completedAt;
+  const context = onboardingContextFromInput(input.context);
+  const event = await recordGrowEvent(
+    {
+      source: input.source ?? "onboarding",
+      type: "onboarding.completed",
+      category: "usage",
+      userId: input.userId,
+      occurredAt: completedAt,
+      payload: {
+        completedAt,
+        ...context,
+      },
+      dedupeKey: `onboarding:completed:${input.userId}`,
+    },
+    { userId: input.userId, source: input.source ?? "onboarding" },
+  );
+
+  if (event.processingStatus !== "processed") {
+    await markGrowEventProcessing(event.id);
+  }
+
+  const sideEffects = await ensureOnboardingSideEffectsForEvent(event);
+  if (sideEffects.curatorOnboarding.status === "skipped" && sideEffects.curatorOnboarding.reason === "loop_failed") {
+    await markGrowEventFailed(event.id, new Error("curator_onboarding_loop_failed"));
+  } else if (event.processingStatus !== "processed") {
+    await markGrowEventProcessed(event.id);
+  }
+
+  return { event, ...sideEffects };
+}

src/events/onboarding-qscore.ts

@@ -9,12 +9,18 @@ function asRecord(value: unknown): Record<string, unknown> {
   return value && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
 }
 
+function parseCompletedAt(value: unknown): Date | null {
+  if (value instanceof Date) {
+    return Number.isNaN(value.getTime()) ? null : value;
+  }
+  if (typeof value !== "string" || !value.trim()) return null;
+  const parsed = new Date(value);
+  return Number.isNaN(parsed.getTime()) ? new Date() : parsed;
+}
+
 function onboardingCompletedAt(preferences: Record<string, unknown> | undefined): Date | null {
   const onboarding = asRecord(preferences?.onboarding);
-  const completedAt = onboarding.completed_at;
-  if (typeof completedAt !== "string" || !completedAt.trim()) return null;
-  const parsed = new Date(completedAt);
-  return Number.isNaN(parsed.getTime()) ? new Date() : parsed;
+  return parseCompletedAt(onboarding.completed_at ?? onboarding.completedAt);
 }
 
 /**
@@ -32,7 +38,15 @@ export async function ensureOnboardingBaselineQscore(
 ): Promise<boolean> {
   const completedAt = onboardingCompletedAt(preferences);
   if (!completedAt) return false;
+  return ensureOnboardingBaselineQscoreForCompletedAt(userId, completedAt);
+}
 
+export async function ensureOnboardingBaselineQscoreForCompletedAt(
+  userId: string,
+  completedAtInput: string | Date,
+): Promise<boolean> {
+  const completedAt = parseCompletedAt(completedAtInput);
+  if (!completedAt) return false;
   const latestSignals = await db
     .select({ signalId: growQscoreLatest.signalId, score: growQscoreLatest.score })
     .from(growQscoreLatest)

src/events/redis-consumer.ts

@@ -1,9 +1,15 @@
 import { randomUUID } from "node:crypto";
 import { config } from "../config.js";
 import { log } from "../log.js";
-import { recordGrowEvent } from "./record-grow-event.js";
+import {
+  markGrowEventFailed,
+  markGrowEventProcessed,
+  markGrowEventProcessing,
+  recordGrowEvent,
+} from "./record-grow-event.js";
 import { routeGrowEventToUserActor } from "./route-to-user-actor.js";
-import { runCuratorOnboardingLoopForEventSafely } from "../v1/curator/curator-onboarding-loop.js";
+import { applyQscoreProjection } from "./projectors/qscore-projector.js";
+import { ensureOnboardingSideEffectsForEvent } from "./onboarding-ledger.js";
 
 // This file has two Redis ingestion modes:
 // 1. Canonical GrowEvent stream: grow.events.raw — future service event bus.
@@ -151,7 +157,23 @@ async function recordAndRoute(input: unknown) {
   await routeGrowEventToUserActor(event).catch((err) => {
     log.warn({ err, eventId: event.id, userId: event.userId }, "failed to route grow event to user actor");
   });
-  await runCuratorOnboardingLoopForEventSafely(event);
+  if (!event.userId || event.processingStatus === "processed") return event;
+
+  await markGrowEventProcessing(event.id);
+  try {
+    await applyQscoreProjection(event);
+    const onboarding = await ensureOnboardingSideEffectsForEvent(event);
+    if (
+      onboarding.curatorOnboarding.status === "skipped" &&
+      onboarding.curatorOnboarding.reason === "loop_failed"
+    ) {
+      throw new Error("curator_onboarding_loop_failed");
+    }
+    await markGrowEventProcessed(event.id);
+  } catch (err) {
+    await markGrowEventFailed(event.id, err);
+    throw err;
+  }
   return event;
 }
 

src/home/home-feed.ts

@@ -17,7 +17,7 @@ import {
 } from "../db/schema.js";
 import { interviewService, resumeService, roleplayService } from "../services/product-service-clients.js";
 import { buildServiceLink } from "../services/service-registry.js";
-import { ensureOnboardingBaselineQscore } from "../events/onboarding-qscore.js";
+import { ensureOnboardingBaselineQscoreFromLedger } from "../events/onboarding-ledger.js";
 import { HomeFeedAgentError, refineHomeNotificationsWithAgent } from "./home-feed-agent.js";
 import { listAvailableMissionDefinitions } from "../missions/registry.js";
 import { listServiceCapabilities } from "../workflows/service-capabilities.js";
@@ -615,7 +615,7 @@ async function buildIdentity(ctx: HomeContext) {
 }
 
 export async function getHomeFeed(userId: string, opts: { refresh?: boolean; userProfile?: Record<string, unknown>; preferences?: Record<string, unknown> } = {}): Promise<HomeFeedResponse> {
-  await ensureOnboardingBaselineQscore(userId, opts.preferences);
+  await ensureOnboardingBaselineQscoreFromLedger(userId);
   const ctx = await collectContext(userId, { userProfile: opts.userProfile, preferences: opts.preferences });
   const persisted = await readPersistedNotifications(userId);
   const newest = persisted[0]?.createdAt?.getTime() ?? 0;

src/routes/events.ts

@@ -4,9 +4,15 @@ import { config } from "../config.js";
 import { db } from "../db/client.js";
 import { growEvents } from "../db/schema.js";
 import { requireUser, type AuthContext } from "../auth/clerk.js";
-import { recordGrowEvent } from "../events/record-grow-event.js";
+import {
+  markGrowEventFailed,
+  markGrowEventProcessed,
+  markGrowEventProcessing,
+  recordGrowEvent,
+} from "../events/record-grow-event.js";
 import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js";
-import { runCuratorOnboardingLoopForEventSafely } from "../v1/curator/curator-onboarding-loop.js";
+import { applyQscoreProjection } from "../events/projectors/qscore-projector.js";
+import { ensureOnboardingSideEffectsForEvent } from "../events/onboarding-ledger.js";
 
 function serviceAuthorized(auth: string | undefined) {
   const token = (auth ?? "").replace(/^Bearer\s+/i, "").trim();
@@ -21,8 +27,41 @@ async function ingest(body: unknown, userId?: string, source?: string) {
     routed: false as const,
     reason: err instanceof Error ? err.message : String(err),
   }));
-  const curatorOnboarding = await runCuratorOnboardingLoopForEventSafely(event);
-  return { event, route, curatorOnboarding };
+  if (event.processingStatus === "processed") {
+    return {
+      event,
+      processingStatus: "processed" as const,
+      route,
+      qscore: { signals: [], score: undefined, idempotent: true },
+      onboarding: { qscoreBaselineSeeded: false, curatorOnboarding: { status: "already_processed" } },
+    };
+  }
+  if (!event.userId) {
+    return {
+      event,
+      processingStatus: event.processingStatus,
+      route,
+      qscore: { signals: [], score: undefined, skipped: "missing_user_id" },
+      onboarding: { qscoreBaselineSeeded: false, curatorOnboarding: { status: "skipped", reason: "missing_user_id" } },
+    };
+  }
+
+  await markGrowEventProcessing(event.id);
+  try {
+    const qscore = await applyQscoreProjection(event);
+    const onboarding = await ensureOnboardingSideEffectsForEvent(event);
+    if (
+      onboarding.curatorOnboarding.status === "skipped" &&
+      onboarding.curatorOnboarding.reason === "loop_failed"
+    ) {
+      throw new Error("curator_onboarding_loop_failed");
+    }
+    await markGrowEventProcessed(event.id);
+    return { event, processingStatus: "processed" as const, route, qscore, onboarding };
+  } catch (err) {
+    await markGrowEventFailed(event.id, err);
+    throw err;
+  }
 }
 
 export function eventRoutes() {
@@ -32,8 +71,8 @@ export function eventRoutes() {
   app.post("/ingest", requireUser, async (c) => {
     const userId = c.get("userId");
     const body = await c.req.json().catch(() => ({}));
-    const { event, route, curatorOnboarding } = await ingest(body, userId);
-    return c.json({ eventId: event.id, processingStatus: event.processingStatus, route, curatorOnboarding }, 202);
+    const { event, processingStatus, route, qscore, onboarding } = await ingest(body, userId);
+    return c.json({ eventId: event.id, processingStatus, route, qscore, onboarding }, 202);
   });
 
   // Service-to-service ingress. Services may include userId directly, or we resolve it from session correlation.
@@ -43,8 +82,8 @@ export function eventRoutes() {
     }
     const body = await c.req.json().catch(() => ({}));
     const source = c.req.header("x-growqr-source") ?? undefined;
-    const { event, route, curatorOnboarding } = await ingest(body, undefined, source);
-    return c.json({ eventId: event.id, processingStatus: event.processingStatus, route, curatorOnboarding }, 202);
+    const { event, processingStatus, route, qscore, onboarding } = await ingest(body, undefined, source);
+    return c.json({ eventId: event.id, processingStatus, route, qscore, onboarding }, 202);
   });
 
   app.get("/", requireUser, async (c) => {

src/routes/services.ts

@@ -10,7 +10,7 @@ import { db } from "../db/client.js";
 import { events, growQscoreLatest, growQscoreProjectionState } from "../db/schema.js";
 import { recordGrowEvent } from "../events/record-grow-event.js";
 import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js";
-import { ensureOnboardingBaselineQscore } from "../events/onboarding-qscore.js";
+import { ensureOnboardingBaselineQscoreFromLedger } from "../events/onboarding-ledger.js";
 import { getRequestUserPreferences, getRequestUserProfile } from "../services/user-context.js";
 import { log } from "../log.js";
 
@@ -655,7 +655,7 @@ export function serviceRoutes() {
   app.get("/qscore/current", async (c) => {
     const userId = c.get("userId");
     try {
-      await ensureOnboardingBaselineQscore(userId, await getRequestUserPreferences(c.req.raw, userId));
+      await ensureOnboardingBaselineQscoreFromLedger(userId);
     } catch (err) {
       log.warn({ err, userId }, "failed to seed onboarding Q Score baseline before current Q Score read");
     }

src/routes/users.ts

@@ -1,22 +1,17 @@
 import { Hono } from "hono";
 import { requireUser, type AuthContext } from "../auth/clerk.js";
 import { db } from "../db/client.js";
-import { growEvents, users, userStacks, type UserStack } from "../db/schema.js";
-import { and, desc, eq, inArray } from "drizzle-orm";
+import { users, userStacks, type UserStack } from "../db/schema.js";
+import { eq } from "drizzle-orm";
 import { log } from "../log.js";
 import { config } from "../config.js";
-import { ensureOnboardingBaselineQscore } from "../events/onboarding-qscore.js";
 import {
   onboardingCompletedAtFromPreferences,
-  runCuratorOnboardingLoopSafely,
 } from "../v1/curator/curator-onboarding-loop.js";
-
-const ONBOARDING_LEDGER_EVENT_TYPES = [
-  "onboarding.snapshot.saved",
-  "onboarding.completed",
-  "user.onboarding.completed",
-  "profile.onboarding.completed",
-] as const;
+import {
+  getLatestValidOnboardingLedgerEvent,
+  recordAndProcessOnboardingCompletion,
+} from "../events/onboarding-ledger.js";
 
 function publicStack(stack: UserStack | null | undefined) {
   if (!stack) return stack;
@@ -62,27 +57,6 @@ async function ensureUserServiceUser(req: Request) {
   return res.json() as Promise<Record<string, unknown>>;
 }
 
-async function getOnboardingEventStatus(userId: string) {
-  const [row] = await db
-    .select({
-      id: growEvents.id,
-      type: growEvents.type,
-      occurredAt: growEvents.occurredAt,
-      processingStatus: growEvents.processingStatus,
-    })
-    .from(growEvents)
-    .where(
-      and(
-        eq(growEvents.userId, userId),
-        inArray(growEvents.type, [...ONBOARDING_LEDGER_EVENT_TYPES]),
-      ),
-    )
-    .orderBy(desc(growEvents.occurredAt))
-    .limit(1);
-
-  return row ?? null;
-}
-
 export function userRoutes() {
   const app = new Hono<AuthContext>();
   app.use("*", requireUser);
@@ -120,7 +94,7 @@ export function userRoutes() {
 
   app.get("/onboarding-status", async (c) => {
     const userId = c.get("userId");
-    const event = await getOnboardingEventStatus(userId);
+    const event = await getLatestValidOnboardingLedgerEvent(userId);
 
     return c.json({
       userId,
@@ -149,13 +123,9 @@ export function userRoutes() {
         const normalizedPreferences = preferences && typeof preferences === "object" && !Array.isArray(preferences)
           ? (preferences as Record<string, unknown>)
           : undefined;
-        await ensureOnboardingBaselineQscore(
-          c.get("userId"),
-          normalizedPreferences,
-        );
         const completedAt = onboardingCompletedAtFromPreferences(normalizedPreferences);
         if (completedAt) {
-          await runCuratorOnboardingLoopSafely({
+          await recordAndProcessOnboardingCompletion({
             userId: c.get("userId"),
             completedAt,
             source: "user-service-profile",

src/v1/curator/curator-onboarding-loop.ts

@@ -12,6 +12,16 @@ import type { CuratorSprintResponse } from "./curator-types.js";
 const CURATOR_SOURCE = "curator-v1";
 const ONBOARDING_READY_EVENT = "curator.onboarding_plan.ready";
 const ONBOARDING_SKIPPED_EVENT = "curator.onboarding_plan.skipped";
+const ONBOARDING_CONTEXT_EVENT_TYPES = [
+  "onboarding.snapshot.saved",
+  "onboarding.completed",
+  "user.onboarding.completed",
+  "profile.onboarding.completed",
+  "onboarding_snapshot_saved",
+  "onboarding_completed",
+  "user_onboarding_completed",
+  "profile_onboarding_completed",
+] as const;
 
 type OnboardingLoopInput = {
   userId: string;
@@ -140,7 +150,7 @@ async function findLatestOnboardingContext(userId: string) {
     .from(growEvents)
     .where(and(
       eq(growEvents.userId, userId),
-      inArray(growEvents.type, ["onboarding.snapshot.saved", "onboarding.completed", "user.onboarding.completed", "profile.onboarding.completed"]),
+      inArray(growEvents.type, [...ONBOARDING_CONTEXT_EVENT_TYPES]),
     ))
     .orderBy(desc(growEvents.occurredAt))
     .limit(1);

src/v1/events/events-routes.ts

@@ -4,7 +4,7 @@ import { requireUser, type AuthContext } from "../../auth/clerk.js";
 import { applyQscoreProjection } from "../../events/projectors/qscore-projector.js";
 import { applyServiceSessionProjection } from "../../events/projectors/service-session-projector.js";
 import { markGrowEventFailed, markGrowEventProcessed, markGrowEventProcessing, recordGrowEvent } from "../../events/record-grow-event.js";
-import { runCuratorOnboardingLoopForEventSafely } from "../curator/curator-onboarding-loop.js";
+import { ensureOnboardingSideEffectsForEvent } from "../../events/onboarding-ledger.js";
 
 const eventTrackSchema = z.object({
   id: z.string().optional(),
@@ -90,14 +90,20 @@ export function v1EventRoutes() {
     try {
       const serviceSession = await applyServiceSessionProjection(event);
       const qscore = await applyQscoreProjection(event);
-      const curatorOnboarding = await runCuratorOnboardingLoopForEventSafely(event);
+      const onboarding = await ensureOnboardingSideEffectsForEvent(event);
+      if (
+        onboarding.curatorOnboarding.status === "skipped" &&
+        onboarding.curatorOnboarding.reason === "loop_failed"
+      ) {
+        throw new Error("curator_onboarding_loop_failed");
+      }
       await markGrowEventProcessed(event.id);
       return c.json({
         eventId: event.id,
         processingStatus: "processed",
         serviceSession,
         qscore,
-        curatorOnboarding,
+        onboarding,
       }, 202);
     } catch (err) {
       await markGrowEventFailed(event.id, err);