From 7e29595a718fd5d839a2802eea005850c9f9f1e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 10:18:52 +0200 Subject: [PATCH] feat(coach): wire the combined memory-refresh worker queue Add the coach-memory-refresh pg-boss queue: a fire-and-forget enqueue contract (coach-memory-shared) and a worker pipeline that runs the conversation-summary refresh then the fact extraction for one long conversation, each fault-isolated and budget-gated. Register the queue in allQueues + a boss.work handler in the worker, with a source-grep inventory guard so the queue can never ship unregistered. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ai/coach/coach-memory-refresh-worker.ts | 55 ++++++++++++++++ src/lib/ai/coach/coach-memory-shared.ts | 64 +++++++++++++++++++ .../coach-memory-refresh-queue.test.ts | 46 +++++++++++++ src/lib/jobs/reminder-worker.ts | 30 +++++++++ 4 files changed, 195 insertions(+) create mode 100644 src/lib/ai/coach/coach-memory-refresh-worker.ts create mode 100644 src/lib/ai/coach/coach-memory-shared.ts create mode 100644 src/lib/jobs/__tests__/coach-memory-refresh-queue.test.ts diff --git a/src/lib/ai/coach/coach-memory-refresh-worker.ts b/src/lib/ai/coach/coach-memory-refresh-worker.ts new file mode 100644 index 000000000..92176444b --- /dev/null +++ b/src/lib/ai/coach/coach-memory-refresh-worker.ts @@ -0,0 +1,55 @@ +/** + * v1.11.1 — worker pipeline for the combined Coach memory-refresh queue. + * + * Runs both background generators for one conversation, sequentially so they + * share the wake-up but each pays its own budget check inside + * `runStatusCompletion`: the rolling conversation summary first, then durable + * fact extraction. Each step is fault-isolated — a failure or no-provider in + * one never sinks the other or the job. Kept out of the route bundle (the + * route imports only `enqueueCoachMemoryRefresh` from `coach-memory-shared`). + */ +import { annotate } from "@/lib/logging/context"; + +import type { CoachMemoryRefreshPayload } from "./coach-memory-shared"; +import { extractAndStoreFacts } from "./facts"; +import { refreshConversationSummary } from "./conversation-summary"; + +export async function runCoachMemoryRefresh( + payload: CoachMemoryRefreshPayload, +): Promise { + const { conversationId, userId } = payload; + const locale = payload.locale ?? "de"; + + let summaryStatus = "error"; + try { + const result = await refreshConversationSummary(conversationId, userId, { + locale, + }); + summaryStatus = result.status; + } catch (err) { + annotate({ + action: { name: "coach.memory.refresh.summary_failed" }, + meta: { error: err instanceof Error ? err.message : String(err) }, + }); + } + + let factsStatus = "error"; + let factsCount = 0; + try { + const result = await extractAndStoreFacts(conversationId, userId, { + locale, + }); + factsStatus = result.status; + factsCount = result.count; + } catch (err) { + annotate({ + action: { name: "coach.memory.refresh.facts_failed" }, + meta: { error: err instanceof Error ? err.message : String(err) }, + }); + } + + annotate({ + action: { name: "coach.memory.refresh.done" }, + meta: { summaryStatus, factsStatus, factsCount }, + }); +} diff --git a/src/lib/ai/coach/coach-memory-shared.ts b/src/lib/ai/coach/coach-memory-shared.ts new file mode 100644 index 000000000..31fc00b2e --- /dev/null +++ b/src/lib/ai/coach/coach-memory-shared.ts @@ -0,0 +1,64 @@ +/** + * v1.11.1 — generator-free contract for the Coach long-term-memory refresh + * queue. The chat route enqueues a single-conversation refresh here without + * importing the concrete generators (which would pull the provider chain into + * the route bundle). The worker handler (`runCoachMemoryRefresh`) calls the + * summary + fact generators; it re-uses the queue name from here so there is + * one source of truth. + * + * One combined queue does BOTH conversation-summary refresh and durable-fact + * extraction in a single worker job: when a long conversation crosses the + * history-window cap, both are usually due together, so one job + one + * singleton window avoids a second queue and a redundant wake-up. + * + * Mirrors `period-narrative-shared.ts`: queue name + payload type + enqueue + * helper here, the concrete dispatch in the worker. + */ +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { annotate } from "@/lib/logging/context"; + +export const COACH_MEMORY_REFRESH_QUEUE = "coach-memory-refresh"; + +export interface CoachMemoryRefreshPayload { + conversationId: string; + userId: string; + /** Locale to compose the summary / facts prose in; defaults to "de". */ + locale?: "de" | "en"; +} + +/** + * Fire-and-forget enqueue from the chat turn once a conversation has grown past + * the history-window cap. A `singletonKey` per conversation collapses repeated + * turns within a short window into one queued job. No-ops cleanly when the + * global boss is unavailable (a web process without an embedded worker) — the + * memory simply stays as-is until the next eligible turn. + */ +export async function enqueueCoachMemoryRefresh(payload: { + conversationId: string; + userId: string; + locale: "de" | "en"; +}): Promise { + const boss = getGlobalBoss(); + if (!boss) return; + try { + await boss.send( + COACH_MEMORY_REFRESH_QUEUE, + { + conversationId: payload.conversationId, + userId: payload.userId, + locale: payload.locale, + } satisfies CoachMemoryRefreshPayload, + { + singletonKey: `refresh:${payload.conversationId}`, + singletonSeconds: 120, + }, + ); + annotate({ + action: { name: "coach.memory.refresh.enqueued" }, + meta: { locale: payload.locale }, + }); + } catch { + // Best-effort — a failure just leaves the Coach memory unchanged until the + // next eligible turn re-enqueues. + } +} diff --git a/src/lib/jobs/__tests__/coach-memory-refresh-queue.test.ts b/src/lib/jobs/__tests__/coach-memory-refresh-queue.test.ts new file mode 100644 index 000000000..c52b8eada --- /dev/null +++ b/src/lib/jobs/__tests__/coach-memory-refresh-queue.test.ts @@ -0,0 +1,46 @@ +/** + * v1.11.1 — queue-registration guard for the combined Coach memory-refresh + * job (rolling conversation summary + durable fact extraction). + * + * Same source-text-grep approach as the period-narrative / stress-strain + * guards: assert the queue is imported, registered in `allQueues`, and wired + * to a `boss.work` handler that runs the refresh pipeline — without booting + * pg-boss + Prisma. An unregistered queue silently never runs (the recurring + * past bug this guards against; v1.4.37 W10 caught exactly this class). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts"); +const source = readFileSync(REMINDER_WORKER_PATH, "utf8"); + +function allQueuesBlock(): string { + const m = source.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/); + expect(m).not.toBeNull(); + return m![1]; +} + +describe("reminder-worker — coach-memory-refresh wiring", () => { + it("imports the queue symbol + worker from the coach-memory modules", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/ai\/coach\/coach-memory-shared["']/, + ); + expect(source).toMatch(/\bCOACH_MEMORY_REFRESH_QUEUE\b/); + expect(source).toMatch( + /from\s*["']@\/lib\/ai\/coach\/coach-memory-refresh-worker["']/, + ); + expect(source).toMatch(/\brunCoachMemoryRefresh\b/); + }); + + it("registers the coach-memory-refresh queue in allQueues", () => { + expect(allQueuesBlock()).toMatch(/\bCOACH_MEMORY_REFRESH_QUEUE\b/); + }); + + it("registers a boss.work handler that runs the refresh pipeline", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}COACH_MEMORY_REFRESH_QUEUE[\s\S]{0,400}runCoachMemoryRefresh/, + ); + }); +}); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 7fdf857fb..5ac070485 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -102,6 +102,11 @@ import { warmOneNarrative, type PeriodNarrativePayload, } from "@/lib/jobs/period-narrative-warm"; +import { + COACH_MEMORY_REFRESH_QUEUE, + type CoachMemoryRefreshPayload, +} from "@/lib/ai/coach/coach-memory-shared"; +import { runCoachMemoryRefresh } from "@/lib/ai/coach/coach-memory-refresh-worker"; import { DENSE_INTRADAY_RETENTION_QUEUE, DENSE_INTRADAY_RETENTION_CONCURRENCY, @@ -2329,6 +2334,11 @@ export async function startReminderWorker() { // the read-only narrative GET. The queue MUST be registered here or the // GET-miss enqueue silently never warms. PERIOD_NARRATIVE_QUEUE, + // v1.11.1 — combined Coach memory-refresh (rolling conversation summary + + // durable fact extraction), enqueued fire-and-forget from a long chat + // turn. The queue MUST be registered here or the enqueue silently never + // runs. + COACH_MEMORY_REFRESH_QUEUE, ]; for (const q of allQueues) { @@ -2737,6 +2747,26 @@ export async function startReminderWorker() { } }, ); + // v1.11.1 — combined Coach memory refresh: rolling conversation summary + + // durable fact extraction for one long conversation. localConcurrency 1 so a + // burst of long-conversation turns can't fan out concurrent provider calls; + // each step is budget-gated inside runStatusCompletion and fault-isolated. + await boss.work( + COACH_MEMORY_REFRESH_QUEUE, + { localConcurrency: 1 }, + async (jobs) => { + for (const job of jobs) { + if (!job.data?.conversationId || !job.data?.userId) continue; + try { + await runCoachMemoryRefresh(job.data); + } catch (err) { + recordError(); + workerLog("error", "[coach-memory-refresh] failed", err); + throw err; + } + } + }, + ); // v1.8.3 — on-demand per-metric status generation enqueued by the // read-only status route on a cold card. Low concurrency so a first // visit that cold-misses several cards can't saturate the Prisma pool