diff --git a/README.md b/README.md index 26afe18..ec7d06a 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,11 @@ A high-performance, distributed music orchestration and recommendation platform. - Docker & Docker Compose ### Running Locally + +Set `MUZICK_VIBE_USER_ID` in `.env` to the UUID of the local Muzick user before +using Vibe. Durable Vibe session routes intentionally reject client-supplied +identities, so this is the trusted single-user binding for a self-hosted stack. + ```bash docker-compose up -d ``` diff --git a/backend/src/routes/vibe-sessions.routes.test.ts b/backend/src/routes/vibe-sessions.routes.test.ts index 0d03bf2..a9300b5 100644 --- a/backend/src/routes/vibe-sessions.routes.test.ts +++ b/backend/src/routes/vibe-sessions.routes.test.ts @@ -22,6 +22,7 @@ async function appWithCoordinator(identityResolver: VibeIdentityResolver = () => appendEvent: vi.fn().mockResolvedValue({ ...response(), event: { id: 'event-1' }, idempotent: false }), end: vi.fn().mockResolvedValue(response()), serveNext: vi.fn().mockResolvedValue(response()), + advancePastUnplayable: vi.fn().mockResolvedValue(response()), } as any; const app = Fastify(); await app.register(vibeSessionsRoutes, { coordinator, identityResolver }); @@ -133,4 +134,31 @@ describe('durable Vibe session routes', () => { expect(coordinator.serveNext).toHaveBeenCalledTimes(1); await app.close(); }); + + it('uses the explicit versioned advancement protocol for a served unplayable item', async () => { + const { app, coordinator } = await appWithCoordinator(); + const result = await app.inject({ + method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`, + payload: { + expectedPlanVersion: 2, + unplayable: { + eventId: '33333333-3333-4333-8333-333333333333', + planVersionId: '44444444-4444-4444-8444-444444444444', + ordinal: 0, + trackId: TRACK_ID, + }, + }, + }); + + expect(result.statusCode).toBe(200); + expect(coordinator.advancePastUnplayable).toHaveBeenCalledWith(USER_ID, SESSION_ID, { + expectedPlanVersion: 2, + eventId: '33333333-3333-4333-8333-333333333333', + planVersionId: '44444444-4444-4444-8444-444444444444', + ordinal: 0, + trackId: TRACK_ID, + }); + expect(coordinator.serveNext).not.toHaveBeenCalled(); + await app.close(); + }); }); diff --git a/backend/src/routes/vibe-sessions.routes.ts b/backend/src/routes/vibe-sessions.routes.ts index 44762e9..a477340 100644 --- a/backend/src/routes/vibe-sessions.routes.ts +++ b/backend/src/routes/vibe-sessions.routes.ts @@ -156,8 +156,33 @@ export default async function vibeSessionsRoutes( && (!Number.isInteger(body.expectedPlanVersion) || (body.expectedPlanVersion as number) < 1)) { return reply.code(400).send({ error: 'expectedPlanVersion must be a positive integer' }); } + const unplayable = body.unplayable; + if (unplayable !== undefined && !isObject(unplayable)) { + return reply.code(400).send({ error: 'unplayable must be an object' }); + } + if (isObject(unplayable)) { + if (body.expectedPlanVersion === undefined) { + return reply.code(400).send({ error: 'expectedPlanVersion is required when advancing an unplayable item' }); + } + if (!validUuid(unplayable.eventId) + || !validUuid(unplayable.planVersionId) + || !validUuid(unplayable.trackId) + || !Number.isInteger(unplayable.ordinal) + || (unplayable.ordinal as number) < 0) { + return reply.code(400).send({ error: 'unplayable requires UUID eventId, planVersionId, trackId and a non-negative integer ordinal' }); + } + } try { const expectedPlanVersion = body.expectedPlanVersion as number | undefined; + if (isObject(unplayable)) { + return reply.send(await coordinator.advancePastUnplayable(userId, sessionId, { + expectedPlanVersion: expectedPlanVersion as number, + eventId: unplayable.eventId as string, + planVersionId: unplayable.planVersionId as string, + ordinal: unplayable.ordinal as number, + trackId: unplayable.trackId as string, + })); + } return reply.send(expectedPlanVersion === undefined ? await coordinator.serveNext(userId, sessionId) : await coordinator.serveNext(userId, sessionId, expectedPlanVersion)); diff --git a/backend/src/services/db.service.test.ts b/backend/src/services/db.service.test.ts index c660eeb..8ce3ace 100644 --- a/backend/src/services/db.service.test.ts +++ b/backend/src/services/db.service.test.ts @@ -361,6 +361,95 @@ describe('DbService v2 methods', () => { expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('SET committed = true')); }); + it('advances a served unplayable item with a separate idempotent event and commits one replacement', async () => { + const { service, clientQuery } = makeTransactionalService(); + const replacement = { + plan_version_id: 'plan-1', ordinal: 1, track_id: 'track-2', slot_role: null, + candidate_source: 'discovery', score: 0.8, 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: [] }) // no prior playback_error event + .mockResolvedValueOnce({ rows: [{ track_id: 'track-1', ordinal: 0 }] }) // current served cursor + .mockResolvedValueOnce({ rows: [{ id: 'error-1' }] }) // playback_error event + .mockResolvedValueOnce({ rows: [replacement] }) // commit replacement + .mockResolvedValueOnce({ rows: [] }) // replacement track_served event + .mockResolvedValueOnce({ rows: [] }) // playback_error result payload + .mockResolvedValueOnce({ rows: [] }) // session timestamp + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.advancePastUnplayableVibePlanItem('session-1', 'user-1', { + expectedPlanVersion: 1, + planVersionId: 'plan-1', + ordinal: 0, + trackId: 'track-1', + eventId: 'event-1', + })).resolves.toEqual({ item: replacement, stale: false }); + + expect(clientQuery.mock.calls[5][0]).toContain("'playback_error'"); + expect(clientQuery.mock.calls[6][0]).toContain('SET committed = true'); + expect(clientQuery.mock.calls[7][0]).toContain("'track_served'"); + expect(clientQuery.mock.calls[8][0]).toContain('UPDATE vibe_events SET payload'); + }); + + it('refuses an old served cursor when events share a timestamp by ordering the immutable plan ordinal', async () => { + const { service, clientQuery } = makeTransactionalService(); + clientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [{ status: 'active' }] }) // session lock + .mockResolvedValueOnce({ rows: [{ id: 'plan-1', version: 1 }] }) // latest plan + .mockResolvedValueOnce({ rows: [] }) // no prior playback_error event + // The lower-ordinal event can have a lexically greater UUID at the + // same occurred_at. The cursor must still be the highest immutable + // plan ordinal, never whichever UUID sorts last. + .mockResolvedValueOnce({ rows: [{ track_id: 'track-2', ordinal: 1 }] }) // current served cursor + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK + + await expect(service.advancePastUnplayableVibePlanItem('session-1', 'user-1', { + expectedPlanVersion: 1, + planVersionId: 'plan-1', + ordinal: 0, + trackId: 'track-1', + eventId: 'event-1', + })).rejects.toThrow('not the current served cursor'); + + expect(clientQuery.mock.calls[4][0]).toContain('JOIN vibe_plan_items'); + expect(clientQuery.mock.calls[4][0]).toContain('ORDER BY i.ordinal DESC'); + expect(clientQuery.mock.calls[4][0]).not.toContain('id DESC'); + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('SET committed = true')); + }); + + it('retries an unplayable advancement with the same event id without consuming another item', async () => { + const { service, clientQuery } = makeTransactionalService(); + const replacement = { + plan_version_id: 'plan-1', ordinal: 1, track_id: 'track-2', slot_role: null, + candidate_source: 'discovery', score: 0.8, 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: [{ type: 'playback_error', payload: { + planVersionId: 'plan-1', ordinal: 0, trackId: 'track-1', + advancedTo: { planVersionId: 'plan-1', ordinal: 1 }, + } }] }) // prior explicit advancement + .mockResolvedValueOnce({ rows: [replacement] }) // canonical replacement + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await expect(service.advancePastUnplayableVibePlanItem('session-1', 'user-1', { + expectedPlanVersion: 1, + planVersionId: 'plan-1', + ordinal: 0, + trackId: 'track-1', + eventId: 'event-1', + })).resolves.toEqual({ item: replacement, stale: false }); + + expect(clientQuery.mock.calls.map(([sql]) => sql)).not.toContain(expect.stringContaining('SET committed = true')); + expect(clientQuery.mock.calls).toHaveLength(6); + }); + 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 a256ebb..7c3cb60 100644 --- a/backend/src/services/db.service.ts +++ b/backend/src/services/db.service.ts @@ -2073,6 +2073,153 @@ export class DbService { }); } + /** + * Advance an already-served item which the player could not resolve (for + * example, its file was hidden after the revision was published). Unlike a + * normal version-aware /next retry, this has a distinct client event id and + * therefore intentionally moves beyond the item previously served for that + * revision. The event records both the rejected item and the replacement so + * a lost response can be retried without consuming another plan item. + */ + async advancePastUnplayableVibePlanItem( + sessionId: string, + userId: string, + input: { + expectedPlanVersion: number; + planVersionId: string; + ordinal: number; + trackId: string; + eventId: string; + }, + ): 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 sessionRow = session.rows[0] as Pick | undefined; + if (!sessionRow) throw new Error('Vibe session was not found or is not owned by this user'); + + 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 (plan.version !== input.expectedPlanVersion || plan.id !== input.planVersionId) { + return { item: null, stale: true }; + } + + // Idempotency is scoped to this explicit advancement operation, rather + // than overloading the version-aware /next retry which must keep + // returning the originally served item. + const prior = await client.query( + `SELECT type, payload + FROM vibe_events + WHERE session_id = $1 AND client_event_id = $2::uuid`, + [sessionId, input.eventId], + ); + const priorEvent = prior.rows[0] as Pick | undefined; + if (priorEvent && priorEvent.type !== 'playback_error') { + throw new Error('Vibe client event id was already used for a different event'); + } + const priorPayload = priorEvent?.payload as Record | undefined; + if (priorPayload) { + if (priorPayload.planVersionId !== input.planVersionId + || priorPayload.ordinal !== input.ordinal + || priorPayload.trackId !== input.trackId) { + throw new Error('Vibe playback-error event does not match the served plan item'); + } + const advancedTo = priorPayload.advancedTo as { planVersionId?: unknown; ordinal?: unknown } | null | undefined; + if (!advancedTo || typeof advancedTo.planVersionId !== 'string' || !Number.isInteger(advancedTo.ordinal)) { + return { item: null, stale: false }; + } + const replacement = await client.query( + `SELECT * FROM vibe_plan_items WHERE plan_version_id = $1 AND ordinal = $2`, + [advancedTo.planVersionId, advancedTo.ordinal], + ); + return { item: (replacement.rows[0] as VibePlanItem | undefined) ?? null, stale: false }; + } + + if (sessionRow.status !== 'active') { + throw new Error(`Cannot record a new event for ${sessionRow.status} Vibe session`); + } + + // A client may advance only the cursor it was just served. Checking for + // any historical serve event would let an old version-aware /next + // response consume whichever future item happens to be uncommitted. + const current = await client.query( + `SELECT i.track_id, i.ordinal + FROM vibe_events e + JOIN vibe_plan_items i + ON i.plan_version_id = $3::uuid + AND i.ordinal = (e.payload->>'ordinal')::integer + AND i.track_id = e.track_id + WHERE e.session_id = $1 + AND e.type = 'track_served' + AND e.payload->>'planVersion' = $2::text + AND e.payload->>'planVersionId' = $3 + ORDER BY i.ordinal DESC + LIMIT 1`, + [sessionId, input.expectedPlanVersion, input.planVersionId], + ); + const currentCursor = current.rows[0] as Pick | undefined; + if (currentCursor?.track_id !== input.trackId || currentCursor.ordinal !== input.ordinal) { + throw new Error('Vibe plan item is not the current served cursor for this session revision'); + } + + const payload = { + kind: 'unplayable_plan_item', + planVersion: input.expectedPlanVersion, + planVersionId: input.planVersionId, + ordinal: input.ordinal, + trackId: input.trackId, + }; + const playbackError = await client.query( + `INSERT INTO vibe_events (client_event_id, session_id, user_id, track_id, type, occurred_at, payload) + VALUES ($1::uuid, $2, $3, $4::uuid, 'playback_error', NOW(), $5::jsonb) + RETURNING id`, + [input.eventId, sessionId, userId, input.trackId, JSON.stringify(payload)], + ); + const eventId = playbackError.rows[0]?.id as string | undefined; + if (!eventId) throw new Error('Vibe playback-error event could not be recorded'); + + 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 = $1 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.*`, + [plan.id], + ); + const replacement = item.rows[0] as VibePlanItem | undefined; + if (replacement) { + 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, replacement.track_id, JSON.stringify({ + planVersion: plan.version, + planVersionId: replacement.plan_version_id, + ordinal: replacement.ordinal, + })], + ); + } + await client.query( + `UPDATE vibe_events SET payload = $2::jsonb WHERE id = $1`, + [eventId, JSON.stringify({ + ...payload, + advancedTo: replacement + ? { planVersionId: replacement.plan_version_id, ordinal: replacement.ordinal } + : null, + })], + ); + await client.query(`UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`, [sessionId]); + return { item: replacement ?? null, 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( diff --git a/backend/src/services/vibe-session-coordinator.service.test.ts b/backend/src/services/vibe-session-coordinator.service.test.ts index bd05625..a1f5e4f 100644 --- a/backend/src/services/vibe-session-coordinator.service.test.ts +++ b/backend/src/services/vibe-session-coordinator.service.test.ts @@ -47,6 +47,7 @@ function setup() { 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 }), + advancePastUnplayableVibePlanItem: 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; @@ -179,6 +180,25 @@ describe('VibeSessionCoordinator', () => { expect(result).toMatchObject({ planVersion: 1, now: { track_id: TRACK_ID } }); }); + it('advances past an unplayable served item using a distinct idempotent event protocol', async () => { + const { db, coordinator } = setup(); + const planVersionId = '44444444-4444-4444-8444-444444444444'; + const eventId = '55555555-5555-4555-8555-555555555555'; + + const result = await coordinator.advancePastUnplayable('user-1', SESSION_ID, { + expectedPlanVersion: 1, + planVersionId, + ordinal: 0, + trackId: TRACK_ID, + eventId, + }); + + expect(db.advancePastUnplayableVibePlanItem).toHaveBeenCalledWith(SESSION_ID, 'user-1', { + expectedPlanVersion: 1, planVersionId, ordinal: 0, trackId: TRACK_ID, eventId, + }); + expect(result.now).toMatchObject({ 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); diff --git a/backend/src/services/vibe-session-coordinator.service.ts b/backend/src/services/vibe-session-coordinator.service.ts index 79efeaa..c52f846 100644 --- a/backend/src/services/vibe-session-coordinator.service.ts +++ b/backend/src/services/vibe-session-coordinator.service.ts @@ -36,6 +36,19 @@ export interface AppendVibeEventInput { payload?: Record; } +/** + * An explicit advancement protocol for a plan item that was durably served + * but cannot be played locally. `eventId` is the idempotency key for this + * state transition; it is intentionally separate from a retry of /next. + */ +export interface AdvanceUnplayableVibeItemInput { + expectedPlanVersion: number; + planVersionId: string; + ordinal: number; + trackId: string; + eventId: string; +} + export class VibeSessionNotFoundError extends Error {} export class VibeSessionLifecycleError extends Error {} export class VibePlanNotFoundError extends Error {} @@ -142,6 +155,21 @@ export class VibeSessionCoordinator { } } + async advancePastUnplayable( + userId: string, + sessionId: string, + input: AdvanceUnplayableVibeItemInput, + ): Promise { + try { + const served = await this.db.advancePastUnplayableVibePlanItem(sessionId, userId, input); + const response = await this.getPlan(userId, sessionId); + 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, diff --git a/docker-compose.yml b/docker-compose.yml index eb5ccf4..f9265a6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -32,6 +32,10 @@ services: TYPESENSE_API_KEY: ${TYPESENSE_API_KEY} MUZICK_API_KEY: ${MUZICK_API_KEY} MUZICK_ADMIN_KEY: ${MUZICK_ADMIN_KEY} + # Durable Vibe sessions are intentionally bound to this configured, + # server-trusted owner instead of accepting a client-supplied user id. + # Set it to the UUID of the local Muzick user in .env. + MUZICK_VIBE_USER_ID: ${MUZICK_VIBE_USER_ID} MUSIC_DIR: /music volumes: # READ-ONLY, deliberately. Nothing in the API request path may write to diff --git a/frontend/src/components/AudioEngine.test.tsx b/frontend/src/components/AudioEngine.test.tsx new file mode 100644 index 0000000..d37e2c4 --- /dev/null +++ b/frontend/src/components/AudioEngine.test.tsx @@ -0,0 +1,45 @@ +import { render, waitFor } from '@testing-library/react'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { Track } from '../types'; +import { usePlaybackStore } from '../store/usePlaybackStore'; +import { useVibeStore } from '../store/useVibeStore'; + +const { advancePastUnplayableVibeTrack, reportVibeEvent } = vi.hoisted(() => ({ + advancePastUnplayableVibeTrack: vi.fn().mockResolvedValue(undefined), + reportVibeEvent: vi.fn().mockResolvedValue(undefined), +})); +vi.mock('../services/vibeSession', () => ({ advancePastUnplayableVibeTrack, reportVibeEvent })); + +import { AudioEngine } from './AudioEngine'; + +const track = (id: string): Track => ({ + id, path: `/music/${id}.mp3`, hash: id, title: id, artist: 'Artist', album_id: 'album', + duration: 180, state: 'LIBRARY', source_type: 'MANUAL', play_count: 0, skip_count: 0, dislike_count: 0, +}); + +describe('AudioEngine', () => { + beforeEach(() => { + vi.spyOn(HTMLMediaElement.prototype, 'load').mockImplementation(() => undefined); + vi.spyOn(HTMLMediaElement.prototype, 'play').mockResolvedValue(undefined); + useVibeStore.getState().reset(); + useVibeStore.getState().setActiveSession({ sessionId: 'session-a', seedTrackId: 'song' }); + usePlaybackStore.setState({ + currentTrack: track('song'), queue: [track('song')], currentIndex: 0, isPlaying: false, + queueOwner: 'vibe', vibeAdvanceHandler: () => undefined, + }); + }); + + afterEach(() => vi.restoreAllMocks()); + + it('uses the durable unplayable advancement when a Vibe stream errors after metadata resolved', async () => { + const { container } = render(); + const audio = container.querySelector('audio')!; + + audio.dispatchEvent(new Event('error')); + audio.dispatchEvent(new Event('error')); + + await waitFor(() => expect(advancePastUnplayableVibeTrack).toHaveBeenCalledWith('song')); + expect(advancePastUnplayableVibeTrack).toHaveBeenCalledTimes(1); + expect(reportVibeEvent).not.toHaveBeenCalledWith('skipped', 'song'); + }); +}); diff --git a/frontend/src/components/AudioEngine.tsx b/frontend/src/components/AudioEngine.tsx index 1d85399..e68ac84 100644 --- a/frontend/src/components/AudioEngine.tsx +++ b/frontend/src/components/AudioEngine.tsx @@ -2,17 +2,9 @@ import { useEffect, useRef } from 'react'; import { usePlaybackStore } from '../store/usePlaybackStore'; import { useVibeStore } from '../store/useVibeStore'; import { trackService } from '../services/trackService'; -import { vibeService } from '../services/vibeService'; +import { advancePastUnplayableVibeTrack, reportVibeEvent } from '../services/vibeSession'; import type { Track } from '../types'; -// Track ids whose next natural feedback transition should be skipped because -// the caller (e.g. Vibe.tsx's dislike button) already recorded feedback for -// them explicitly. Consumed once, then cleared. -const suppressedFeedbackIds = new Set(); -export function suppressAutoFeedback(trackId: string): void { - suppressedFeedbackIds.add(trackId); -} - // Threshold (seconds) above which a store position change is treated as a user // scrub and applied to the audio element. Keeps the timeupdate -> setPosition -> // effect loop from fighting itself. @@ -48,6 +40,8 @@ export const AudioEngine = () => { const endedNaturallyRef = useRef(false); // Track whether the current track has crossed the completion threshold. const crossedThresholdRef = useRef(false); + const lastProgressSecondRef = useRef(-1); + const streamErrorTrackIdRef = useRef(null); // --- DOM -> store: media events ----------------------------------------- useEffect(() => { @@ -66,6 +60,13 @@ export const AudioEngine = () => { ) { crossedThresholdRef.current = true; } + const vibe = useVibeStore.getState(); + const track = store().currentTrack; + const elapsed = Math.floor(audio.currentTime); + if (vibe.activeSessionId && store().queueOwner === 'vibe' && track && elapsed > 0 && elapsed % 30 === 0 && elapsed !== lastProgressSecondRef.current) { + lastProgressSecondRef.current = elapsed; + void reportVibeEvent('progress', track.id, Math.round(audio.currentTime * 1000), Math.round((audio.duration || 0) * 1000)).catch(() => undefined); + } }; const onLoadedMetadata = () => { if (Number.isFinite(audio.duration)) store().setDuration(audio.duration); @@ -83,7 +84,20 @@ export const AudioEngine = () => { // feedback, on the resulting track-change, so completion is recorded // exactly once per track. endedNaturallyRef.current = true; - store().next(); + store().nextWithReason('completed'); + }; + const onError = () => { + const playback = store(); + const track = playback.currentTrack; + const vibe = useVibeStore.getState(); + // Metadata can be available while the stream itself is no longer + // readable. Vibe must advance that exact durable cursor, not fall back + // to ordinary queue navigation or feedback-driven replanning. + if (!track || !vibe.activeSessionId || playback.queueOwner !== 'vibe' || streamErrorTrackIdRef.current === track.id) return; + streamErrorTrackIdRef.current = track.id; + void advancePastUnplayableVibeTrack(track.id) + .catch(() => undefined) + .finally(() => { streamErrorTrackIdRef.current = null; }); }; audio.addEventListener('timeupdate', onTimeUpdate); @@ -91,6 +105,7 @@ export const AudioEngine = () => { audio.addEventListener('play', onPlay); audio.addEventListener('pause', onPause); audio.addEventListener('ended', onEnded); + audio.addEventListener('error', onError); return () => { audio.removeEventListener('timeupdate', onTimeUpdate); @@ -98,6 +113,7 @@ export const AudioEngine = () => { audio.removeEventListener('play', onPlay); audio.removeEventListener('pause', onPause); audio.removeEventListener('ended', onEnded); + audio.removeEventListener('error', onError); }; }, []); @@ -109,28 +125,19 @@ export const AudioEngine = () => { const applyTrack = (id: string | null) => { if (id === loadedIdRef.current) return; - // The previously loaded track is changing. If it didn't end naturally and - // hadn't crossed the completion threshold, record a skip (best-effort). - // If it crossed the threshold OR ended naturally, record as completed. + // The durable Vibe controller owns normal next/ended navigation. It + // records the outcome, receives a new plan revision, then calls the raw + // advance method. Do not emit a second event here after that transition. const prevId = loadedIdRef.current; const completed = endedNaturallyRef.current || crossedThresholdRef.current; - // Only vibe sessions want this feedback — plain library browsing - // shouldn't write skip/completed evidence for tracks merely sampled. - const inVibeSession = !!useVibeStore.getState().activeSessionId; - if (prevId && inVibeSession) { - if (suppressedFeedbackIds.delete(prevId)) { - // Caller already recorded explicit feedback (e.g. dislike) for - // this track — don't also record the implicit transition. - } else { - try { - void vibeService.feedback(prevId, completed ? 'completed' : 'skipped', useVibeStore.getState().activeSessionId ?? undefined).catch(() => {}); - } catch { - /* best-effort */ - } - } + const playback = usePlaybackStore.getState(); + const inVibePlayback = !!useVibeStore.getState().activeSessionId && playback.queueOwner === 'vibe'; + if (prevId && inVibePlayback && !playback.vibeAdvanceHandler) { + void reportVibeEvent(completed ? 'completed' : 'skipped', prevId).catch(() => undefined); } endedNaturallyRef.current = false; crossedThresholdRef.current = false; + lastProgressSecondRef.current = -1; loadedIdRef.current = id; if (!id) { @@ -141,6 +148,7 @@ export const AudioEngine = () => { audio.src = trackService.getStreamUrl(id); audio.load(); + if (inVibePlayback) void reportVibeEvent('playback_started', id).catch(() => undefined); if (usePlaybackStore.getState().isPlaying) { void audio.play().catch(() => {}); } diff --git a/frontend/src/components/TrackRow.tsx b/frontend/src/components/TrackRow.tsx index 3ef7484..aa4a2f1 100644 --- a/frontend/src/components/TrackRow.tsx +++ b/frontend/src/components/TrackRow.tsx @@ -26,9 +26,11 @@ interface TrackRowProps { showVibe?: boolean; /** Override ordinary queue playback, for contextual actions such as Vibe seed rows. */ onSelect?: (track: Track) => void; + /** Display-only rows keep their surrounding playback controller authoritative. */ + playable?: boolean; } -export function TrackRow({ track, queue, showActions = true, variant = 'default', showVibe = false, onSelect }: TrackRowProps) { +export function TrackRow({ track, queue, showActions = true, variant = 'default', showVibe = false, onSelect, playable = true }: TrackRowProps) { const { setQueue, playTrack, play, pause, currentTrack, isPlaying } = usePlaybackStore(); const dislikeTrack = useDislikeTrack(); const router = useRouter(); @@ -36,6 +38,7 @@ export function TrackRow({ track, queue, showActions = true, variant = 'default' const compact = variant === 'compact'; const handlePlay = () => { + if (!playable) return; if (onSelect) { onSelect(track); return; @@ -47,7 +50,9 @@ export function TrackRow({ track, queue, showActions = true, variant = 'default' playTrack(track); }; - const playLabel = isCurrent && isPlaying + const playLabel = !playable + ? `${track.title || 'Track'} is queued by Vibe` + : isCurrent && isPlaying ? `Pause ${track.title || 'track'}` : `Play ${track.title || 'track'}`; @@ -81,8 +86,9 @@ export function TrackRow({ track, queue, showActions = true, variant = 'default' - - )} + {planVersion &&

