From 515cab2f890c50b44f4c380828a2504508cb4e99 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 22:32:42 +0400 Subject: [PATCH] feat(vibe): persist durable session plans and events --- backend/src/db/migrations.test.ts | 23 ++ backend/src/db/migrations.ts | 64 ++++++ backend/src/db/schema.sql | 62 ++++++ backend/src/db/types.ts | 64 ++++++ backend/src/services/db.service.test.ts | 160 ++++++++++++++ backend/src/services/db.service.ts | 266 ++++++++++++++++++++++++ 6 files changed, 639 insertions(+) diff --git a/backend/src/db/migrations.test.ts b/backend/src/db/migrations.test.ts index edf3bd0..148c336 100644 --- a/backend/src/db/migrations.test.ts +++ b/backend/src/db/migrations.test.ts @@ -20,3 +20,26 @@ describe('track release-date migration', () => { expect(migration!.sql).toContain('WHEN (OLD.release_date IS DISTINCT FROM NEW.release_date)'); }); }); + +describe('Vibe durable session migration', () => { + const migration = MIGRATIONS.find( + ({ id }) => id === '20260801_vibe_session_persistence', + ); + + it('creates the event ledger, retry key, and revisioned plan tables', () => { + expect(migration).toBeDefined(); + expect(migration!.sql).toContain('CREATE TABLE IF NOT EXISTS vibe_sessions'); + expect(migration!.sql).toContain('CREATE TABLE IF NOT EXISTS vibe_events'); + expect(migration!.sql).toContain('UNIQUE (session_id, client_event_id)'); + expect(migration!.sql).toContain('CREATE TABLE IF NOT EXISTS vibe_plan_versions'); + expect(migration!.sql).toContain('UNIQUE (session_id, version)'); + expect(migration!.sql).toContain('CREATE TABLE IF NOT EXISTS vibe_plan_items'); + expect(migration!.sql).toContain('UNIQUE (plan_version_id, track_id)'); + }); + + it('indexes session and event reads along their required time axes', () => { + expect(migration!.sql).toContain('idx_vibe_sessions_user_last_event'); + expect(migration!.sql).toContain('idx_vibe_events_session_occurred'); + expect(migration!.sql).toContain('idx_vibe_events_user_occurred'); + }); +}); diff --git a/backend/src/db/migrations.ts b/backend/src/db/migrations.ts index f0a4bc9..c68d5c8 100644 --- a/backend/src/db/migrations.ts +++ b/backend/src/db/migrations.ts @@ -644,4 +644,68 @@ export const MIGRATIONS: Migration[] = [ EXECUTE FUNCTION propagate_album_release_date_to_tracks(); `, }, + { + // Vibe v2 needs an immutable event ledger and revisioned plans. The + // legacy session_state table remains in place as a derived-state cache so + // existing v2 endpoints can migrate independently. + id: '20260801_vibe_session_persistence', + sql: ` + CREATE TABLE IF NOT EXISTS vibe_sessions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL, + status TEXT NOT NULL CHECK (status IN ('active', 'paused', 'ended', 'expired', 'replaced')), + seed_track_id UUID REFERENCES tracks(id) ON DELETE SET NULL, + context JSONB NOT NULL DEFAULT '{}'::jsonb, + policy_version TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_event_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ended_at TIMESTAMPTZ + ); + CREATE INDEX IF NOT EXISTS idx_vibe_sessions_user_last_event + ON vibe_sessions (user_id, last_event_at DESC); + + CREATE TABLE IF NOT EXISTS vibe_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + client_event_id UUID, + session_id UUID NOT NULL REFERENCES vibe_sessions(id) ON DELETE CASCADE, + user_id UUID NOT NULL, + track_id UUID REFERENCES tracks(id) ON DELETE SET NULL, + type TEXT NOT NULL, + occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + position_ms INTEGER, + duration_ms INTEGER, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + UNIQUE (session_id, client_event_id) + ); + CREATE INDEX IF NOT EXISTS idx_vibe_events_session_occurred + ON vibe_events (session_id, occurred_at); + CREATE INDEX IF NOT EXISTS idx_vibe_events_user_occurred + ON vibe_events (user_id, occurred_at DESC); + + CREATE TABLE IF NOT EXISTS vibe_plan_versions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + session_id UUID NOT NULL REFERENCES vibe_sessions(id) ON DELETE CASCADE, + version INTEGER NOT NULL CHECK (version > 0), + reason TEXT NOT NULL, + state_snapshot JSONB NOT NULL, + objective_snapshot JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (session_id, version) + ); + + CREATE TABLE IF NOT EXISTS vibe_plan_items ( + plan_version_id UUID NOT NULL REFERENCES vibe_plan_versions(id) ON DELETE CASCADE, + ordinal INTEGER NOT NULL CHECK (ordinal >= 0), + track_id UUID NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + slot_role TEXT, + candidate_source TEXT NOT NULL, + score REAL NOT NULL, + score_breakdown JSONB NOT NULL, + explanation JSONB NOT NULL, + committed BOOLEAN NOT NULL DEFAULT false, + PRIMARY KEY (plan_version_id, ordinal), + UNIQUE (plan_version_id, track_id) + ); + `, + }, ]; diff --git a/backend/src/db/schema.sql b/backend/src/db/schema.sql index 6ca5be5..3a48764 100644 --- a/backend/src/db/schema.sql +++ b/backend/src/db/schema.sql @@ -525,6 +525,68 @@ CREATE TABLE IF NOT EXISTS session_state ( CREATE INDEX IF NOT EXISTS idx_session_state_user ON session_state (user_id, last_interaction DESC); +-- Durable Vibe v2 session ledger. session_state remains a rebuildable cache for +-- the existing director; these tables are the authoritative record for the +-- next-generation, versioned planner. +CREATE TABLE IF NOT EXISTS vibe_sessions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL, + status TEXT NOT NULL CHECK (status IN ('active', 'paused', 'ended', 'expired', 'replaced')), + seed_track_id UUID REFERENCES tracks(id) ON DELETE SET NULL, + context JSONB NOT NULL DEFAULT '{}'::jsonb, + policy_version TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_event_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ended_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_vibe_sessions_user_last_event + ON vibe_sessions (user_id, last_event_at DESC); + +CREATE TABLE IF NOT EXISTS vibe_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + client_event_id UUID, + session_id UUID NOT NULL REFERENCES vibe_sessions(id) ON DELETE CASCADE, + user_id UUID NOT NULL, + track_id UUID REFERENCES tracks(id) ON DELETE SET NULL, + type TEXT NOT NULL, + occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + position_ms INTEGER, + duration_ms INTEGER, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + UNIQUE (session_id, client_event_id) +); + +CREATE INDEX IF NOT EXISTS idx_vibe_events_session_occurred + ON vibe_events (session_id, occurred_at); +CREATE INDEX IF NOT EXISTS idx_vibe_events_user_occurred + ON vibe_events (user_id, occurred_at DESC); + +CREATE TABLE IF NOT EXISTS vibe_plan_versions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + session_id UUID NOT NULL REFERENCES vibe_sessions(id) ON DELETE CASCADE, + version INTEGER NOT NULL CHECK (version > 0), + reason TEXT NOT NULL, + state_snapshot JSONB NOT NULL, + objective_snapshot JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (session_id, version) +); + +CREATE TABLE IF NOT EXISTS vibe_plan_items ( + plan_version_id UUID NOT NULL REFERENCES vibe_plan_versions(id) ON DELETE CASCADE, + ordinal INTEGER NOT NULL CHECK (ordinal >= 0), + track_id UUID NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + slot_role TEXT, + candidate_source TEXT NOT NULL, + score REAL NOT NULL, + score_breakdown JSONB NOT NULL, + explanation JSONB NOT NULL, + committed BOOLEAN NOT NULL DEFAULT false, + PRIMARY KEY (plan_version_id, ordinal), + UNIQUE (plan_version_id, track_id) +); + -- Diversity budgets for the session director's planner. CREATE TABLE IF NOT EXISTS diversity_budgets ( user_id UUID NOT NULL, diff --git a/backend/src/db/types.ts b/backend/src/db/types.ts index c45eda7..fc51708 100644 --- a/backend/src/db/types.ts +++ b/backend/src/db/types.ts @@ -194,3 +194,67 @@ export interface RepetitionRule { dimension: string; min_distance: number; } + +// --------------------------------------------------------------------------- +// Vibe v2 durable session ledger +// --------------------------------------------------------------------------- + +export const VIBE_SESSION_STATUSES = ['active', 'paused', 'ended', 'expired', 'replaced'] as const; +export type VibeSessionStatus = (typeof VIBE_SESSION_STATUSES)[number]; + +export interface VibeSession { + id: string; + user_id: string; + status: VibeSessionStatus; + seed_track_id: string | null; + context: Record; + policy_version: string; + started_at: Date; + last_event_at: Date; + ended_at: Date | null; +} + +export interface VibeEvent { + id: string; + client_event_id: string | null; + session_id: string; + user_id: string; + track_id: string | null; + type: string; + occurred_at: Date; + position_ms: number | null; + duration_ms: number | null; + payload: Record; +} + +export interface RecordedVibeEvent { + event: VibeEvent; + /** False when a retried client_event_id returned the original event. */ + inserted: boolean; +} + +export interface VibePlanVersion { + id: string; + session_id: string; + version: number; + reason: string; + state_snapshot: Record; + objective_snapshot: Record; + created_at: Date; +} + +export interface VibePlanItem { + plan_version_id: string; + ordinal: number; + track_id: string; + slot_role: string | null; + candidate_source: string; + score: number; + score_breakdown: Record; + explanation: unknown; + committed: boolean; +} + +export interface VibePlan extends VibePlanVersion { + items: VibePlanItem[]; +} diff --git a/backend/src/services/db.service.test.ts b/backend/src/services/db.service.test.ts index 1eff997..6bd5595 100644 --- a/backend/src/services/db.service.test.ts +++ b/backend/src/services/db.service.test.ts @@ -7,7 +7,167 @@ function makeService(): { service: DbService; mockQuery: ReturnType; clientQuery: ReturnType } { + const poolQuery = vi.fn(); + const clientQuery = vi.fn(); + const service = new DbService({ + query: poolQuery, + connect: vi.fn().mockResolvedValue({ query: clientQuery, release: vi.fn() }), + } as any); + return { service, poolQuery, clientQuery }; +} + describe('DbService v2 methods', () => { + describe('durable Vibe sessions', () => { + it('creates, reads, and ends sessions scoped to their user', async () => { + const { service, mockQuery } = makeService(); + const session = { + id: 'session-1', user_id: 'user-1', status: 'active', seed_track_id: null, + context: { activity: 'focus' }, policy_version: 'v2.1', + }; + mockQuery + .mockResolvedValueOnce({ rows: [session] }) + .mockResolvedValueOnce({ rows: [session] }) + .mockResolvedValueOnce({ rows: [{ ...session, status: 'ended' }] }); + + await expect(service.createVibeSession({ + userId: 'user-1', policyVersion: 'v2.1', context: { activity: 'focus' }, + })).resolves.toEqual(session); + await expect(service.getVibeSession('session-1', 'user-1')).resolves.toEqual(session); + await expect(service.endVibeSession('session-1', 'user-1')).resolves.toMatchObject({ status: 'ended' }); + + expect(mockQuery.mock.calls[0][0]).toContain('INSERT INTO vibe_sessions'); + expect(mockQuery.mock.calls[0][1]).toEqual(['user-1', null, JSON.stringify({ activity: 'focus' }), 'v2.1']); + expect(mockQuery.mock.calls[1][0]).toContain('id = $1 AND user_id = $2'); + expect(mockQuery.mock.calls[2][0]).toContain('COALESCE(ended_at, NOW())'); + expect(mockQuery.mock.calls[2][0]).toContain('CASE WHEN ended_at IS NULL THEN $3 ELSE status END'); + }); + + it('records retry-safe events and reports whether the event was inserted', async () => { + const { service, clientQuery } = makeTransactionalService(); + const event = { + id: 'event-1', client_event_id: 'client-event-1', session_id: 'session-1', + user_id: 'user-1', track_id: 'track-1', type: 'skipped', occurred_at: new Date(), + position_ms: 1_500, duration_ms: 10_000, payload: { reason: 'next' }, + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) // lock session + .mockResolvedValueOnce({ rows: [event] }) // existing retry + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + const result = await service.recordVibeEvent({ + sessionId: 'session-1', userId: 'user-1', clientEventId: 'client-event-1', + trackId: 'track-1', type: 'skipped', positionMs: 1_500, durationMs: 10_000, + payload: { reason: 'next' }, + }); + + expect(result.inserted).toBe(false); + expect(result.event.id).toBe('event-1'); + const [sql, values] = clientQuery.mock.calls[1]; + expect(sql).toContain('FOR UPDATE'); + expect(values).toEqual([ + 'session-1', 'user-1', + ]); + expect(clientQuery.mock.calls).toHaveLength(4); + expect(clientQuery.mock.calls.map(([query]) => query)).not.toContain( + expect.stringContaining('UPDATE vibe_sessions') + ); + }); + + it('rejects an event when no owned session is returned', async () => { + const { service, clientQuery } = makeTransactionalService(); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // session lookup + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK + await expect(service.recordVibeEvent({ + sessionId: 'session-1', userId: 'other-user', type: 'completed', + })).rejects.toThrow('not found or is not owned'); + }); + + it('rejects new events for terminal sessions but returns an existing idempotent retry', async () => { + const { service, clientQuery } = makeTransactionalService(); + const event = { + id: 'event-1', client_event_id: 'client-event-1', session_id: 'session-1', + user_id: 'user-1', track_id: null, type: 'completed', occurred_at: new Date(), + position_ms: null, duration_ms: null, payload: {}, + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN: idempotent retry + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'ended' }] }) + .mockResolvedValueOnce({ rows: [event] }) + .mockResolvedValueOnce({ rows: [] }) // COMMIT + .mockResolvedValueOnce({ rows: [] }) // BEGIN: new event + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'ended' }] }) + .mockResolvedValueOnce({ rows: [] }) // no matching retry + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK + + await expect(service.recordVibeEvent({ + sessionId: 'session-1', userId: 'user-1', clientEventId: 'client-event-1', type: 'completed', + })).resolves.toEqual({ event, inserted: false }); + await expect(service.recordVibeEvent({ + sessionId: 'session-1', userId: 'user-1', clientEventId: 'new-event-1', type: 'completed', + })).rejects.toThrow('Cannot record a new event for ended Vibe session'); + + expect(clientQuery.mock.calls.map(([query]) => query)).not.toContain( + expect.stringContaining('INSERT INTO vibe_events') + ); + }); + }); + + describe('durable Vibe plans', () => { + it('writes a header and all items in one transaction', async () => { + const { service, clientQuery } = makeTransactionalService(); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ + id: 'plan-1', session_id: 'session-1', version: 1, reason: 'session_started', + state_snapshot: { energy: 0.5 }, objective_snapshot: { freshness: 0.4 }, created_at: new Date(), + }] }) + .mockResolvedValueOnce({ rows: [] }) // item + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + const plan = await service.persistVibePlan({ + sessionId: 'session-1', userId: 'user-1', version: 1, reason: 'session_started', + stateSnapshot: { energy: 0.5 }, objectiveSnapshot: { freshness: 0.4 }, + items: [{ + ordinal: 0, track_id: 'track-1', slot_role: 'anchor', candidate_source: 'comfort', + score: 0.91, score_breakdown: { affinity: 0.8 }, explanation: [{ because: 'favourite' }], committed: true, + }], + }); + + expect(plan.items[0].plan_version_id).toBe('plan-1'); + expect(clientQuery.mock.calls[1][0]).toContain('INSERT INTO vibe_plan_versions'); + expect(clientQuery.mock.calls[2][0]).toContain('INSERT INTO vibe_plan_items'); + expect(clientQuery.mock.calls[3][0]).toBe('COMMIT'); + }); + + it('reads the latest revision and reconstructs ordered plan items', async () => { + const { service, mockQuery } = makeService(); + mockQuery.mockResolvedValue({ rows: [{ + id: 'plan-2', session_id: 'session-1', version: 2, reason: 'feedback', + state_snapshot: { energy: 0.7 }, objective_snapshot: { freshness: 0.5 }, created_at: new Date(), + item_plan_version_id: 'plan-2', ordinal: 0, track_id: 'track-2', slot_role: 'next', + candidate_source: 'adjacent', score: 0.8, score_breakdown: { transition: 0.7 }, + explanation: [{ because: 'similar artist' }], committed: true, + }, { + id: 'plan-2', session_id: 'session-1', version: 2, reason: 'feedback', + state_snapshot: { energy: 0.7 }, objective_snapshot: { freshness: 0.5 }, created_at: new Date(), + item_plan_version_id: 'plan-2', ordinal: 1, track_id: 'track-3', slot_role: null, + candidate_source: 'discovery', score: 0.6, score_breakdown: { novelty: 0.5 }, + explanation: [], committed: false, + }] }); + + const plan = await service.getVibePlan('session-1', 'user-1'); + + expect(plan?.version).toBe(2); + expect(plan?.items.map((item) => item.track_id)).toEqual(['track-2', 'track-3']); + expect(mockQuery.mock.calls[0][0]).toContain('SELECT MAX(version) FROM vibe_plan_versions'); + expect(mockQuery.mock.calls[0][1]).toEqual(['session-1', 'user-1', null]); + }); + }); + describe('createAlbum', () => { it('persists the canonical release date instead of discarding it', async () => { const { service, mockQuery } = makeService(); diff --git a/backend/src/services/db.service.ts b/backend/src/services/db.service.ts index e987ce0..85cf023 100644 --- a/backend/src/services/db.service.ts +++ b/backend/src/services/db.service.ts @@ -10,6 +10,18 @@ import { SearchService } from './search.service.js'; /** Anything with a `.query()` — either the shared Pool or a checked-out client. */ type Queryable = Pool | PoolClient; +type VibePlanVersionRow = Omit & { + item_plan_version_id: string | null; + ordinal: number | null; + track_id: string | null; + slot_role: string | null; + candidate_source: string | null; + score: number | null; + score_breakdown: Record | null; + explanation: unknown | null; + committed: boolean | null; +}; + import { MIGRATIONS } from '../db/migrations.js'; import { allowedFields } from '../db/updatable-columns.js'; @@ -33,6 +45,12 @@ import type { SessionState, DiversityBudget, RepetitionRule, + VibeSession, + VibeSessionStatus, + VibeEvent, + RecordedVibeEvent, + VibePlan, + VibePlanItem, } from '../db/types.js'; import { AUDIO_PREFERENCE_BUCKETS, FEEDBACK_ACTIONS } from '../db/types.js'; export * from '../db/types.js'; @@ -1479,6 +1497,254 @@ export class DbService { return (res.rows[0] as SessionState) || null; } + /** + * Create the authoritative Vibe v2 session record. This intentionally does + * not create a legacy session_state row: callers can migrate to the durable + * ledger without changing the existing v2 endpoint contract first. + */ + async createVibeSession(params: { + userId: string; + policyVersion: string; + seedTrackId?: string | null; + context?: Record; + }): Promise { + const res = await this.pgClient.query( + `INSERT INTO vibe_sessions (user_id, status, seed_track_id, context, policy_version) + VALUES ($1, 'active', $2, $3::jsonb, $4) + RETURNING *`, + [ + params.userId, + params.seedTrackId ?? null, + JSON.stringify(params.context ?? {}), + params.policyVersion, + ] + ); + return res.rows[0] as VibeSession; + } + + /** Fetch a Vibe session only when it belongs to the requesting user. */ + async getVibeSession(sessionId: string, userId: string): Promise { + const res = await this.pgClient.query( + 'SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2', + [sessionId, userId] + ); + return (res.rows[0] as VibeSession) ?? null; + } + + /** + * End (or expire/replace) a session without changing its original end time + * when a client retries the same request. + */ + async endVibeSession( + sessionId: string, + userId: string, + status: Extract = 'ended' + ): Promise { + const res = await this.pgClient.query( + `UPDATE vibe_sessions + SET status = CASE WHEN ended_at IS NULL THEN $3 ELSE status END, + ended_at = COALESCE(ended_at, NOW()), + last_event_at = CASE WHEN ended_at IS NULL THEN NOW() ELSE last_event_at END + WHERE id = $1 AND user_id = $2 + RETURNING *`, + [sessionId, userId, status] + ); + return (res.rows[0] as VibeSession) ?? null; + } + + /** + * Append an immutable Vibe event. A supplied clientEventId is idempotent per + * session: a retry returns the original event and does not advance the + * session timestamp a second time. A missing id deliberately means a new, + * server-originated event. + */ + async recordVibeEvent(params: { + sessionId: string; + userId: string; + type: string; + clientEventId?: string | null; + trackId?: string | null; + occurredAt?: Date; + positionMs?: number | null; + durationMs?: number | null; + payload?: Record; + }): Promise { + 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 + // race where ON CONFLICT observes a concurrent event but a later CTE + // cannot yet read it. The duplicate lookup happens after the lock, so an + // idempotent retry remains valid even after the session has ended. + const sessionRes = await client.query( + `SELECT id, status + FROM vibe_sessions + WHERE id = $1 AND user_id = $2 + FOR UPDATE`, + [params.sessionId, params.userId] + ); + const session = sessionRes.rows[0] as Pick | undefined; + if (!session) { + throw new Error('Vibe session was not found or is not owned by this user'); + } + + if (params.clientEventId) { + const existingRes = await client.query( + `SELECT * + FROM vibe_events + WHERE session_id = $1 AND client_event_id = $2::uuid`, + [params.sessionId, params.clientEventId] + ); + const existing = existingRes.rows[0] as VibeEvent | undefined; + if (existing) { + return { event: existing, inserted: false }; + } + } + + if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') { + throw new Error(`Cannot record a new event for ${session.status} Vibe session`); + } + + const occurredAt = params.occurredAt?.toISOString() ?? null; + const insertRes = await client.query( + `INSERT INTO vibe_events + (client_event_id, session_id, user_id, track_id, type, occurred_at, position_ms, duration_ms, payload) + VALUES ($1::uuid, $2, $3, $4::uuid, $5, COALESCE($6::timestamptz, NOW()), $7, $8, $9::jsonb) + RETURNING *`, + [ + params.clientEventId ?? null, + params.sessionId, + params.userId, + params.trackId ?? null, + params.type, + occurredAt, + params.positionMs ?? null, + params.durationMs ?? null, + JSON.stringify(params.payload ?? {}), + ] + ); + const event = insertRes.rows[0] as VibeEvent | undefined; + if (!event) { + throw new Error('Vibe event could not be recorded'); + } + + await client.query( + `UPDATE vibe_sessions + SET last_event_at = GREATEST(last_event_at, $2::timestamptz) + WHERE id = $1`, + [params.sessionId, event.occurred_at] + ); + return { event, inserted: true }; + }); + } + + /** + * Persist one complete revision of a session plan atomically. The caller + * supplies the monotonically increasing version; session-level scheduling + * will own version allocation when the director is migrated to this ledger. + */ + async persistVibePlan(params: { + sessionId: string; + userId: string; + version: number; + reason: string; + stateSnapshot: Record; + objectiveSnapshot: Record; + items: Array>; + }): Promise { + return this.withTransaction(async (client) => { + const header = await client.query( + `INSERT INTO vibe_plan_versions + (session_id, version, reason, state_snapshot, objective_snapshot) + SELECT s.id, $3, $4, $5::jsonb, $6::jsonb + FROM vibe_sessions s + WHERE s.id = $1 AND s.user_id = $2 + RETURNING *`, + [ + params.sessionId, + params.userId, + params.version, + params.reason, + JSON.stringify(params.stateSnapshot), + JSON.stringify(params.objectiveSnapshot), + ] + ); + const planVersion = header.rows[0] as VibePlan | undefined; + if (!planVersion) { + throw new Error('Vibe session was not found or is not owned by this user'); + } + + for (const item of params.items) { + await client.query( + `INSERT INTO vibe_plan_items + (plan_version_id, ordinal, track_id, slot_role, candidate_source, score, score_breakdown, explanation, committed) + VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9)`, + [ + planVersion.id, + item.ordinal, + item.track_id, + item.slot_role, + item.candidate_source, + item.score, + JSON.stringify(item.score_breakdown), + JSON.stringify(item.explanation), + item.committed, + ] + ); + } + + return { ...planVersion, items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })) }; + }); + } + + /** Read a specific plan revision, or the latest revision for a session. */ + async getVibePlan(sessionId: string, userId: string, version?: number): Promise { + const res = await this.pgClient.query( + `SELECT p.*, i.plan_version_id AS item_plan_version_id, i.ordinal, i.track_id, + i.slot_role, i.candidate_source, i.score, i.score_breakdown, + i.explanation, i.committed + FROM vibe_plan_versions p + JOIN vibe_sessions s ON s.id = p.session_id + LEFT JOIN vibe_plan_items i ON i.plan_version_id = p.id + WHERE p.session_id = $1 AND s.user_id = $2 + AND ( + ($3::integer IS NOT NULL AND p.version = $3) + OR ($3::integer IS NULL AND p.version = ( + SELECT MAX(version) FROM vibe_plan_versions WHERE session_id = $1 + )) + ) + ORDER BY i.ordinal ASC`, + [sessionId, userId, version ?? null] + ); + if (!res.rows[0]) return null; + + const first = res.rows[0] as VibePlanVersionRow; + const plan: VibePlan = { + id: first.id, + session_id: first.session_id, + version: first.version, + reason: first.reason, + state_snapshot: first.state_snapshot, + objective_snapshot: first.objective_snapshot, + created_at: first.created_at, + items: [], + }; + for (const row of res.rows as VibePlanVersionRow[]) { + if (!row.item_plan_version_id) continue; + plan.items.push({ + plan_version_id: row.item_plan_version_id, + ordinal: row.ordinal!, + track_id: row.track_id!, + slot_role: row.slot_role, + candidate_source: row.candidate_source!, + score: row.score!, + score_breakdown: row.score_breakdown!, + explanation: row.explanation, + committed: row.committed!, + }); + } + return plan; + } + /** * Upsert a diversity budget for a user. */