feat(memory): implement stability stage

Co-authored-by: multica-agent <github@multica.ai>

apps/worker/src/index.ts

@@ -3,8 +3,8 @@ import { Worker } from 'bullmq'
 import IORedis from 'ioredis'
 import { applyMechanicalEvents, bindMechanicalEventsToRolls, resolveChecks } from '@dng/game-engine'
 import { CharacterSchema, PlayerIntentSchema, makeId } from '@dng/shared'
-import { generateWorld, narrateRound, planRound, summarizeStory } from './ai'
-import { buildActionSequence, buildRoundContext, sanitizeRoundMemory, shouldUpdateStorySummary, validateAndOrderRoundPlan } from './orchestration'
+import { generateWorld, narrateRound, planRound, summarizeStory, type AiUsageMeasurement } from './ai'
+import { buildActionSequence, buildRetrievalText, buildRoundContext, sanitizeRoundMemory, shouldUpdateStorySummary, validateAndOrderRoundPlan } from './orchestration'
 
 const redisUrl = process.env.REDIS_URL
 const supabaseUrl = process.env.SUPABASE_URL
@@ -28,6 +28,12 @@ if (!supabaseUrl || !serviceKey) {
     job_type: JobType
     entity_id: string
   }
+  interface RetrievedContext {
+    memories?: Array<{ summary: string; importance: number; tags: string[]; entityIds: string[] }>
+    entities?: Array<{ id: string; kind: 'location' | 'npc' | 'faction' | 'quest'; name: string; summary: string; tags: string[] }>
+    relationships?: Array<{ sourceEntityId: string; targetEntityId: string; score: number; notes: string }>
+    activeGoals?: Array<{ questEntityId: string; name: string; status: 'hidden' | 'active' | 'completed' | 'failed'; summary: string; tags: string[] }>
+  }
 
   class SupabaseRequestError extends Error {
     constructor(
@@ -67,38 +73,42 @@ if (!supabaseUrl || !serviceKey) {
     return databaseRequest<T[]>(`${table}?${new URLSearchParams(query)}`)
   }
 
-  async function persistStorySummary(jobId: string, campaignId: string, throughRound: number, summary: string): Promise<void> {
-    let lastError: unknown
-    for (let attempt = 1; attempt <= 3; attempt += 1) {
+  function usageRecorder(job: JobPayload, userId: string | null, campaignId: string | null) {
+    return async (measurement: AiUsageMeasurement): Promise<void> => {
       try {
-        await databaseRequest('rpc/stage_two_upsert_story_summary', {
+        await databaseRequest('ai_usage?on_conflict=usage_key', {
           method: 'POST',
+          headers: { Prefer: 'resolution=merge-duplicates,return=minimal' },
           body: JSON.stringify({
-            p_job_id: jobId,
-            p_campaign_id: campaignId,
-            p_through_round: throughRound,
-            p_summary: summary,
+            usage_key: `${job.id}:${measurement.operation}:${makeId('usage')}`,
+            user_id: userId,
+            campaign_id: campaignId,
+            job_id: job.id,
+            provider: measurement.provider,
+            request_kind: measurement.operation,
+            model: measurement.model,
+            input_tokens: measurement.inputTokens,
+            output_tokens: measurement.outputTokens,
+            cost_usd: measurement.costUsd,
+            latency_ms: measurement.latencyMs,
           }),
         })
-        return
       } catch (error) {
-        lastError = error
-        if (attempt < 3) await new Promise(resolve => setTimeout(resolve, attempt * 250))
+        // Usage telemetry must never trigger a second paid model call. Quota
+        // reservations are persisted before the job is queued, so a temporary
+        // telemetry failure cannot bypass rate limits.
+        console.error(`[worker] could not record ${measurement.operation} usage for job ${job.id}:`, error)
       }
     }
-    throw lastError
   }
 
   async function processWorld(job: JobPayload) {
-    let messages = job.messages
-    if (!messages) {
-      const [session] = await selectRows<{ messages: Array<{ role: 'user' | 'assistant'; content: string }> }>('coauthor_sessions', {
-        select: 'messages', id: `eq.${job.entityId}`, limit: '1',
-      })
-      if (!session) throw new Error('Coauthor session not found')
-      messages = session.messages
-    }
-    const world = await generateWorld(messages)
+    const [session] = await selectRows<{ owner_id: string; messages: Array<{ role: 'user' | 'assistant'; content: string }> }>('coauthor_sessions', {
+      select: 'owner_id,messages', id: `eq.${job.entityId}`, limit: '1',
+    })
+    if (!session) throw new Error('Coauthor session not found')
+    const messages = job.messages ?? session.messages
+    const world = await generateWorld(messages, { onUsage: usageRecorder(job, session.owner_id, null) })
     await databaseRequest(`coauthor_sessions?id=eq.${encodeURIComponent(job.entityId)}`, {
       method: 'PATCH',
       body: JSON.stringify({ status: 'ready', generated_world: world, updated_at: new Date().toISOString() }),
@@ -118,12 +128,9 @@ if (!supabaseUrl || !serviceKey) {
     if (!round) throw new Error('Round not found')
     if (round.status === 'resolved') return { duplicate: true }
 
-    const [rawCharacters, rawIntents, rawMemories, rawRecentRounds, rawStorySummaries] = await Promise.all([
+    const [rawCharacters, rawIntents, rawRecentRounds, rawStorySummaries] = await Promise.all([
       selectRows<Record<string, any>>('characters', { select: '*', campaign_id: `eq.${round.campaign_id}` }),
       selectRows<Record<string, any>>('player_intents', { select: '*', round_id: `eq.${round.id}` }),
-      selectRows<{ summary: string; importance: number; tags: string[]; entity_ids: string[] }>('memories', {
-        select: 'summary,importance,tags,entity_ids', campaign_id: `eq.${round.campaign_id}`, order: 'importance.desc,created_at.desc', limit: '24',
-      }),
       selectRows<{ number: number; narration: string; next_prompt: string | null }>('rounds', {
         select: 'number,narration,next_prompt', campaign_id: `eq.${round.campaign_id}`, status: 'eq.resolved', order: 'number.desc', limit: '2',
       }),
@@ -143,20 +150,30 @@ if (!supabaseUrl || !serviceKey) {
       action: row.action, ready: row.ready, createdAt: row.created_at, updatedAt: row.updated_at,
     }))
     const scene = String(round.campaigns.current_scene ?? '')
+    const retrieval = await databaseRequest<RetrievedContext>('rpc/stage_six_retrieve_context', {
+      method: 'POST',
+      body: JSON.stringify({
+        p_campaign_id: round.campaign_id,
+        p_search_text: buildRetrievalText({ scene, intents, characters }),
+        p_limit: 8,
+      }),
+    })
     const context = buildRoundContext({
       scene,
       storySummary: rawStorySummaries[0]?.summary ?? null,
       recentRounds: rawRecentRounds.map(previous => ({ number: previous.number, narration: previous.narration, nextPrompt: previous.next_prompt })),
       intents,
       characters,
-      memories: rawMemories.map(memory => ({
-        summary: memory.summary, importance: memory.importance, tags: memory.tags, entityIds: memory.entity_ids,
-      })),
+      memories: retrieval.memories ?? [],
+      entities: retrieval.entities ?? [],
+      relationships: retrieval.relationships ?? [],
+      activeGoals: retrieval.activeGoals ?? [],
     })
-    const plan = validateAndOrderRoundPlan(await planRound(context), context)
+    const recordUsage = usageRecorder(job, String(round.campaigns.owner_id), String(round.campaign_id))
+    const plan = validateAndOrderRoundPlan(await planRound(context, { onUsage: recordUsage }), context)
     const rolls = resolveChecks(plan.checks, characters)
     const permittedEvents = bindMechanicalEventsToRolls(plan.proposedEvents, rolls)
-    const resolution = await narrateRound({ context, actionSequence: buildActionSequence(context, plan), rolls, permittedEvents })
+    const resolution = await narrateRound({ context, actionSequence: buildActionSequence(context, plan), rolls, permittedEvents }, { onUsage: recordUsage })
     const safeEvents = resolution.events.filter(event => permittedEvents.some(permitted => JSON.stringify(permitted) === JSON.stringify(event)))
     const applied = applyMechanicalEvents(characters, safeEvents, rolls)
     const shouldSummarize = shouldUpdateStorySummary(Number(round.number))
@@ -164,10 +181,10 @@ if (!supabaseUrl || !serviceKey) {
       ? await summarizeStory({
           previousSummary: context.storySummary,
           rounds: [...context.recentRounds.map(previous => ({ number: previous.number, narration: previous.narration })), { number: Number(round.number), narration: resolution.narration }],
-        })
+        }, { onUsage: recordUsage })
       : null
 
-    await databaseRequest(job.claimToken ? 'rpc/commit_claimed_srd_round_resolution' : 'rpc/commit_srd_round_resolution', {
+    await databaseRequest(job.claimToken ? 'rpc/commit_claimed_stage_six_round_resolution' : 'rpc/commit_stage_six_round_resolution', {
       method: 'POST',
       body: JSON.stringify({
         p_round_id: round.id,
@@ -182,18 +199,15 @@ if (!supabaseUrl || !serviceKey) {
           statuses: character.statuses,
           rulesState: character.rules,
         })),
-        p_memory: sanitizeRoundMemory(resolution.memory),
+        p_memory: sanitizeRoundMemory(resolution.memory, new Set([
+          ...context.activeCharacters.map(character => character.id),
+          ...context.entities.map(entity => entity.id),
+        ])),
+        p_story_summary: summary?.summary ?? null,
         p_idempotency_key: job.id,
         ...(job.claimToken ? { p_worker_id: job.claimToken } : {}),
       }),
     })
-    if (summary) {
-      try {
-        await persistStorySummary(job.id, round.campaign_id, Number(round.number), summary.summary)
-      } catch (error) {
-        console.error(`[worker] round ${round.id} committed but story summary persistence failed:`, error)
-      }
-    }
     return { narration: resolution.narration, rolls: rolls.length }
   }