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) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.8 committed 2026-06-04 10:18:52 +02:00
1 parent 5591858f16
commit 7e29595a71
4 files changed
+195

No files matched your search

@@ -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<void> {
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 },
});
}
+64
View File
@@ -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<void> {
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.
}
}
@@ -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/,
);
});
});
+30
View File
@@ -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<CoachMemoryRefreshPayload>(
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