diff --git a/package.json b/package.json index 455f320..2d732ac 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "dev": "tsx watch src/index.ts", "build": "tsc -p tsconfig.json", "test:onboarding": "tsx scripts/onboarding-ledger.test.ts", + "test:missions": "tsx scripts/mission-lifecycle.test.ts", "start": "node dist/index.js", "typecheck": "tsc -p tsconfig.json --noEmit", "workflows:smoke": "tsx src/workflows/smoke-test.ts", diff --git a/scripts/mission-lifecycle.test.ts b/scripts/mission-lifecycle.test.ts new file mode 100644 index 0000000..cdd677c --- /dev/null +++ b/scripts/mission-lifecycle.test.ts @@ -0,0 +1,46 @@ +import assert from "node:assert/strict"; +import { + onboardingMissionInstanceId, + selectOnboardingMissionIds, +} from "../src/missions/lifecycle.js"; + +const userA = "user_abc123"; + +assert.deepEqual( + selectOnboardingMissionIds({ onboarding: { goal: "I need internship interview prep" } }), + ["interview-to-offer", "personal-brand-opportunity-engine"], + "default onboarding should start interview-to-offer plus personal brand", +); + +assert.deepEqual( + selectOnboardingMissionIds({ onboarding: { goal: "I want to negotiate my offer and compensation" } }), + ["salary-negotiation-war-room", "personal-brand-opportunity-engine"], + "salary/offer context should prioritize the negotiation mission", +); + +assert.deepEqual( + selectOnboardingMissionIds({ preferences: { onboarding: { goal: "I need a career transition into product" } } }), + ["career-transition", "personal-brand-opportunity-engine"], + "preferences onboarding context should be read when selecting missions", +); + +assert.deepEqual( + selectOnboardingMissionIds({ preferences: { onboarding: { goal: "Build LinkedIn visibility and network" } } }), + ["personal-brand-opportunity-engine", "interview-to-offer"], + "brand/networking context should not duplicate the personal-brand mission", +); + +assert.equal( + onboardingMissionInstanceId(userA, "interview-to-offer"), + onboardingMissionInstanceId(userA, "interview-to-offer"), + "onboarding mission instance ids must be deterministic for idempotent retries", +); + +assert.notEqual( + onboardingMissionInstanceId(userA, "interview-to-offer"), + onboardingMissionInstanceId("user_other", "interview-to-offer"), + "onboarding mission instance ids must be scoped by user", +); + +console.log("mission-lifecycle tests passed"); +process.exit(0); diff --git a/src/config.ts b/src/config.ts index 573d20e..ffa7237 100644 --- a/src/config.ts +++ b/src/config.ts @@ -144,6 +144,12 @@ export const config = { .map((s) => s.trim()) .filter(Boolean), + // Passive mission refresh loop. Dedupe keys make this safe across retries and + // multiple staging replicas; set MISSION_PASSIVE_LOOP_ENABLED=false to disable. + missionPassiveLoopEnabled: (process.env.MISSION_PASSIVE_LOOP_ENABLED ?? "true").toLowerCase() !== "false", + missionPassiveLoopIntervalMs: Number(process.env.MISSION_PASSIVE_LOOP_INTERVAL_MS ?? 60 * 60 * 1000), + missionPassiveLoopBatchSize: Number(process.env.MISSION_PASSIVE_LOOP_BATCH_SIZE ?? 100), + // Used by LLM requests. maxAgentTokens: Number(process.env.MAX_AGENT_TOKENS ?? 4096), diff --git a/src/events/onboarding-ledger.ts b/src/events/onboarding-ledger.ts index 09b03e1..9314745 100644 --- a/src/events/onboarding-ledger.ts +++ b/src/events/onboarding-ledger.ts @@ -13,6 +13,7 @@ import { onboardingCompletedAtFromEvent, runCuratorOnboardingLoopSafely, } from "../v1/curator/curator-onboarding-loop.js"; +import { ensureOnboardingActiveMissions } from "../missions/lifecycle.js"; export const ONBOARDING_LEDGER_EVENT_TYPES = [ "onboarding.snapshot.saved", @@ -130,6 +131,7 @@ export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow) { return { qscoreBaselineSeeded: false, curatorOnboarding: { status: "skipped" as const, reason: event.userId ? "not_onboarding_completion" : "missing_user_id" }, + missions: { status: "skipped" as const, reason: event.userId ? "not_onboarding_completion" : "missing_user_id", started: [], existing: [] }, }; } @@ -146,8 +148,15 @@ export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow) { source: event.source, context: onboardingContextFromInput(event.payload), }); + const missions = await ensureOnboardingActiveMissions({ + userId: event.userId, + completedAt, + sourceEventId: event.id, + source: event.source, + context: onboardingContextFromInput(event.payload), + }); - return { qscoreBaselineSeeded, curatorOnboarding }; + return { qscoreBaselineSeeded, curatorOnboarding, missions }; } export async function recordAndProcessOnboardingCompletion(input: { diff --git a/src/grow/persistence.ts b/src/grow/persistence.ts index d9f344f..fee80e3 100644 --- a/src/grow/persistence.ts +++ b/src/grow/persistence.ts @@ -274,6 +274,20 @@ export async function listActiveMissionsPg(userId: string) { return rows.map((row) => ({ mission: activeMissionFromRow(row), snapshot: missionSnapshotFromRow(row) })); } +export async function listActiveMissionsForPassiveReviewPg(opts: { userId?: string; limit?: number } = {}) { + const conditions = [eq(growActiveMissions.status, "active")]; + if (opts.userId) conditions.push(eq(growActiveMissions.userId, opts.userId)); + const query = db + .select() + .from(growActiveMissions) + .where(and(...conditions)) + .orderBy(asc(growActiveMissions.updatedAt)); + const rows = typeof opts.limit === "number" && opts.limit > 0 + ? await query.limit(opts.limit) + : await query; + return rows.map((row) => ({ userId: row.userId, mission: activeMissionFromRow(row), snapshot: missionSnapshotFromRow(row) })); +} + export async function getActiveMissionPg(userId: string, instanceId: string) { const [row] = await db.select().from(growActiveMissions).where(and(eq(growActiveMissions.userId, userId), eq(growActiveMissions.instanceId, instanceId))).limit(1); return row ? { mission: activeMissionFromRow(row), snapshot: missionSnapshotFromRow(row) } : null; diff --git a/src/index.ts b/src/index.ts index 61057a3..38b9ff4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -23,6 +23,7 @@ import { analyticsRoutes } from "./routes/analytics.js"; import { logRoutes } from "./routes/logs.js"; import { v1Routes } from "./v1/index.js"; import { startGrowEventsRedisConsumer } from "./events/redis-consumer.js"; +import { startPassiveMissionReviewLoop } from "./missions/passive-runner.js"; import { db } from "./db/client.js"; import { ensureRuntimeSchema } from "./db/ensure-runtime-schema.js"; import { hydratePortAllocator, reconcileOnBoot, ensureCentralGiteaReady } from "./docker/manager.js"; @@ -47,6 +48,7 @@ async function main() { await reconcileOnBoot(); startGrowEventsRedisConsumer().catch((err) => log.error({ err }, "failed to start grow events redis consumer")); + startPassiveMissionReviewLoop(); const app = new Hono(); diff --git a/src/missions/lifecycle.ts b/src/missions/lifecycle.ts new file mode 100644 index 0000000..9ac6613 --- /dev/null +++ b/src/missions/lifecycle.ts @@ -0,0 +1,354 @@ +import crypto from "node:crypto"; +import { createClient, type Client } from "rivetkit/client"; +import { eq } from "drizzle-orm"; +import { config } from "../config.js"; +import { db } from "../db/client.js"; +import { growEvents, users } from "../db/schema.js"; +import { + completeMissionCoachRunPg, + createMissionCoachRunPg, + getActiveMissionPg, + listActiveMissionsForPassiveReviewPg, + listActiveMissionsPg, + replaceMissionSuggestionsPg, + upsertActiveMissionPg, +} from "../grow/persistence.js"; +import { recordGrowEvent, markGrowEventFailed, markGrowEventProcessed, markGrowEventProcessing } from "../events/record-grow-event.js"; +import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js"; +import { getPersistedMissionDefinition } from "./postgres-registry.js"; +import { buildDeterministicMissionSuggestions } from "./suggestions.js"; +import type { Registry } from "../actors/registry.js"; +import type { GrowActiveMission, MissionActorType, MissionId, MissionSnapshot } from "../actors/missions/types.js"; +import { log } from "../log.js"; + +const ONBOARDING_MISSION_LIMIT = 2; +const PASSIVE_REVIEW_SOURCE = "growqr-backend:mission-passive"; + +let _client: Client | null = null; +function getClient(): Client { + return (_client ??= createClient(config.rivetClientEndpoint)); +} + +function missionActorFor(userId: string, instanceId: string, actorType: MissionActorType) { + const client = getClient(); + switch (actorType) { + case "interviewToOfferMissionActor": + return client.interviewToOfferMissionActor.getOrCreate([userId, instanceId]); + case "careerTransitionMissionActor": + return client.careerTransitionMissionActor.getOrCreate([userId, instanceId]); + case "salaryNegotiationWarRoomMissionActor": + return client.salaryNegotiationWarRoomMissionActor.getOrCreate([userId, instanceId]); + case "promotionReadinessMissionActor": + return client.promotionReadinessMissionActor.getOrCreate([userId, instanceId]); + case "personalBrandOpportunityEngineMissionActor": + return client.personalBrandOpportunityEngineMissionActor.getOrCreate([userId, instanceId]); + } +} + +function activeMissionFromSnapshot(snapshot: MissionSnapshot): GrowActiveMission { + return { + instanceId: snapshot.instanceId, + missionId: snapshot.missionId, + workflowId: snapshot.workflowId, + title: snapshot.title, + shortTitle: snapshot.shortTitle, + status: snapshot.status, + progressPercent: snapshot.progressPercent, + currentStageId: snapshot.currentStageId, + goal: snapshot.goal, + actorType: actorTypeFor(snapshot.missionId), + createdAt: new Date(snapshot.createdAt).getTime(), + updatedAt: new Date(snapshot.updatedAt).getTime(), + }; +} + +function actorTypeFor(missionId: MissionId): MissionActorType | undefined { + if (missionId === "interview-to-offer") return "interviewToOfferMissionActor"; + if (missionId === "career-transition") return "careerTransitionMissionActor"; + if (missionId === "salary-negotiation-war-room") return "salaryNegotiationWarRoomMissionActor"; + if (missionId === "promotion-readiness") return "promotionReadinessMissionActor"; + if (missionId === "personal-brand-opportunity-engine") return "personalBrandOpportunityEngineMissionActor"; + return undefined; +} + +function hashUser(userId: string) { + return crypto.createHash("sha256").update(userId).digest("hex").slice(0, 12); +} + +export function onboardingMissionInstanceId(userId: string, missionId: MissionId) { + return `mission-${missionId}-${hashUser(userId)}`; +} + +function asRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function stringValues(value: unknown): string[] { + if (Array.isArray(value)) return value.filter((item): item is string => typeof item === "string" && Boolean(item.trim())).map((item) => item.trim()); + if (typeof value === "string" && value.trim()) return [value.trim()]; + return []; +} + +function onboardingText(context?: Record) { + const source = asRecord(context); + const preferences = asRecord(source.preferences); + const onboarding = asRecord(source.onboarding ?? preferences.onboarding); + const mission = asRecord(preferences.mission_preferences); + const resume = asRecord(preferences.resume_preferences); + const interview = asRecord(preferences.interview_preferences); + const values = [ + ...stringValues(onboarding.goal), + ...stringValues(onboarding.target_role ?? onboarding.targetRole ?? onboarding.role ?? onboarding.current_role), + ...stringValues(onboarding.timeline), + ...stringValues(mission.active_goal), + ...stringValues(resume.target_title), + ...stringValues(interview.job_description), + ...stringValues(preferences.target_roles), + ...stringValues(preferences.target_companies), + ]; + return values.join(" ").toLowerCase(); +} + +export function selectOnboardingMissionIds(context?: Record, limit = ONBOARDING_MISSION_LIMIT): MissionId[] { + const text = onboardingText(context); + const primary: MissionId = + /salary|compensation|negotiat/.test(text) ? "salary-negotiation-war-room" : + /promot|manager|leadership|level up|level-up/.test(text) ? "promotion-readiness" : + /transition|switch|pivot|career change|new field/.test(text) ? "career-transition" : + /brand|linkedin|network|visibility|opportunit/.test(text) ? "personal-brand-opportunity-engine" : + "interview-to-offer"; + + const ordered: MissionId[] = [ + primary, + "personal-brand-opportunity-engine", + "interview-to-offer", + "career-transition", + "promotion-readiness", + "salary-negotiation-war-room", + ]; + return Array.from(new Set(ordered)).slice(0, Math.max(1, limit)); +} + +async function ensureUser(userId: string) { + await db + .insert(users) + .values({ id: userId, email: `${userId}@service.local`, displayName: userId }) + .onConflictDoNothing(); +} + +export async function ensureOnboardingActiveMissions(input: { + userId: string; + context?: Record; + completedAt?: string | Date | null; + sourceEventId?: string; + source?: string; + limit?: number; +}) { + const userId = input.userId.trim(); + if (!userId) return { status: "skipped" as const, reason: "missing_user_id" as const, started: [], existing: [] }; + + await ensureUser(userId); + const missionIds = selectOnboardingMissionIds(input.context, input.limit ?? ONBOARDING_MISSION_LIMIT); + const activeRows = await listActiveMissionsPg(userId); + const started: GrowActiveMission[] = []; + const existing: GrowActiveMission[] = []; + + for (const missionId of missionIds) { + const alreadyActive = activeRows.find((item) => item.mission.missionId === missionId && ["active", "paused"].includes(item.mission.status)); + if (alreadyActive) { + existing.push(alreadyActive.mission); + continue; + } + + const mission = await getPersistedMissionDefinition(missionId); + if (!mission?.actorType) continue; + + const instanceId = onboardingMissionInstanceId(userId, missionId); + const existingInstance = await getActiveMissionPg(userId, instanceId); + if (existingInstance) { + existing.push(existingInstance.mission); + continue; + } + + const actor = missionActorFor(userId, instanceId, mission.actorType); + const completedAt = input.completedAt instanceof Date ? input.completedAt.toISOString() : input.completedAt ?? new Date().toISOString(); + const snapshot = await actor.init({ + userId, + instanceId, + missionId, + goal: onboardingText(input.context) || undefined, + input: { + source: "onboarding", + sourceEventId: input.sourceEventId, + completedAt, + context: input.context ?? {}, + }, + }); + const activeMission = activeMissionFromSnapshot(snapshot); + await upsertActiveMissionPg(userId, activeMission, snapshot); + started.push(activeMission); + + const event = await recordGrowEvent({ + source: input.source ?? "onboarding", + type: "mission.started", + category: "mission", + userId, + occurredAt: completedAt, + mission: { instanceId, missionId, stageId: snapshot.currentStageId }, + correlation: { sourceEventId: input.sourceEventId }, + payload: { + trigger: "onboarding", + title: snapshot.title, + goal: snapshot.goal, + selectedMissionIds: missionIds, + }, + dedupeKey: `mission-onboarding-start:${userId}:${missionId}`, + }, { userId, source: input.source ?? "onboarding" }); + await routeGrowEventToUserActor(event).catch((err) => log.warn({ err, userId, missionId }, "failed to route onboarding mission start event")); + } + + return { + status: started.length ? "started" as const : "already_ready" as const, + selectedMissionIds: missionIds, + started, + existing, + }; +} + +function passiveReviewDate(input?: string | Date) { + const date = input instanceof Date ? input : input ? new Date(input) : new Date(); + return Number.isNaN(date.getTime()) ? new Date().toISOString().slice(0, 10) : date.toISOString().slice(0, 10); +} + +async function passiveReviewAlreadyRan(instanceId: string, date: string) { + const [existing] = await db + .select({ id: growEvents.id, processingStatus: growEvents.processingStatus }) + .from(growEvents) + .where(eq(growEvents.dedupeKey, `mission-passive-review:${instanceId}:${date}`)) + .limit(1); + return existing ?? null; +} + +export async function runPassiveMissionReviewForMission(input: { + userId: string; + mission: GrowActiveMission; + snapshot?: MissionSnapshot | null; + date?: string | Date; + force?: boolean; +}) { + const date = passiveReviewDate(input.date); + const existing = input.force ? null : await passiveReviewAlreadyRan(input.mission.instanceId, date); + if (existing) { + return { status: "skipped" as const, reason: "already_ran" as const, eventId: existing.id, mission: input.mission }; + } + if (!input.mission.actorType) { + return { status: "skipped" as const, reason: "missing_actor" as const, mission: input.mission }; + } + + const dedupeKey = input.force + ? `mission-passive-review:${input.mission.instanceId}:${date}:${Date.now()}` + : `mission-passive-review:${input.mission.instanceId}:${date}`; + const event = await recordGrowEvent({ + source: PASSIVE_REVIEW_SOURCE, + type: "mission.passive_review.completed", + category: "mission", + userId: input.userId, + occurredAt: new Date().toISOString(), + mission: { instanceId: input.mission.instanceId, missionId: input.mission.missionId, stageId: input.mission.currentStageId }, + payload: { + reviewDate: date, + status: "started", + }, + dedupeKey, + }, { userId: input.userId, source: PASSIVE_REVIEW_SOURCE }); + await markGrowEventProcessing(event.id); + + try { + const actor = missionActorFor(input.userId, input.mission.instanceId, input.mission.actorType); + const scrum = await actor.runDailyScrum({ trigger: "nightly" }); + const snapshot = scrum.snapshot ?? input.snapshot; + if (!snapshot) { + await markGrowEventFailed(event.id, new Error("mission_passive_review_missing_snapshot")); + return { status: "skipped" as const, reason: "missing_snapshot" as const, eventId: event.id, mission: input.mission }; + } + + const activeMission = activeMissionFromSnapshot(snapshot); + await upsertActiveMissionPg(input.userId, activeMission, snapshot); + const windowEnd = new Date(`${date}T23:59:59.999Z`); + const windowStart = new Date(`${date}T00:00:00.000Z`); + const run = await createMissionCoachRunPg({ + userId: input.userId, + missionInstanceId: activeMission.instanceId, + missionId: activeMission.missionId, + windowStart, + windowEnd, + skillVersion: snapshot.skillVersion, + inputDigest: { + passive: true, + reviewDate: date, + trigger: "nightly", + stageCount: snapshot.stages.length, + currentStageId: snapshot.currentStageId, + progressPercent: snapshot.progressPercent, + artifactCount: snapshot.artifacts.length, + eventCount: snapshot.events.length, + }, + }); + + const snapshotContext = asRecord(snapshot.input?.context); + const suggestions = await replaceMissionSuggestionsPg({ + userId: input.userId, + missionInstanceId: activeMission.instanceId, + missionId: activeMission.missionId, + coachRunId: run.id, + suggestions: buildDeterministicMissionSuggestions(snapshot, { preferences: asRecord(snapshotContext.preferences) }), + }); + const summary = suggestions[0] + ? `Passive mission review refreshed ${suggestions.length} suggestion${suggestions.length === 1 ? "" : "s"}. Top action: ${suggestions[0].title}` + : "Passive mission review found no open action."; + await completeMissionCoachRunPg({ id: run.id, summary, output: { suggestions, passive: true, reviewDate: date } }); + + await db.update(growEvents).set({ + mission: { instanceId: activeMission.instanceId, missionId: activeMission.missionId, stageId: activeMission.currentStageId }, + payload: { + reviewDate: date, + status: "completed", + coachRunId: run.id, + suggestionIds: suggestions.map((item) => item.id), + summary, + }, + }).where(eq(growEvents.id, event.id)); + await markGrowEventProcessed(event.id); + + return { status: "reviewed" as const, eventId: event.id, coachRunId: run.id, mission: activeMission, suggestionCount: suggestions.length, summary }; + } catch (err) { + await markGrowEventFailed(event.id, err); + throw err; + } +} + +export async function runPassiveMissionReviews(input: { userId?: string; date?: string | Date; force?: boolean; limit?: number } = {}) { + const rows = await listActiveMissionsForPassiveReviewPg({ userId: input.userId, limit: input.limit }); + const results = []; + for (const row of rows) { + try { + results.push(await runPassiveMissionReviewForMission({ + userId: row.userId, + mission: row.mission, + snapshot: row.snapshot, + date: input.date, + force: input.force, + })); + } catch (err) { + log.error({ err, userId: row.userId, missionInstanceId: row.mission.instanceId }, "passive mission review failed"); + results.push({ status: "failed" as const, mission: row.mission, error: err instanceof Error ? err.message : String(err) }); + } + } + return { + date: passiveReviewDate(input.date), + reviewed: results.filter((item) => item.status === "reviewed").length, + skipped: results.filter((item) => item.status === "skipped").length, + failed: results.filter((item) => item.status === "failed").length, + results, + }; +} diff --git a/src/missions/passive-runner.ts b/src/missions/passive-runner.ts new file mode 100644 index 0000000..8539e21 --- /dev/null +++ b/src/missions/passive-runner.ts @@ -0,0 +1,42 @@ +import { config } from "../config.js"; +import { log } from "../log.js"; +import { runPassiveMissionReviews } from "./lifecycle.js"; + +let timer: NodeJS.Timeout | undefined; +let running = false; + +async function runOnce() { + if (running) return; + running = true; + try { + const result = await runPassiveMissionReviews({ limit: config.missionPassiveLoopBatchSize }); + if (result.reviewed || result.failed) { + log.info({ + reviewed: result.reviewed, + skipped: result.skipped, + failed: result.failed, + date: result.date, + }, "passive mission review loop completed"); + } + } catch (err) { + log.error({ err }, "passive mission review loop failed"); + } finally { + running = false; + } +} + +export function startPassiveMissionReviewLoop() { + if (!config.missionPassiveLoopEnabled) { + log.info("passive mission review loop disabled"); + return; + } + if (timer) return; + + const intervalMs = Math.max(5 * 60 * 1000, config.missionPassiveLoopIntervalMs); + const firstDelayMs = Math.min(60_000, intervalMs); + const first = setTimeout(() => void runOnce(), firstDelayMs); + first.unref?.(); + timer = setInterval(() => void runOnce(), intervalMs); + timer.unref?.(); + log.info({ intervalMs, batchSize: config.missionPassiveLoopBatchSize }, "passive mission review loop scheduled"); +} diff --git a/src/routes/missions.ts b/src/routes/missions.ts index 59c10a3..7700fe5 100644 --- a/src/routes/missions.ts +++ b/src/routes/missions.ts @@ -14,6 +14,7 @@ import { recordGrowEvent } from "../events/record-grow-event.js"; import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js"; import { getRequestUserPreferences } from "../services/user-context.js"; import { missionDetailHref } from "../missions/reducer-helpers.js"; +import { runPassiveMissionReviews } from "../missions/lifecycle.js"; let _client: Client | null = null; function getClient(): Client { @@ -69,6 +70,12 @@ const snoozeActionSchema = z.object({ until: z.string().datetime().optional(), }); +const passiveReviewSchema = z.object({ + date: z.string().optional(), + force: z.boolean().optional(), + limit: z.number().int().min(1).max(50).optional(), +}); + const createInstanceId = (missionId: string) => `${missionId}-${Date.now()}-${Math.random().toString(16).slice(2)}`; @@ -285,6 +292,17 @@ export function missionRoutes() { return c.json({ action }); }); + app.post("/passive/run", async (c) => { + const userId = c.get("userId"); + const body = passiveReviewSchema.parse(await c.req.json().catch(() => ({}))); + return c.json(await runPassiveMissionReviews({ + userId, + date: body.date, + force: body.force, + limit: body.limit, + })); + }); + app.post("/:missionId/start", async (c) => { const userId = c.get("userId"); const missionId = c.req.param("missionId");