refactor(vibe): simplify durable session flow
This commit is contained in:
@@ -40,9 +40,9 @@ A high-performance, distributed music orchestration and recommendation platform.
|
|||||||
|
|
||||||
### Running Locally
|
### Running Locally
|
||||||
|
|
||||||
Set `MUZICK_VIBE_USER_ID` in `.env` to the UUID of the local Muzick user before
|
Vibe uses the same per-user identity convention as the rest of the API:
|
||||||
using Vibe. Durable Vibe session routes intentionally reject client-supplied
|
`x-user-id` when supplied, otherwise the local default user. Each user's Vibe
|
||||||
identities, so this is the trusted single-user binding for a self-hosted stack.
|
session and listening history are isolated from other users.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
docker-compose up -d
|
docker-compose up -d
|
||||||
|
|||||||
@@ -14,7 +14,6 @@ import quarantineRoutes from './routes/quarantine.routes.js';
|
|||||||
import settingsRoutes from './routes/settings.routes.js';
|
import settingsRoutes from './routes/settings.routes.js';
|
||||||
import graphRoutes from './routes/graph.routes.js';
|
import graphRoutes from './routes/graph.routes.js';
|
||||||
import { SessionDirector } from './services/session-director.service.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 vibeSessionsRoutes from './routes/vibe-sessions.routes.js';
|
||||||
import discoveryRoutes from './routes/discovery.routes.js';
|
import discoveryRoutes from './routes/discovery.routes.js';
|
||||||
import imagesRoutes from './routes/images.routes.js';
|
import imagesRoutes from './routes/images.routes.js';
|
||||||
@@ -161,16 +160,10 @@ export async function buildApp(config: AppConfig) {
|
|||||||
|
|
||||||
const sessionDirector = new SessionDirector(dbService);
|
const sessionDirector = new SessionDirector(dbService);
|
||||||
const vibeSessionCoordinator = new VibeSessionCoordinator(dbService, sessionDirector);
|
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, {
|
fastify.register(vibeSessionsRoutes, {
|
||||||
prefix: '/api',
|
prefix: '/api',
|
||||||
coordinator: vibeSessionCoordinator,
|
coordinator: vibeSessionCoordinator,
|
||||||
identityResolver: () => vibeOwnerId,
|
|
||||||
});
|
});
|
||||||
fastify.register(discoveryRoutes, { prefix: '/api', dbService, jobService });
|
fastify.register(discoveryRoutes, { prefix: '/api', dbService, jobService });
|
||||||
// ponytail: /api/test/enqueue-job (manual job-enqueue test endpoint) removed —
|
// ponytail: /api/test/enqueue-job (manual job-enqueue test endpoint) removed —
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ function response() {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
async function appWithCoordinator(identityResolver: VibeIdentityResolver = () => USER_ID) {
|
async function appWithCoordinator(identityResolver?: VibeIdentityResolver) {
|
||||||
const coordinator = {
|
const coordinator = {
|
||||||
start: vi.fn().mockResolvedValue(response()),
|
start: vi.fn().mockResolvedValue(response()),
|
||||||
getPlan: vi.fn().mockResolvedValue(response()),
|
getPlan: vi.fn().mockResolvedValue(response()),
|
||||||
@@ -25,36 +25,36 @@ async function appWithCoordinator(identityResolver: VibeIdentityResolver = () =>
|
|||||||
advancePastUnplayable: vi.fn().mockResolvedValue(response()),
|
advancePastUnplayable: vi.fn().mockResolvedValue(response()),
|
||||||
} as any;
|
} as any;
|
||||||
const app = Fastify();
|
const app = Fastify();
|
||||||
await app.register(vibeSessionsRoutes, { coordinator, identityResolver });
|
await app.register(vibeSessionsRoutes, { coordinator, ...(identityResolver ? { identityResolver } : {}) });
|
||||||
await app.ready();
|
await app.ready();
|
||||||
return { app, coordinator };
|
return { app, coordinator };
|
||||||
}
|
}
|
||||||
|
|
||||||
describe('durable Vibe session routes', () => {
|
describe('durable Vibe session routes', () => {
|
||||||
it('uses the trusted identity resolver and never accepts a spoofed x-user-id header', async () => {
|
it('uses the caller identity so concurrent listeners receive separate sessions', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator();
|
||||||
const result = await app.inject({ method: 'POST', url: '/v2/vibe/sessions', payload: {} });
|
const result = await app.inject({ method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': USER_ID }, payload: {} });
|
||||||
|
|
||||||
expect(result.statusCode).toBe(201);
|
expect(result.statusCode).toBe(201);
|
||||||
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.any(Object));
|
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.any(Object));
|
||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
it('rejects requests when no trusted identity is configured instead of defaulting a shared user', async () => {
|
it('uses the existing application default when no user header is provided', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator(() => null);
|
const { app, coordinator } = await appWithCoordinator();
|
||||||
const result = await app.inject({
|
const result = await app.inject({
|
||||||
method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': USER_ID }, payload: {},
|
method: 'POST', url: '/v2/vibe/sessions', payload: {},
|
||||||
});
|
});
|
||||||
expect(result.statusCode).toBe(401);
|
expect(result.statusCode).toBe(201);
|
||||||
expect(coordinator.start).not.toHaveBeenCalled();
|
expect(coordinator.start).toHaveBeenCalledWith('00000000-0000-0000-0000-000000000000', expect.any(Object));
|
||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
it('creates a session and validates event payloads before touching the coordinator', async () => {
|
it('creates a session and validates event payloads before touching the coordinator', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator(() => USER_ID);
|
||||||
const created = await app.inject({
|
const created = await app.inject({
|
||||||
method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': 'spoofed' },
|
method: 'POST', url: '/v2/vibe/sessions', headers: { 'x-user-id': 'spoofed' },
|
||||||
payload: { seedTrackId: TRACK_ID, context: { activity: 'focus' }, policyVersion: 'test-policy' },
|
payload: { seedTrackId: TRACK_ID },
|
||||||
});
|
});
|
||||||
const invalidEvent = await app.inject({
|
const invalidEvent = await app.inject({
|
||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`, headers: { 'x-user-id': 'spoofed' },
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`, headers: { 'x-user-id': 'spoofed' },
|
||||||
@@ -62,13 +62,13 @@ describe('durable Vibe session routes', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
expect(created.statusCode).toBe(201);
|
expect(created.statusCode).toBe(201);
|
||||||
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ policyVersion: 'test-policy' }));
|
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ seedTrackId: TRACK_ID }));
|
||||||
expect(invalidEvent.statusCode).toBe(400);
|
expect(invalidEvent.statusCode).toBe(400);
|
||||||
expect(coordinator.appendEvent).not.toHaveBeenCalled();
|
expect(coordinator.appendEvent).not.toHaveBeenCalled();
|
||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
it('reserves track_served for the authoritative /next operation', async () => {
|
it('rejects server-only event types from the client event ledger', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator();
|
||||||
const result = await app.inject({
|
const result = await app.inject({
|
||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`,
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`,
|
||||||
@@ -80,7 +80,7 @@ describe('durable Vibe session routes', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
expect(result.statusCode).toBe(400);
|
expect(result.statusCode).toBe(400);
|
||||||
expect(result.json()).toEqual({ error: 'track_served is reserved for the server /next operation' });
|
expect(result.json()).toEqual({ error: 'type must be a supported client Vibe event type' });
|
||||||
expect(coordinator.appendEvent).not.toHaveBeenCalled();
|
expect(coordinator.appendEvent).not.toHaveBeenCalled();
|
||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
@@ -102,7 +102,7 @@ describe('durable Vibe session routes', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it('passes a requested plan revision and validated idempotent event through to the coordinator', async () => {
|
it('passes a requested plan revision and validated idempotent event through to the coordinator', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator(() => USER_ID);
|
||||||
const plan = await app.inject({
|
const plan = await app.inject({
|
||||||
method: 'GET', url: `/v2/vibe/sessions/${SESSION_ID}/plans?version=2`, headers: { 'x-user-id': 'spoofed' },
|
method: 'GET', url: `/v2/vibe/sessions/${SESSION_ID}/plans?version=2`, headers: { 'x-user-id': 'spoofed' },
|
||||||
});
|
});
|
||||||
@@ -118,8 +118,8 @@ describe('durable Vibe session routes', () => {
|
|||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
it('validates occurredAt and exposes owned resume and next operations', async () => {
|
it('validates occurredAt and exposes owned resume and advance operations', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator(() => USER_ID);
|
||||||
const resume = await app.inject({
|
const resume = await app.inject({
|
||||||
method: 'POST', url: '/v2/vibe/sessions', payload: { resumeSessionId: SESSION_ID },
|
method: 'POST', url: '/v2/vibe/sessions', payload: { resumeSessionId: SESSION_ID },
|
||||||
});
|
});
|
||||||
@@ -127,7 +127,7 @@ describe('durable Vibe session routes', () => {
|
|||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`,
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/events`,
|
||||||
payload: { type: 'completed', occurredAt: 'not-a-date' },
|
payload: { type: 'completed', occurredAt: 'not-a-date' },
|
||||||
});
|
});
|
||||||
const next = await app.inject({ method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next` });
|
const next = await app.inject({ method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/advance` });
|
||||||
expect(resume.statusCode).toBe(201);
|
expect(resume.statusCode).toBe(201);
|
||||||
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ resumeSessionId: SESSION_ID }));
|
expect(coordinator.start).toHaveBeenCalledWith(USER_ID, expect.objectContaining({ resumeSessionId: SESSION_ID }));
|
||||||
expect(badTime.statusCode).toBe(400);
|
expect(badTime.statusCode).toBe(400);
|
||||||
@@ -136,13 +136,13 @@ describe('durable Vibe session routes', () => {
|
|||||||
await app.close();
|
await app.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
it('passes a version-aware next request through and rejects an invalid expected version', async () => {
|
it('passes a version-aware advance request through and rejects an invalid expected version', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator(() => USER_ID);
|
||||||
const valid = await app.inject({
|
const valid = await app.inject({
|
||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`, payload: { expectedPlanVersion: 2 },
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/advance`, payload: { expectedPlanVersion: 2 },
|
||||||
});
|
});
|
||||||
const invalid = await app.inject({
|
const invalid = await app.inject({
|
||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`, payload: { expectedPlanVersion: 0 },
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/advance`, payload: { expectedPlanVersion: 0 },
|
||||||
});
|
});
|
||||||
|
|
||||||
expect(valid.statusCode).toBe(200);
|
expect(valid.statusCode).toBe(200);
|
||||||
@@ -153,9 +153,9 @@ describe('durable Vibe session routes', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it('uses the explicit versioned advancement protocol for a served unplayable item', async () => {
|
it('uses the explicit versioned advancement protocol for a served unplayable item', async () => {
|
||||||
const { app, coordinator } = await appWithCoordinator();
|
const { app, coordinator } = await appWithCoordinator(() => USER_ID);
|
||||||
const result = await app.inject({
|
const result = await app.inject({
|
||||||
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/next`,
|
method: 'POST', url: `/v2/vibe/sessions/${SESSION_ID}/advance`,
|
||||||
payload: {
|
payload: {
|
||||||
expectedPlanVersion: 2,
|
expectedPlanVersion: 2,
|
||||||
unplayable: {
|
unplayable: {
|
||||||
|
|||||||
@@ -6,89 +6,125 @@ import {
|
|||||||
VibeSessionNotFoundError,
|
VibeSessionNotFoundError,
|
||||||
VibePlanNotFoundError,
|
VibePlanNotFoundError,
|
||||||
} from '../services/vibe-session-coordinator.service.js';
|
} from '../services/vibe-session-coordinator.service.js';
|
||||||
import { isValidVibeContext } from '../services/vibe-context.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;
|
const DEFAULT_USER_ID = '00000000-0000-0000-0000-000000000000';
|
||||||
|
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
||||||
|
const ISO_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,3})?(?:Z|[+-]\d{2}:\d{2})$/i;
|
||||||
|
const CLIENT_VIBE_EVENT_TYPES = VIBE_EVENT_TYPES.filter((type) => ![
|
||||||
|
'session_started', 'session_resumed', 'session_ended', 'plan_published', 'track_served',
|
||||||
|
].includes(type));
|
||||||
|
|
||||||
/** Identity must come from authenticated server configuration/middleware, never a client header. */
|
type Reply = { code: (statusCode: number) => { send: (payload: unknown) => unknown } };
|
||||||
|
type Body = Record<string, unknown>;
|
||||||
|
|
||||||
|
/** This mirrors the rest of the application until authentication owns identity. */
|
||||||
export type VibeIdentityResolver = (request: FastifyRequest) => string | null;
|
export type VibeIdentityResolver = (request: FastifyRequest) => string | null;
|
||||||
|
|
||||||
function isObject(value: unknown): value is Record<string, unknown> {
|
function isObject(value: unknown): value is Body {
|
||||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||||
}
|
}
|
||||||
|
|
||||||
function validUuid(value: unknown): value is string {
|
function validUuid(value: unknown): value is string {
|
||||||
return typeof value === 'string' && UUID_RE.test(value);
|
return value === DEFAULT_USER_ID || (typeof value === 'string' && UUID_RE.test(value));
|
||||||
|
}
|
||||||
|
|
||||||
|
function requestUser(request: FastifyRequest, resolveIdentity?: VibeIdentityResolver): string | null {
|
||||||
|
const resolved = resolveIdentity?.(request);
|
||||||
|
if (resolved !== undefined) return validUuid(resolved) ? resolved : null;
|
||||||
|
const header = request.headers['x-user-id'];
|
||||||
|
const userId = typeof header === 'string' && header ? header : DEFAULT_USER_ID;
|
||||||
|
return validUuid(userId) ? userId : null;
|
||||||
}
|
}
|
||||||
|
|
||||||
function validOccurredAt(value: unknown): value is string {
|
function validOccurredAt(value: unknown): value is string {
|
||||||
// Require an actual offset-bearing timestamp, rather than Date.parse's
|
return typeof value === 'string' && ISO_TIMESTAMP_RE.test(value) && !Number.isNaN(Date.parse(value));
|
||||||
// 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)
|
function validationError(reply: Reply, message: string) {
|
||||||
&& !Number.isNaN(Date.parse(value));
|
return reply.code(400).send({ error: message });
|
||||||
|
}
|
||||||
|
|
||||||
|
function sessionIdFrom(request: FastifyRequest, reply: Reply): string | null {
|
||||||
|
const { sessionId } = request.params as { sessionId: string };
|
||||||
|
return validUuid(sessionId) ? sessionId : (validationError(reply, 'sessionId must be a UUID'), null);
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseStart(body: unknown):
|
||||||
|
| { seedTrackId?: string; resumeSessionId?: string }
|
||||||
|
| { error: string } {
|
||||||
|
const input = isObject(body) ? body : {};
|
||||||
|
if (input.resumeSessionId !== undefined && !validUuid(input.resumeSessionId)) return { error: 'resumeSessionId must be a UUID' };
|
||||||
|
if (input.seedTrackId !== undefined && !validUuid(input.seedTrackId)) return { error: 'seedTrackId must be a UUID' };
|
||||||
|
if (input.resumeSessionId !== undefined && input.seedTrackId !== undefined) return { error: 'resumeSessionId cannot be combined with seedTrackId' };
|
||||||
|
return { seedTrackId: input.seedTrackId as string | undefined, resumeSessionId: input.resumeSessionId as string | undefined };
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseEvent(body: unknown):
|
||||||
|
| { type: typeof CLIENT_VIBE_EVENT_TYPES[number]; eventId?: string; trackId?: string; occurredAt?: Date; positionMs?: number; durationMs?: number; payload?: Body }
|
||||||
|
| { error: string } {
|
||||||
|
if (!isObject(body)) return { error: 'event body must be an object' };
|
||||||
|
if (!CLIENT_VIBE_EVENT_TYPES.includes(body.type as typeof CLIENT_VIBE_EVENT_TYPES[number])) return { error: 'type must be a supported client Vibe event type' };
|
||||||
|
if (body.eventId !== undefined && !validUuid(body.eventId)) return { error: 'eventId must be a UUID' };
|
||||||
|
if (body.trackId !== undefined && !validUuid(body.trackId)) return { error: 'trackId must be a UUID' };
|
||||||
|
if (body.occurredAt !== undefined && !validOccurredAt(body.occurredAt)) return { error: 'occurredAt must be an ISO-8601 timestamp' };
|
||||||
|
if (body.positionMs !== undefined && (typeof body.positionMs !== 'number' || !Number.isInteger(body.positionMs) || body.positionMs < 0)) return { error: 'positionMs must be a non-negative integer' };
|
||||||
|
if (body.durationMs !== undefined && (typeof body.durationMs !== 'number' || !Number.isInteger(body.durationMs) || body.durationMs < 0)) return { error: 'durationMs must be a non-negative integer' };
|
||||||
|
if (body.payload !== undefined && !isObject(body.payload)) return { error: 'payload must be an object' };
|
||||||
|
return {
|
||||||
|
type: body.type as typeof CLIENT_VIBE_EVENT_TYPES[number],
|
||||||
|
eventId: body.eventId as string | undefined,
|
||||||
|
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 Body | undefined,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseAdvance(body: unknown):
|
||||||
|
| { expectedPlanVersion?: number; unplayable?: { eventId: string; planVersionId: string; ordinal: number; trackId: string } }
|
||||||
|
| { error: string } {
|
||||||
|
const input = isObject(body) ? body : {};
|
||||||
|
const version = input.expectedPlanVersion;
|
||||||
|
if (version !== undefined && (typeof version !== 'number' || !Number.isInteger(version) || version < 1)) return { error: 'expectedPlanVersion must be a positive integer' };
|
||||||
|
if (input.unplayable === undefined) return { expectedPlanVersion: version as number | undefined };
|
||||||
|
if (!isObject(input.unplayable)) return { error: 'unplayable must be an object' };
|
||||||
|
const unplayable = input.unplayable;
|
||||||
|
if (version === undefined) return { error: 'expectedPlanVersion is required when advancing an unplayable item' };
|
||||||
|
if (!validUuid(unplayable.eventId) || !validUuid(unplayable.planVersionId) || !validUuid(unplayable.trackId) || typeof unplayable.ordinal !== 'number' || !Number.isInteger(unplayable.ordinal) || unplayable.ordinal < 0) {
|
||||||
|
return { error: 'unplayable requires UUID eventId, planVersionId, trackId and a non-negative integer ordinal' };
|
||||||
|
}
|
||||||
|
return { expectedPlanVersion: version as number, unplayable: unplayable as { eventId: string; planVersionId: string; ordinal: number; trackId: string } };
|
||||||
}
|
}
|
||||||
|
|
||||||
export default async function vibeSessionsRoutes(
|
export default async function vibeSessionsRoutes(
|
||||||
fastify: FastifyInstance,
|
fastify: FastifyInstance,
|
||||||
options: { coordinator: VibeSessionCoordinator; identityResolver: VibeIdentityResolver },
|
options: { coordinator: VibeSessionCoordinator; identityResolver?: VibeIdentityResolver },
|
||||||
) {
|
) {
|
||||||
const { coordinator, identityResolver } = options;
|
const { coordinator, identityResolver } = options;
|
||||||
const requireUser = (request: FastifyRequest, reply: { code: (statusCode: number) => { send: (payload: unknown) => unknown } }): string | null => {
|
const userFor = (request: FastifyRequest, reply: Reply) => {
|
||||||
const userId = identityResolver(request);
|
const userId = requestUser(request, identityResolver);
|
||||||
if (userId && validUuid(userId)) return userId;
|
return userId ? userId : (validationError(reply, 'x-user-id must be a UUID'), null);
|
||||||
reply.code(401).send({ error: 'A trusted Vibe identity is required' });
|
|
||||||
return null;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
fastify.post('/v2/vibe/sessions', async (request, reply) => {
|
fastify.post('/v2/vibe/sessions', async (request, reply) => {
|
||||||
const userId = requireUser(request, reply);
|
const userId = userFor(request, reply);
|
||||||
if (!userId) return;
|
const input = parseStart(request.body);
|
||||||
const body = isObject(request.body) ? request.body : {};
|
if (!userId || 'error' in input) return 'error' in input ? validationError(reply, input.error) : undefined;
|
||||||
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 (isObject(body.context) && !isValidVibeContext(body.context)) {
|
|
||||||
return reply.code(400).send({ error: 'context contains an invalid structured Vibe value' });
|
|
||||||
}
|
|
||||||
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 {
|
try {
|
||||||
return reply.code(201).send(await coordinator.start(userId, {
|
return reply.code(201).send(await coordinator.start(userId, input));
|
||||||
seedTrackId: body.seedTrackId as string | undefined,
|
|
||||||
context: body.context as Record<string, unknown> | undefined,
|
|
||||||
intent: body.intent as string | undefined,
|
|
||||||
policyVersion: body.policyVersion as string | undefined,
|
|
||||||
resumeSessionId: body.resumeSessionId as string | undefined,
|
|
||||||
}));
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return sendCoordinatorError(reply, error);
|
return sendCoordinatorError(reply, error);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
fastify.get('/v2/vibe/sessions/:sessionId/plans', async (request, reply) => {
|
fastify.get('/v2/vibe/sessions/:sessionId/plans', async (request, reply) => {
|
||||||
const userId = requireUser(request, reply);
|
const userId = userFor(request, reply);
|
||||||
if (!userId) return;
|
const sessionId = sessionIdFrom(request, reply);
|
||||||
const { sessionId } = request.params as { sessionId: string };
|
|
||||||
const { version } = request.query as { version?: 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);
|
const parsedVersion = version === undefined ? undefined : Number(version);
|
||||||
if (version !== undefined && (!Number.isInteger(parsedVersion) || parsedVersion! < 1)) {
|
if (!userId || !sessionId) return;
|
||||||
return reply.code(400).send({ error: 'version must be a positive integer' });
|
if (version !== undefined && (!Number.isInteger(parsedVersion) || parsedVersion! < 1)) return validationError(reply, 'version must be a positive integer');
|
||||||
}
|
|
||||||
try {
|
try {
|
||||||
return reply.send(await coordinator.getPlan(userId, sessionId, parsedVersion));
|
return reply.send(await coordinator.getPlan(userId, sessionId, parsedVersion));
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -97,63 +133,21 @@ export default async function vibeSessionsRoutes(
|
|||||||
});
|
});
|
||||||
|
|
||||||
fastify.post('/v2/vibe/sessions/:sessionId/events', async (request, reply) => {
|
fastify.post('/v2/vibe/sessions/:sessionId/events', async (request, reply) => {
|
||||||
const userId = requireUser(request, reply);
|
const userId = userFor(request, reply);
|
||||||
if (!userId) return;
|
const sessionId = sessionIdFrom(request, reply);
|
||||||
const { sessionId } = request.params as { sessionId: string };
|
const input = parseEvent(request.body);
|
||||||
const body = isObject(request.body) ? request.body : null;
|
if (!userId || !sessionId || 'error' in input) return 'error' in input ? validationError(reply, input.error) : undefined;
|
||||||
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' });
|
|
||||||
}
|
|
||||||
// Delivery is an authoritative state transition performed only by /next.
|
|
||||||
// Accepting this event from the public ledger endpoint would let a client
|
|
||||||
// fabricate exposure rows and consume the server-side surprise budget.
|
|
||||||
if (body.type === 'track_served') {
|
|
||||||
return reply.code(400).send({ error: 'track_served is reserved for the server /next operation' });
|
|
||||||
}
|
|
||||||
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' });
|
|
||||||
}
|
|
||||||
if (body.type === 'context_changed' && isObject(body.payload)
|
|
||||||
&& body.payload.context !== undefined
|
|
||||||
&& (!isObject(body.payload.context) || !isValidVibeContext(body.payload.context))) {
|
|
||||||
return reply.code(400).send({ error: 'context_changed payload.context must be structured Vibe context' });
|
|
||||||
}
|
|
||||||
try {
|
try {
|
||||||
return reply.send(await coordinator.appendEvent(userId, sessionId, {
|
return reply.send(await coordinator.appendEvent(userId, sessionId, input));
|
||||||
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<string, unknown> | undefined,
|
|
||||||
}));
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return sendCoordinatorError(reply, error);
|
return sendCoordinatorError(reply, error);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
fastify.post('/v2/vibe/sessions/:sessionId/end', async (request, reply) => {
|
fastify.post('/v2/vibe/sessions/:sessionId/end', async (request, reply) => {
|
||||||
const userId = requireUser(request, reply);
|
const userId = userFor(request, reply);
|
||||||
if (!userId) return;
|
const sessionId = sessionIdFrom(request, reply);
|
||||||
const { sessionId } = request.params as { sessionId: string };
|
if (!userId || !sessionId) return;
|
||||||
if (!validUuid(sessionId)) return reply.code(400).send({ error: 'sessionId must be a UUID' });
|
|
||||||
try {
|
try {
|
||||||
return reply.send(await coordinator.end(userId, sessionId));
|
return reply.send(await coordinator.end(userId, sessionId));
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -161,55 +155,25 @@ export default async function vibeSessionsRoutes(
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
fastify.post('/v2/vibe/sessions/:sessionId/next', async (request, reply) => {
|
fastify.post('/v2/vibe/sessions/:sessionId/advance', async (request, reply) => {
|
||||||
const userId = requireUser(request, reply);
|
const userId = userFor(request, reply);
|
||||||
if (!userId) return;
|
const sessionId = sessionIdFrom(request, reply);
|
||||||
const { sessionId } = request.params as { sessionId: string };
|
const input = parseAdvance(request.body);
|
||||||
const body = isObject(request.body) ? request.body : {};
|
if (!userId || !sessionId || 'error' in input) return 'error' in input ? validationError(reply, input.error) : undefined;
|
||||||
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' });
|
|
||||||
}
|
|
||||||
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 {
|
try {
|
||||||
const expectedPlanVersion = body.expectedPlanVersion as number | undefined;
|
return reply.send(input.unplayable
|
||||||
if (isObject(unplayable)) {
|
? await coordinator.advancePastUnplayable(userId, sessionId, { expectedPlanVersion: input.expectedPlanVersion!, ...input.unplayable })
|
||||||
return reply.send(await coordinator.advancePastUnplayable(userId, sessionId, {
|
: input.expectedPlanVersion === undefined
|
||||||
expectedPlanVersion: expectedPlanVersion as number,
|
? await coordinator.serveNext(userId, sessionId)
|
||||||
eventId: unplayable.eventId as string,
|
: await coordinator.serveNext(userId, sessionId, input.expectedPlanVersion));
|
||||||
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));
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return sendCoordinatorError(reply, error);
|
return sendCoordinatorError(reply, error);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
function sendCoordinatorError(reply: { code: (statusCode: number) => { send: (payload: unknown) => unknown } }, error: unknown) {
|
function sendCoordinatorError(reply: Reply, error: unknown) {
|
||||||
if (error instanceof VibeSessionNotFoundError) return reply.code(404).send({ error: error.message });
|
if (error instanceof VibeSessionNotFoundError || error instanceof VibePlanNotFoundError) 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' });
|
if (error instanceof VibeSessionLifecycleError) return reply.code(409).send({ error: error.message, code: 'VIBE_SESSION_NOT_ACTIVE' });
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -75,9 +75,7 @@ describe('DbService v2 methods', () => {
|
|||||||
.mockResolvedValueOnce({ rows: [session] })
|
.mockResolvedValueOnce({ rows: [session] })
|
||||||
.mockResolvedValueOnce({ rows: [{ ...session, status: 'ended' }] });
|
.mockResolvedValueOnce({ rows: [{ ...session, status: 'ended' }] });
|
||||||
|
|
||||||
await expect(service.createVibeSession({
|
await expect(service.createVibeSession({ userId: 'user-1', policyVersion: 'v2.1' })).resolves.toEqual(session);
|
||||||
userId: 'user-1', policyVersion: 'v2.1', context: { activity: 'focus' },
|
|
||||||
})).resolves.toEqual(session);
|
|
||||||
await expect(service.getVibeSession('session-1', 'user-1')).resolves.toEqual(session);
|
await expect(service.getVibeSession('session-1', 'user-1')).resolves.toEqual(session);
|
||||||
await expect(service.endVibeSession('session-1', 'user-1')).resolves.toMatchObject({ status: 'ended' });
|
await expect(service.endVibeSession('session-1', 'user-1')).resolves.toMatchObject({ status: 'ended' });
|
||||||
|
|
||||||
@@ -88,17 +86,13 @@ describe('DbService v2 methods', () => {
|
|||||||
'user-1', null, expect.any(String), 'v2.1',
|
'user-1', null, expect.any(String), 'v2.1',
|
||||||
JSON.stringify({ type: 'discovery', target: 1, progress: 0 }), 0.3, 0.38,
|
JSON.stringify({ type: 'discovery', target: 1, progress: 0 }), 0.3, 0.38,
|
||||||
]));
|
]));
|
||||||
expect(JSON.parse(clientQuery.mock.calls[3][1][2])).toEqual(expect.objectContaining({
|
expect(JSON.parse(clientQuery.mock.calls[3][1][2])).toEqual({});
|
||||||
activity: 'focus',
|
|
||||||
}));
|
|
||||||
expect(poolQuery.mock.calls[0][0]).toContain('id = $1 AND user_id = $2');
|
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('COALESCE(ended_at, NOW())');
|
||||||
expect(poolQuery.mock.calls[1][0]).toContain('CASE WHEN ended_at IS NULL THEN $3 ELSE status END');
|
expect(poolQuery.mock.calls[1][0]).toContain('CASE WHEN ended_at IS NULL THEN $3 ELSE status END');
|
||||||
});
|
});
|
||||||
|
|
||||||
it('normalizes context at the persistence boundary for direct callers', async () => {
|
it('starts sessions without inventing client context', async () => {
|
||||||
vi.useFakeTimers();
|
|
||||||
vi.setSystemTime(new Date('2026-08-02T19:00:00.000Z'));
|
|
||||||
const { service, clientQuery } = makeTransactionalService();
|
const { service, clientQuery } = makeTransactionalService();
|
||||||
const session = { id: 'session-1', user_id: 'user-1', status: 'active' };
|
const session = { id: 'session-1', user_id: 'user-1', status: 'active' };
|
||||||
clientQuery
|
clientQuery
|
||||||
@@ -108,33 +102,13 @@ describe('DbService v2 methods', () => {
|
|||||||
.mockResolvedValueOnce({ rows: [session] }) // insert
|
.mockResolvedValueOnce({ rows: [session] }) // insert
|
||||||
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
||||||
|
|
||||||
try {
|
await service.createVibeSession({ userId: 'user-1', policyVersion: 'v2.1' });
|
||||||
await service.createVibeSession({
|
const insertParameters = clientQuery.mock.calls[3][1];
|
||||||
userId: 'user-1',
|
expect(insertParameters.slice(0, 2)).toEqual(['user-1', null]);
|
||||||
policyVersion: 'v2.1',
|
expect(JSON.parse(insertParameters[2])).toEqual({});
|
||||||
context: {
|
expect(insertParameters.slice(3)).toEqual([
|
||||||
timeZone: 'UTC',
|
'v2.1', JSON.stringify({ type: 'discovery', target: 1, progress: 0 }), 0.3, 0.38,
|
||||||
activity: 'walking',
|
]);
|
||||||
device: 'phone',
|
|
||||||
exactCoordinates: '53.1959,50.1002',
|
|
||||||
browserTelemetry: { batteryPercent: 4, ipAddress: '192.0.2.1' },
|
|
||||||
localHour: 3,
|
|
||||||
weekday: 1,
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
const insertParameters = clientQuery.mock.calls[3][1];
|
|
||||||
expect(insertParameters.slice(0, 2)).toEqual(['user-1', null]);
|
|
||||||
expect(JSON.parse(insertParameters[2])).toEqual({
|
|
||||||
timeZone: 'UTC', localHour: 19, weekday: 0, dayKind: 'weekend',
|
|
||||||
activity: 'walking', device: 'phone',
|
|
||||||
});
|
|
||||||
expect(insertParameters.slice(3)).toEqual([
|
|
||||||
'v2.1', JSON.stringify({ type: 'discovery', target: 1, progress: 0 }), 0.3, 0.38,
|
|
||||||
]);
|
|
||||||
} finally {
|
|
||||||
vi.useRealTimers();
|
|
||||||
}
|
|
||||||
});
|
});
|
||||||
|
|
||||||
it('replaces an owned active session and writes its terminal event before starting another', async () => {
|
it('replaces an owned active session and writes its terminal event before starting another', async () => {
|
||||||
@@ -189,45 +163,38 @@ describe('DbService v2 methods', () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
it('sanitizes and projects an inserted context change atomically with its ledger event', async () => {
|
it('records a non-material event without mutating session context', async () => {
|
||||||
const { service, clientQuery } = makeTransactionalService();
|
const { service, clientQuery } = makeTransactionalService();
|
||||||
const event = {
|
const event = {
|
||||||
id: 'event-1', client_event_id: null, session_id: 'session-1', user_id: 'user-1', track_id: null,
|
id: 'event-1', client_event_id: null, session_id: 'session-1', user_id: 'user-1', track_id: null,
|
||||||
type: 'context_changed', occurred_at: new Date(), position_ms: null, duration_ms: null,
|
type: 'progress', occurred_at: new Date(), position_ms: null, duration_ms: null,
|
||||||
payload: { context: { activity: 'walking', localHour: 12 } },
|
payload: { source: 'player' },
|
||||||
};
|
};
|
||||||
clientQuery
|
clientQuery
|
||||||
.mockResolvedValueOnce({ rows: [] }) // BEGIN
|
.mockResolvedValueOnce({ rows: [] }) // BEGIN
|
||||||
.mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) // lock
|
.mockResolvedValueOnce({ rows: [{ id: 'session-1', status: 'active' }] }) // lock
|
||||||
.mockResolvedValueOnce({ rows: [event] }) // insert
|
.mockResolvedValueOnce({ rows: [event] }) // insert
|
||||||
.mockResolvedValueOnce({ rows: [] }) // session context projection
|
|
||||||
.mockResolvedValueOnce({ rows: [] }) // legacy session-state projection
|
|
||||||
.mockResolvedValueOnce({ rows: [] }) // last event timestamp
|
.mockResolvedValueOnce({ rows: [] }) // last event timestamp
|
||||||
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
||||||
|
|
||||||
await service.recordVibeEvent({
|
await service.recordVibeEvent({
|
||||||
sessionId: 'session-1', userId: 'user-1', type: 'context_changed',
|
sessionId: 'session-1', userId: 'user-1', type: 'progress', payload: { source: 'player' },
|
||||||
payload: {
|
|
||||||
context: { activity: 'walking', exactCoordinates: '53.2,50.1' },
|
|
||||||
rawBrowserTelemetry: { battery: 4 },
|
|
||||||
},
|
|
||||||
});
|
});
|
||||||
|
|
||||||
const values = clientQuery.mock.calls[2][1];
|
const values = clientQuery.mock.calls[2][1];
|
||||||
const storedPayload = JSON.parse(values[8]);
|
const storedPayload = JSON.parse(values[8]);
|
||||||
expect(storedPayload).toEqual({ context: expect.objectContaining({ activity: 'walking' }) });
|
expect(storedPayload).toEqual({ source: 'player' });
|
||||||
expect(storedPayload.context).not.toHaveProperty('exactCoordinates');
|
expect(clientQuery.mock.calls.map(([sql]) => String(sql))).not.toContain(
|
||||||
expect(storedPayload).not.toHaveProperty('rawBrowserTelemetry');
|
expect.stringContaining('SET context = $3::jsonb'),
|
||||||
expect(clientQuery.mock.calls[3][0]).toContain('SET context = $3::jsonb');
|
);
|
||||||
expect(clientQuery.mock.calls[4][0]).toContain("jsonb_build_object('context'");
|
|
||||||
});
|
});
|
||||||
|
|
||||||
it('does not apply a retry body to an existing context-change event', async () => {
|
it('does not apply a retry body to an existing event', async () => {
|
||||||
const { service, clientQuery } = makeTransactionalService();
|
const { service, clientQuery } = makeTransactionalService();
|
||||||
const canonicalEvent = {
|
const canonicalEvent = {
|
||||||
id: 'event-1', client_event_id: 'client-event-1', session_id: 'session-1', user_id: 'user-1', track_id: null,
|
id: 'event-1', client_event_id: 'client-event-1', session_id: 'session-1', user_id: 'user-1', track_id: null,
|
||||||
type: 'context_changed', occurred_at: new Date(), position_ms: null, duration_ms: null,
|
type: 'progress', occurred_at: new Date(), position_ms: null, duration_ms: null,
|
||||||
payload: { context: { activity: 'focus', localHour: 12 } },
|
payload: { source: 'player' },
|
||||||
};
|
};
|
||||||
clientQuery
|
clientQuery
|
||||||
.mockResolvedValueOnce({ rows: [] }) // BEGIN
|
.mockResolvedValueOnce({ rows: [] }) // BEGIN
|
||||||
@@ -236,8 +203,8 @@ describe('DbService v2 methods', () => {
|
|||||||
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
.mockResolvedValueOnce({ rows: [] }); // COMMIT
|
||||||
|
|
||||||
const result = await service.recordVibeEvent({
|
const result = await service.recordVibeEvent({
|
||||||
sessionId: 'session-1', userId: 'user-1', clientEventId: 'client-event-1', type: 'context_changed',
|
sessionId: 'session-1', userId: 'user-1', clientEventId: 'client-event-1', type: 'progress',
|
||||||
payload: { context: { activity: 'workout', exactCoordinates: '53.2,50.1' } },
|
payload: { source: 'retry' },
|
||||||
});
|
});
|
||||||
|
|
||||||
expect(result).toEqual({ event: canonicalEvent, inserted: false });
|
expect(result).toEqual({ event: canonicalEvent, inserted: false });
|
||||||
|
|||||||
@@ -6,7 +6,6 @@ import { fileURLToPath } from 'url';
|
|||||||
import { dirname, join } from 'path';
|
import { dirname, join } from 'path';
|
||||||
import { Pool, PoolClient } from 'pg';
|
import { Pool, PoolClient } from 'pg';
|
||||||
import { SearchService } from './search.service.js';
|
import { SearchService } from './search.service.js';
|
||||||
import { normalizeVibeContext, normalizeVibeEventPayload } from './vibe-context.service.js';
|
|
||||||
|
|
||||||
/** Anything with a `.query()` — either the shared Pool or a checked-out client. */
|
/** Anything with a `.query()` — either the shared Pool or a checked-out client. */
|
||||||
type Queryable = Pool | PoolClient;
|
type Queryable = Pool | PoolClient;
|
||||||
@@ -1515,18 +1514,12 @@ export class DbService {
|
|||||||
userId: string;
|
userId: string;
|
||||||
policyVersion: string;
|
policyVersion: string;
|
||||||
seedTrackId?: string | null;
|
seedTrackId?: string | null;
|
||||||
context?: Record<string, unknown>;
|
|
||||||
profile?: {
|
profile?: {
|
||||||
goals: Record<string, unknown>;
|
goals: Record<string, unknown>;
|
||||||
explorationCoefficient: number;
|
explorationCoefficient: number;
|
||||||
discoveryRadius: number;
|
discoveryRadius: number;
|
||||||
};
|
};
|
||||||
}): Promise<VibeSession> {
|
}): Promise<VibeSession> {
|
||||||
// DbService is also used directly by workers and migrations. Keep the
|
|
||||||
// durable storage boundary canonical even when callers bypass the HTTP
|
|
||||||
// coordinator, so opaque or precise client telemetry can never become
|
|
||||||
// session context.
|
|
||||||
const canonicalContext = normalizeVibeContext(params.context ?? {});
|
|
||||||
return this.withTransaction(async (client) => {
|
return this.withTransaction(async (client) => {
|
||||||
// Serialize starts for one listener even when there is no active row to
|
// 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.
|
// lock yet. The row lock below then safely replaces any prior session.
|
||||||
@@ -1568,7 +1561,7 @@ export class DbService {
|
|||||||
[
|
[
|
||||||
params.userId,
|
params.userId,
|
||||||
params.seedTrackId ?? null,
|
params.seedTrackId ?? null,
|
||||||
JSON.stringify(canonicalContext),
|
'{}',
|
||||||
params.policyVersion,
|
params.policyVersion,
|
||||||
JSON.stringify(params.profile?.goals ?? { type: 'discovery', target: 1, progress: 0 }),
|
JSON.stringify(params.profile?.goals ?? { type: 'discovery', target: 1, progress: 0 }),
|
||||||
params.profile?.explorationCoefficient ?? 0.3,
|
params.profile?.explorationCoefficient ?? 0.3,
|
||||||
@@ -1615,29 +1608,6 @@ export class DbService {
|
|||||||
return (res.rows[0] as VibeSessionProfile | undefined) ?? null;
|
return (res.rows[0] as VibeSessionProfile | undefined) ?? null;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Replace only the coarse, sanitised context attached to an active session.
|
|
||||||
* The immutable context_changed event remains the audit trail. */
|
|
||||||
async updateVibeSessionContext(sessionId: string, userId: string, context: Record<string, unknown>): Promise<void> {
|
|
||||||
const canonicalContext = normalizeVibeContext(context);
|
|
||||||
await this.withTransaction(async client => {
|
|
||||||
const updated = await client.query(
|
|
||||||
`UPDATE vibe_sessions SET context = $3::jsonb, last_event_at = NOW()
|
|
||||||
WHERE id = $1 AND user_id = $2 AND status = 'active'
|
|
||||||
RETURNING id`,
|
|
||||||
[sessionId, userId, JSON.stringify(canonicalContext)],
|
|
||||||
);
|
|
||||||
if (!updated.rows[0]) throw new Error('Vibe session was not found or is not owned by this user');
|
|
||||||
await client.query(
|
|
||||||
`UPDATE session_state
|
|
||||||
SET context = COALESCE($3::jsonb->>'activity', $3::jsonb->>'device'),
|
|
||||||
state_vector = state_vector || jsonb_build_object('context', $3::jsonb),
|
|
||||||
last_interaction = NOW()
|
|
||||||
WHERE session_id = $1 AND user_id = $2`,
|
|
||||||
[sessionId, userId, JSON.stringify(canonicalContext)],
|
|
||||||
);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Project unknown-track feedback into the session exploration controls.
|
* Project unknown-track feedback into the session exploration controls.
|
||||||
* This deliberately runs behind its own projection marker: the immutable
|
* This deliberately runs behind its own projection marker: the immutable
|
||||||
@@ -1825,10 +1795,6 @@ export class DbService {
|
|||||||
durationMs?: number | null;
|
durationMs?: number | null;
|
||||||
payload?: Record<string, unknown>;
|
payload?: Record<string, unknown>;
|
||||||
}): Promise<RecordedVibeEvent> {
|
}): Promise<RecordedVibeEvent> {
|
||||||
// This service is also called by jobs and tests which bypass the HTTP
|
|
||||||
// route. Preserve the context privacy boundary at the final point before
|
|
||||||
// an immutable ledger write.
|
|
||||||
const payload = normalizeVibeEventPayload(params.type, params.payload);
|
|
||||||
return this.withTransaction(async (client) => {
|
return this.withTransaction(async (client) => {
|
||||||
// A session-row lock serializes both event writes and terminal state
|
// A session-row lock serializes both event writes and terminal state
|
||||||
// transitions. In particular, it avoids the READ COMMITTED CTE snapshot
|
// transitions. In particular, it avoids the READ COMMITTED CTE snapshot
|
||||||
@@ -1880,7 +1846,7 @@ export class DbService {
|
|||||||
occurredAt,
|
occurredAt,
|
||||||
params.positionMs ?? null,
|
params.positionMs ?? null,
|
||||||
params.durationMs ?? null,
|
params.durationMs ?? null,
|
||||||
JSON.stringify(payload ?? {}),
|
JSON.stringify(params.payload ?? {}),
|
||||||
]
|
]
|
||||||
);
|
);
|
||||||
const event = insertRes.rows[0] as VibeEvent | undefined;
|
const event = insertRes.rows[0] as VibeEvent | undefined;
|
||||||
@@ -1889,8 +1855,6 @@ export class DbService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
await this.projectVibeFeedback(event, client);
|
await this.projectVibeFeedback(event, client);
|
||||||
await this.projectVibeContextChanged(event, client);
|
|
||||||
|
|
||||||
await client.query(
|
await client.query(
|
||||||
`UPDATE vibe_sessions
|
`UPDATE vibe_sessions
|
||||||
SET last_event_at = GREATEST(last_event_at, $2::timestamptz)
|
SET last_event_at = GREATEST(last_event_at, $2::timestamptz)
|
||||||
@@ -1901,30 +1865,6 @@ export class DbService {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Apply the context projection in the same transaction as its *inserted*
|
|
||||||
* ledger event. A client-event retry returns before this method, so its body
|
|
||||||
* can never overwrite session state with a different context. */
|
|
||||||
private async projectVibeContextChanged(event: VibeEvent, client: PoolClient): Promise<void> {
|
|
||||||
if (event.type !== 'context_changed') return;
|
|
||||||
const context = event.payload?.context;
|
|
||||||
if (!context || typeof context !== 'object' || Array.isArray(context)) return;
|
|
||||||
const canonicalContext = context as Record<string, unknown>;
|
|
||||||
await client.query(
|
|
||||||
`UPDATE vibe_sessions
|
|
||||||
SET context = $3::jsonb
|
|
||||||
WHERE id = $1 AND user_id = $2 AND status = 'active'`,
|
|
||||||
[event.session_id, event.user_id, JSON.stringify(canonicalContext)],
|
|
||||||
);
|
|
||||||
await client.query(
|
|
||||||
`UPDATE session_state
|
|
||||||
SET context = COALESCE($3::jsonb->>'activity', $3::jsonb->>'device'),
|
|
||||||
state_vector = state_vector || jsonb_build_object('context', $3::jsonb),
|
|
||||||
last_interaction = NOW()
|
|
||||||
WHERE session_id = $1 AND user_id = $2`,
|
|
||||||
[event.session_id, event.user_id, JSON.stringify(canonicalContext)],
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Materialize Vibe feedback into the listener inputs used by the incumbent
|
* Materialize Vibe feedback into the listener inputs used by the incumbent
|
||||||
* director. The projection marker and every write share the event's
|
* director. The projection marker and every write share the event's
|
||||||
|
|||||||
@@ -1636,7 +1636,7 @@ export class SessionDirector {
|
|||||||
return scored.map(s => s.candidate);
|
return scored.map(s => s.candidate);
|
||||||
}
|
}
|
||||||
|
|
||||||
// session_state is otherwise only written once at /v2/vibe/start — persist the
|
// session_state is otherwise only written once when a Vibe session starts — persist the
|
||||||
// freshly-computed state vector here so it evolves across the session instead of
|
// freshly-computed state vector here so it evolves across the session instead of
|
||||||
// buildState always reading back the boot defaults.
|
// buildState always reading back the boot defaults.
|
||||||
async persistState(sessionId: string, userId: string, state: GeneratorContext['state']): Promise<void> {
|
async persistState(sessionId: string, userId: string, state: GeneratorContext['state']): Promise<void> {
|
||||||
|
|||||||
@@ -1,36 +0,0 @@
|
|||||||
import { describe, expect, it } from 'vitest';
|
|
||||||
import { initialVibeState, normalizeVibeContext, normalizeVibeEventPayload } from './vibe-context.service.js';
|
|
||||||
|
|
||||||
describe('Vibe context', () => {
|
|
||||||
it('keeps only coarse structured values and derives time on the server', () => {
|
|
||||||
const context = normalizeVibeContext({
|
|
||||||
timeZone: 'UTC', activity: 'workout', device: 'headphones',
|
|
||||||
exactCoordinates: '53.2,50.1', localHour: 3,
|
|
||||||
}, new Date('2026-08-02T19:00:00.000Z'));
|
|
||||||
|
|
||||||
expect(context).toMatchObject({ localHour: 19, weekday: 0, dayKind: 'weekend', activity: 'workout' });
|
|
||||||
expect(context).not.toHaveProperty('exactCoordinates');
|
|
||||||
expect(context).not.toHaveProperty('localHour', 3);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('uses context only as a bounded initial prior and gives focus a comfort goal', () => {
|
|
||||||
const state = initialVibeState(normalizeVibeContext({ activity: 'focus' }, new Date('2026-08-03T12:00:00.000Z')));
|
|
||||||
expect(state.energy).toBeGreaterThan(0);
|
|
||||||
expect(state.energy).toBeLessThan(1);
|
|
||||||
expect(state.sessionGoal).toEqual({ type: 'familiar', target: 1, progress: 0 });
|
|
||||||
});
|
|
||||||
|
|
||||||
it('persists only canonical context for context-change events', () => {
|
|
||||||
const payload = normalizeVibeEventPayload('context_changed', {
|
|
||||||
context: { activity: 'walking', exactCoordinates: '53.2,50.1', adId: 'do-not-store' },
|
|
||||||
rawBrowserTelemetry: { battery: 4 },
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(payload).toEqual({
|
|
||||||
context: expect.objectContaining({ activity: 'walking' }),
|
|
||||||
});
|
|
||||||
expect(payload).not.toHaveProperty('rawBrowserTelemetry');
|
|
||||||
expect((payload?.context as Record<string, unknown>)).not.toHaveProperty('exactCoordinates');
|
|
||||||
expect((payload?.context as Record<string, unknown>)).not.toHaveProperty('adId');
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -1,120 +0,0 @@
|
|||||||
/**
|
|
||||||
* Coarse, opt-in context accepted by the durable Vibe API. It deliberately
|
|
||||||
* has no precise location, identifiers, or browser telemetry: clients can
|
|
||||||
* supply a hint, but the server owns the time fields and can ignore all of it.
|
|
||||||
*/
|
|
||||||
export const VIBE_CONTEXT_VALUES = {
|
|
||||||
device: ['desktop', 'phone', 'speaker', 'car', 'headphones'] as const,
|
|
||||||
activity: ['focus', 'relax', 'walking', 'workout', 'social', 'unknown'] as const,
|
|
||||||
locationCategory: ['home', 'work', 'gym', 'travel', 'unknown'] as const,
|
|
||||||
weather: ['clear', 'rain', 'snow', 'hot', 'cold', 'unknown'] as const,
|
|
||||||
source: ['current_track', 'artist', 'genre', 'surprise', 'resume'] as const,
|
|
||||||
};
|
|
||||||
|
|
||||||
export interface VibeContext {
|
|
||||||
timeZone?: string;
|
|
||||||
localHour?: number;
|
|
||||||
weekday?: number;
|
|
||||||
dayKind?: 'weekday' | 'weekend' | 'holiday';
|
|
||||||
device?: (typeof VIBE_CONTEXT_VALUES.device)[number];
|
|
||||||
activity?: (typeof VIBE_CONTEXT_VALUES.activity)[number];
|
|
||||||
locationCategory?: (typeof VIBE_CONTEXT_VALUES.locationCategory)[number];
|
|
||||||
weather?: (typeof VIBE_CONTEXT_VALUES.weather)[number];
|
|
||||||
source?: (typeof VIBE_CONTEXT_VALUES.source)[number];
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface InitialVibeState {
|
|
||||||
contextLabel: string | undefined;
|
|
||||||
energy: number;
|
|
||||||
noveltyHunger: number;
|
|
||||||
explorationCoefficient: number;
|
|
||||||
discoveryRadius: number;
|
|
||||||
sessionGoal: { type: 'surprise' | 'familiar' | 'discovery' | 'artist_introduction'; target: number; progress: number };
|
|
||||||
}
|
|
||||||
|
|
||||||
const hasValue = <T extends readonly string[]>(values: T, value: unknown): value is T[number] =>
|
|
||||||
typeof value === 'string' && (values as readonly string[]).includes(value);
|
|
||||||
|
|
||||||
function serverTime(timeZone?: string, now = new Date()): Pick<VibeContext, 'timeZone' | 'localHour' | 'weekday' | 'dayKind'> {
|
|
||||||
// Intl rejects bad IANA names. Falling back to the server clock is safe and
|
|
||||||
// still makes time a weak prior rather than client-controlled fact.
|
|
||||||
let zone: string | undefined;
|
|
||||||
try {
|
|
||||||
if (timeZone) new Intl.DateTimeFormat('en-US', { timeZone }).format(now);
|
|
||||||
zone = timeZone;
|
|
||||||
} catch { /* server-local fallback */ }
|
|
||||||
const parts = new Intl.DateTimeFormat('en-US', {
|
|
||||||
timeZone: zone, hour: 'numeric', weekday: 'short', hourCycle: 'h23',
|
|
||||||
}).formatToParts(now);
|
|
||||||
const hour = Number(parts.find(part => part.type === 'hour')?.value ?? now.getHours());
|
|
||||||
const weekdayName = parts.find(part => part.type === 'weekday')?.value;
|
|
||||||
const weekday = ['Sun', 'Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat'].indexOf(weekdayName ?? '');
|
|
||||||
return {
|
|
||||||
...(zone ? { timeZone: zone } : {}),
|
|
||||||
localHour: Number.isInteger(hour) ? hour : now.getHours(),
|
|
||||||
weekday: weekday >= 0 ? weekday : now.getDay(),
|
|
||||||
dayKind: ([0, 6].includes(weekday >= 0 ? weekday : now.getDay()) ? 'weekend' : 'weekday'),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Remove unknown fields and derive time server-side. This preserves old
|
|
||||||
* clients that send an empty object while preventing opaque context blobs from
|
|
||||||
* becoming a permanent behavioural profile. */
|
|
||||||
export function normalizeVibeContext(input: Record<string, unknown> = {}, now = new Date()): VibeContext {
|
|
||||||
const time = serverTime(typeof input.timeZone === 'string' ? input.timeZone : undefined, now);
|
|
||||||
return {
|
|
||||||
...time,
|
|
||||||
...(hasValue(VIBE_CONTEXT_VALUES.device, input.device) ? { device: input.device } : {}),
|
|
||||||
...(hasValue(VIBE_CONTEXT_VALUES.activity, input.activity) ? { activity: input.activity } : {}),
|
|
||||||
...(hasValue(VIBE_CONTEXT_VALUES.locationCategory, input.locationCategory) ? { locationCategory: input.locationCategory } : {}),
|
|
||||||
...(hasValue(VIBE_CONTEXT_VALUES.weather, input.weather) ? { weather: input.weather } : {}),
|
|
||||||
...(hasValue(VIBE_CONTEXT_VALUES.source, input.source) ? { source: input.source } : {}),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Context changes are the only event payload with a structured, durable
|
|
||||||
* context body. Keep their ledger representation intentionally tiny: callers
|
|
||||||
* cannot smuggle precise location or arbitrary browser telemetry into an
|
|
||||||
* immutable event by adding sibling fields or unknown context keys.
|
|
||||||
*/
|
|
||||||
export function normalizeVibeEventPayload(
|
|
||||||
type: string,
|
|
||||||
payload?: Record<string, unknown>,
|
|
||||||
): Record<string, unknown> | undefined {
|
|
||||||
if (type !== 'context_changed') return payload;
|
|
||||||
const context = payload?.context;
|
|
||||||
if (!context || typeof context !== 'object' || Array.isArray(context)) return {};
|
|
||||||
return { context: normalizeVibeContext(context as Record<string, unknown>) };
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Context is intentionally a gentle prior. It can nudge the initial arc but
|
|
||||||
* never overrides observed playback behaviour. */
|
|
||||||
export function initialVibeState(context: VibeContext): InitialVibeState {
|
|
||||||
const activityEnergy: Record<string, number> = {
|
|
||||||
focus: 0.42, relax: 0.34, walking: 0.58, workout: 0.72, social: 0.62, unknown: 0.5,
|
|
||||||
};
|
|
||||||
const hour = context.localHour ?? 12;
|
|
||||||
const hourEnergy = hour < 6 ? 0.32 : hour < 10 ? 0.46 : hour >= 22 ? 0.38 : 0.52;
|
|
||||||
const activity = context.activity ?? 'unknown';
|
|
||||||
const energy = Math.max(0, Math.min(1, activityEnergy[activity] * 0.7 + hourEnergy * 0.3));
|
|
||||||
const goal = activity === 'focus' || activity === 'relax'
|
|
||||||
? 'familiar'
|
|
||||||
: activity === 'workout' || activity === 'walking' ? 'surprise' : 'discovery';
|
|
||||||
return {
|
|
||||||
contextLabel: context.activity ?? context.device,
|
|
||||||
energy,
|
|
||||||
noveltyHunger: 0.3,
|
|
||||||
explorationCoefficient: 0.3,
|
|
||||||
discoveryRadius: 0.38,
|
|
||||||
sessionGoal: { type: goal, target: 1, progress: 0 },
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isValidVibeContext(input: Record<string, unknown>): boolean {
|
|
||||||
const scalar = (key: keyof typeof VIBE_CONTEXT_VALUES) => input[key] === undefined
|
|
||||||
|| hasValue(VIBE_CONTEXT_VALUES[key], input[key]);
|
|
||||||
return (input.timeZone === undefined || typeof input.timeZone === 'string')
|
|
||||||
&& scalar('device') && scalar('activity') && scalar('locationCategory')
|
|
||||||
&& scalar('weather') && scalar('source');
|
|
||||||
}
|
|
||||||
@@ -14,7 +14,7 @@ const EVENT_ID = '33333333-3333-4333-8333-333333333333';
|
|||||||
function session(status: 'active' | 'ended' = 'active') {
|
function session(status: 'active' | 'ended' = 'active') {
|
||||||
return {
|
return {
|
||||||
id: SESSION_ID, user_id: 'user-1', status, seed_track_id: null,
|
id: SESSION_ID, user_id: 'user-1', status, seed_track_id: null,
|
||||||
context: { activity: 'focus' }, policy_version: DEFAULT_VIBE_POLICY_VERSION,
|
context: {}, policy_version: DEFAULT_VIBE_POLICY_VERSION,
|
||||||
started_at: new Date('2026-01-01T00:00:00.000Z'),
|
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,
|
last_event_at: new Date('2026-01-01T00:00:00.000Z'), ended_at: status === 'ended' ? new Date() : null,
|
||||||
} as any;
|
} as any;
|
||||||
@@ -63,15 +63,13 @@ describe('VibeSessionCoordinator', () => {
|
|||||||
it('creates an authoritative session, shadow state, initial plan revision, and ledger events', async () => {
|
it('creates an authoritative session, shadow state, initial plan revision, and ledger events', async () => {
|
||||||
const { db, director, coordinator } = setup();
|
const { db, director, coordinator } = setup();
|
||||||
|
|
||||||
const response = await coordinator.start('user-1', {
|
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(response).toMatchObject({ sessionId: SESSION_ID, planVersion: 1, now: { track_id: TRACK_ID } });
|
||||||
expect(db.createVibeSession).toHaveBeenCalledWith(expect.objectContaining({
|
expect(db.createVibeSession).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
userId: 'user-1', policyVersion: DEFAULT_VIBE_POLICY_VERSION,
|
userId: 'user-1', policyVersion: DEFAULT_VIBE_POLICY_VERSION,
|
||||||
}));
|
}));
|
||||||
expect(db.createSessionState).toHaveBeenCalledWith('user-1', 'focus', expect.any(Object), SESSION_ID);
|
expect(db.createSessionState).toHaveBeenCalledWith('user-1', undefined, expect.any(Object), SESSION_ID);
|
||||||
expect(director.buildPlan).toHaveBeenCalledWith('user-1', SESSION_ID, undefined);
|
expect(director.buildPlan).toHaveBeenCalledWith('user-1', SESSION_ID, undefined);
|
||||||
expect(db.publishVibePlan).toHaveBeenCalledWith(expect.objectContaining({
|
expect(db.publishVibePlan).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
sessionId: SESSION_ID, version: 1, reason: 'session_started',
|
sessionId: SESSION_ID, version: 1, reason: 'session_started',
|
||||||
@@ -154,38 +152,14 @@ describe('VibeSessionCoordinator', () => {
|
|||||||
expect(response).toMatchObject({ replanned: true, replanReason: 'feedback:completed', planVersion: 2 });
|
expect(response).toMatchObject({ replanned: true, replanReason: 'feedback:completed', planVersion: 2 });
|
||||||
});
|
});
|
||||||
|
|
||||||
it('normalizes context before the immutable event write', async () => {
|
it('creates a neutral durable profile until listening behaviour provides evidence', async () => {
|
||||||
const { db, coordinator } = setup();
|
|
||||||
(db.recordVibeEvent as any).mockResolvedValueOnce({
|
|
||||||
event: { id: 'event-1', type: 'context_changed', payload: {} }, inserted: true,
|
|
||||||
});
|
|
||||||
|
|
||||||
await coordinator.appendEvent('user-1', SESSION_ID, {
|
|
||||||
type: 'context_changed',
|
|
||||||
payload: {
|
|
||||||
context: { activity: 'walking', exactCoordinates: '53.2,50.1' },
|
|
||||||
rawBrowserTelemetry: { battery: 4 },
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(db.recordVibeEvent).toHaveBeenCalledWith(expect.objectContaining({
|
|
||||||
payload: {
|
|
||||||
context: expect.objectContaining({ activity: 'walking' }),
|
|
||||||
},
|
|
||||||
}));
|
|
||||||
const payload = (db.recordVibeEvent as any).mock.calls[0][0].payload;
|
|
||||||
expect(payload).not.toHaveProperty('rawBrowserTelemetry');
|
|
||||||
expect(payload.context).not.toHaveProperty('exactCoordinates');
|
|
||||||
});
|
|
||||||
|
|
||||||
it('creates the durable profile from the initial context goal', async () => {
|
|
||||||
const { db, coordinator } = setup();
|
const { db, coordinator } = setup();
|
||||||
|
|
||||||
await coordinator.start('user-1', { context: { activity: 'focus' } });
|
await coordinator.start('user-1', {});
|
||||||
|
|
||||||
expect(db.createVibeSession).toHaveBeenCalledWith(expect.objectContaining({
|
expect(db.createVibeSession).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
profile: expect.objectContaining({
|
profile: expect.objectContaining({
|
||||||
goals: { type: 'familiar', target: 1, progress: 0 },
|
goals: { type: 'discovery', target: 1, progress: 0 },
|
||||||
explorationCoefficient: 0.3,
|
explorationCoefficient: 0.3,
|
||||||
discoveryRadius: 0.38,
|
discoveryRadius: 0.38,
|
||||||
}),
|
}),
|
||||||
|
|||||||
@@ -1,12 +1,6 @@
|
|||||||
import { DbService, VibeEvent, VibePlan, VibeSession } from './db.service.js';
|
import { DbService, VibeEvent, VibePlan, VibeSession } from './db.service.js';
|
||||||
import { SessionDirector } from './session-director.service.js';
|
import { SessionDirector } from './session-director.service.js';
|
||||||
import { Candidate } from './generators.service.js';
|
import { Candidate } from './generators.service.js';
|
||||||
import {
|
|
||||||
initialVibeState,
|
|
||||||
normalizeVibeContext,
|
|
||||||
normalizeVibeEventPayload,
|
|
||||||
VibeContext,
|
|
||||||
} from './vibe-context.service.js';
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This is deliberately a narrow bridge between the durable Vibe ledger and
|
* This is deliberately a narrow bridge between the durable Vibe ledger and
|
||||||
@@ -16,7 +10,7 @@ import {
|
|||||||
export const DEFAULT_VIBE_POLICY_VERSION = 'vibe-v2-initial';
|
export const DEFAULT_VIBE_POLICY_VERSION = 'vibe-v2-initial';
|
||||||
|
|
||||||
export const VIBE_EVENT_TYPES = [
|
export const VIBE_EVENT_TYPES = [
|
||||||
'session_started', 'session_resumed', 'session_ended', 'context_changed',
|
'session_started', 'session_resumed', 'session_ended',
|
||||||
'plan_published', 'track_served', 'playback_started', 'progress', 'completed',
|
'plan_published', 'track_served', 'playback_started', 'progress', 'completed',
|
||||||
'skipped', 'disliked', 'kept', 'favourite_added', 'queue_removed',
|
'skipped', 'disliked', 'kept', 'favourite_added', 'queue_removed',
|
||||||
'manual_search', 'album_opened', 'artist_opened', 'playlist_added',
|
'manual_search', 'album_opened', 'artist_opened', 'playlist_added',
|
||||||
@@ -27,9 +21,6 @@ export type VibeEventType = (typeof VIBE_EVENT_TYPES)[number];
|
|||||||
|
|
||||||
export interface StartVibeSessionInput {
|
export interface StartVibeSessionInput {
|
||||||
seedTrackId?: string;
|
seedTrackId?: string;
|
||||||
context?: VibeContext | Record<string, unknown>;
|
|
||||||
intent?: string;
|
|
||||||
policyVersion?: string;
|
|
||||||
resumeSessionId?: string;
|
resumeSessionId?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -79,14 +70,19 @@ export class VibeSessionCoordinator {
|
|||||||
|
|
||||||
async start(userId: string, input: StartVibeSessionInput): Promise<VibeSessionResponse> {
|
async start(userId: string, input: StartVibeSessionInput): Promise<VibeSessionResponse> {
|
||||||
if (input.resumeSessionId) return this.resume(userId, input.resumeSessionId);
|
if (input.resumeSessionId) return this.resume(userId, input.resumeSessionId);
|
||||||
const context = normalizeVibeContext({ ...(input.context ?? {}) });
|
// Vibe has no reliable device/activity/location signal. Start from neutral
|
||||||
const initialState = initialVibeState(context);
|
// recommendation state and let actual listening behaviour shape the plan.
|
||||||
const policyVersion = input.policyVersion ?? DEFAULT_VIBE_POLICY_VERSION;
|
const initialState = {
|
||||||
|
energy: 0.5,
|
||||||
|
noveltyHunger: 0.3,
|
||||||
|
explorationCoefficient: 0.3,
|
||||||
|
discoveryRadius: 0.38,
|
||||||
|
sessionGoal: { type: 'discovery' as const, target: 1, progress: 0 },
|
||||||
|
};
|
||||||
const session = await this.db.createVibeSession({
|
const session = await this.db.createVibeSession({
|
||||||
userId,
|
userId,
|
||||||
policyVersion,
|
policyVersion: DEFAULT_VIBE_POLICY_VERSION,
|
||||||
seedTrackId: input.seedTrackId ?? null,
|
seedTrackId: input.seedTrackId ?? null,
|
||||||
context: { ...context },
|
|
||||||
profile: {
|
profile: {
|
||||||
goals: initialState.sessionGoal,
|
goals: initialState.sessionGoal,
|
||||||
explorationCoefficient: initialState.explorationCoefficient,
|
explorationCoefficient: initialState.explorationCoefficient,
|
||||||
@@ -99,14 +95,13 @@ export class VibeSessionCoordinator {
|
|||||||
// session while the durable tables remain the source of truth.
|
// session while the durable tables remain the source of truth.
|
||||||
await this.db.createSessionState(
|
await this.db.createSessionState(
|
||||||
userId,
|
userId,
|
||||||
initialState.contextLabel ?? input.intent,
|
undefined,
|
||||||
{
|
{
|
||||||
energy: initialState.energy,
|
energy: initialState.energy,
|
||||||
noveltyHunger: initialState.noveltyHunger,
|
noveltyHunger: initialState.noveltyHunger,
|
||||||
explorationCoefficient: initialState.explorationCoefficient,
|
explorationCoefficient: initialState.explorationCoefficient,
|
||||||
discoveryRadius: initialState.discoveryRadius,
|
discoveryRadius: initialState.discoveryRadius,
|
||||||
sessionGoal: initialState.sessionGoal,
|
sessionGoal: initialState.sessionGoal,
|
||||||
context,
|
|
||||||
},
|
},
|
||||||
session.id,
|
session.id,
|
||||||
);
|
);
|
||||||
@@ -114,7 +109,7 @@ export class VibeSessionCoordinator {
|
|||||||
sessionId: session.id,
|
sessionId: session.id,
|
||||||
userId,
|
userId,
|
||||||
type: 'session_started',
|
type: 'session_started',
|
||||||
payload: { policyVersion, context, intent: input.intent ?? null },
|
payload: { policyVersion: DEFAULT_VIBE_POLICY_VERSION },
|
||||||
});
|
});
|
||||||
|
|
||||||
const candidates = await this.director.buildPlan(userId, session.id, input.seedTrackId);
|
const candidates = await this.director.buildPlan(userId, session.id, input.seedTrackId);
|
||||||
@@ -128,8 +123,7 @@ export class VibeSessionCoordinator {
|
|||||||
reason: 'session_started',
|
reason: 'session_started',
|
||||||
stateSnapshot: state,
|
stateSnapshot: state,
|
||||||
objectiveSnapshot: {
|
objectiveSnapshot: {
|
||||||
policyVersion,
|
policyVersion: DEFAULT_VIBE_POLICY_VERSION,
|
||||||
intent: input.intent ?? null,
|
|
||||||
horizonTracks: candidates.length,
|
horizonTracks: candidates.length,
|
||||||
...(candidates[0]?.plan?.objective ?? {}),
|
...(candidates[0]?.plan?.objective ?? {}),
|
||||||
},
|
},
|
||||||
@@ -203,7 +197,6 @@ export class VibeSessionCoordinator {
|
|||||||
// events are immutable. The DB repeats this boundary for non-HTTP
|
// events are immutable. The DB repeats this boundary for non-HTTP
|
||||||
// callers; keeping it here also makes coordinator callers see exactly
|
// callers; keeping it here also makes coordinator callers see exactly
|
||||||
// what will be persisted.
|
// what will be persisted.
|
||||||
const payload = normalizeVibeEventPayload(input.type, input.payload);
|
|
||||||
const result = await this.db.recordVibeEvent({
|
const result = await this.db.recordVibeEvent({
|
||||||
sessionId,
|
sessionId,
|
||||||
userId,
|
userId,
|
||||||
@@ -213,7 +206,7 @@ export class VibeSessionCoordinator {
|
|||||||
occurredAt: input.occurredAt,
|
occurredAt: input.occurredAt,
|
||||||
positionMs: input.positionMs,
|
positionMs: input.positionMs,
|
||||||
durationMs: input.durationMs,
|
durationMs: input.durationMs,
|
||||||
payload,
|
payload: input.payload,
|
||||||
});
|
});
|
||||||
// The ledger write is authoritative; this idempotent projection updates
|
// The ledger write is authoritative; this idempotent projection updates
|
||||||
// exploration only after the exact event exists. Keep the compatibility
|
// exploration only after the exact event exists. Keep the compatibility
|
||||||
|
|||||||
@@ -32,10 +32,6 @@ services:
|
|||||||
TYPESENSE_API_KEY: ${TYPESENSE_API_KEY}
|
TYPESENSE_API_KEY: ${TYPESENSE_API_KEY}
|
||||||
MUZICK_API_KEY: ${MUZICK_API_KEY}
|
MUZICK_API_KEY: ${MUZICK_API_KEY}
|
||||||
MUZICK_ADMIN_KEY: ${MUZICK_ADMIN_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
|
MUSIC_DIR: /music
|
||||||
volumes:
|
volumes:
|
||||||
# READ-ONLY, deliberately. Nothing in the API request path may write to
|
# READ-ONLY, deliberately. Nothing in the API request path may write to
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ describe('durable vibe service', () => {
|
|||||||
|
|
||||||
await vibeService.next('session', 3);
|
await vibeService.next('session', 3);
|
||||||
|
|
||||||
expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/next', { expectedPlanVersion: 3 });
|
expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/advance', { expectedPlanVersion: 3 });
|
||||||
});
|
});
|
||||||
|
|
||||||
it('uses an explicit idempotency key to advance a served but unplayable item', async () => {
|
it('uses an explicit idempotency key to advance a served but unplayable item', async () => {
|
||||||
@@ -21,7 +21,7 @@ describe('durable vibe service', () => {
|
|||||||
eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track',
|
eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track',
|
||||||
});
|
});
|
||||||
|
|
||||||
expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/next', {
|
expect(post).toHaveBeenCalledWith('/v2/vibe/sessions/session/advance', {
|
||||||
expectedPlanVersion: 3,
|
expectedPlanVersion: 3,
|
||||||
unplayable: { eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track' },
|
unplayable: { eventId: 'event', planVersionId: 'plan', ordinal: 4, trackId: 'track' },
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -71,7 +71,7 @@ export const vibeService = {
|
|||||||
},
|
},
|
||||||
|
|
||||||
async next(sessionId: string, expectedPlanVersion: number): Promise<DurableVibeSessionResponse> {
|
async next(sessionId: string, expectedPlanVersion: number): Promise<DurableVibeSessionResponse> {
|
||||||
const res = await api.post<DurableVibeSessionResponse>(`/v2/vibe/sessions/${sessionId}/next`, {
|
const res = await api.post<DurableVibeSessionResponse>(`/v2/vibe/sessions/${sessionId}/advance`, {
|
||||||
expectedPlanVersion,
|
expectedPlanVersion,
|
||||||
});
|
});
|
||||||
return res.data;
|
return res.data;
|
||||||
@@ -82,7 +82,7 @@ export const vibeService = {
|
|||||||
expectedPlanVersion: number,
|
expectedPlanVersion: number,
|
||||||
unplayable: VibeUnplayableItemInput,
|
unplayable: VibeUnplayableItemInput,
|
||||||
): Promise<DurableVibeSessionResponse> {
|
): Promise<DurableVibeSessionResponse> {
|
||||||
const res = await api.post<DurableVibeSessionResponse>(`/v2/vibe/sessions/${sessionId}/next`, {
|
const res = await api.post<DurableVibeSessionResponse>(`/v2/vibe/sessions/${sessionId}/advance`, {
|
||||||
expectedPlanVersion,
|
expectedPlanVersion,
|
||||||
unplayable,
|
unplayable,
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -202,7 +202,7 @@ function isSessionTerminalError(error: unknown): boolean {
|
|||||||
export function vibeErrorMessage(error: unknown): string {
|
export function vibeErrorMessage(error: unknown): string {
|
||||||
if (!axios.isAxiosError(error)) return 'Could not refresh this Vibe. Please try again.';
|
if (!axios.isAxiosError(error)) return 'Could not refresh this Vibe. Please try again.';
|
||||||
switch (error.response?.status) {
|
switch (error.response?.status) {
|
||||||
case 401: return 'Vibe needs a trusted local user identity. Set MUZICK_VIBE_USER_ID and try again.';
|
case 400: return 'Vibe needs a valid user identity.';
|
||||||
case 404: return 'This Vibe session is no longer available.';
|
case 404: return 'This Vibe session is no longer available.';
|
||||||
case 409: return 'This Vibe session has already ended or was replaced.';
|
case 409: return 'This Vibe session has already ended or was replaced.';
|
||||||
default: return 'Could not refresh this Vibe. Please try again.';
|
default: return 'Could not refresh this Vibe. Please try again.';
|
||||||
|
|||||||
@@ -75,8 +75,8 @@ export interface HealthResponse {
|
|||||||
redis: 'ok' | 'error' | 'unknown';
|
redis: 'ok' | 'error' | 'unknown';
|
||||||
}
|
}
|
||||||
|
|
||||||
// Active v2 recommendation session. The sessionId comes from
|
// Active durable recommendation session. The sessionId comes from
|
||||||
// POST /api/v2/vibe/start and identifies the Redis-stored plan.
|
// POST /api/v2/vibe/sessions and identifies its persisted plan.
|
||||||
export interface VibeSession {
|
export interface VibeSession {
|
||||||
sessionId: string;
|
sessionId: string;
|
||||||
seedTrackId: string | null;
|
seedTrackId: string | null;
|
||||||
|
|||||||
Reference in New Issue
Block a user