Important Fixes, New mechanics, and many more

apps/worker/src/index.ts

@@ -3,7 +3,8 @@ import { Worker } from 'bullmq'
 import IORedis from 'ioredis'
 import { applyMechanicalEvents, bindMechanicalEventsToRolls, resolveCheck } from '@dng/game-engine'
 import { CharacterSchema, PlayerIntentSchema, makeId } from '@dng/shared'
-import { generateWorld, narrateRound, planRound } from './ai'
+import { generateWorld, narrateRound, planRound, summarizeStory } from './ai'
+import { buildActionSequence, buildRoundContext, sanitizeRoundMemory, shouldUpdateStorySummary, validateAndOrderRoundPlan } from './orchestration'
 
 const redisUrl = process.env.REDIS_URL
 const supabaseUrl = process.env.SUPABASE_URL
@@ -36,7 +37,7 @@ if (!supabaseUrl || !serviceKey) {
     ) {
       const rpcName = path.startsWith('rpc/') ? path.slice('rpc/'.length) : null
       const message = status === 404 && rpcName
-        ? `Supabase RPC "${rpcName}" was not found. Apply supabase/migrations/0002_supabase_ai_queue.sql in the Supabase SQL Editor, then restart the worker.`
+        ? `Supabase RPC "${rpcName}" was not found. The database schema is incomplete. Run the generated supabase/bootstrap.sql in the Supabase SQL Editor, then restart the worker.`
         : `Supabase returned ${status} for ${path}: ${detail}`
       super(message)
       this.name = 'SupabaseRequestError'
@@ -57,14 +58,37 @@ if (!supabaseUrl || !serviceKey) {
       const detail = await response.text()
       throw new SupabaseRequestError(response.status, path, detail)
     }
-    if (response.status === 204) return undefined as T
-    return response.json() as Promise<T>
+    const responseBody = await response.text()
+    if (!responseBody) return undefined as T
+    return JSON.parse(responseBody) as T
   }
 
   async function selectRows<T>(table: string, query: Record<string, string>): Promise<T[]> {
     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) {
+      try {
+        await databaseRequest('rpc/stage_two_upsert_story_summary', {
+          method: 'POST',
+          body: JSON.stringify({
+            p_job_id: jobId,
+            p_campaign_id: campaignId,
+            p_through_round: throughRound,
+            p_summary: summary,
+          }),
+        })
+        return
+      } catch (error) {
+        lastError = error
+        if (attempt < 3) await new Promise(resolve => setTimeout(resolve, attempt * 250))
+      }
+    }
+    throw lastError
+  }
+
   async function processWorld(job: JobPayload) {
     let messages = job.messages
     if (!messages) {
@@ -94,32 +118,57 @@ if (!supabaseUrl || !serviceKey) {
     if (!round) throw new Error('Round not found')
     if (round.status === 'resolved') return { duplicate: true }
 
-    const [rawCharacters, rawIntents, rawMemories] = await Promise.all([
+    const [rawCharacters, rawIntents, rawMemories, 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 }>('memories', { select: 'summary', campaign_id: `eq.${round.campaign_id}`, order: 'importance.desc', limit: '12' }),
+      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',
+      }),
+      selectRows<{ through_round: number; summary: string }>('story_summaries', {
+        select: 'through_round,summary', campaign_id: `eq.${round.campaign_id}`, order: 'through_round.desc', limit: '1',
+      }),
     ])
 
     const characters = rawCharacters.map(row => CharacterSchema.parse({
       id: row.id, name: row.name, concept: row.concept, controller: row.controller,
       userId: row.user_id, abilities: row.abilities, hp: row.hp, maxHp: row.max_hp,
-      defense: row.defense, proficiency: row.proficiency, inventory: row.inventory, statuses: row.statuses,
+      defense: row.defense, proficiency: row.proficiency, inventory: row.inventory, statuses: row.statuses, persona: row.persona,
     }))
     const intents = rawIntents.map(row => PlayerIntentSchema.parse({
       id: row.id, roundId: row.round_id, memberId: row.member_id, characterId: row.character_id,
       action: row.action, ready: row.ready, createdAt: row.created_at, updatedAt: row.updated_at,
     }))
     const scene = String(round.campaigns.current_scene ?? '')
-    const plan = await planRound({ scene, intents, characters, memories: rawMemories.map(row => String(row.summary)) })
+    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,
+      })),
+    })
+    const plan = validateAndOrderRoundPlan(await planRound(context), context)
     const rolls = plan.checks.map(check => {
       const actor = characters.find(character => character.id === check.actorId)
       if (!actor) throw new Error(`Unknown check actor ${check.actorId}`)
       return resolveCheck(check, actor)
     })
     const permittedEvents = bindMechanicalEventsToRolls(plan.proposedEvents, rolls)
