feat(vibe): adapt sessions to context and exploration

backend/src/services/db.service.ts

@@ -6,6 +6,7 @@ import { fileURLToPath } from 'url';
 import { dirname, join } from 'path';
 import { Pool, PoolClient } from 'pg';
 import { SearchService } from './search.service.js';
+import { normalizeVibeContext, normalizeVibeEventPayload } from './vibe-context.service.js';
 
 /** Anything with a `.query()` — either the shared Pool or a checked-out client. */
 type Queryable = Pool | PoolClient;
@@ -51,6 +52,7 @@ import type {
   RecordedVibeEvent,
   VibePlan,
   VibePlanItem,
+  VibeSessionProfile,
 } from '../db/types.js';
 import { AUDIO_PREFERENCE_BUCKETS, FEEDBACK_ACTIONS } from '../db/types.js';
 export * from '../db/types.js';
@@ -1514,7 +1516,17 @@ export class DbService {
     policyVersion: string;
     seedTrackId?: string | null;
     context?: Record<string, unknown>;
+    profile?: {
+      goals: Record<string, unknown>;
+      explorationCoefficient: number;
+      discoveryRadius: number;
+    };
   }): Promise<VibeSession> {
+    // DbService is also used directly by workers and migrations. Keep the
+    // durable storage boundary canonical even when callers bypass the HTTP
+    // coordinator, so opaque or precise client telemetry can never become
+    // session context.
+    const canonicalContext = normalizeVibeContext(params.context ?? {});
     return this.withTransaction(async (client) => {
       // Serialize starts for one listener even when there is no active row to
       // lock yet. The row lock below then safely replaces any prior session.
@@ -1542,14 +1554,25 @@ export class DbService {
         }
       }
       const res = await client.query(
-        `INSERT INTO vibe_sessions (user_id, status, seed_track_id, context, policy_version)
-         VALUES ($1, 'active', $2, $3::jsonb, $4)
-         RETURNING *`,
+        `WITH created AS (
+           INSERT INTO vibe_sessions (user_id, status, seed_track_id, context, policy_version)
+           VALUES ($1, 'active', $2, $3::jsonb, $4)
+           RETURNING *
+         ), profile AS (
+           INSERT INTO vibe_session_profiles
+             (session_id, user_id, goals, exploration_coefficient, discovery_radius)
+           SELECT id, user_id, $5::jsonb, $6::real, $7::real FROM created
+           ON CONFLICT (session_id) DO NOTHING
+         )
+         SELECT * FROM created`,
         [
           params.userId,
           params.seedTrackId ?? null,
-          JSON.stringify(params.context ?? {}),
+          JSON.stringify(canonicalContext),
           params.policyVersion,
+          JSON.stringify(params.profile?.goals ?? { type: 'discovery', target: 1, progress: 0 }),
+          params.profile?.explorationCoefficient ?? 0.3,
+          params.profile?.discoveryRadius ?? 0.38,
         ],
       );
       return res.rows[0] as VibeSession;
@@ -1565,6 +1588,141 @@ export class DbService {
     return (res.rows[0] as VibeSession) ?? null;
   }
 
+  /** Recent session shapes, excluding the active session. The fingerprint is
+   * intentionally aggregate-only and is used as a soft planning penalty. */
+  async getRecentVibeSessionFingerprints(userId: string, sessionId: string, limit = 8): Promise<Record<string, unknown>[]> {
+    const res = await this.pgClient.query(
+      `SELECT p.fingerprint
+       FROM vibe_session_profiles p
+       JOIN vibe_sessions s ON s.id = p.session_id
+       WHERE p.user_id = $1 AND p.session_id <> $2::uuid
+         AND s.status IN ('ended', 'expired', 'replaced')
+         AND p.fingerprint <> '{}'::jsonb
+       ORDER BY p.updated_at DESC
+       LIMIT $3`,
+      [userId, sessionId, limit],
+    );
+    return res.rows.map((row: { fingerprint: Record<string, unknown> }) => row.fingerprint ?? {});
+  }
+
+  async getVibeSessionProfile(sessionId: string, userId: string): Promise<VibeSessionProfile | null> {
+    const res = await this.pgClient.query(
+      `SELECT p.* FROM vibe_session_profiles p
+       JOIN vibe_sessions s ON s.id = p.session_id
+       WHERE p.session_id = $1 AND s.user_id = $2`,
+      [sessionId, userId],
+    );
+    return (res.rows[0] as VibeSessionProfile | undefined) ?? null;
+  }
+
+  /** Replace only the coarse, sanitised context attached to an active session.
+   * The immutable context_changed event remains the audit trail. */
+  async updateVibeSessionContext(sessionId: string, userId: string, context: Record<string, unknown>): Promise<void> {
+    const canonicalContext = normalizeVibeContext(context);
+    await this.withTransaction(async client => {
+      const updated = await client.query(
+        `UPDATE vibe_sessions SET context = $3::jsonb, last_event_at = NOW()
+         WHERE id = $1 AND user_id = $2 AND status = 'active'
+         RETURNING id`,
+        [sessionId, userId, JSON.stringify(canonicalContext)],
+      );
+      if (!updated.rows[0]) throw new Error('Vibe session was not found or is not owned by this user');
+      await client.query(
+        `UPDATE session_state
+         SET context = COALESCE($3::jsonb->>'activity', $3::jsonb->>'device'),
+             state_vector = state_vector || jsonb_build_object('context', $3::jsonb),
+             last_interaction = NOW()
+         WHERE session_id = $1 AND user_id = $2`,
+        [sessionId, userId, JSON.stringify(canonicalContext)],
+      );
+    });
+  }
+
+  /**
+   * Project unknown-track feedback into the session exploration controls.
+   * This deliberately runs behind its own projection marker: the immutable
+   * event has already committed, so retries after a transient failure are
+   * safe and converge on one evidence row and one coefficient adjustment.
+   */
+  async projectVibeSessionFeedback(event: VibeEvent): Promise<void> {
+    if (!event.track_id || !['completed', 'skipped', 'kept'].includes(event.type)) return;
+    const trackId = event.track_id;
+    await this.withTransaction(async client => {
+      // Old active/resumable sessions predate vibe_session_profiles. Create a
+      // neutral profile before claiming the exactly-once marker: otherwise the
+      // marker could permanently consume a feedback event without adapting its
+      // session. Existing goals are deliberately never overwritten here.
+      await client.query(
+        `INSERT INTO vibe_session_profiles (session_id, user_id)
+         VALUES ($1, $2)
+         ON CONFLICT (session_id) DO NOTHING`,
+        [event.session_id, event.user_id],
+      );
+      const marker = await client.query(
+        `INSERT INTO vibe_session_feedback_projections (event_id)
+         VALUES ($1) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,
+        [event.id],
+      );
+      if (!marker.rows[0]) return;
+
+      // This query occurs before a completed event's play_history projection
+      // can be considered. Favourites and prior evidence count as familiarity
+      // too, avoiding a false “new discovery” on a locally known track.
+      const familiarity = await client.query(
+        `SELECT (
+           EXISTS (SELECT 1 FROM play_history WHERE user_id = $1 AND track_id = $2 AND completed = true AND played_at < $3::timestamptz)
+           OR EXISTS (SELECT 1 FROM favorites WHERE user_id = $1 AND track_id = $2)
+           OR EXISTS (SELECT 1 FROM evidence WHERE user_id = $1 AND entity_type = 'track' AND entity_id = $2 AND created_at < $3::timestamptz)
+         ) AS familiar`,
+        [event.user_id, trackId, event.occurred_at],
+      );
+      const familiar = Boolean(familiarity.rows[0]?.familiar);
+
+      const delta = event.type === 'skipped' ? -0.08 : event.type === 'completed' ? 0.06 : 0.03;
+      const signal = event.type === 'skipped' ? 'skip_quick' : 'play_of_never_seen';
+      const weight = event.type === 'skipped' ? -0.05 : delta;
+      if (!familiar) {
+        await this.recordTrackEvidence({
+          user_id: event.user_id,
+          track_id: trackId,
+          signal,
+          profile: event.type === 'skipped' ? 'negative' : 'discovery',
+          weight,
+          context: { vibe_event_id: event.id, session_id: event.session_id, unfamiliar: true },
+        }, client);
+      }
+
+      const profile = await client.query(
+        `UPDATE vibe_session_profiles
+         SET exploration_coefficient = GREATEST(0, LEAST(1, exploration_coefficient + CASE WHEN $4 THEN 0 ELSE $3 END)),
+             discovery_radius = GREATEST(0.15, LEAST(0.9, 0.2 + (GREATEST(0, LEAST(1, exploration_coefficient + CASE WHEN $4 THEN 0 ELSE $3 END)) * 0.65))),
+             goals = CASE
+               WHEN goals->>'type' = 'familiar' AND $4 AND $5
+                 THEN jsonb_set(goals, '{progress}', to_jsonb(LEAST(COALESCE((goals->>'progress')::int, 0) + 1, COALESCE((goals->>'target')::int, 1))))
+               WHEN goals->>'type' IN ('discovery', 'surprise', 'artist_introduction') AND NOT $4 AND $3 > 0
+                 THEN jsonb_set(goals, '{progress}', to_jsonb(LEAST(COALESCE((goals->>'progress')::int, 0) + 1, COALESCE((goals->>'target')::int, 1))))
+               ELSE goals END,
+             updated_at = NOW()
+         WHERE session_id = $1 AND user_id = $2
+         RETURNING exploration_coefficient, discovery_radius, goals`,
+        [event.session_id, event.user_id, delta, familiar, event.type === 'completed' || event.type === 'kept'],
+      );
+      const row = profile.rows[0] as Pick<VibeSessionProfile, 'exploration_coefficient' | 'discovery_radius' | 'goals'> | undefined;
+      if (row) {
+        await client.query(
+          `UPDATE session_state
+           SET state_vector = state_vector || jsonb_build_object(
+                 'explorationCoefficient', $3::real,
+                 'discoveryRadius', $4::real,
+                 'sessionGoal', $5::jsonb
+               ), last_interaction = NOW()
+           WHERE session_id = $1 AND user_id = $2`,
+          [event.session_id, event.user_id, row.exploration_coefficient, row.discovery_radius, JSON.stringify(row.goals)],
+        );
+      }
+    });
+  }
+
   /**
    * End (or expire/replace) a session without changing its original end time
    * when a client retries the same request.
@@ -1667,6 +1825,10 @@ export class DbService {
     durationMs?: number | null;
     payload?: Record<string, unknown>;
   }): Promise<RecordedVibeEvent> {
+    // This service is also called by jobs and tests which bypass the HTTP
+    // route. Preserve the context privacy boundary at the final point before
+    // an immutable ledger write.
+    const payload = normalizeVibeEventPayload(params.type, params.payload);
     return this.withTransaction(async (client) => {
       // A session-row lock serializes both event writes and terminal state
       // transitions. In particular, it avoids the READ COMMITTED CTE snapshot
@@ -1718,7 +1880,7 @@ export class DbService {
           occurredAt,
           params.positionMs ?? null,
           params.durationMs ?? null,
-          JSON.stringify(params.payload ?? {}),
+          JSON.stringify(payload ?? {}),
         ]
       );
       const event = insertRes.rows[0] as VibeEvent | undefined;
@@ -1727,6 +1889,7 @@ export class DbService {
       }
 
       await this.projectVibeFeedback(event, client);
+      await this.projectVibeContextChanged(event, client);
 
       await client.query(
         `UPDATE vibe_sessions
@@ -1738,6 +1901,30 @@ export class DbService {
     });
   }
 
+  /** Apply the context projection in the same transaction as its *inserted*
+   * ledger event. A client-event retry returns before this method, so its body
+   * can never overwrite session state with a different context. */
+  private async projectVibeContextChanged(event: VibeEvent, client: PoolClient): Promise<void> {
+    if (event.type !== 'context_changed') return;
+    const context = event.payload?.context;
+    if (!context || typeof context !== 'object' || Array.isArray(context)) return;
+    const canonicalContext = context as Record<string, unknown>;
+    await client.query(
+      `UPDATE vibe_sessions
+       SET context = $3::jsonb
+       WHERE id = $1 AND user_id = $2 AND status = 'active'`,
+      [event.session_id, event.user_id, JSON.stringify(canonicalContext)],
+    );
+    await client.query(
+      `UPDATE session_state
+       SET context = COALESCE($3::jsonb->>'activity', $3::jsonb->>'device'),
+           state_vector = state_vector || jsonb_build_object('context', $3::jsonb),
+           last_interaction = NOW()
+       WHERE session_id = $1 AND user_id = $2`,
+      [event.session_id, event.user_id, JSON.stringify(canonicalContext)],
+    );
+  }
+
   /**
    * Materialize Vibe feedback into the listener inputs used by the incumbent
    * director. The projection marker and every write share the event's
@@ -1899,6 +2086,32 @@ export class DbService {
           feedbackEventId: params.objectiveSnapshot.feedbackEventId ?? null,
         })],
       );
+      // Store a compact session shape rather than a replayable queue. It is
+      // overwritten on each revision so a recently adapted session represents
+      // its current direction when tomorrow's session asks for freshness.
+      await client.query(
+        `WITH selected AS (
+           SELECT i.track_id, i.candidate_source
+           FROM vibe_plan_items i WHERE i.plan_version_id = $1
+         ), artists AS (
+           SELECT DISTINCT ta.artist_id::text AS value FROM selected s
+           JOIN track_artists_v2 ta ON ta.track_id = s.track_id AND ta.role = 'main'
+         ), genres AS (
+           SELECT DISTINCT tg.genre_id::text AS value FROM selected s
+           JOIN track_genre tg ON tg.track_id = s.track_id
+         ), sources AS (
+           SELECT DISTINCT candidate_source AS value FROM selected
+         )
+         UPDATE vibe_session_profiles
+         SET fingerprint = jsonb_build_object(
+               'artists', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM artists), '[]'::jsonb),
+               'genres', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM genres), '[]'::jsonb),
+               'sources', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM sources), '[]'::jsonb),
+               'context', (SELECT context FROM vibe_sessions WHERE id = $2)
+             ), updated_at = NOW()
+         WHERE session_id = $2 AND user_id = $3`,
+        [planVersion.id, params.sessionId, params.userId],
+      );
       await client.query(
         `UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`,
         [params.sessionId],