diff --git a/backend/src/app.ts b/backend/src/app.ts index 081a0fa..42ecdb0 100644 --- a/backend/src/app.ts +++ b/backend/src/app.ts @@ -15,8 +15,10 @@ import settingsRoutes from './routes/settings.routes.js'; import graphRoutes from './routes/graph.routes.js'; import { SessionDirector } from './services/session-director.service.js'; import v2Routes from './routes/v2.routes.js'; +import vibeSessionsRoutes from './routes/vibe-sessions.routes.js'; import discoveryRoutes from './routes/discovery.routes.js'; import imagesRoutes from './routes/images.routes.js'; +import { VibeSessionCoordinator } from './services/vibe-session-coordinator.service.js'; export interface AppConfig { port: number; @@ -158,8 +160,18 @@ export async function buildApp(config: AppConfig) { fastify.register(graphRoutes, { prefix: '/api', dbService }); const sessionDirector = new SessionDirector(dbService); + const vibeSessionCoordinator = new VibeSessionCoordinator(dbService, sessionDirector); + // Durable sessions intentionally do not trust x-user-id. A self-hosted + // deployment may configure one owner through MUZICK_VIBE_USER_ID today; + // an authenticated deployment can replace this resolver at registration. + const vibeOwnerId = process.env.MUZICK_VIBE_USER_ID?.trim() || null; fastify.register(v2Routes, { prefix: '/api', dbService, sessionDirector }); + fastify.register(vibeSessionsRoutes, { + prefix: '/api', + coordinator: vibeSessionCoordinator, + identityResolver: () => vibeOwnerId, + }); fastify.register(discoveryRoutes, { prefix: '/api', dbService, jobService }); // ponytail: /api/test/enqueue-job (manual job-enqueue test endpoint) removed — // nothing in the deployed app or its tests called it, and deployment never diff --git a/backend/src/db/migrations.ts b/backend/src/db/migrations.ts index c68d5c8..ca57ab3 100644 --- a/backend/src/db/migrations.ts +++ b/backend/src/db/migrations.ts @@ -708,4 +708,16 @@ export const MIGRATIONS: Migration[] = [ ); `, }, + { + // A material Vibe event is projected into the legacy listener inputs in + // the same transaction as its ledger write. This marker makes that bridge + // auditable and exactly-once even when a client retries an event id. + id: '20260801_vibe_event_projections', + sql: ` + CREATE TABLE IF NOT EXISTS vibe_event_projections ( + event_id UUID PRIMARY KEY REFERENCES vibe_events(id) ON DELETE CASCADE, + projected_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + `, + }, ]; diff --git a/backend/src/db/schema.sql b/backend/src/db/schema.sql index 3a48764..105936a 100644 --- a/backend/src/db/schema.sql +++ b/backend/src/db/schema.sql @@ -562,6 +562,14 @@ CREATE INDEX IF NOT EXISTS idx_vibe_events_session_occurred CREATE INDEX IF NOT EXISTS idx_vibe_events_user_occurred ON vibe_events (user_id, occurred_at DESC); +-- Exactly-once projection marker for material Vibe feedback. The immutable +-- event remains authoritative; this row proves its effect was applied to the +-- listener inputs without double-counting an idempotent client retry. +CREATE TABLE IF NOT EXISTS vibe_event_projections ( + event_id UUID PRIMARY KEY REFERENCES vibe_events(id) ON DELETE CASCADE, + projected_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + 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, diff --git a/backend/src/routes/vibe-sessions.routes.test.ts b/backend/src/routes/vibe-sessions.routes.test.ts new file mode 100644 index 0000000..0d03bf2 --- /dev/null +++ b/backend/src/routes/vibe-sessions.routes.test.ts @@ -0,0 +1,136 @@ +import Fastify from 'fastify'; +import { describe, expect, it, vi } from 'vitest'; +import vibeSessionsRoutes, { VibeIdentityResolver } from './vibe-sessions.routes.js'; +import { VibeSessionLifecycleError } from '../services/vibe-session-coordinator.service.js'; + +const SESSION_ID = '11111111-1111-4111-8111-111111111111'; +const TRACK_ID = '22222222-2222-4222-8222-222222222222'; +const USER_ID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; + +function response() { + return { + sessionId: SESSION_ID, planVersion: 1, now: null, preview: [], state: {}, + replanned: false, replanReason: null, + session: { id: SESSION_ID, status: 'active' }, + }; +} + +async function appWithCoordinator(identityResolver: VibeIdentityResolver = () => USER_ID) { + const coordinator = { + start: vi.fn().mockResolvedValue(response()), + getPlan: vi.fn().mockResolvedValue(response()), + appendEvent: vi.fn().mockResolvedValue({ ...response(), event: { id: 'event-1' }, idempotent: false }), + end: vi.fn().mockResolvedValue(response()), + serveNext: vi.fn().mockResolvedValue(response()), + } as any; + const app = Fastify(); + await app.register(vibeSessionsRoutes, { coordinator, identityResolver }); + await app.ready(); + return { app, coordinator }; +} + +describe('durable Vibe session routes', () => { + it('uses the trusted identity resolver and never accepts a spoofed x-user-id header', async () => { + const { app, coordinator } = await appWithCoordinator(); + const result = await app.inject({ method: 'POST', url: '/v2/vibe/sessions', payload: {} }); + + expect(result.statusCode).toBe(201); + expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.any(Object)); + await app.close(); + }); + + it('rejects requests when no trusted identity is configured instead of defaulting a shared user', async () => { + const { app, coordinator } = await appWithCoordinator(() => null); + const result = await app.inject({ + method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': USER_ID }, payload: {}, + }); + expect(result.statusCode).toBe(401); + expect(coordinator.start).not.toHaveBeenCalled(); + await app.close(); + }); + + it('creates a session and validates event payloads before touching the coordinator', async () => { + const { app, coordinator } = await appWithCoordinator(); + const created = await app.inject({ + method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': 'spoofed' }, + payload: { seedTrackId: TRACK_ID, context: { activity: 'focus' }, policyVersion: 'test-policy' }, + }); + const invalidEvent = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`, headers: { 'x-user-id': 'spoofed' }, + payload: { type: 'definitely-not-an-event' }, + }); + + expect(created.statusCode).toBe(201); + expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ policyVersion: 'test-policy' })); + expect(invalidEvent.statusCode).toBe(400); + expect(coordinator.appendEvent).not.toHaveBeenCalled(); + await app.close(); + }); + + it('returns a lifecycle conflict when an initial plan race ends or replaces the session', async () => { + const { app, coordinator } = await appWithCoordinator(); + coordinator.start.mockRejectedValueOnce( + new VibeSessionLifecycleError('Cannot publish a plan for ended Vibe session'), + ); + + const result = await app.inject({ method: 'POST', url: '/v2/vibe/sessions', payload: {} }); + + expect(result.statusCode).toBe(409); + expect(result.json()).toEqual({ + error: 'Cannot publish a plan for ended Vibe session', + code: 'VIBE_SESSION_NOT_ACTIVE', + }); + await app.close(); + }); + + it('passes a requested plan revision and validated idempotent event through to the coordinator', async () => { + const { app, coordinator } = await appWithCoordinator(); + const plan = await app.inject({ + method: 'GET', url: `/v2/vibe/sessions/${SESSION_ID}/plans?version=2`, headers: { 'x-user-id': 'spoofed' }, + }); + const event = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`, headers: { 'x-user-id': 'spoofed' }, + payload: { eventId: '33333333-3333-4333-8333-333333333333', type: 'completed', trackId: TRACK_ID, positionMs: 5_000 }, + }); + + expect(plan.statusCode).toBe(200); + expect(coordinator.getPlan).toHaveBeenCalledWith(USER_ID, SESSION_ID, 2); + expect(event.statusCode).toBe(200); + expect(coordinator.appendEvent).toHaveBeenCalledWith(USER_ID, SESSION_ID, expect.objectContaining({ type: 'completed', positionMs: 5_000 })); + await app.close(); + }); + + it('validates occurredAt and exposes owned resume and next operations', async () => { + const { app, coordinator } = await appWithCoordinator(); + const resume = await app.inject({ + method: 'POST', url: '/v2/vibe/sessions', payload: { resumeSessionId: SESSION_ID }, + }); + const badTime = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`, + payload: { type: 'completed', occurredAt: 'not-a-date' }, + }); + const next = await app.inject({ method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next` }); + expect(resume.statusCode).toBe(201); + expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ resumeSessionId: SESSION_ID })); + expect(badTime.statusCode).toBe(400); + expect(next.statusCode).toBe(200); + expect(coordinator.serveNext).toHaveBeenCalledWith(USER_ID, SESSION_ID); + await app.close(); + }); + + it('passes a version-aware next request through and rejects an invalid expected version', async () => { + const { app, coordinator } = await appWithCoordinator(); + const valid = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`, payload: { expectedPlanVersion: 2 }, + }); + const invalid = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`, payload: { expectedPlanVersion: 0 }, + }); + + expect(valid.statusCode).toBe(200); + expect(coordinator.serveNext).toHaveBeenCalledWith(USER_ID, SESSION_ID, 2); + expect(invalid.statusCode).toBe(400); + expect(coordinator.serveNext).toHaveBeenCalledTimes(1); + await app.close(); + }); +}); diff --git a/backend/src/routes/vibe-sessions.routes.ts b/backend/src/routes/vibe-sessions.routes.ts new file mode 100644 index 0000000..44762e9 --- /dev/null +++ b/backend/src/routes/vibe-sessions.routes.ts @@ -0,0 +1,175 @@ +import { FastifyInstance, FastifyRequest } from 'fastify'; +import { + VIBE_EVENT_TYPES, + VibeSessionCoordinator, + VibeSessionLifecycleError, + VibeSessionNotFoundError, + VibePlanNotFoundError, +} from '../services/vibe-session-coordinator.service.js'; + +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +/** Identity must come from authenticated server configuration/middleware, never a client header. */ +export type VibeIdentityResolver = (request: FastifyRequest) => string | null; + +function isObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function validUuid(value: unknown): value is string { + return typeof value === 'string' && UUID_RE.test(value); +} + +function validOccurredAt(value: unknown): value is string { + // Require an actual offset-bearing timestamp, rather than Date.parse's + // permissive inputs such as "2026" or locale-dependent strings. + return typeof value === 'string' + && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,3})?(?:Z|[+-]\d{2}:\d{2})$/i.test(value) + && !Number.isNaN(Date.parse(value)); +} + +export default async function vibeSessionsRoutes( + fastify: FastifyInstance, + options: { coordinator: VibeSessionCoordinator; identityResolver: VibeIdentityResolver }, +) { + const { coordinator, identityResolver } = options; + const requireUser = (request: FastifyRequest, reply: { code: (statusCode: number) => { send: (payload: unknown) => unknown } }): string | null => { + const userId = identityResolver(request); + if (userId && validUuid(userId)) return userId; + reply.code(401).send({ error: 'A trusted Vibe identity is required' }); + return null; + }; + + fastify.post('/v2/vibe/sessions', async (request, reply) => { + const userId = requireUser(request, reply); + if (!userId) return; + const body = isObject(request.body) ? request.body : {}; + if (body.resumeSessionId !== undefined && !validUuid(body.resumeSessionId)) { + return reply.code(400).send({ error: 'resumeSessionId must be a UUID' }); + } + if (body.resumeSessionId !== undefined && body.seedTrackId !== undefined) { + return reply.code(400).send({ error: 'resumeSessionId cannot be combined with seedTrackId' }); + } + if (body.seedTrackId !== undefined && !validUuid(body.seedTrackId)) { + return reply.code(400).send({ error: 'seedTrackId must be a UUID' }); + } + if (body.context !== undefined && !isObject(body.context)) { + return reply.code(400).send({ error: 'context must be an object' }); + } + if (body.intent !== undefined && typeof body.intent !== 'string') { + return reply.code(400).send({ error: 'intent must be a string' }); + } + if (body.policyVersion !== undefined && (typeof body.policyVersion !== 'string' || !body.policyVersion.trim())) { + return reply.code(400).send({ error: 'policyVersion must be a non-empty string' }); + } + try { + return reply.code(201).send(await coordinator.start(userId, { + seedTrackId: body.seedTrackId as string | undefined, + context: body.context as Record | undefined, + intent: body.intent as string | undefined, + policyVersion: body.policyVersion as string | undefined, + resumeSessionId: body.resumeSessionId as string | undefined, + })); + } catch (error) { + return sendCoordinatorError(reply, error); + } + }); + + fastify.get('/v2/vibe/sessions/:sessionId/plans', async (request, reply) => { + const userId = requireUser(request, reply); + if (!userId) return; + const { sessionId } = request.params as { sessionId: string }; + const { version } = request.query as { version?: string }; + if (!validUuid(sessionId)) return reply.code(400).send({ error: 'sessionId must be a UUID' }); + const parsedVersion = version === undefined ? undefined : Number(version); + if (version !== undefined && (!Number.isInteger(parsedVersion) || parsedVersion! < 1)) { + return reply.code(400).send({ error: 'version must be a positive integer' }); + } + try { + return reply.send(await coordinator.getPlan(userId, sessionId, parsedVersion)); + } catch (error) { + return sendCoordinatorError(reply, error); + } + }); + + fastify.post('/v2/vibe/sessions/:sessionId/events', async (request, reply) => { + const userId = requireUser(request, reply); + if (!userId) return; + const { sessionId } = request.params as { sessionId: string }; + const body = isObject(request.body) ? request.body : null; + if (!validUuid(sessionId)) return reply.code(400).send({ error: 'sessionId must be a UUID' }); + if (!body || !VIBE_EVENT_TYPES.includes(body.type as typeof VIBE_EVENT_TYPES[number])) { + return reply.code(400).send({ error: 'type must be a supported Vibe event type' }); + } + if (body.eventId !== undefined && !validUuid(body.eventId)) { + return reply.code(400).send({ error: 'eventId must be a UUID' }); + } + if (body.trackId !== undefined && !validUuid(body.trackId)) { + return reply.code(400).send({ error: 'trackId must be a UUID' }); + } + if (body.occurredAt !== undefined && !validOccurredAt(body.occurredAt)) { + return reply.code(400).send({ error: 'occurredAt must be an ISO-8601 timestamp' }); + } + if (body.positionMs !== undefined && (!Number.isInteger(body.positionMs) || (body.positionMs as number) < 0)) { + return reply.code(400).send({ error: 'positionMs must be a non-negative integer' }); + } + if (body.durationMs !== undefined && (!Number.isInteger(body.durationMs) || (body.durationMs as number) < 0)) { + return reply.code(400).send({ error: 'durationMs must be a non-negative integer' }); + } + if (body.payload !== undefined && !isObject(body.payload)) { + return reply.code(400).send({ error: 'payload must be an object' }); + } + try { + return reply.send(await coordinator.appendEvent(userId, sessionId, { + eventId: body.eventId as string | undefined, + type: body.type as typeof VIBE_EVENT_TYPES[number], + trackId: body.trackId as string | undefined, + occurredAt: body.occurredAt === undefined ? undefined : new Date(body.occurredAt as string), + positionMs: body.positionMs as number | undefined, + durationMs: body.durationMs as number | undefined, + payload: body.payload as Record | undefined, + })); + } catch (error) { + return sendCoordinatorError(reply, error); + } + }); + + fastify.post('/v2/vibe/sessions/:sessionId/end', async (request, reply) => { + const userId = requireUser(request, reply); + if (!userId) return; + const { sessionId } = request.params as { sessionId: string }; + if (!validUuid(sessionId)) return reply.code(400).send({ error: 'sessionId must be a UUID' }); + try { + return reply.send(await coordinator.end(userId, sessionId)); + } catch (error) { + return sendCoordinatorError(reply, error); + } + }); + + fastify.post('/v2/vibe/sessions/:sessionId/next', async (request, reply) => { + const userId = requireUser(request, reply); + if (!userId) return; + const { sessionId } = request.params as { sessionId: string }; + const body = isObject(request.body) ? request.body : {}; + if (!validUuid(sessionId)) return reply.code(400).send({ error: 'sessionId must be a UUID' }); + if (body.expectedPlanVersion !== undefined + && (!Number.isInteger(body.expectedPlanVersion) || (body.expectedPlanVersion as number) < 1)) { + return reply.code(400).send({ error: 'expectedPlanVersion must be a positive integer' }); + } + try { + const expectedPlanVersion = body.expectedPlanVersion as number | undefined; + return reply.send(expectedPlanVersion === undefined + ? await coordinator.serveNext(userId, sessionId) + : await coordinator.serveNext(userId, sessionId, expectedPlanVersion)); + } catch (error) { + return sendCoordinatorError(reply, error); + } + }); +} + +function sendCoordinatorError(reply: { code: (statusCode: number) => { send: (payload: unknown) => unknown } }, error: unknown) { + if (error instanceof VibeSessionNotFoundError) return reply.code(404).send({ error: error.message }); + if (error instanceof VibePlanNotFoundError) return reply.code(404).send({ error: error.message }); + if (error instanceof VibeSessionLifecycleError) return reply.code(409).send({ error: error.message, code: 'VIBE_SESSION_NOT_ACTIVE' }); + throw error; +} diff --git a/backend/src/services/db.service.test.ts b/backend/src/services/db.service.test.ts index 6bd5595..c660eeb 100644 --- a/backend/src/services/db.service.test.ts +++ b/backend/src/services/db.service.test.ts @@ -20,13 +20,18 @@ function makeTransactionalService(): { service: DbService; poolQuery: ReturnType describe('DbService v2 methods', () => { describe('durable Vibe sessions', () => { it('creates, reads, and ends sessions scoped to their user', async () => { - const { service, mockQuery } = makeService(); + const { service, poolQuery, clientQuery } = makeTransactionalService(); 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] }) + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // user advisory lock + .mockResolvedValueOnce({ rows: [] }) // active-session lock + .mockResolvedValueOnce({ rows: [session] }) // insert + .mockResolvedValueOnce({ rows: [] }); // COMMIT + poolQuery .mockResolvedValueOnce({ rows: [session] }) .mockResolvedValueOnce({ rows: [{ ...session, status: 'ended' }] }); @@ -36,11 +41,32 @@ describe('DbService v2 methods', () => { 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'); + expect(clientQuery.mock.calls[1][0]).toContain('pg_advisory_xact_lock'); + expect(clientQuery.mock.calls[2][0]).toContain("status = 'active' FOR UPDATE"); + expect(clientQuery.mock.calls[3][0]).toContain('INSERT INTO vibe_sessions'); + expect(clientQuery.mock.calls[3][1]).toEqual(['user-1', null, JSON.stringify({ activity: 'focus' }), 'v2.1']); + expect(poolQuery.mock.calls[0][0]).toContain('id = $1 AND user_id = $2'); + expect(poolQuery.mock.calls[1][0]).toContain('COALESCE(ended_at, NOW())'); + expect(poolQuery.mock.calls[1][0]).toContain('CASE WHEN ended_at IS NULL THEN $3 ELSE status END'); + }); + + it('replaces an owned active session and writes its terminal event before starting another', async () => { + const { service, clientQuery } = makeTransactionalService(); + const replacement = { id: 'session-2', user_id: 'user-1', status: 'active' }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // user advisory lock + .mockResolvedValueOnce({ rows: [{ id: 'session-1' }] }) // active lock + .mockResolvedValueOnce({ rows: [{ id: 'session-1' }] }) // replace + .mockResolvedValueOnce({ rows: [] }) // terminal event + .mockResolvedValueOnce({ rows: [replacement] }) // new session + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.createVibeSession({ userId: 'user-1', policyVersion: 'v2.1' })) + .resolves.toEqual(replacement); + expect(clientQuery.mock.calls[3][0]).toContain("status = 'replaced'"); + expect(clientQuery.mock.calls[4][0]).toContain("'session_ended'"); + expect(clientQuery.mock.calls[5][0]).toContain('INSERT INTO vibe_sessions'); }); it('records retry-safe events and reports whether the event was inserted', async () => { @@ -69,12 +95,44 @@ describe('DbService v2 methods', () => { expect(values).toEqual([ 'session-1', 'user-1', ]); - expect(clientQuery.mock.calls).toHaveLength(4); + expect(clientQuery.mock.calls).toHaveLength(5); + expect(clientQuery.mock.calls[3][0]).toContain('vibe_event_projections'); expect(clientQuery.mock.calls.map(([query]) => query)).not.toContain( expect.stringContaining('UPDATE vibe_sessions') ); }); + it('projects material feedback once with the durable event transaction', 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: 'completed', occurred_at: new Date(), + position_ms: null, duration_ms: null, payload: {}, + }; + const evidence = vi.spyOn(service as any, 'recordTrackEvidence').mockResolvedValue('evidence-1'); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) + .mockResolvedValueOnce({ rows: [] }) // no existing idempotency key + .mockResolvedValueOnce({ rows: [event] }) // event insert + .mockResolvedValueOnce({ rows: [{ event_id: 'event-1' }] }) // projection marker + .mockResolvedValueOnce({ rows: [] }) // play history + .mockResolvedValueOnce({ rows: [] }) // track counter + .mockResolvedValueOnce({ rows: [] }) // session timestamp + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.recordVibeEvent({ + sessionId: 'session-1', userId: 'user-1', clientEventId: 'client-event-1', + type: 'completed', trackId: 'track-1', + })).resolves.toMatchObject({ inserted: true, event: { id: 'event-1' } }); + + expect(clientQuery.mock.calls.map(([sql]) => sql)).toEqual(expect.arrayContaining([ + expect.stringContaining('vibe_event_projections'), + expect.stringContaining('INSERT INTO play_history'), + ])); + expect(evidence).toHaveBeenCalledTimes(1); + }); + it('rejects an event when no owned session is returned', async () => { const { service, clientQuery } = makeTransactionalService(); clientQuery @@ -114,9 +172,116 @@ describe('DbService v2 methods', () => { expect.stringContaining('INSERT INTO vibe_events') ); }); + + it('locks the terminal transition with its event and makes terminal retries no-ops', async () => { + const { service, clientQuery } = makeTransactionalService(); + const active = { id: 'session-1', user_id: 'user-1', status: 'active' }; + const ended = { ...active, status: 'ended', ended_at: new Date() }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [active] }) // lock + .mockResolvedValueOnce({ rows: [] }) // terminal event + .mockResolvedValueOnce({ rows: [ended] }) // status transition + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.endVibeSessionWithEvent('session-1', 'user-1')) + .resolves.toEqual({ session: ended, ended: true }); + expect(clientQuery.mock.calls[1][0]).toContain('FOR UPDATE'); + expect(clientQuery.mock.calls[2][0]).toContain("'session_ended'"); + expect(clientQuery.mock.calls[3][0]).toContain("status = 'ended'"); + }); + + it('resumes an owned session once and records session_resumed in the same lock', async () => { + const { service, clientQuery } = makeTransactionalService(); + const active = { id: 'session-1', user_id: 'user-1', status: 'active' }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ ...active, status: 'paused' }] }) // lock + .mockResolvedValueOnce({ rows: [] }) // no old resume event + .mockResolvedValueOnce({ rows: [active] }) // activate + .mockResolvedValueOnce({ rows: [] }) // ledger event + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.resumeVibeSession('session-1', 'user-1')) + .resolves.toEqual({ session: active, resumed: true }); + expect(clientQuery.mock.calls[1][0]).toContain('FOR UPDATE'); + expect(clientQuery.mock.calls[4][0]).toContain("'session_resumed'"); + }); }); describe('durable Vibe plans', () => { + it('publishes a revision and its plan_published event in one transaction', async () => { + const { service, clientQuery } = makeTransactionalService(); + const published = { + id: 'plan-1', session_id: 'session-1', version: 1, reason: 'session_started', + state_snapshot: {}, objective_snapshot: {}, created_at: new Date(), + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) // lock + .mockResolvedValueOnce({ rows: [published] }) // header + .mockResolvedValueOnce({ rows: [] }) // item + .mockResolvedValueOnce({ rows: [] }) // plan_published + .mockResolvedValueOnce({ rows: [] }) // timestamp + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.publishVibePlan({ + sessionId: 'session-1', userId: 'user-1', version: 1, reason: 'session_started', + stateSnapshot: {}, objectiveSnapshot: {}, + items: [{ ordinal: 0, track_id: 'track-1', slot_role: 'next', candidate_source: 'comfort', score: 1, score_breakdown: {}, explanation: [], committed: false }], + })).resolves.toMatchObject({ version: 1, items: [{ track_id: 'track-1' }] }); + + expect(clientQuery.mock.calls[1][0]).toContain('FOR UPDATE'); + expect(clientQuery.mock.calls[4][0]).toContain("'plan_published'"); + expect(clientQuery.mock.calls.at(-1)?.[0]).toBe('COMMIT'); + }); + + it('rolls back the plan header and items if writing plan_published fails', async () => { + const { service, clientQuery } = makeTransactionalService(); + const published = { + id: 'plan-1', session_id: 'session-1', version: 1, reason: 'session_started', + state_snapshot: {}, objective_snapshot: {}, created_at: new Date(), + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) + .mockResolvedValueOnce({ rows: [published] }) // header + .mockResolvedValueOnce({ rows: [] }) // item + .mockRejectedValueOnce(new Error('ledger write failed')) + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK + + await expect(service.publishVibePlan({ + sessionId: 'session-1', userId: 'user-1', version: 1, reason: 'session_started', + stateSnapshot: {}, objectiveSnapshot: {}, + items: [{ ordinal: 0, track_id: 'track-1', slot_role: 'next', candidate_source: 'comfort', score: 1, score_breakdown: {}, explanation: [], committed: false }], + })).rejects.toThrow('ledger write failed'); + expect(clientQuery.mock.calls.at(-1)?.[0]).toBe('ROLLBACK'); + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain('COMMIT'); + }); + + it('refuses a delayed initial publication after a concurrent start replaced its session', async () => { + const { service, clientQuery } = makeTransactionalService(); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'replaced' }] }) + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK + + await expect(service.publishVibePlan({ + sessionId: 'session-1', userId: 'user-1', version: 1, reason: 'session_started', + stateSnapshot: {}, objectiveSnapshot: {}, items: [], + })).rejects.toThrow('Cannot publish a plan for replaced Vibe session'); + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('INSERT INTO vibe_plan_versions')); + }); + + it('reads every durable session track as a replacement-plan exclusion', async () => { + const { service, mockQuery } = makeService(); + mockQuery.mockResolvedValueOnce({ rows: [{ track_id: 'served' }, { track_id: 'skipped' }, { track_id: 'disliked' }] }); + await expect(service.getVibeSessionTrackIds('session-1', 'user-1')) + .resolves.toEqual(['served', 'skipped', 'disliked']); + expect(mockQuery.mock.calls[0][0]).toContain('SELECT DISTINCT e.track_id'); + expect(mockQuery.mock.calls[0][0]).toContain('e.track_id IS NOT NULL'); + }); + it('writes a header and all items in one transaction', async () => { const { service, clientQuery } = makeTransactionalService(); clientQuery @@ -143,6 +308,59 @@ describe('DbService v2 methods', () => { expect(clientQuery.mock.calls[3][0]).toBe('COMMIT'); }); + it('serves and commits one next item under the session lock', async () => { + const { service, clientQuery } = makeTransactionalService(); + const item = { + plan_version_id: 'plan-1', ordinal: 0, track_id: 'track-1', slot_role: 'next', + candidate_source: 'comfort', score: 0.9, score_breakdown: {}, explanation: [], committed: true, + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ status: 'active' }] }) // session lock + .mockResolvedValueOnce({ rows: [{ id: 'plan-1', version: 1 }] }) // latest plan + .mockResolvedValueOnce({ rows: [item] }) // commit item + .mockResolvedValueOnce({ rows: [] }) // track_served event + .mockResolvedValueOnce({ rows: [] }) // timestamp + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.serveNextVibePlanItem('session-1', 'user-1')).resolves.toEqual({ item, stale: false }); + expect(clientQuery.mock.calls[1][0]).toContain('FOR UPDATE'); + expect(clientQuery.mock.calls[3][0]).toContain('SET committed = true'); + expect(clientQuery.mock.calls[4][0]).toContain("'track_served'"); + }); + + it('returns a newer preview signal without committing when the expected plan is stale', async () => { + const { service, clientQuery } = makeTransactionalService(); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ status: 'active' }] }) // session lock + .mockResolvedValueOnce({ rows: [{ id: 'plan-2', version: 2 }] }) // latest + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.serveNextVibePlanItem('session-1', 'user-1', 1)) + .resolves.toEqual({ item: null, stale: true }); + expect(clientQuery.mock.calls).toHaveLength(4); + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('SET committed = true')); + }); + + it('returns the original item when a version-aware next request is retried', async () => { + const { service, clientQuery } = makeTransactionalService(); + const item = { + plan_version_id: 'plan-1', ordinal: 0, track_id: 'track-1', slot_role: 'next', + candidate_source: 'comfort', score: 0.9, score_breakdown: {}, explanation: [], committed: true, + }; + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ status: 'active' }] }) // session lock + .mockResolvedValueOnce({ rows: [{ id: 'plan-1', version: 1 }] }) // latest + .mockResolvedValueOnce({ rows: [item] }) // prior served item + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.serveNextVibePlanItem('session-1', 'user-1', 1)) + .resolves.toEqual({ item, stale: false }); + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('SET committed = true')); + }); + it('reads the latest revision and reconstructs ordered plan items', async () => { const { service, mockQuery } = makeService(); mockQuery.mockResolvedValue({ rows: [{ diff --git a/backend/src/services/db.service.ts b/backend/src/services/db.service.ts index 85cf023..a256ebb 100644 --- a/backend/src/services/db.service.ts +++ b/backend/src/services/db.service.ts @@ -1478,10 +1478,17 @@ export class DbService { /** * Create a new session state row. */ - async createSessionState(userId: string, context?: string, stateVector?: Record): Promise { + async createSessionState( + userId: string, + context?: string, + stateVector?: Record, + sessionId?: string + ): Promise { const res = await this.pgClient.query( - `INSERT INTO session_state (user_id, context, state_vector) VALUES ($1, $2, $3) RETURNING session_id`, - [userId, context ?? null, stateVector ? JSON.stringify(stateVector) : '{}'] + `INSERT INTO session_state (session_id, user_id, context, state_vector) + VALUES (COALESCE($1::uuid, gen_random_uuid()), $2, $3, $4::jsonb) + RETURNING session_id`, + [sessionId ?? null, userId, context ?? null, stateVector ? JSON.stringify(stateVector) : '{}'] ); return res.rows[0].session_id as string; } @@ -1508,18 +1515,45 @@ export class DbService { 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; + 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. + await client.query(`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, [params.userId]); + // Lock every active session first. This makes concurrent starts converge + // on one active durable session instead of creating overlapping streams. + const active = await client.query( + `SELECT id FROM vibe_sessions WHERE user_id = $1 AND status = 'active' FOR UPDATE`, + [params.userId], + ); + if (active.rows.length > 0) { + const replaced = await client.query( + `UPDATE vibe_sessions + SET status = 'replaced', ended_at = NOW(), last_event_at = NOW() + WHERE user_id = $1 AND status = 'active' + RETURNING id`, + [params.userId], + ); + for (const session of replaced.rows as Array<{ id: string }>) { + await client.query( + `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload) + VALUES ($1, $2, 'session_ended', NOW(), '{"reason":"replaced"}'::jsonb)`, + [session.id, params.userId], + ); + } + } + 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 *`, + [ + 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. */ @@ -1552,6 +1586,70 @@ export class DbService { return (res.rows[0] as VibeSession) ?? null; } + /** + * Resume an owned paused/active session exactly once. The row lock makes + * the transition and its ledger entry inseparable and prevents retries from + * manufacturing a stream of session_resumed events. + */ + async resumeVibeSession(sessionId: string, userId: string): Promise<{ session: VibeSession; resumed: boolean }> { + return this.withTransaction(async (client) => { + const result = await client.query( + `SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`, + [sessionId, userId] + ); + const session = result.rows[0] as VibeSession | undefined; + if (!session) throw new Error('Vibe session was not found or is not owned by this user'); + if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') { + throw new Error(`Cannot resume ${session.status} Vibe session`); + } + + const prior = await client.query( + `SELECT 1 FROM vibe_events WHERE session_id = $1 AND type = 'session_resumed' LIMIT 1`, + [sessionId] + ); + if (prior.rowCount) return { session, resumed: false }; + + const updated = await client.query( + `UPDATE vibe_sessions SET status = 'active', ended_at = NULL, last_event_at = NOW() + WHERE id = $1 RETURNING *`, + [sessionId] + ); + const resumed = updated.rows[0] as VibeSession; + await client.query( + `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload) + VALUES ($1, $2, 'session_resumed', NOW(), '{}'::jsonb)`, + [sessionId, userId] + ); + return { session: resumed, resumed: true }; + }); + } + + /** End a session and append its terminal event under one session-row lock. */ + async endVibeSessionWithEvent(sessionId: string, userId: string): Promise<{ session: VibeSession; ended: boolean }> { + return this.withTransaction(async (client) => { + const result = await client.query( + `SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`, + [sessionId, userId] + ); + const session = result.rows[0] as VibeSession | undefined; + if (!session) throw new Error('Vibe session was not found or is not owned by this user'); + if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') { + return { session, ended: false }; + } + await client.query( + `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload) + VALUES ($1, $2, 'session_ended', NOW(), '{}'::jsonb)`, + [sessionId, userId] + ); + const updated = await client.query( + `UPDATE vibe_sessions SET status = 'ended', ended_at = NOW(), last_event_at = NOW() + WHERE id = $1 RETURNING *`, + [sessionId] + ); + return { session: updated.rows[0] as VibeSession, ended: true }; + }); + } + /** * Append an immutable Vibe event. A supplied clientEventId is idempotent per * session: a retry returns the original event and does not advance the @@ -1596,6 +1694,7 @@ export class DbService { ); const existing = existingRes.rows[0] as VibeEvent | undefined; if (existing) { + await this.projectVibeFeedback(existing, client); return { event: existing, inserted: false }; } } @@ -1627,6 +1726,8 @@ export class DbService { throw new Error('Vibe event could not be recorded'); } + await this.projectVibeFeedback(event, client); + await client.query( `UPDATE vibe_sessions SET last_event_at = GREATEST(last_event_at, $2::timestamptz) @@ -1637,11 +1738,178 @@ export class DbService { }); } + /** + * Materialize Vibe feedback into the listener inputs used by the incumbent + * director. The projection marker and every write share the event's + * transaction, so a retry either sees the completed projection or performs + * it once; it can never double-count a completed/skip/dislike/kept signal. + */ + private async projectVibeFeedback(event: VibeEvent, client: PoolClient): Promise { + if (!event.track_id || !['completed', 'skipped', 'disliked', 'kept'].includes(event.type)) return; + const projection = await client.query( + `INSERT INTO vibe_event_projections (event_id) + VALUES ($1) + ON CONFLICT (event_id) DO NOTHING + RETURNING event_id`, + [event.id], + ); + if (!projection.rows[0]) return; + + const occurredAt = event.occurred_at?.toISOString?.() ?? new Date().toISOString(); + switch (event.type) { + case 'completed': + await client.query( + `INSERT INTO play_history (user_id, track_id, completed, played_at) + VALUES ($1, $2, true, $3::timestamptz)`, + [event.user_id, event.track_id, occurredAt], + ); + await client.query( + `UPDATE tracks + SET play_count = play_count + 1, last_played_at = $2::timestamptz + WHERE id = $1`, + [event.track_id, occurredAt], + ); + await this.recordTrackEvidence({ + user_id: event.user_id, + track_id: event.track_id, + signal: 'playback_completed', + profile: 'longterm', + weight: 0.10, + context: { vibe_event_id: event.id, session_id: event.session_id }, + }, client); + break; + case 'skipped': + await client.query('UPDATE tracks SET skip_count = skip_count + 1 WHERE id = $1', [event.track_id]); + await client.query( + `INSERT INTO feedback (user_id, track_id, action, created_at) + VALUES ($1, $2, 'skipped', $3::timestamptz)`, + [event.user_id, event.track_id, occurredAt], + ); + await this.recordTrackEvidence({ + user_id: event.user_id, + track_id: event.track_id, + signal: 'skip_quick', + profile: 'negative', + weight: -0.20, + context: { vibe_event_id: event.id, session_id: event.session_id }, + }, client); + break; + case 'disliked': + await client.query('UPDATE tracks SET dislike_count = dislike_count + 1 WHERE id = $1', [event.track_id]); + await client.query( + `INSERT INTO feedback (user_id, track_id, action, created_at) + VALUES ($1, $2, 'disliked', $3::timestamptz)`, + [event.user_id, event.track_id, occurredAt], + ); + await this.recordTrackEvidence({ + user_id: event.user_id, + track_id: event.track_id, + signal: 'hidden', + profile: 'negative', + weight: -0.60, + context: { vibe_event_id: event.id, session_id: event.session_id }, + }, client); + break; + case 'kept': + await this.recordTrackEvidence({ + user_id: event.user_id, + track_id: event.track_id, + signal: 'kept', + profile: 'longterm', + weight: 0.05, + context: { vibe_event_id: event.id, session_id: event.session_id }, + }, client); + break; + } + } + /** * 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 publishVibePlan(params: { + sessionId: string; + userId: string; + /** Supplying one is for the first revision; otherwise allocate the next. */ + version?: number; + reason: string; + stateSnapshot: Record; + objectiveSnapshot: Record; + items: Array>; + }): Promise { + return this.withTransaction(async (client) => { + // The session lock is also the concurrency boundary for starts/ends and + // plan revisions. In particular, a slow initial planner cannot publish + // into a session a newer start has already replaced. + 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 (session.status !== 'active') { + throw new Error(`Cannot publish a plan for ${session.status} Vibe session`); + } + const version = params.version ?? Number((await client.query( + `SELECT COALESCE(MAX(version), 0) + 1 AS version + FROM vibe_plan_versions WHERE session_id = $1`, + [params.sessionId], + )).rows[0].version); + const header = await client.query( + `INSERT INTO vibe_plan_versions + (session_id, version, reason, state_snapshot, objective_snapshot) + VALUES ($1, $2, $3, $4::jsonb, $5::jsonb) + RETURNING *`, + [ + params.sessionId, + 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 plan could not be published'); + 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, + ], + ); + } + // Header/items and this ledger event deliberately commit together. A + // client retry can therefore find either neither or the same canonical + // revision; it can never observe a published event without its plan. + await client.query( + `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload) + VALUES ($1, $2, 'plan_published', NOW(), $3::jsonb)`, + [params.sessionId, params.userId, JSON.stringify({ + planVersion: planVersion.version, + planVersionId: planVersion.id, + reason: params.reason, + itemCount: params.items.length, + feedbackEventId: params.objectiveSnapshot.feedbackEventId ?? null, + })], + ); + await client.query( + `UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`, + [params.sessionId], + ); + return { + ...planVersion, + items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })), + }; + }); + } + async persistVibePlan(params: { sessionId: string; userId: string; @@ -1696,6 +1964,115 @@ export class DbService { }); } + /** Allocate and persist the next immutable revision while holding the session lock. */ + async persistNextVibePlan(params: Omit[0], 'version'>): Promise { + return this.withTransaction(async (client) => { + const session = await client.query( + `SELECT id, status FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`, + [params.sessionId, params.userId] + ); + if (!session.rows[0]) throw new Error('Vibe session was not found or is not owned by this user'); + if ((session.rows[0] as Pick).status !== 'active') { + throw new Error(`Cannot record a new event for ${(session.rows[0] as Pick).status} Vibe session`); + } + const versionResult = await client.query( + `SELECT COALESCE(MAX(version), 0) + 1 AS version FROM vibe_plan_versions WHERE session_id = $1`, + [params.sessionId] + ); + const version = Number(versionResult.rows[0].version); + const header = await client.query( + `INSERT INTO vibe_plan_versions + (session_id, version, reason, state_snapshot, objective_snapshot) + VALUES ($1, $2, $3, $4::jsonb, $5::jsonb) RETURNING *`, + [params.sessionId, version, params.reason, JSON.stringify(params.stateSnapshot), JSON.stringify(params.objectiveSnapshot)] + ); + const planVersion = header.rows[0] as VibePlan; + 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 })) }; + }); + } + + /** + * Atomically commit one item from the latest plan. A version-aware request + * acts as an idempotency key: retrying the same expected version receives + * the original item, while a replaced plan is returned as stale without + * committing any old item. Calls without an expected version preserve the + * original legacy cursor behaviour. + */ + async serveNextVibePlanItem( + sessionId: string, + userId: string, + expectedPlanVersion?: number, + ): Promise<{ item: VibePlanItem | null; stale: boolean }> { + return this.withTransaction(async (client) => { + const session = await client.query( + `SELECT status FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`, + [sessionId, userId] + ); + const row = session.rows[0] as Pick | undefined; + if (!row) throw new Error('Vibe session was not found or is not owned by this user'); + if (row.status !== 'active') throw new Error(`Cannot record a new event for ${row.status} Vibe session`); + const latest = await client.query( + `SELECT id, version FROM vibe_plan_versions WHERE session_id = $1 ORDER BY version DESC LIMIT 1 FOR UPDATE`, + [sessionId], + ); + const plan = latest.rows[0] as Pick | undefined; + if (!plan) return { item: null, stale: false }; + if (expectedPlanVersion !== undefined && expectedPlanVersion !== plan.version) { + return { item: null, stale: true }; + } + + if (expectedPlanVersion !== undefined) { + const prior = await client.query( + `SELECT i.* + FROM vibe_events e + JOIN vibe_plan_items i + ON i.plan_version_id = (e.payload->>'planVersionId')::uuid + AND i.ordinal = (e.payload->>'ordinal')::integer + WHERE e.session_id = $1 + AND e.type = 'track_served' + AND e.payload->>'planVersion' = $2::text + ORDER BY e.occurred_at ASC + LIMIT 1`, + [sessionId, expectedPlanVersion], + ); + const servedPreviously = prior.rows[0] as VibePlanItem | undefined; + if (servedPreviously) return { item: servedPreviously, stale: false }; + } + const item = await client.query( + `WITH next_item AS ( + SELECT i.plan_version_id, i.ordinal FROM vibe_plan_items i + WHERE i.plan_version_id = $2 AND NOT i.committed ORDER BY i.ordinal ASC LIMIT 1 FOR UPDATE + ) + UPDATE vibe_plan_items i SET committed = true + FROM next_item n WHERE i.plan_version_id = n.plan_version_id AND i.ordinal = n.ordinal + RETURNING i.*`, + [sessionId, plan.id] + ); + const served = item.rows[0] as VibePlanItem | undefined; + if (!served) return { item: null, stale: false }; + await client.query( + `INSERT INTO vibe_events (session_id, user_id, track_id, type, occurred_at, payload) + VALUES ($1, $2, $3, 'track_served', NOW(), $4::jsonb)`, + [sessionId, userId, served.track_id, JSON.stringify({ + planVersion: plan.version, + planVersionId: served.plan_version_id, + ordinal: served.ordinal, + })] + ); + await client.query(`UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`, [sessionId]); + return { item: served, stale: false }; + }); + } + /** 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( @@ -1745,6 +2122,41 @@ export class DbService { return plan; } + /** Return the replacement revision caused by a material feedback event. */ + async getVibePlanForFeedbackEvent(sessionId: string, userId: string, eventId: string): Promise { + const res = await this.pgClient.query( + `SELECT version + FROM vibe_plan_versions p + JOIN vibe_sessions s ON s.id = p.session_id + WHERE p.session_id = $1 + AND s.user_id = $2 + AND p.objective_snapshot->>'feedbackEventId' = $3 + ORDER BY p.version DESC + LIMIT 1`, + [sessionId, userId, eventId], + ); + const version = res.rows[0]?.version as number | undefined; + return version === undefined ? null : this.getVibePlan(sessionId, userId, version); + } + + /** + * Tracks exposed by a durable session are never eligible for another + * revision of that same session. This includes served items and every + * explicit feedback target, not just completed play history. + */ + async getVibeSessionTrackIds(sessionId: string, userId: string): Promise { + const res = await this.pgClient.query( + `SELECT DISTINCT e.track_id + FROM vibe_events e + JOIN vibe_sessions s ON s.id = e.session_id + WHERE e.session_id = $1 + AND s.user_id = $2 + AND e.track_id IS NOT NULL`, + [sessionId, userId], + ); + return res.rows.map((row: { track_id: string }) => row.track_id); + } + /** * Upsert a diversity budget for a user. */ diff --git a/backend/src/services/session-director.service.ts b/backend/src/services/session-director.service.ts index 0f1c1cf..296c5e0 100644 --- a/backend/src/services/session-director.service.ts +++ b/backend/src/services/session-director.service.ts @@ -835,6 +835,11 @@ export class SessionDirector { seedTrackId?: string, options: PlanBuildOptions = {} ): Promise { + // Durable events outlive any process-local queue. Fetch them here rather + // than trusting callers to remember the boundary, so a served, skipped, + // disliked, or otherwise exposed track can never leak into a replacement + // revision for this session. + const durableSessionTrackIds = await this.db.getVibeSessionTrackIds(sessionId, userId); // Do not let abundant track-level beliefs crowd out the artist/genre // affinities required by the discovery generators. const beliefGroups = await Promise.all([ @@ -895,6 +900,7 @@ export class SessionDirector { // a replacement plan while its Vibe session is active. const recentExclusionSet = new Set([ ...recentPlays.map(p => p.trackId), + ...durableSessionTrackIds, ...(options.excludedTrackIds ?? []), ]); if (seedTrackId) recentExclusionSet.add(seedTrackId); diff --git a/backend/src/services/vibe-session-coordinator.service.test.ts b/backend/src/services/vibe-session-coordinator.service.test.ts new file mode 100644 index 0000000..bd05625 --- /dev/null +++ b/backend/src/services/vibe-session-coordinator.service.test.ts @@ -0,0 +1,198 @@ +import { describe, expect, it, vi } from 'vitest'; +import { DbService } from './db.service.js'; +import { + DEFAULT_VIBE_POLICY_VERSION, + VibeSessionCoordinator, + VibeSessionLifecycleError, + VibePlanNotFoundError, +} from './vibe-session-coordinator.service.js'; + +const SESSION_ID = '11111111-1111-4111-8111-111111111111'; +const TRACK_ID = '22222222-2222-4222-8222-222222222222'; +const EVENT_ID = '33333333-3333-4333-8333-333333333333'; + +function session(status: 'active' | 'ended' = 'active') { + return { + id: SESSION_ID, user_id: 'user-1', status, seed_track_id: null, + context: { activity: 'focus' }, policy_version: DEFAULT_VIBE_POLICY_VERSION, + started_at: new Date('2026-01-01T00:00:00.000Z'), + last_event_at: new Date('2026-01-01T00:00:00.000Z'), ended_at: status === 'ended' ? new Date() : null, + } as any; +} + +function plan() { + return { + id: 'plan-1', session_id: SESSION_ID, version: 1, reason: 'session_started', + state_snapshot: { energy: 0.5 }, objective_snapshot: {}, created_at: new Date(), + items: [{ + plan_version_id: 'plan-1', ordinal: 0, track_id: TRACK_ID, slot_role: 'next', + candidate_source: 'comfort', score: 0.8, score_breakdown: { relevance: 0.8 }, + explanation: [], committed: false, + }], + } as any; +} + +function setup() { + const db = { + createVibeSession: vi.fn().mockResolvedValue(session()), + createSessionState: vi.fn().mockResolvedValue(SESSION_ID), + recordVibeEvent: vi.fn().mockResolvedValue({ event: { id: 'event-1' }, inserted: true }), + persistVibePlan: vi.fn().mockResolvedValue(plan()), + publishVibePlan: vi.fn().mockImplementation((input: { version?: number; reason: string }) => Promise.resolve({ + ...plan(), version: input.version ?? 2, reason: input.reason, + })), + getVibeSession: vi.fn().mockResolvedValue(session()), + getVibePlan: vi.fn().mockResolvedValue(plan()), + endVibeSession: vi.fn().mockResolvedValue(session('ended')), + endVibeSessionWithEvent: vi.fn().mockResolvedValue({ session: session('ended'), ended: true }), + resumeVibeSession: vi.fn().mockResolvedValue({ session: session(), resumed: true }), + serveNextVibePlanItem: vi.fn().mockResolvedValue({ item: plan().items[0], stale: false }), + persistNextVibePlan: vi.fn().mockResolvedValue({ ...plan(), version: 2, reason: 'feedback:completed' }), + getVibePlanForFeedbackEvent: vi.fn().mockResolvedValue(null), + } as unknown as DbService; + const director = { + buildPlan: vi.fn().mockResolvedValue([{ trackId: TRACK_ID, generatorId: 'comfort', relevance: 0.8, explanation: [] }]), + buildState: vi.fn().mockResolvedValue({ energy: 0.5, noveltyHunger: 0.3 }), + } as any; + return { db, director, coordinator: new VibeSessionCoordinator(db, director) }; +} + +describe('VibeSessionCoordinator', () => { + it('creates an authoritative session, shadow state, initial plan revision, and ledger events', async () => { + const { db, director, coordinator } = setup(); + + const response = await coordinator.start('user-1', { + context: { activity: 'focus' }, intent: 'deep-work', + }); + + expect(response).toMatchObject({ sessionId: SESSION_ID, planVersion: 1, now: { track_id: TRACK_ID } }); + expect(db.createVibeSession).toHaveBeenCalledWith(expect.objectContaining({ + userId: 'user-1', policyVersion: DEFAULT_VIBE_POLICY_VERSION, + })); + expect(db.createSessionState).toHaveBeenCalledWith('user-1', 'focus', expect.any(Object), SESSION_ID); + expect(director.buildPlan).toHaveBeenCalledWith('user-1', SESSION_ID, undefined); + expect(db.publishVibePlan).toHaveBeenCalledWith(expect.objectContaining({ + sessionId: SESSION_ID, version: 1, reason: 'session_started', + items: [expect.objectContaining({ track_id: TRACK_ID, committed: false })], + })); + expect((db.recordVibeEvent as any).mock.calls.map(([input]: any[]) => input.type)) + .toEqual(['session_started']); + }); + + it('returns the canonical replacement on an idempotent material-event retry without replanning', async () => { + const { db, coordinator } = setup(); + (db.recordVibeEvent as any).mockResolvedValueOnce({ + event: { id: 'event-1', client_event_id: EVENT_ID, type: 'skipped' }, inserted: false, + }); + (db.getVibePlanForFeedbackEvent as any).mockResolvedValueOnce({ ...plan(), version: 2 }); + + const response = await coordinator.appendEvent('user-1', SESSION_ID, { + eventId: EVENT_ID, type: 'skipped', trackId: TRACK_ID, + }); + + expect(response).toMatchObject({ idempotent: true, planVersion: 2, replanned: false, replanReason: null }); + expect(db.recordVibeEvent).toHaveBeenCalledWith(expect.objectContaining({ clientEventId: EVENT_ID })); + expect(db.persistNextVibePlan).not.toHaveBeenCalled(); + }); + + it('recovers a material feedback replan when its first persistence attempt failed', async () => { + const { db, coordinator } = setup(); + (db.publishVibePlan as any).mockRejectedValueOnce(new Error('temporary database failure')); + await expect(coordinator.appendEvent('user-1', SESSION_ID, { + eventId: EVENT_ID, type: 'completed', trackId: TRACK_ID, + })).rejects.toThrow('temporary database failure'); + + (db.recordVibeEvent as any).mockResolvedValueOnce({ + event: { id: 'event-1', client_event_id: EVENT_ID, type: 'completed' }, inserted: false, + }); + const recovered = await coordinator.appendEvent('user-1', SESSION_ID, { + eventId: EVENT_ID, type: 'completed', trackId: TRACK_ID, + }); + + expect(db.publishVibePlan).toHaveBeenCalledTimes(2); + expect(recovered).toMatchObject({ idempotent: true, replanned: true, planVersion: 2 }); + }); + + it('persists a replacement revision for material feedback and returns its preview', async () => { + const { db, director, coordinator } = setup(); + const response = await coordinator.appendEvent('user-1', SESSION_ID, { type: 'completed', trackId: TRACK_ID }); + + expect(director.buildPlan).toHaveBeenCalledWith('user-1', SESSION_ID, TRACK_ID); + expect(db.publishVibePlan).toHaveBeenCalledWith(expect.objectContaining({ + sessionId: SESSION_ID, reason: 'feedback:completed', items: [expect.objectContaining({ committed: false })], + })); + expect(response).toMatchObject({ replanned: true, replanReason: 'feedback:completed', planVersion: 2 }); + }); + + it('keeps the durable seed excluded when feedback supplies a different local replan anchor', async () => { + const { db, director, coordinator } = setup(); + const seedTrackId = '44444444-4444-4444-8444-444444444444'; + (db.getVibeSession as any).mockResolvedValue({ ...session(), seed_track_id: seedTrackId }); + + await coordinator.appendEvent('user-1', SESSION_ID, { type: 'completed', trackId: TRACK_ID }); + + expect(director.buildPlan).toHaveBeenCalledWith( + 'user-1', + SESSION_ID, + TRACK_ID, + { excludedTrackIds: new Set([seedTrackId]) }, + ); + }); + + it('resumes only the caller-owned session and serves a plan item through the durable API', async () => { + const { db, coordinator } = setup(); + const resumed = await coordinator.start('user-1', { resumeSessionId: SESSION_ID }); + const served = await coordinator.serveNext('user-1', SESSION_ID); + + expect(db.resumeVibeSession).toHaveBeenCalledWith(SESSION_ID, 'user-1'); + expect(resumed.sessionId).toBe(SESSION_ID); + expect(db.serveNextVibePlanItem).toHaveBeenCalledWith(SESSION_ID, 'user-1'); + expect(served.now).toMatchObject({ track_id: TRACK_ID }); + }); + + it('returns a lifecycle conflict when the ledger rejects a new terminal-session event', async () => { + const { db, coordinator } = setup(); + (db.recordVibeEvent as any).mockRejectedValueOnce(new Error('Cannot record a new event for ended Vibe session')); + + await expect(coordinator.appendEvent('user-1', SESSION_ID, { type: 'completed' })) + .rejects.toBeInstanceOf(VibeSessionLifecycleError); + }); + + it('distinguishes a missing requested revision from an empty latest plan', async () => { + const { db, coordinator } = setup(); + (db.getVibePlan as any).mockResolvedValueOnce(null); + await expect(coordinator.getPlan('user-1', SESSION_ID, 99)).rejects.toBeInstanceOf(VibePlanNotFoundError); + }); + + it('returns 404-worthy failure when a session has no latest plan', async () => { + const { db, coordinator } = setup(); + (db.getVibePlan as any).mockResolvedValueOnce(null); + await expect(coordinator.getPlan('user-1', SESSION_ID)).rejects.toBeInstanceOf(VibePlanNotFoundError); + }); + + it('does not commit a stale version-aware next request and returns the current preview', async () => { + const { db, coordinator } = setup(); + (db.serveNextVibePlanItem as any).mockResolvedValueOnce({ item: null, stale: true }); + const result = await coordinator.serveNext('user-1', SESSION_ID, 1); + + expect(db.serveNextVibePlanItem).toHaveBeenCalledWith(SESSION_ID, 'user-1', 1); + expect(result).toMatchObject({ planVersion: 1, now: { track_id: TRACK_ID } }); + }); + + it('ends an active session once and preserves its latest persisted plan', async () => { + const { db, coordinator } = setup(); + const response = await coordinator.end('user-1', SESSION_ID); + + expect(db.endVibeSessionWithEvent).toHaveBeenCalledWith(SESSION_ID, 'user-1'); + expect(response.session.status).toBe('ended'); + expect(response.planVersion).toBe(1); + }); + + it('does not publish an initial plan when another start replaced the session while planning', async () => { + const { db, coordinator } = setup(); + (db.publishVibePlan as any).mockRejectedValueOnce(new Error('Cannot publish a plan for replaced Vibe session')); + + await expect(coordinator.start('user-1', {})).rejects.toBeInstanceOf(VibeSessionLifecycleError); + expect(db.publishVibePlan).toHaveBeenCalledWith(expect.objectContaining({ version: 1 })); + }); +}); diff --git a/backend/src/services/vibe-session-coordinator.service.ts b/backend/src/services/vibe-session-coordinator.service.ts new file mode 100644 index 0000000..79efeaa --- /dev/null +++ b/backend/src/services/vibe-session-coordinator.service.ts @@ -0,0 +1,293 @@ +import { DbService, VibeEvent, VibePlan, VibeSession } from './db.service.js'; +import { SessionDirector } from './session-director.service.js'; + +/** + * This is deliberately a narrow bridge between the durable Vibe ledger and + * the current deterministic director. It writes authoritative revisions and + * lets the playback client replace only its unserved preview after feedback. + */ +export const DEFAULT_VIBE_POLICY_VERSION = 'vibe-v2-initial'; + +export const VIBE_EVENT_TYPES = [ + 'session_started', 'session_resumed', 'session_ended', 'context_changed', + 'plan_published', 'track_served', 'playback_started', 'progress', 'completed', + 'skipped', 'disliked', 'kept', 'favourite_added', 'queue_removed', + 'manual_search', 'album_opened', 'artist_opened', 'playlist_added', + 'track_replayed', 'volume_changed', 'playback_error', +] as const; + +export type VibeEventType = (typeof VIBE_EVENT_TYPES)[number]; + +export interface StartVibeSessionInput { + seedTrackId?: string; + context?: Record; + intent?: string; + policyVersion?: string; + resumeSessionId?: string; +} + +export interface AppendVibeEventInput { + eventId?: string; + type: VibeEventType; + trackId?: string; + occurredAt?: Date; + positionMs?: number; + durationMs?: number; + payload?: Record; +} + +export class VibeSessionNotFoundError extends Error {} +export class VibeSessionLifecycleError extends Error {} +export class VibePlanNotFoundError extends Error {} + +export interface VibeSessionResponse { + session: VibeSession; + sessionId: string; + planVersion: number | null; + now: VibePlan['items'][number] | null; + preview: VibePlan['items']; + state: Record; + replanned: boolean; + replanReason: string | null; +} + +export class VibeSessionCoordinator { + constructor( + private readonly db: DbService, + private readonly director: Pick, + ) {} + + async start(userId: string, input: StartVibeSessionInput): Promise { + if (input.resumeSessionId) return this.resume(userId, input.resumeSessionId); + const context = input.context ?? {}; + const policyVersion = input.policyVersion ?? DEFAULT_VIBE_POLICY_VERSION; + const session = await this.db.createVibeSession({ + userId, + policyVersion, + seedTrackId: input.seedTrackId ?? null, + context, + }); + + // session_state is a derived cache used by the current director. Give it + // the durable ID so director state cannot accidentally bleed into another + // session while the durable tables remain the source of truth. + await this.db.createSessionState( + userId, + typeof context.activity === 'string' ? context.activity : input.intent, + { energy: 0.5, noveltyHunger: 0.3 }, + session.id, + ); + await this.db.recordVibeEvent({ + sessionId: session.id, + userId, + type: 'session_started', + payload: { policyVersion, context, intent: input.intent ?? null }, + }); + + const candidates = await this.director.buildPlan(userId, session.id, input.seedTrackId); + const state = await this.director.buildState(userId, session.id); + let plan: VibePlan; + try { + plan = await this.db.publishVibePlan({ + sessionId: session.id, + userId, + version: 1, + reason: 'session_started', + stateSnapshot: state, + objectiveSnapshot: { + policyVersion, + intent: input.intent ?? null, + horizonTracks: candidates.length, + }, + items: candidates.map((candidate, ordinal) => ({ + ordinal, + track_id: candidate.trackId, + slot_role: ordinal === 0 ? 'next' : null, + candidate_source: candidate.generatorId, + score: candidate.relevance, + score_breakdown: { relevance: candidate.relevance }, + explanation: candidate.explanation, + committed: false, + })), + }); + } catch (error) { + throw this.mapLifecycleError(error); + } + + return this.toResponse(session, plan, state); + } + + async getPlan(userId: string, sessionId: string, version?: number): Promise { + const session = await this.requireSession(userId, sessionId); + const plan = await this.db.getVibePlan(sessionId, userId, version); + if (!plan) throw new VibePlanNotFoundError( + version === undefined ? 'Vibe session does not have a published plan' : 'Vibe plan revision was not found', + ); + return this.toResponse(session, plan, plan?.state_snapshot ?? {}); + } + + async serveNext(userId: string, sessionId: string, expectedPlanVersion?: number): Promise { + try { + const served = expectedPlanVersion === undefined + ? await this.db.serveNextVibePlanItem(sessionId, userId) + : await this.db.serveNextVibePlanItem(sessionId, userId, expectedPlanVersion); + const response = await this.getPlan(userId, sessionId); + // A plan may be replaced between the client's preview and this request. + // In that case the database does not commit anything and this is the + // current, revisable preview the client must reconcile to. + if (served.stale) return response; + return { ...response, now: served.item, preview: response.preview }; + } catch (error) { + throw this.mapLifecycleError(error); + } + } + + async appendEvent( + userId: string, + sessionId: string, + input: AppendVibeEventInput, + ): Promise { + try { + const result = await this.db.recordVibeEvent({ + sessionId, + userId, + clientEventId: input.eventId, + type: input.type, + trackId: input.trackId, + occurredAt: input.occurredAt, + positionMs: input.positionMs, + durationMs: input.durationMs, + payload: input.payload, + }); + if (!isMaterialFeedback(input.type)) { + const response = await this.getPlan(userId, sessionId); + return { ...response, event: result.event, idempotent: !result.inserted }; + } + + // A material event is durable before its computed replacement can be + // written. If planning/persistence failed after that event committed, a + // retry must finish the missing replacement instead of permanently + // returning an obsolete preview. Once a replacement exists, a duplicate + // retry returns that canonical revision without doing work again. + if (!result.inserted) { + const existingReplacement = await this.db.getVibePlanForFeedbackEvent(sessionId, userId, result.event.id); + if (existingReplacement) { + const session = await this.requireSession(userId, sessionId); + return { + ...this.toResponse(session, existingReplacement, existingReplacement.state_snapshot), + event: result.event, + idempotent: true, + }; + } + } + + const session = await this.requireSession(userId, sessionId); + const state = await this.director.buildState(userId, sessionId); + // A feedback target is useful as the local replan anchor, but it must + // never displace the durable seed from the exclusion boundary. Unlike + // feedback tracks, the seed is not necessarily present in the event + // ledger, so carry it explicitly into every replacement request. + const seedTrackId = session.seed_track_id ?? undefined; + const candidates = seedTrackId + ? await this.director.buildPlan(userId, sessionId, input.trackId ?? seedTrackId, { + excludedTrackIds: new Set([seedTrackId]), + }) + : await this.director.buildPlan(userId, sessionId, input.trackId); + const reason = `feedback:${input.type}`; + const plan = await this.db.publishVibePlan({ + sessionId, + userId, + reason, + stateSnapshot: state, + objectiveSnapshot: { + policyVersion: session.policy_version, + feedbackEventId: result.event.id, + feedbackType: input.type, + horizonTracks: candidates.length, + }, + items: candidates.map((candidate, ordinal) => ({ + ordinal, + track_id: candidate.trackId, + slot_role: ordinal === 0 ? 'next' : null, + candidate_source: candidate.generatorId, + score: candidate.relevance, + score_breakdown: { relevance: candidate.relevance }, + explanation: candidate.explanation, + committed: false, + })), + }); + return { + ...this.toResponse(session, plan, state), + event: result.event, + idempotent: !result.inserted, + replanned: true, + replanReason: reason, + }; + } catch (error) { + throw this.mapLifecycleError(error); + } + } + + async end(userId: string, sessionId: string): Promise { + let ended: VibeSession; + try { + ended = (await this.db.endVibeSessionWithEvent(sessionId, userId)).session; + } catch (error) { + throw this.mapLifecycleError(error); + } + const plan = await this.db.getVibePlan(sessionId, userId); + return this.toResponse(ended, plan, plan?.state_snapshot ?? {}); + } + + private async resume(userId: string, sessionId: string): Promise { + try { + const resumed = await this.db.resumeVibeSession(sessionId, userId); + const plan = await this.db.getVibePlan(sessionId, userId); + return this.toResponse(resumed.session, plan, plan?.state_snapshot ?? {}); + } catch (error) { + throw this.mapLifecycleError(error); + } + } + + private async requireSession(userId: string, sessionId: string): Promise { + const session = await this.db.getVibeSession(sessionId, userId); + if (!session) throw new VibeSessionNotFoundError('Vibe session was not found'); + return session; + } + + private toResponse( + session: VibeSession, + plan: VibePlan | null, + state: Record, + ): VibeSessionResponse { + // A revision is immutable, but clients need a live future: already served + // rows stay in the ledger and are excluded from the replacement preview. + const preview = plan?.items.filter((item) => !item.committed).slice(0, 8) ?? []; + return { + session, + sessionId: session.id, + planVersion: plan?.version ?? null, + now: preview[0] ?? null, + preview, + state, + replanned: false, + replanReason: null, + }; + } + + private mapLifecycleError(error: unknown): Error { + if (error instanceof Error && (error.message.includes('Cannot record a new event for') || error.message.includes('Cannot resume ') || error.message.includes('Cannot publish a plan for'))) { + return new VibeSessionLifecycleError(error.message); + } + if (error instanceof Error && error.message.includes('not found or is not owned')) { + return new VibeSessionNotFoundError('Vibe session was not found'); + } + return error instanceof Error ? error : new Error(String(error)); + } +} + +const MATERIAL_FEEDBACK_EVENTS = new Set(['skipped', 'disliked', 'completed', 'kept']); + +function isMaterialFeedback(type: VibeEventType): boolean { + return MATERIAL_FEEDBACK_EVENTS.has(type); +}