Session 7b: Fix pipeline - body parser, Redis queueing, poller visibility, auto-start

Dockerfile

@@ -7,9 +7,9 @@
 #   2. runner — copy src/, poller/, scripts/, node_modules and start
 #
 # The Next.js frontend ships in a separate image (web/Dockerfile). PM2
-# pollers can run inside this image via `pm2-runtime` or as a sibling
+# pollers run INSIDE this image and are auto-started by
-# container — production deploys use a sibling so a poller crash doesn't
+# scripts/docker-entrypoint.sh so a Coolify redeploy reseeds them on
-# restart the API.
+# every container start.
 #
 # Build:  docker build -t vyndr-api .
 # Run:    docker run -p 3001:3001 --env-file .env vyndr-api
@@ -26,12 +26,16 @@ RUN npm ci --omit=dev --no-audit --no-fund
 FROM node:20-alpine AS runner
 WORKDIR /app
 
-# curl is used by the /api/health smoke check (Coolify HEALTHCHECK), and
+# curl is used by the /api/health smoke check (Coolify HEALTHCHECK).
-# postgresql-client lets the in-container migrations script run if needed.
 RUN apk add --no-cache curl tini
 
+# PM2 is installed globally so the entrypoint can call `pm2 start` to
+# boot all three pollers (NBA / WNBA / MLB) alongside the Express API.
+RUN npm install -g pm2@latest --no-audit --no-fund
+
 ENV NODE_ENV=production \
-    PORT=3001
+    PORT=3001 \
+    PM2_HOME=/app/.pm2
 
 # Non-root user — the container should never run as uid 0 even if the
 # host accidentally maps a privileged port.
@@ -44,19 +48,22 @@ COPY poller ./poller
 COPY scripts ./scripts
 COPY supabase ./supabase
 
-# Persistent volume for JSONL training data. Coolify mounts this path so
+# Persistent volume for JSONL training data (resolutions survive
-# resolutions survive redeploys.
+# redeploys via the Coolify mount). PM2_HOME lives outside it so
-RUN mkdir -p /app/data/training && chown -R vyndr:vyndr /app/data
+# supervisor state is local to the container.
+RUN mkdir -p /app/data/training /app/.pm2 \
+  && chown -R vyndr:vyndr /app/data /app/.pm2 \
+  && chmod +x /app/scripts/docker-entrypoint.sh
 
 USER vyndr
 
 EXPOSE 3001
 
-# tini reaps zombies cleanly when Node spawns child processes (e.g., the
+# tini reaps zombies — important now that we spawn pm2 as a child of
-# embedded Python pre-checks the orchestrator may run during a slate).
+# this entrypoint.
 ENTRYPOINT ["/sbin/tini", "--"]
 
-HEALTHCHECK --interval=30s --timeout=10s --start-period=20s --retries=3 \
+HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \
   CMD curl -fsS http://127.0.0.1:3001/api/health || exit 1
 
-CMD ["node", "src/server.js"]
+CMD ["sh", "scripts/docker-entrypoint.sh"]

data/training/resolutions-2026-06.jsonl