-    const resolution = await narrateRound({ scene, intents, aiActions: plan.aiActions, rolls, permittedEvents })
+    const resolution = await narrateRound({ context, actionSequence: buildActionSequence(context, plan), rolls, permittedEvents })
     const safeEvents = resolution.events.filter(event => permittedEvents.some(permitted => JSON.stringify(permitted) === JSON.stringify(event)))
     const applied = applyMechanicalEvents(characters, safeEvents)
+    const shouldSummarize = shouldUpdateStorySummary(Number(round.number))
+    const summary = shouldSummarize
+      ? await summarizeStory({
+          previousSummary: context.storySummary,
+          rounds: [...context.recentRounds.map(previous => ({ number: previous.number, narration: previous.narration })), { number: Number(round.number), narration: resolution.narration }],
+        })
+      : null
 
     await databaseRequest(job.claimToken ? 'rpc/commit_claimed_round_resolution' : 'rpc/commit_round_resolution', {
       method: 'POST',
@@ -130,11 +179,18 @@ if (!supabaseUrl || !serviceKey) {
         p_rolls: rolls,
         p_events: safeEvents,
         p_character_states: applied.characters.map(character => ({ id: character.id, hp: character.hp, inventory: character.inventory, statuses: character.statuses })),
-        p_memory: resolution.memory,
+        p_memory: sanitizeRoundMemory(resolution.memory),
         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 }
   }
 
@@ -201,6 +257,7 @@ if (!supabaseUrl || !serviceKey) {
     const pollInterval = Math.max(250, Number(process.env.AI_JOB_POLL_INTERVAL_MS ?? 1_000))
     const workerId = process.env.RAILWAY_REPLICA_ID ?? makeId('worker')
     let stopped = false
+    let consecutivePollFailures = 0
 
     const stop = () => { stopped = true }
     process.once('SIGTERM', stop)
@@ -212,6 +269,7 @@ if (!supabaseUrl || !serviceKey) {
         const claimed = await databaseRequest<ClaimedJob[]>('rpc/claim_ai_job', {
           method: 'POST', body: JSON.stringify({ p_worker_id: workerId }),
         })
+        consecutivePollFailures = 0
         const job = claimed[0]
         if (!job) {
           await new Promise(resolve => setTimeout(resolve, pollInterval))
@@ -233,8 +291,15 @@ if (!supabaseUrl || !serviceKey) {
           process.exitCode = 1
           break
         }
-        console.error('[worker] polling failed:', error)
-        await new Promise(resolve => setTimeout(resolve, pollInterval))
+        consecutivePollFailures += 1
+        const retryDelay = Math.min(30_000, pollInterval * 2 ** Math.min(consecutivePollFailures - 1, 5))
+        const message = error instanceof Error ? error.message : String(error)
+        // Avoid flooding the terminal and Supabase during a network outage.
+        // Log the first failure and then only when the backoff duration grows.
+        if (consecutivePollFailures === 1 || (consecutivePollFailures & (consecutivePollFailures - 1)) === 0) {
+          console.error(`[worker] polling failed; retrying in ${retryDelay}ms: ${message}`)
+        }
+        await new Promise(resolve => setTimeout(resolve, retryDelay))
       }
     }
   }