Plan revision {planVersion}; upcoming tracks may change as you listen.

} {currentTrack && (
diff --git a/frontend/src/services/vibeService.test.ts b/frontend/src/services/vibeService.test.ts index 0f45bfa..6542ac2 100644 --- a/frontend/src/services/vibeService.test.ts +++ b/frontend/src/services/vibeService.test.ts @@ -1,32 +1,41 @@ -import { AxiosError } from 'axios'; import { describe, expect, it, vi } from 'vitest'; -import type { Track } from '../types'; -import { fetchNextBatch, vibeService } from './vibeService'; -const track = (id: string): Track => ({ - id, path: `/music/${id}.mp3`, hash: id, title: id, artist: 'Artist', - album_id: 'album', duration: 180, state: 'LIBRARY', source_type: 'MANUAL', - play_count: 0, skip_count: 0, dislike_count: 0, -}); +const { post, get } = vi.hoisted(() => ({ post: vi.fn(), get: vi.fn() })); +vi.mock('./api', () => ({ default: { post, get } })); -function responseError(status: number, code?: string) { - return new AxiosError('request failed', undefined, undefined, undefined, { - data: code ? { code } : {}, status, statusText: 'error', headers: {}, config: {} as never, - }); -} +import { vibeService } from './vibeService'; -describe('fetchNextBatch', () => { - it('uses the supplied session id and treats VIBE_PLAN_EXHAUSTED as terminal', async () => { - const next = vi.spyOn(vibeService, 'next') - .mockResolvedValueOnce({ track: track('one'), explanation: null, planRemaining: 0 }) - .mockRejectedValueOnce(responseError(409, 'VIBE_PLAN_EXHAUSTED')); +describe('durable vibe service', () => { + it('serves the next item with the caller plan version', async () => { + post.mockResolvedValue({ data: { sessionId: 'session', planVersion: 3, now: null, preview: [] } }); - await expect(fetchNextBatch(3, 'session-a')).resolves.toEqual({ tracks: [track('one')], status: 'exhausted' }); - expect(next).toHaveBeenCalledWith('session-a'); + await vibeService.next('session', 3); + + expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/next', { expectedPlanVersion: 3 }); }); - it('does not disguise a missing or replaced session as normal exhaustion', async () => { - vi.spyOn(vibeService, 'next').mockRejectedValue(responseError(404)); - await expect(fetchNextBatch(1, 'expired-session')).resolves.toEqual({ tracks: [], status: 'failed' }); + it('uses an explicit idempotency key to advance a served but unplayable item', async () => { + post.mockResolvedValue({ data: { sessionId: 'session', planVersion: 3, now: null, preview: [] } }); + + await vibeService.advancePastUnplayable('session', 3, { + eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track', + }); + + expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/next', { + expectedPlanVersion: 3, + unplayable: { eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track' }, + }); + }); + + it('sends client event ids to the durable event ledger', async () => { + post.mockResolvedValue({ data: { sessionId: 'session', planVersion: 2, now: null, preview: [] } }); + + await vibeService.event('session', { + eventId: 'event', type: 'progress', occurredAt: '2026-01-01T00:00:00.000Z', trackId: 'track', positionMs: 30000, + }); + + expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/events', expect.objectContaining({ + eventId: 'event', type: 'progress', trackId: 'track', positionMs: 30000, + })); }); }); diff --git a/frontend/src/services/vibeService.ts b/frontend/src/services/vibeService.ts index 600e02e..7b22343 100644 --- a/frontend/src/services/vibeService.ts +++ b/frontend/src/services/vibeService.ts @@ -1,88 +1,101 @@ import api from './api'; -import axios from 'axios'; -import type { Track } from '../types'; -// A candidate from the v2 recommendation plan. The plan is stored server-side -// in Redis; the frontend only needs trackId + explanation for display. +// These types intentionally mirror the durable session API. Tracks are not +// embedded in a plan revision: the client resolves ids through the normal +// library endpoint so a deleted/hidden track can never become playable merely +// because an older plan mentioned it. export interface VibePlanItem { - trackId: string; - generatorId: string; - explanation: unknown[]; - relevance: number; + 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 VibeStartResponse { +export interface DurableVibeSessionResponse { sessionId: string; - plan: VibePlanItem[]; + planVersion: number | null; + now: VibePlanItem | null; + preview: VibePlanItem[]; + state: Record; + replanned: boolean; + replanReason: string | null; } -export interface VibeNextResponse { - track: Track; - explanation: unknown[] | null; - planRemaining: number; +export type VibeEventType = + | 'playback_started' + | 'progress' + | 'completed' + | 'skipped' + | 'disliked' + | 'kept'; + +export interface VibeEventInput { + eventId: string; + type: VibeEventType; + trackId?: string; + occurredAt: string; + positionMs?: number; + durationMs?: number; + payload?: Record; } -export type VibeBatchStatus = 'complete' | 'exhausted' | 'failed'; - -export interface VibeBatchResult { - tracks: Track[]; - status: VibeBatchStatus; +export interface VibeEventResponse extends DurableVibeSessionResponse { + event: { id: string; client_event_id: string | null; type: string }; + idempotent: boolean; } -export type VibeFeedbackAction = 'completed' | 'skipped' | 'promoted' | 'disliked'; +/** A durable, idempotent advancement past a plan item the player cannot load. */ +export interface VibeUnplayableItemInput { + eventId: string; + planVersionId: string; + ordinal: number; + trackId: string; +} -// V2 recommendation engine. The backend stores the plan in Redis (2h TTL) and -// serves tracks one at a time via GET /next. Feedback triggers replanning. export const vibeService = { - // POST /api/v2/vibe/start { seedTrackId? } -> { sessionId, plan } - async start(seedTrackId?: string): Promise { - const res = await api.post('/v2/vibe/start', { seedTrackId }); + async start(seedTrackId?: string): Promise { + const res = await api.post('/v2/vibe/sessions', { seedTrackId }); return res.data; }, - // GET /api/v2/vibe/next -> { track, explanation, planRemaining } - // Returns one track at a time, shifting the server-side plan. - // 404 if no active plan — caller should handle gracefully. - async next(sessionId: string): Promise { - const res = await api.get('/v2/vibe/next', { params: { sessionId } }); + async getPlan(sessionId: string, version?: number): Promise { + const res = await api.get(`/v2/vibe/sessions/${sessionId}/plans`, { + params: version === undefined ? undefined : { version }, + }); return res.data; }, - // POST /api/v2/vibe/feedback { trackId, action } -> { status, planRemaining } - // Action 'promoted' also calls addFavorite; 'disliked' also calls dislikeTrack. - // Triggers replan of the remaining plan. - async feedback(trackId: string, action: VibeFeedbackAction, sessionId?: string): Promise<{ status: string; planRemaining: number }> { - const res = await api.post<{ status: string; planRemaining: number }>('/v2/vibe/feedback', { trackId, action, sessionId }); + async next(sessionId: string, expectedPlanVersion: number): Promise { + const res = await api.post(`/v2/vibe/sessions/${sessionId}/next`, { + expectedPlanVersion, + }); return res.data; }, - // GET /api/v2/vibe/plan -> { sessionId, planRemaining, plan } - // Debug endpoint — returns the full remaining plan. - async getPlan(): Promise<{ sessionId: string; planRemaining: number; plan: VibePlanItem[] }> { - const res = await api.get('/v2/vibe/plan'); + async advancePastUnplayable( + sessionId: string, + expectedPlanVersion: number, + unplayable: VibeUnplayableItemInput, + ): Promise { + const res = await api.post(`/v2/vibe/sessions/${sessionId}/next`, { + expectedPlanVersion, + unplayable, + }); + return res.data; + }, + + async event(sessionId: string, event: VibeEventInput): Promise { + const res = await api.post(`/v2/vibe/sessions/${sessionId}/events`, event); + return res.data; + }, + + async end(sessionId: string): Promise { + const res = await api.post(`/v2/vibe/sessions/${sessionId}/end`); return res.data; }, }; - -// Fetch N tracks from the v2 plan sequentially. Each call to /next shifts the -// server-side plan, so calls must be sequential (not parallel). Stops early on -// 409/VIBE_PLAN_EXHAUSTED is a normal terminal condition. A missing/replaced -// session is intentionally reported as a failure so callers can preserve the -// current playback state rather than pretending the plan completed cleanly. -export async function fetchNextBatch(count: number, sessionId: string): Promise { - const tracks: Track[] = []; - for (let i = 0; i < count; i++) { - try { - const { track } = await vibeService.next(sessionId); - tracks.push(track); - } catch (error) { - return { - tracks, - status: axios.isAxiosError(error) && error.response?.status === 409 && - (error.response.data as { code?: string } | undefined)?.code === 'VIBE_PLAN_EXHAUSTED' - ? 'exhausted' : 'failed', - }; - } - } - return { tracks, status: 'complete' }; -} diff --git a/frontend/src/services/vibeSession.test.ts b/frontend/src/services/vibeSession.test.ts index f2895a4..6a610ea 100644 --- a/frontend/src/services/vibeSession.test.ts +++ b/frontend/src/services/vibeSession.test.ts @@ -1,21 +1,16 @@ +import { AxiosError } from 'axios'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import type { Track } from '../types'; import { usePlaybackStore } from '../store/usePlaybackStore'; import { useVibeStore } from '../store/useVibeStore'; -const { next, start } = vi.hoisted(() => ({ next: vi.fn(), start: vi.fn() })); -vi.mock('./vibeService', () => ({ - vibeService: { start, next }, - fetchNextBatch: async (count: number, sessionId: string) => { - const tracks: Track[] = []; - for (let index = 0; index < count; index++) { - try { tracks.push((await next(sessionId)).track); } catch { return { tracks, status: 'failed' as const }; } - } - return { tracks, status: 'complete' as const }; - }, +const { start, next, advancePastUnplayable, event, end, getTrack } = vi.hoisted(() => ({ + start: vi.fn(), next: vi.fn(), advancePastUnplayable: vi.fn(), event: vi.fn(), end: vi.fn(), getTrack: vi.fn(), })); +vi.mock('./vibeService', () => ({ vibeService: { start, next, advancePastUnplayable, event, end } })); +vi.mock('./trackService', () => ({ trackService: { getTrack } })); -import { startVibeSession } from './vibeSession'; +import { advancePastUnplayableVibeTrack, advanceVibe, endVibeSession, reportVibeEvent, startVibeSession } from './vibeSession'; const track = (id: string): Track => ({ id, path: `/music/${id}.mp3`, hash: id, title: id, artist: 'Artist', @@ -23,34 +18,237 @@ const track = (id: string): Track => ({ play_count: 0, skip_count: 0, dislike_count: 0, }); -describe('startVibeSession', () => { +const item = (track_id: string, committed = false, ordinal = 0) => ({ + plan_version_id: 'plan', ordinal, track_id, slot_role: null, candidate_source: 'test', + score: 1, score_breakdown: {}, explanation: [], committed, +}); + +const response = (planVersion: number, now = item('one', true), preview = [item('two')]) => ({ + sessionId: 'session-a', planVersion, now, preview, state: {}, replanned: false, replanReason: null, +}); + +describe('durable Vibe session client', () => { beforeEach(() => { vi.clearAllMocks(); useVibeStore.getState().reset(); - usePlaybackStore.setState({ currentTrack: null, queue: [], currentIndex: -1, isPlaying: false }); + usePlaybackStore.setState({ + currentTrack: null, queue: [], currentIndex: -1, isPlaying: false, vibeAdvanceHandler: null, queueOwner: 'ordinary', + }); + getTrack.mockImplementation((id: string) => Promise.resolve(track(id))); }); - it('does not replace a working Vibe when the new plan cannot hydrate', async () => { - const old = track('old'); - useVibeStore.getState().setActiveSession({ sessionId: 'old-session', seedTrackId: old.id }); - usePlaybackStore.getState().setQueue([old]); - usePlaybackStore.getState().playTrack(old); - start.mockResolvedValue({ sessionId: 'new-session', plan: [] }); - next.mockRejectedValue(new Error('missing session')); + it('starts by version-serving and hydrating the first durable plan item', async () => { + start.mockResolvedValue(response(1, item('one'), [item('two')])); + next.mockResolvedValue(response(1, item('one', true), [item('two')])); - await expect(startVibeSession(track('seed'))).resolves.toMatchObject({ tracks: [], status: 'failed' }); - expect(useVibeStore.getState().activeSessionId).toBe('old-session'); - expect(usePlaybackStore.getState().currentTrack?.id).toBe('old'); + await expect(startVibeSession(track('seed'))).resolves.toMatchObject({ status: 'complete', tracks: [track('one'), track('two')] }); + + expect(next).toHaveBeenCalledWith('session-a', 1); + expect(useVibeStore.getState()).toMatchObject({ activeSessionId: 'session-a', planVersion: 1, buffer: [track('two')] }); + expect(usePlaybackStore.getState().currentTrack).toEqual(track('one')); }); - it('serializes rapid starts and hydrates only one session', async () => { - const recommended = track('recommended'); - start.mockResolvedValue({ sessionId: 'session-a', plan: [] }); - next.mockResolvedValue({ track: recommended }); + it('replans, version-serves, and removes stale prefetched tracks before advancing', async () => { + start.mockResolvedValue(response(1, item('one'), [item('stale')])); + next + .mockResolvedValueOnce(response(1, item('one', true), [item('stale')])) + .mockResolvedValueOnce(response(2, item('two', true), [item('three')])); + event.mockResolvedValue({ ...response(2, item('two'), [item('three')]), replanned: true, event: {}, idempotent: false }); + await startVibeSession(track('seed')); - await Promise.all([startVibeSession(track('seed-a')), startVibeSession(track('seed-b'))]); - expect(start).toHaveBeenCalledTimes(1); - expect(next).toHaveBeenCalledWith('session-a'); - expect(useVibeStore.getState().activeSessionId).toBe('session-a'); + await advanceVibe('skipped'); + + expect(next).toHaveBeenLastCalledWith('session-a', 2); + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).toEqual(['one', 'two', 'three']); + expect(usePlaybackStore.getState().currentTrack?.id).toBe('two'); + expect(useVibeStore.getState().buffer.map((entry) => entry.id)).toEqual(['three']); + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).not.toContain('stale'); + expect(event).toHaveBeenCalledWith('session-a', expect.objectContaining({ type: 'skipped', trackId: 'one', eventId: expect.any(String) })); + }); + + it('replaces only the future when a keep event replans', async () => { + start.mockResolvedValue(response(1, item('one'), [item('stale')])); + next.mockResolvedValue(response(1, item('one', true), [item('stale')])); + event.mockResolvedValue({ ...response(2, item('fresh'), [item('fresh'), item('later')]), replanned: true, event: {}, idempotent: false }); + await startVibeSession(track('seed')); + + await reportVibeEvent('kept', 'one'); + + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).toEqual(['one', 'fresh', 'later']); + expect(usePlaybackStore.getState().currentTrack?.id).toBe('one'); + expect(useVibeStore.getState().planVersion).toBe(2); + }); + + it('reconciles the canonical replacement returned by an idempotent material-event retry', async () => { + start.mockResolvedValue(response(1, item('one'), [item('stale')])); + next.mockResolvedValue(response(1, item('one', true), [item('stale')])); + // The first response was lost after it published revision 2. Retrying the + // same client event returns that revision with replanned=false. + event.mockResolvedValue({ + ...response(2, item('fresh'), [item('fresh'), item('later')]), + replanned: false, + event: {}, + idempotent: true, + }); + await startVibeSession(track('seed')); + + await reportVibeEvent('kept', 'one'); + + expect(useVibeStore.getState()).toMatchObject({ planVersion: 2, buffer: [track('fresh'), track('later')] }); + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).toEqual(['one', 'fresh', 'later']); + }); + + it('cleans up local playback when the durable session is gone', async () => { + start.mockResolvedValue(response(1, item('one'), [item('stale')])); + next.mockResolvedValue(response(1, item('one', true), [item('stale')])); + event.mockRejectedValue(new AxiosError('gone', undefined, undefined, undefined, { + data: {}, status: 404, statusText: 'Not Found', headers: {}, config: {} as never, + })); + await startVibeSession(track('seed')); + + await expect(reportVibeEvent('kept', 'one')).rejects.toThrow('gone'); + + expect(useVibeStore.getState().activeSessionId).toBeNull(); + expect(usePlaybackStore.getState()).toMatchObject({ currentTrack: null, queue: [], isPlaying: false }); + }); + + it('hands ordinary playback back to browse queues without Vibe reporting or next interception', async () => { + start.mockResolvedValue(response(1, item('one'), [item('two')])); + next.mockResolvedValue(response(1, item('one', true), [item('two')])); + await startVibeSession(track('seed')); + + const ordinary = track('ordinary'); + const playback = usePlaybackStore.getState(); + playback.setQueue([ordinary, track('ordinary-next')]); + playback.playTrack(ordinary); + playback.nextWithReason('completed'); + + expect(usePlaybackStore.getState()).toMatchObject({ + queueOwner: 'ordinary', currentTrack: track('ordinary-next'), vibeAdvanceHandler: null, + }); + expect(event).not.toHaveBeenCalled(); + }); + + it('serializes material events and ignores an older plan revision', async () => { + start.mockResolvedValue(response(1, item('one'), [item('old')])); + next.mockResolvedValue(response(1, item('one', true), [item('old')])); + await startVibeSession(track('seed')); + + let resolveFirst!: (value: ReturnType & { event: object; idempotent: boolean }) => void; + event.mockImplementationOnce(() => new Promise((resolve) => { resolveFirst = resolve; })); + event.mockResolvedValueOnce({ ...response(1, item('stale'), [item('stale')]), replanned: true, event: {}, idempotent: false }); + + const first = reportVibeEvent('kept', 'one'); + const second = reportVibeEvent('completed', 'one'); + await Promise.resolve(); + expect(event).toHaveBeenCalledTimes(1); + resolveFirst({ ...response(2, item('fresh'), [item('fresh')]), replanned: true, event: {}, idempotent: false }); + await Promise.all([first, second]); + + expect(event).toHaveBeenCalledTimes(2); + expect(useVibeStore.getState()).toMatchObject({ planVersion: 2, buffer: [track('fresh')] }); + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).toEqual(['one', 'fresh']); + }); + + it('retries a failed event with the same idempotency key until it is acknowledged', async () => { + start.mockResolvedValue(response(1, item('one'), [item('two')])); + next.mockResolvedValue(response(1, item('one', true), [item('two')])); + event.mockRejectedValueOnce(new Error('network dropped')).mockResolvedValueOnce({ + ...response(1), event: {}, idempotent: true, + }); + await startVibeSession(track('seed')); + + await reportVibeEvent('progress', 'one', 30000, 180000); + + expect(event).toHaveBeenCalledTimes(2); + expect(event.mock.calls[0][1].eventId).toBe(event.mock.calls[1][1].eventId); + }); + + it('skips a hidden plan item and starts from the next playable item', async () => { + start.mockResolvedValue(response(1, item('hidden'), [item('good')])); + next + .mockResolvedValueOnce(response(1, item('hidden', true), [item('good')])); + advancePastUnplayable.mockResolvedValueOnce(response(1, item('good', true), [item('later')])); + getTrack.mockImplementation((id: string) => id === 'hidden' + ? Promise.resolve({ ...track(id), state: 'HIDDEN' }) + : Promise.resolve(track(id))); + + await expect(startVibeSession(track('seed'))).resolves.toMatchObject({ status: 'complete', tracks: [track('good'), track('later')] }); + + expect(next).toHaveBeenCalledTimes(1); + expect(advancePastUnplayable).toHaveBeenCalledWith('session-a', 1, expect.objectContaining({ + planVersionId: 'plan', ordinal: 0, trackId: 'hidden', eventId: expect.any(String), + })); + expect(usePlaybackStore.getState().currentTrack?.id).toBe('good'); + expect(usePlaybackStore.getState().queue.map((entry) => entry.id)).not.toContain('hidden'); + }); + + it('advances consecutive hidden replacements directly without replaying an older served cursor', async () => { + start.mockResolvedValue(response(1, item('hidden-one'), [item('hidden-two', false, 1)])); + next.mockResolvedValueOnce(response(1, item('hidden-one', true), [item('hidden-two', false, 1)])); + advancePastUnplayable + .mockResolvedValueOnce(response(1, item('hidden-two', true, 1), [item('good', false, 2)])) + .mockResolvedValueOnce(response(1, item('good', true, 2), [item('later', false, 3)])); + getTrack.mockImplementation((id: string) => ['hidden-one', 'hidden-two'].includes(id) + ? Promise.resolve({ ...track(id), state: 'HIDDEN' }) + : Promise.resolve(track(id))); + + await expect(startVibeSession(track('seed'))).resolves.toMatchObject({ + status: 'complete', tracks: [track('good'), track('later')], + }); + + expect(next).toHaveBeenCalledTimes(1); + expect(advancePastUnplayable).toHaveBeenCalledTimes(2); + expect(advancePastUnplayable.mock.calls.map(([, , input]) => [input.ordinal, input.trackId])) + .toEqual([[0, 'hidden-one'], [1, 'hidden-two']]); + expect(usePlaybackStore.getState().currentTrack?.id).toBe('good'); + }); + + it('retries an unplayable advancement with its original event id after a lost response', async () => { + start.mockResolvedValue(response(1, item('hidden'), [item('good')])); + next.mockResolvedValueOnce(response(1, item('hidden', true), [item('good')])); + advancePastUnplayable + .mockRejectedValueOnce(new Error('response dropped')) + .mockResolvedValueOnce(response(1, item('good', true), [item('later')])); + getTrack.mockImplementation((id: string) => id === 'hidden' + ? Promise.resolve({ ...track(id), state: 'MISSING' }) + : Promise.resolve(track(id))); + + await startVibeSession(track('seed')); + + expect(advancePastUnplayable).toHaveBeenCalledTimes(2); + expect(advancePastUnplayable.mock.calls[0][2].eventId) + .toBe(advancePastUnplayable.mock.calls[1][2].eventId); + }); + + it('advances a stream-error track through its stored durable cursor without ordinary feedback', async () => { + start.mockResolvedValue(response(1, item('one'), [item('two', false, 1)])); + next.mockResolvedValueOnce(response(1, item('one', true), [item('two', false, 1)])); + advancePastUnplayable.mockResolvedValueOnce(response(1, item('two', true, 1), [item('later', false, 2)])); + await startVibeSession(track('seed')); + + await advancePastUnplayableVibeTrack('one'); + + expect(advancePastUnplayable).toHaveBeenCalledWith('session-a', 1, expect.objectContaining({ + planVersionId: 'plan', ordinal: 0, trackId: 'one', eventId: expect.any(String), + })); + expect(event).not.toHaveBeenCalledWith('session-a', expect.objectContaining({ type: 'skipped', trackId: 'one' })); + expect(usePlaybackStore.getState().currentTrack?.id).toBe('two'); + expect(useVibeStore.getState().currentPlanItem).toMatchObject({ track_id: 'two', ordinal: 1 }); + }); + + it('ends a Vibe by removing Vibe ownership and clearing the local queue', async () => { + start.mockResolvedValue(response(1, item('one'), [item('two')])); + next.mockResolvedValue(response(1, item('one', true), [item('two')])); + end.mockResolvedValue(response(1)); + await startVibeSession(track('seed')); + + await endVibeSession(); + + expect(end).toHaveBeenCalledWith('session-a'); + expect(useVibeStore.getState().activeSessionId).toBeNull(); + expect(usePlaybackStore.getState()).toMatchObject({ + queueOwner: 'ordinary', vibeAdvanceHandler: null, currentTrack: null, queue: [], isPlaying: false, + }); }); }); diff --git a/frontend/src/services/vibeSession.ts b/frontend/src/services/vibeSession.ts index 30128a6..b6332da 100644 --- a/frontend/src/services/vibeSession.ts +++ b/frontend/src/services/vibeSession.ts @@ -1,25 +1,417 @@ +import axios from 'axios'; import type { Track } from '../types'; -import { usePlaybackStore } from '../store/usePlaybackStore'; +import { usePlaybackStore, type VibeAdvanceReason } from '../store/usePlaybackStore'; import { useVibeStore } from '../store/useVibeStore'; -import { fetchNextBatch, vibeService, type VibeBatchStatus } from './vibeService'; - -export const INITIAL_VIBE_BATCH_SIZE = 5; +import { + vibeService, + type DurableVibeSessionResponse, + type VibeEventType, + type VibePlanItem, +} from './vibeService'; +import { trackService } from './trackService'; export interface StartedVibeSession { - status: VibeBatchStatus; + status: 'complete' | 'exhausted' | 'failed'; tracks: Track[]; } let startInFlight: Promise | null = null; +let advanceInFlight: Promise | null = null; +let materialTail: Promise = Promise.resolve(); + +interface PendingEvent { + sessionId: string; + input: Parameters[1]; + retried: boolean; + settled: boolean; + resolve: (response: DurableVibeSessionResponse) => void; + reject: (error: unknown) => void; +} + +// The event ledger deduplicates client_event_id. Keep an event in this ordered +// outbox until the server acknowledges it so a transient failure never turns a +// retry into a second listener action. +const eventOutbox: PendingEvent[] = []; +let flushingOutbox = false; + +function serializeMaterial(operation: () => Promise): Promise { + const result = materialTail.then(operation, operation); + materialTail = result.then(() => undefined, () => undefined); + return result; +} + +function newEventId(): string { + if (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function') return crypto.randomUUID(); + // UUID v4-shaped fallback for older embedded webviews. The server only uses + // this as an idempotency key, not as a source of entropy. + return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (letter) => { + const value = Math.floor(Math.random() * 16); + return (letter === 'x' ? value : (value & 0x3) | 0x8).toString(16); + }); +} + +function isPlayable(track: Track): boolean { + return !['HIDDEN', 'MISSING', 'DELETED'].includes(track.state); +} + +async function hydrateItem(item: VibePlanItem | null): Promise { + if (!item) return null; + try { + const track = await trackService.getTrack(item.track_id); + return isPlayable(track) ? track : null; + } catch { + // A plan can outlive a hidden/deleted file. Never substitute another item + // for this ordinal: keeping the remaining order is safer than a mismatch. + return null; + } +} + +async function hydratePreview(items: VibePlanItem[]): Promise { + const uniqueIds = [...new Set(items.map((item) => item.track_id))]; + const loaded = await Promise.all(uniqueIds.map(async (id) => { + try { + const track = await trackService.getTrack(id); + return [id, isPlayable(track) ? track : null] as const; + } catch { + return [id, null] as const; + } + })); + const byId = new Map(loaded.filter((entry): entry is readonly [string, Track] => entry[1] !== null)); + return items.flatMap((item) => { + const track = byId.get(item.track_id); + return track ? [track] : []; + }); +} + +/** Replace only the queue after the currently playing Vibe track. */ +function replaceUnplayedQueue(preview: Track[]): void { + const playback = usePlaybackStore.getState(); + const current = playback.currentTrack; + const queueIndex = current + ? (playback.currentIndex >= 0 && playback.queue[playback.currentIndex]?.id === current.id + ? playback.currentIndex + : playback.queue.findIndex((track) => track.id === current.id)) + : -1; + const history = queueIndex >= 0 + ? playback.queue.slice(0, queueIndex + 1) + : current ? [current] : []; + const seen = new Set(history.map((track) => track.id)); + const future = preview.filter((track) => !seen.has(track.id)); + playback.setVibeQueue([...history, ...future]); +} + +function isCurrentVibeOwner(sessionId: string): boolean { + return useVibeStore.getState().activeSessionId === sessionId + && usePlaybackStore.getState().queueOwner === 'vibe'; +} + +async function reconcilePreview(sessionId: string, response: DurableVibeSessionResponse): Promise { + if (!isCurrentVibeOwner(sessionId)) return []; + const preview = await hydratePreview(response.preview); + if (!isCurrentVibeOwner(sessionId) || !useVibeStore.getState().setPlan(response.planVersion, preview)) return []; + replaceUnplayedQueue(preview); + return preview; +} + +async function serveNextCurrent(sessionId: string, version: number): Promise { + // A concurrent device or a feedback replan can make a version stale between + // the event response and /next. A stale response has an uncommitted `now`; + // refresh once with its latest version before admitting a track to playback. + let response = await vibeService.next(sessionId, version); + if (response.now?.committed) return response; + if (!response.planVersion || response.planVersion === version) return response; + response = await vibeService.next(sessionId, response.planVersion); + return response; +} + +async function serveNextPlayable( + sessionId: string, + version: number, +): Promise<{ response: DurableVibeSessionResponse; now: Track; preview: Track[] } | null> { + return resolvePlayableResponse(sessionId, await serveNextCurrent(sessionId, version)); +} + +async function advanceResponsePastUnplayable( + sessionId: string, + response: DurableVibeSessionResponse, +): Promise { + if (!response.now?.committed || !response.planVersion) return response; + const unplayable = { + eventId: newEventId(), + planVersionId: response.now.plan_version_id, + ordinal: response.now.ordinal, + trackId: response.now.track_id, + }; + try { + return await vibeService.advancePastUnplayable(sessionId, response.planVersion, unplayable); + } catch (error) { + // A response may have been lost after the server committed the advance. + // Retry the same event id so it returns the same replacement rather than + // consuming another future item. + if (isSessionTerminalError(error)) throw error; + return vibeService.advancePastUnplayable(sessionId, response.planVersion, unplayable); + } +} + +async function resolvePlayableResponse( + sessionId: string, + initialResponse: DurableVibeSessionResponse, +): Promise<{ response: DurableVibeSessionResponse; now: Track; preview: Track[] } | null> { + let response = initialResponse; + // A durable plan can reference a file which has since become hidden. Commit + // past such entries but never load one into the player. An unplayable + // advancement already returns and commits its replacement, so process that + // response directly: asking ordinary /next again would replay the original + // served cursor rather than advancing through consecutive hidden entries. + for (let attempts = 0; attempts < 20; attempts++) { + if (!response.now?.committed || !response.planVersion) { + if (!response.planVersion) return null; + response = await serveNextCurrent(sessionId, response.planVersion); + continue; + } + let now = await hydrateItem(response.now); + let preview = await hydratePreview(response.preview); + if (now) return { response, now, preview }; + const advanced = await advanceResponsePastUnplayable(sessionId, response); + if (!advanced.planVersion) return null; + // The unplayable transition may itself race a feedback replan. Its stale + // response did not advance the old revision, so version-serve the current + // revision normally rather than treating a preview item as committed. + if (!advanced.now?.committed) { + response = await serveNextCurrent(sessionId, advanced.planVersion); + continue; + } + response = advanced; + } + return null; +} + +function deactivateBrokenSession(): void { + const playback = usePlaybackStore.getState(); + playback.setVibeAdvanceHandler(null); + useVibeStore.getState().reset(); + playback.pause(); + playback.setQueue([]); + playback.setCurrentTrack(null); +} + +function isSessionTerminalError(error: unknown): boolean { + return axios.isAxiosError(error) && [401, 404, 409].includes(error.response?.status ?? 0); +} + +export function vibeErrorMessage(error: unknown): string { + if (!axios.isAxiosError(error)) return 'Could not refresh this Vibe. Please try again.'; + switch (error.response?.status) { + case 401: return 'Vibe needs a trusted local user identity. Set MUZICK_VIBE_USER_ID and try again.'; + case 404: return 'This Vibe session is no longer available.'; + case 409: return 'This Vibe session has already ended or was replaced.'; + default: return 'Could not refresh this Vibe. Please try again.'; + } +} + +async function sendEvent( + type: VibeEventType, + trackId?: string, + positionMs?: number, + durationMs?: number, +): Promise { + const sessionId = useVibeStore.getState().activeSessionId; + if (!sessionId) return null; + const input = { + eventId: newEventId(), + type, + trackId, + occurredAt: new Date().toISOString(), + positionMs, + durationMs, + }; + return new Promise((resolve, reject) => { + eventOutbox.push({ sessionId, input, retried: false, settled: false, resolve, reject }); + void flushEventOutbox(); + }); +} + +function retryableEventError(error: unknown): boolean { + return !isSessionTerminalError(error); +} + +async function flushEventOutbox(): Promise { + if (flushingOutbox) return; + flushingOutbox = true; + try { + while (eventOutbox.length > 0) { + const entry = eventOutbox[0]; + try { + const response = await vibeService.event(entry.sessionId, entry.input); + eventOutbox.shift(); + entry.settled = true; + entry.resolve(response); + } catch (error) { + // Retry once immediately using the exact same client event id. After + // that leave it at the head for a later retry, rather than discarding + // the idempotency key or allowing newer material events to overtake it. + if (!entry.retried && retryableEventError(error)) { + entry.retried = true; + continue; + } + // A session that is gone/ended can never acknowledge this event. Do + // not let an irrecoverable old-session entry block a later session. + if (isSessionTerminalError(error)) eventOutbox.shift(); + if (!entry.settled) { + entry.settled = true; + entry.reject(error); + } + return; + } + } + } finally { + flushingOutbox = false; + } +} /** - * Start a V2 plan and immediately hand its first recommendations to playback. - * Keeping this in one place prevents entry points from accidentally replacing a - * generated Vibe queue with a normal browse queue. + * Send a non-navigation event. Material feedback reconciles the future before + * it resolves, so no old prefetch remains after Keep or an implicit update. */ +export async function reportVibeEvent( + type: VibeEventType, + trackId?: string, + positionMs?: number, + durationMs?: number, +): Promise { + return serializeMaterial(async () => { + const sessionId = useVibeStore.getState().activeSessionId; + if (!sessionId || !isCurrentVibeOwner(sessionId)) return; + try { + const response = await sendEvent(type, trackId, positionMs, durationMs); + // Even a duplicate material event can acknowledge a canonical + // replacement revision (replanned=false). Reconcile every valid + // revision so a response lost after its original replan cannot leave a + // stale locally-prefetched future behind. + if (response?.planVersion !== null && response?.planVersion !== undefined) { + await reconcilePreview(sessionId, response); + } + } catch (error) { + if (isSessionTerminalError(error)) deactivateBrokenSession(); + throw error; + } + }); +} + +/** Advance only after the prior track's durable outcome has produced a new plan. */ +export function advanceVibe(reason: VibeAdvanceReason): Promise { + if (advanceInFlight) return advanceInFlight; + advanceInFlight = serializeMaterial(async () => { + const vibe = useVibeStore.getState(); + const current = usePlaybackStore.getState().currentTrack; + if (!vibe.activeSessionId || !current || !isCurrentVibeOwner(vibe.activeSessionId)) return; + + try { + const feedback = await sendEvent(reason, current.id); + if (!feedback?.planVersion) { + usePlaybackStore.getState().pause(); + return; + } + const served = await serveNextPlayable(vibe.activeSessionId, feedback.planVersion); + if (!served || served.response.sessionId !== vibe.activeSessionId || !isCurrentVibeOwner(vibe.activeSessionId)) { + // Never play an uncommitted or unresolvable plan item. The user can + // retry from the page after the director publishes another revision. + replaceUnplayedQueue([]); + usePlaybackStore.getState().pause(); + return; + } + if (!useVibeStore.getState().setPlan(served.response.planVersion, served.preview)) return; + useVibeStore.getState().setCurrentPlanItem(served.response.now); + replaceUnplayedQueue([served.now, ...served.preview]); + usePlaybackStore.getState().advance(); + } catch (error) { + // Clearing the future is deliberate: carrying on with stale prefetches + // after a rejected feedback/replan would violate the plan boundary. + replaceUnplayedQueue([]); + if (isSessionTerminalError(error)) deactivateBrokenSession(); + else usePlaybackStore.getState().pause(); + throw error; + } + }).finally(() => { advanceInFlight = null; }); + return advanceInFlight; +} + +/** + * A stream can fail after its track metadata was successfully hydrated. This + * advances the exact durable cursor through the explicit unplayable protocol, + * rather than treating it as ordinary feedback and allowing a replan to hide + * the failure. + */ +export function advancePastUnplayableVibeTrack(trackId: string): Promise { + if (advanceInFlight) return advanceInFlight; + advanceInFlight = serializeMaterial(async () => { + const vibe = useVibeStore.getState(); + const playback = usePlaybackStore.getState(); + const currentItem = vibe.currentPlanItem; + if (!vibe.activeSessionId || !currentItem || currentItem.track_id !== trackId || !isCurrentVibeOwner(vibe.activeSessionId)) return; + + try { + const advanced = await advanceResponsePastUnplayable(vibe.activeSessionId, { + sessionId: vibe.activeSessionId, + planVersion: vibe.planVersion, + now: currentItem, + preview: [], + state: {}, + replanned: false, + replanReason: null, + }); + const served = await resolvePlayableResponse(vibe.activeSessionId, advanced); + if (!served || served.response.sessionId !== vibe.activeSessionId || !isCurrentVibeOwner(vibe.activeSessionId)) { + replaceUnplayedQueue([]); + playback.pause(); + return; + } + if (!useVibeStore.getState().setPlan(served.response.planVersion, served.preview)) return; + useVibeStore.getState().setCurrentPlanItem(served.response.now); + replaceUnplayedQueue([served.now, ...served.preview]); + playback.advance(); + } catch (error) { + replaceUnplayedQueue([]); + if (isSessionTerminalError(error)) deactivateBrokenSession(); + else playback.pause(); + throw error; + } + }).finally(() => { advanceInFlight = null; }); + return advanceInFlight; +} + +function installVibeAdvanceHandler(): void { + usePlaybackStore.getState().setVibeAdvanceHandler((reason) => { + void advanceVibe(reason).catch(() => undefined); + }); +} + +/** Start, version-serve and hydrate the first durable Vibe track. */ export async function startVibeSession(seed: Track): Promise { if (startInFlight) return startInFlight; - startInFlight = beginVibeSession(seed); + startInFlight = serializeMaterial(async () => { + const started = await vibeService.start(seed.id); + if (!started.planVersion) return { status: 'exhausted', tracks: [] }; + const served = await serveNextPlayable(started.sessionId, started.planVersion); + if (!served) return { status: 'exhausted', tracks: [] }; + + const vibe = useVibeStore.getState(); + // A newly started session has its own revision sequence. Drop the old + // local revision before admitting revision 1 from this new session. + vibe.reset(); + vibe.setInitialBatchStatus('loading'); + vibe.setActiveSession({ sessionId: started.sessionId, seedTrackId: seed.id }); + vibe.setCenterTrack(seed); + vibe.setPlan(served.response.planVersion, served.preview); + vibe.setCurrentPlanItem(served.response.now); + vibe.setInitialBatchStatus('idle'); + + const playback = usePlaybackStore.getState(); + playback.setVibeQueue([served.now, ...served.preview]); + playback.playTrack(served.now); + installVibeAdvanceHandler(); + return { status: 'complete', tracks: [served.now, ...served.preview] }; + }); try { return await startInFlight; } finally { @@ -27,25 +419,18 @@ export async function startVibeSession(seed: Track): Promise } } -async function beginVibeSession(seed: Track): Promise { - const { sessionId } = await vibeService.start(seed.id); - const vibe = useVibeStore.getState(); - // Do not replace a working Vibe until the new session has produced a usable - // initial batch. This also keeps the page prefetcher attached to the old - // session while this request is in flight. - const result = await fetchNextBatch(INITIAL_VIBE_BATCH_SIZE, sessionId); - if (result.tracks.length === 0) return result; - - vibe.setInitialBatchStatus('loading'); - vibe.setActiveSession({ sessionId, seedTrackId: seed.id }); - vibe.setSeedTrackId(seed.id); - vibe.setCenterTrack(seed); - vibe.setBuffer(result.tracks); - vibe.setInitialBatchStatus('idle'); - - const playback = usePlaybackStore.getState(); - playback.setQueue(result.tracks); - playback.playTrack(result.tracks[0]); - - return result; +export async function endVibeSession(): Promise { + return serializeMaterial(async () => { + const sessionId = useVibeStore.getState().activeSessionId; + try { + if (sessionId) await vibeService.end(sessionId); + } finally { + const playback = usePlaybackStore.getState(); + playback.setVibeAdvanceHandler(null); + useVibeStore.getState().reset(); + playback.pause(); + playback.setQueue([]); + playback.setCurrentTrack(null); + } + }); } diff --git a/frontend/src/store/usePlaybackStore.ts b/frontend/src/store/usePlaybackStore.ts index 01615e8..ce75c32 100644 --- a/frontend/src/store/usePlaybackStore.ts +++ b/frontend/src/store/usePlaybackStore.ts @@ -2,6 +2,8 @@ import { create } from 'zustand'; import type { Track } from '../types'; export type RepeatMode = 'none' | 'all' | 'one'; +export type VibeAdvanceReason = 'skipped' | 'completed' | 'disliked'; +export type PlaybackOwner = 'ordinary' | 'vibe'; /** * How many already-played tracks to keep behind the cursor. Bounds queue growth @@ -22,12 +24,22 @@ interface PlaybackState { repeat: RepeatMode; /** Ids already played this shuffle "lap" (repeat-all), to avoid bouncing between the same few tracks. */ shufflePlayed: Set; + /** Installed only while a durable Vibe session owns the queue. */ + vibeAdvanceHandler: ((reason: VibeAdvanceReason) => void) | null; + /** Vibe must opt in explicitly; ordinary browsing always owns itself. */ + queueOwner: PlaybackOwner; setQueue: (queue: Track[]) => void; + /** Vibe-only queue replacement. Do not use for library browsing. */ + setVibeQueue: (queue: Track[]) => void; playTrack: (track: Track) => void; play: () => void; pause: () => void; next: () => void; + nextWithReason: (reason: VibeAdvanceReason) => void; + /** Bypass the Vibe controller after it has prepared the next committed track. */ + advance: () => void; + setVibeAdvanceHandler: (handler: ((reason: VibeAdvanceReason) => void) | null) => void; prev: () => void; setPosition: (position: number) => void; setDuration: (duration: number) => void; @@ -71,6 +83,8 @@ export const usePlaybackStore = create((set, get) => ({ shuffle: false, repeat: 'none', shufflePlayed: new Set(), + vibeAdvanceHandler: null, + queueOwner: 'ordinary', setQueue: (queue) => set((state) => ({ @@ -78,6 +92,18 @@ export const usePlaybackStore = create((set, get) => ({ // Keep the cursor pointing at whatever is playing, if it is still queued. currentIndex: state.currentTrack ? queue.findIndex((t) => t.id === state.currentTrack!.id) : -1, shufflePlayed: new Set(), + // Every ordinary queue operation is an explicit ownership handoff. This + // prevents a stale Vibe session from intercepting browser/UI next. + queueOwner: 'ordinary', + vibeAdvanceHandler: null, + })), + + setVibeQueue: (queue) => + set((state) => ({ + queue, + currentIndex: state.currentTrack ? queue.findIndex((t) => t.id === state.currentTrack!.id) : -1, + shufflePlayed: new Set(), + queueOwner: 'vibe', })), playTrack: (track) => @@ -94,6 +120,19 @@ export const usePlaybackStore = create((set, get) => ({ pause: () => set({ isPlaying: false }), next: () => { + get().nextWithReason('skipped'); + }, + + nextWithReason: (reason) => { + const { vibeAdvanceHandler: handler, queueOwner } = get(); + if (queueOwner === 'vibe' && handler) { + handler(reason); + return; + } + get().advance(); + }, + + advance: () => { const { queue, currentTrack, shuffle, repeat, shufflePlayed } = get(); if (queue.length === 0) { set({ isPlaying: false, position: 0 }); @@ -169,6 +208,11 @@ export const usePlaybackStore = create((set, get) => ({ } }, + setVibeAdvanceHandler: (vibeAdvanceHandler) => set((state) => ({ + vibeAdvanceHandler, + queueOwner: vibeAdvanceHandler ? 'vibe' : state.queueOwner, + })), + prev: () => { const { queue, currentTrack, currentIndex } = get(); if (queue.length === 0) return; diff --git a/frontend/src/store/useVibeStore.ts b/frontend/src/store/useVibeStore.ts index fd2272d..e079da3 100644 --- a/frontend/src/store/useVibeStore.ts +++ b/frontend/src/store/useVibeStore.ts @@ -1,30 +1,34 @@ import { create } from 'zustand'; import type { Track, VibeSession } from '../types'; +import type { VibePlanItem } from '../services/vibeService'; -// V2 recommendation session state. The backend stores the plan in Redis -// (keyed by sessionId) and serves tracks one at a time via GET /v2/vibe/next. -// We keep a lookahead buffer of upcoming Track[] to feed playback. +// The durable plan is authoritative. `buffer` is only its currently +// uncommitted, hydrated preview; it may be replaced at any feedback boundary. interface VibeState { activeSessionId: string | null; seedTrackId: string | null; + planVersion: number | null; + /** Durable cursor for the track currently in Vibe playback. */ + currentPlanItem: VibePlanItem | null; centerTrack: Track | null; - buffer: Track[]; // lookahead buffer of upcoming recommended tracks - /** Outcome of the first V2 batch, including sessions initiated from Discover. */ + buffer: Track[]; initialBatchStatus: 'idle' | 'loading' | 'exhausted' | 'failed'; setActiveSession: (session: VibeSession | null) => void; setSeedTrackId: (seedTrackId: string | null) => void; setCenterTrack: (track: Track | null) => void; - setBuffer: (buffer: Track[]) => void; + /** Returns false when a response belongs to an older plan revision. */ + setPlan: (planVersion: number | null, preview: Track[]) => boolean; + setCurrentPlanItem: (item: VibePlanItem | null) => void; setInitialBatchStatus: (status: VibeState['initialBatchStatus']) => void; - appendBuffer: (tracks: Track[]) => void; - shiftBuffer: () => Track | undefined; reset: () => void; } const initialState = { activeSessionId: null as string | null, seedTrackId: null as string | null, + planVersion: null as number | null, + currentPlanItem: null as VibePlanItem | null, centerTrack: null as Track | null, buffer: [] as Track[], initialBatchStatus: 'idle' as const, @@ -33,26 +37,20 @@ const initialState = { export const useVibeStore = create((set, get) => ({ ...initialState, - setActiveSession: (session) => - set( - session - ? { activeSessionId: session.sessionId, seedTrackId: session.seedTrackId } - : { activeSessionId: null, seedTrackId: null } - ), - + setActiveSession: (session) => set( + session + ? { activeSessionId: session.sessionId, seedTrackId: session.seedTrackId } + : { activeSessionId: null, seedTrackId: null }, + ), setSeedTrackId: (seedTrackId) => set({ seedTrackId }), setCenterTrack: (centerTrack) => set({ centerTrack }), - setBuffer: (buffer) => set({ buffer }), - setInitialBatchStatus: (initialBatchStatus) => set({ initialBatchStatus }), - appendBuffer: (tracks) => set((state) => ({ buffer: [...state.buffer, ...tracks] })), - - shiftBuffer: () => { - const { buffer } = get(); - if (buffer.length === 0) return undefined; - const [head, ...rest] = buffer; - set({ buffer: rest }); - return head; + setPlan: (planVersion, buffer) => { + const current = get().planVersion; + if (planVersion === null || (current !== null && planVersion < current)) return false; + set({ planVersion, buffer }); + return true; }, - + setCurrentPlanItem: (currentPlanItem) => set({ currentPlanItem }), + setInitialBatchStatus: (initialBatchStatus) => set({ initialBatchStatus }), reset: () => set({ ...initialState }), }));