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