@@ -118,3 +118,17 @@
 {"ts":"2026-06-08T15:21:50.450Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A"}
 {"ts":"2026-06-08T15:21:50.477Z","sport":"nba","player_espn_id":"999999999","player_name":"Phantom Player","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"C"}
 {"ts":"2026-06-08T15:21:50.537Z","sport":"nba","player_espn_id":"99999","player_name":"Phantom","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"B"}
+{"ts":"2026-06-10T05:11:19.658Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A"}
+{"ts":"2026-06-10T05:11:19.690Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A-"}
+{"ts":"2026-06-10T05:11:19.690Z","sport":"nba","player_espn_id":"3136195","player_name":"Karl-Anthony Towns","stat_type":"rebounds","line":13.5,"direction":"under","actual_value":13,"result":"hit","margin":-0.5,"grade":"B+"}
+{"ts":"2026-06-10T05:11:19.690Z","sport":"nba","player_espn_id":"3062679","player_name":"Josh Hart","stat_type":"pts_reb_ast","line":10.5,"direction":"over","actual_value":10,"result":"miss","margin":-0.5,"grade":"C"}
+{"ts":"2026-06-10T05:11:19.741Z","sport":"nba","player_espn_id":"999999999","player_name":"Phantom Player","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"C"}
+{"ts":"2026-06-10T05:11:19.761Z","sport":"nba","player_espn_id":"99999","player_name":"Phantom","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"B"}
+{"ts":"2026-06-10T05:18:28.032Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A"}
+{"ts":"2026-06-10T05:18:36.253Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A-"}
+{"ts":"2026-06-10T05:18:36.253Z","sport":"nba","player_espn_id":"3136195","player_name":"Karl-Anthony Towns","stat_type":"rebounds","line":13.5,"direction":"under","actual_value":13,"result":"hit","margin":-0.5,"grade":"B+"}
+{"ts":"2026-06-10T05:18:36.253Z","sport":"nba","player_espn_id":"3062679","player_name":"Josh Hart","stat_type":"pts_reb_ast","line":10.5,"direction":"over","actual_value":10,"result":"miss","margin":-0.5,"grade":"C"}
+{"ts":"2026-06-10T05:18:36.305Z","sport":"nba","player_espn_id":"999999999","player_name":"Phantom Player","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"C"}
+{"ts":"2026-06-10T05:18:36.318Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A"}
+{"ts":"2026-06-10T05:18:36.418Z","sport":"nba","player_espn_id":"99999","player_name":"Phantom","stat_type":"points","line":10.5,"direction":"over","actual_value":null,"result":"void","grade":"B"}
+{"ts":"2026-06-10T05:18:36.836Z","sport":"nba","player_espn_id":"3934719","player_name":"OG Anunoby","stat_type":"points","line":16.5,"direction":"over","actual_value":17,"result":"hit","margin":0.5,"grade":"A"}

poller/poller.js

@@ -234,6 +234,20 @@ async function tick(sportCfg) {
     console.warn(`[poller-${SPORT}] scoreboard fetch failed: ${err.message}`);
     return;
   }
+
+  // Bucket games by status so every tick surfaces what we're observing —
+  // without this, an NBA Finals night with one game stuck in 'pre' looks
+  // identical to a silent crash. Visibility > debugging in the dark.
+  const bucket = { pre: 0, in: 0, post: 0, other: 0 };
+  for (const g of games) {
+    const s = (g.state || 'other').toLowerCase();
+    if (bucket[s] !== undefined) bucket[s] += 1;
+    else bucket.other += 1;
+  }
+  console.log(
+    `[poller-${SPORT}] tick — ${games.length} games (pre=${bucket.pre} in=${bucket.in} post=${bucket.post}${bucket.other ? ` other=${bucket.other}` : ''})`
+  );
+
   for (const g of games) {
     try { await handleGame(g, sportCfg); }
     catch (err) { console.warn(`[poller-${SPORT}] game ${g.id} handler error: ${err.message}`); }
@@ -254,15 +268,30 @@ async function main() {
   console.log(`[poller-${SPORT}] starting — pollInterval=${POLL_INTERVAL_MS}ms buffer=${BUFFER_MS}ms`);
 
   // Run forever. The PM2 supervisor restarts on crash; tick errors are
-  // already caught inside.
+  // already caught inside. Wrap the cycle in try/catch so a missed
+  // rejection inside tick() doesn't silently kill the loop.
   /* eslint-disable no-constant-condition */
   while (true) {
+    try {
       await tick(sportCfg);
+    } catch (err) {
+      console.error(`[poller-${SPORT}] tick error:`, err?.stack || err?.message || err);
+    }
     const intervalMs = inGameHours(sportCfg) ? POLL_INTERVAL_MS : OFF_HOURS_POLL_MS;
     await new Promise((r) => setTimeout(r, intervalMs));
   }
 }
 
+// Surface unhandled rejections / uncaught exceptions in PM2 logs. Without
+// these handlers the process can die silently with no stderr — which is
+// the symptom 7b is asked to fix.
+process.on('unhandledRejection', (reason) => {
+  console.error(`[poller-${SPORT}] unhandledRejection:`, reason?.stack || reason);
+});
+process.on('uncaughtException', (err) => {
+  console.error(`[poller-${SPORT}] uncaughtException:`, err?.stack || err?.message || err);
+});
+
 // Surface for tests — they import individual handlers without firing main().
 module.exports = {
   isFinalStatus,

scripts/docker-entrypoint.sh

@@ -0,0 +1,34 @@
+#!/bin/sh
+#
+# VYNDR API container entrypoint.
+#
+# Boots the PM2-managed pollers in daemon mode, then execs the Express
+# server in the foreground so it runs as PID 1 (which Docker's signal
+# forwarding + healthcheck expect). PM2 daemonizes its supervisor inside
+# the container; the pollers keep running while Express handles requests.
+#
+# On container restart (Coolify redeploy) PM2's state is gone — that's
+# fine, this script reseeds it. Pollers are idempotent: status keys live
+# in Redis with TTLs, so a restart at most re-processes one game.
+#
+# Why a separate file from scripts/start.sh: that one is the local-dev
+# launcher (boots the Python NBA service + Node API in the host shell).
+# This one is the production container entrypoint.
+
+set -e
+
+# PM2 needs a writable HOME for its run-dir. The vyndr non-root user
+# can't write to /root, so park PM2's metadata in /app/.pm2.
+export PM2_HOME="${PM2_HOME:-/app/.pm2}"
+mkdir -p "$PM2_HOME"
+
+echo "[docker-entrypoint] PM2_HOME=$PM2_HOME"
+echo "[docker-entrypoint] booting pollers via PM2"
+cd /app/poller
+pm2 start ecosystem.config.js || \
+  echo "[docker-entrypoint] PM2 start failed — continuing without pollers so the API still serves"
+pm2 list || true
+
+cd /app
+echo "[docker-entrypoint] starting Express server (PID 1)"
+exec node src/server.js

scripts/verify-schema.js

@@ -3,7 +3,7 @@ const { createClient } = require('@supabase/supabase-js');
 
 const supabase = createClient(
   process.env.SUPABASE_URL,
-  process.env.SUPABASE_SERVICE_KEY
+  process.env.SUPABASE_SERVICE_ROLE_KEY || process.env.SUPABASE_SERVICE_KEY
 );
 
 const EXPECTED_TABLES = ['users', 'picks', 'scan_sessions', 'bets', 'outcomes', 'performance'];

src/app.js

@@ -51,7 +51,11 @@ app.use(missionHeader);
 // Stripe webhook needs raw body — must be before express.json()
 app.use('/api/stripe/webhook', express.raw({ type: 'application/json' }));
 
-app.use(express.json());
+// Body parser limit raised to 10MB to accommodate full-slate poller
+// payloads. The default 100KB rejected real WNBA slates with 413.
+// Per-route limits below can tighten or loosen this for specific paths.
+app.use(express.json({ limit: '10mb' }));
+app.use(express.urlencoded({ extended: true, limit: '10mb' }));
 
 // Health check — public minimal status (Coolify, uptime monitors). Detailed
 // adapter + Python service status only with X-VYNDR-Internal-Key.
@@ -119,9 +123,11 @@ app.use('/api/waitlist', waitlistRoutes);
 app.use('/api/pipeline', pipelineRoutes);
 app.use('/api/share-card', shareCardRoutes);
 app.use('/api/push', pushRoutes);
-// Resolution payloads carry full ESPN box scores (50-100KB). Scope a larger
+// Resolution payloads carry full ESPN box scores plus per-game prop
-// limit to /api/grading only so the other routes keep the safer 100KB default.
+// arrays. Full-slate WNBA / MLB resolves exceed 2MB in practice — keep
-app.use('/api/grading', express.json({ limit: '2mb' }), gradingRoutes);
+// /api/grading aligned with the global 10MB ceiling. Correction sweep
+// stays small (just a window-hours integer + flags).
+app.use('/api/grading', express.json({ limit: '10mb' }), gradingRoutes);
 app.use('/api/grading', express.json({ limit: '256kb' }), correctionRoutes);
 const widgetRoutes = require('./routes/widget');
 app.use('/api/widget', widgetRoutes);

src/utils/redis.js

@@ -11,10 +11,13 @@ let degraded = false;
 function getRedisClient() {
   if (!client) {
     client = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379', {
-      // Don't block the app starting up if Redis is down at boot. Failed
+      // Eager connect — but allow commands to QUEUE while the TCP/auth
-      // commands fall through to the cache-miss path immediately.
+      // handshake completes. Without offline queueing, pollers booting
+      // alongside the API throw "Stream isn't writeable" before Redis
+      // is ready. With queueing, ioredis flushes the backlog the moment
+      // the connection enters READY state.
       lazyConnect: false,
-      enableOfflineQueue: false,
+      enableOfflineQueue: true,
       maxRetriesPerRequest: 1,
       retryStrategy(times) {
         // Exponential backoff up to 30s. ioredis keeps trying forever on its
@@ -30,7 +33,11 @@ function getRedisClient() {
       }
     });
     client.on('ready', () => {
+      // Surface the ready transition in PM2/container logs so operators
+      // can confirm the connection actually established. Distinct from
+      // reconnect-after-outage which the line below logs at info.
       if (degraded) console.info('[redis] reconnected, leaving degraded mode');
+      else console.log('[redis] connected and ready');
       degraded = false;
     });
   }

src/utils/supabase.js

@@ -15,9 +15,14 @@ function getSupabaseClient() {
 
 function getSupabaseServiceClient() {
   if (!serviceClient) {
+    // SUPABASE_SERVICE_ROLE_KEY is the canonical name across .env.example
+    // and the Next.js side. The fallback to SUPABASE_SERVICE_KEY supports
+    // any deployment still set with the legacy variable so the rename
+    // doesn't take down a running instance — remove the fallback once all
+    // Coolify envs are updated.
     serviceClient = createClient(
       process.env.SUPABASE_URL,
-      process.env.SUPABASE_SERVICE_KEY
+      process.env.SUPABASE_SERVICE_ROLE_KEY || process.env.SUPABASE_SERVICE_KEY
     );
   }
   return serviceClient;

tests/integration/pipeline.test.js

@@ -0,0 +1,193 @@
+/**
+ * Session 7b pipeline regression test.
+ *
+ * Locks in the three fixes that unblock the live deploy:
+ *   1. Body parser accepts payloads up to 10 MB (no more poller 413s).
+ *   2. /api/grading/resolve still grades correctly with the raised limit.
+ *   3. The orchestrator's grading output has the canonical shape:
+ *      { id, grade, prop } for every prop persisted.
+ *
+ * Mocks every upstream (Supabase, distribution channels, intelligence
+ * services) so the test runs offline.
+ */
+
+process.env.VYNDR_INTERNAL_KEY = 'pipeline-7b-key';
+process.env.NODE_ENV = 'test';
+process.env.ENGINE2_ENABLED = 'false';
+
+const mockState = {
+  unresolved: [],
+  inserts: [],
+  updates: [],
+};
+
+jest.mock('../../src/utils/supabase', () => ({
+  getSupabaseServiceClient: () => ({
+    from(table) {
+      const ctx = { table, filters: {} };
+      const proxy = {
+        select(_cols, opts) { ctx.head = !!opts?.head; ctx.countMode = opts?.count; return proxy; },
+        eq(col, val) { ctx.filters[col] = val; return proxy; },
+        limit() { return Promise.resolve({ data: [], error: null }); },
+        is() { return Promise.resolve({ data: mockState.unresolved, error: null }); },
+        in() { return Promise.resolve({ error: null }); },
+        insert(row) {
+          mockState.inserts.push({ table, row });
+          return {
+            select() {
+              return { single: () => Promise.resolve({ data: { id: `gid-${mockState.inserts.length}` }, error: null }) };
+            },
+          };
+        },
+        update(patch) {
+          mockState.updates.push({ table, patch });
+          return {
+            eq() { return Promise.resolve({ error: null }); },
+            in() { return Promise.resolve({ error: null }); },
+          };
+        },
+        upsert() { return Promise.resolve({ error: null }); },
+      };
+      return proxy;
+    },
+  }),
+}));
+
+jest.mock('../../src/services/distribution/webPush', () => ({
+  configured: () => false, sendPushToSport: async () => ({ ok: true, sent: 0 }),
+}));
+jest.mock('../../src/services/distribution/telegram', () => ({
+  configured: () => false, postToTelegram: async () => ({ ok: true }),
+}));
+jest.mock('../../src/services/distribution/discord', () => ({
+  webhookFor: () => null, postToDiscord: async () => ({ ok: true }),
+}));
+
+// Stub the learning-loop hooks so the resolve route doesn't try to
+// recompute CLV / accuracy / weights against the fake DB shape above.
+jest.mock('../../src/services/intelligence/clvTracker', () => ({
+  computeCLV: async () => ({ clv: null }),
+}));
+jest.mock('../../src/services/intelligence/accuracyTracker', () => ({
+  recordResolution: async () => undefined,
+}));
+jest.mock('../../src/services/intelligence/weightAdjuster', () => ({
+  adjustWeights: async () => ({ skipped: true, reason: 'test' }),
+}));
+
+const fs = require('fs');
+const path = require('path');
+const express = require('express');
+const gradingRoutes = require('../../src/routes/grading');
+
+const fixture = JSON.parse(fs.readFileSync(
+  path.join(__dirname, '..', 'fixtures', 'nba-boxscore-sample.json'),
+  'utf8',
+));
+
+// Mirror app.js so the body parser limit under test matches production.
+function makeApp({ limit = '10mb' } = {}) {
+  const app = express();
+  app.use(express.json({ limit }));
+  app.use('/api/grading', express.json({ limit }), gradingRoutes);
+  return app;
+}
+
+function call(app, method, url, body, headers = {}) {
+  return new Promise((resolve, reject) => {
+    const http = require('http');
+    const server = app.listen(0, '127.0.0.1', () => {
+      const port = server.address().port;
+      const data = body ? JSON.stringify(body) : '';
+      const req = http.request({
+        host: '127.0.0.1', port, path: url, method,
+        headers: { 'Content-Type': 'application/json', ...headers },
+      }, (res) => {
+        const chunks = [];
+        res.on('data', (c) => chunks.push(c));
+        res.on('end', () => {
+          server.close();
+          const raw = Buffer.concat(chunks).toString('utf8');
+          let parsed; try { parsed = JSON.parse(raw); } catch { parsed = raw; }
+          resolve({ status: res.statusCode, body: parsed });
+        });
+      });
+      req.on('error', (err) => { server.close(); reject(err); });
+      if (data) req.write(data);
+      req.end();
+    });
+  });
+}
+
+beforeEach(() => {
+  mockState.unresolved = [];
+  mockState.inserts.length = 0;
+  mockState.updates.length = 0;
+});
+
+describe('pipeline body-parser regression (Session 7b)', () => {
+  test('accepts 5MB payloads on /api/grading/resolve (was 413 before fix)', async () => {
+    // Build a payload well over the old 100KB default — pad the body with
+    // a large but ignorable field so the route still has the gameId/sport
+    // it needs to no-op cleanly.
+    const huge = 'X'.repeat(5 * 1024 * 1024);
+    const app = makeApp({ limit: '10mb' });
+    const res = await call(
+      app, 'POST', '/api/grading/resolve',
+      { gameId: 'g-large', sport: 'nba', void: true, reason: 'test_oversize', _padding: huge },
+      { 'X-VYNDR-Internal-Key': 'pipeline-7b-key' },
+    );
+    expect(res.status).not.toBe(413);
+    // void: true path returns 200 with the empty resolution summary.
+    expect(res.status).toBe(200);
+    expect(res.body).toHaveProperty('resolved');
+  });
+
+  test('rejects payloads larger than the 10MB limit with 413 (sanity check)', async () => {
+    // Use a tiny app limit so the test doesn't allocate 11MB just to
+    // prove the rejection path exists.
+    const app = makeApp({ limit: '100kb' });
+    const huge = 'Y'.repeat(200 * 1024);
+    const res = await call(
+      app, 'POST', '/api/grading/resolve',
+      { gameId: 'g', sport: 'nba', _padding: huge },
+      { 'X-VYNDR-Internal-Key': 'pipeline-7b-key' },
+    );
+    expect(res.status).toBe(413);
+  });
+});
+
+describe('pipeline grading shape (Session 7b)', () => {
+  test('grading a fixture box score returns a results array with grade + actual_value', async () => {
+    // Use a real player from the saved fixture so the result is
+    // deterministic without depending on every upstream stub.
+    const team0 = fixture.boxscore.players[0];
+    const athlete = team0.statistics[0].athletes[0];
+    mockState.unresolved = [{
+      id: 'gh-pipeline-1',
+      player_id: String(athlete.athlete.id),
+      player_name: athlete.athlete.displayName,
+      stat_type: 'points',
+      line: Number(athlete.stats[1]) - 0.5,
+      direction: 'over',
+      grade: 'A',
+      sport: 'nba',
+      factors: ['l5_hot_vs_line'],
+    }];
+
+    const app = makeApp();
+    const res = await call(
+      app, 'POST', '/api/grading/resolve',
+      { gameId: '401859964', sport: 'nba', boxScore: fixture },
+      { 'X-VYNDR-Internal-Key': 'pipeline-7b-key' },
+    );
+
+    expect(res.status).toBe(200);
+    expect(Array.isArray(res.body.results)).toBe(true);
+    expect(res.body.results).toHaveLength(1);
+    const r = res.body.results[0];
+    expect(r).toHaveProperty('grade');
+    expect(r).toHaveProperty('actual_value');
+    expect(['hit', 'miss', 'push', 'void']).toContain(r.result);
+  });
+});

tests/unit/redisAuthUrl.test.js

@@ -37,12 +37,12 @@ describe('redis client URL handling', () => {
     expect(mockCtor.mock.calls[0][0]).toBe('redis://127.0.0.1:6379');
   });
 
-  test('passes enableOfflineQueue=false so degraded mode fails fast', () => {
+  test('passes enableOfflineQueue=true so early commands queue until READY', () => {
     process.env.REDIS_URL = 'redis://localhost:6379';
     const { getRedisClient } = require('../../src/utils/redis');
     getRedisClient();
     expect(mockCtor.mock.calls[0][1]).toMatchObject({
-      enableOfflineQueue: false,
+      enableOfflineQueue: true,
       maxRetriesPerRequest: 1,
     });
   });