From b8844e1ff0e8242885361f9aba3516a36a6ad2f8 Mon Sep 17 00:00:00 2001 From: coder-mos2 Date: Sun, 2 Aug 2026 23:16:09 -0500 Subject: [PATCH] fix(sync): close local queue semantic merge gap --- .../commands/command-executor-p8012.spec.ts | 20 ++++++++++++++-- .../gateway/src/gc/session-gc.service.spec.ts | 13 +++++++++++ .../system-override.service.spec.ts | 23 +++++++++++++++++++ apps/gateway/src/queue/queue.service.spec.ts | 1 + apps/gateway/src/queue/queue.service.ts | 7 ++++++ 5 files changed, 62 insertions(+), 2 deletions(-) create mode 100644 apps/gateway/src/preferences/system-override.service.spec.ts diff --git a/apps/gateway/src/commands/command-executor-p8012.spec.ts b/apps/gateway/src/commands/command-executor-p8012.spec.ts index fc5d9660..b22ed05a 100644 --- a/apps/gateway/src/commands/command-executor-p8012.spec.ts +++ b/apps/gateway/src/commands/command-executor-p8012.spec.ts @@ -72,13 +72,13 @@ const mockChatGateway = { broadcastSessionInfo: vi.fn(), }; -function buildService(): CommandExecutorService { +function buildService(redis: typeof mockRedis | null = mockRedis): CommandExecutorService { return new CommandExecutorService( mockRegistry as never, mockAgentService as never, mockSystemOverride as never, mockSessionGC as never, - mockRedis as never, + redis as never, mockBrain as never, null, mockChatGateway as never, @@ -131,6 +131,22 @@ describe('CommandExecutorService — P8-012 commands', () => { expect(ttl).toBe(300); }); + it('/provider login remains available without Redis on the local tier', async () => { + const localService = buildService(null); + const payload: SlashCommandPayload = { + command: 'provider', + args: 'login anthropic', + conversationId, + }; + + const result = await localService.execute(payload, userScope); + + expect(result.success).toBe(true); + expect(result.message).not.toContain('token='); + expect(result.data).toEqual({ provider: 'anthropic' }); + expect(mockRedis.set).not.toHaveBeenCalled(); + }); + // /provider with no args — returns usage it('/provider with no args returns usage message', async () => { const payload: SlashCommandPayload = { command: 'provider', conversationId }; diff --git a/apps/gateway/src/gc/session-gc.service.spec.ts b/apps/gateway/src/gc/session-gc.service.spec.ts index 8c5ab6ad..014df200 100644 --- a/apps/gateway/src/gc/session-gc.service.spec.ts +++ b/apps/gateway/src/gc/session-gc.service.spec.ts @@ -119,6 +119,19 @@ describe('SessionGCService', () => { ).resolves.toEqual({ allowed: true }); }); + it('collect() skips Valkey but still demotes only the requested session on local tier', async () => { + const localService = new SessionGCService(null, mockLogService as unknown as LogService); + + const result = await localService.collect('local-session'); + + expect(result.sessionId).toBe('local-session'); + expect(result.cleaned.valkeyKeys).toBeUndefined(); + expect(mockLogService.logs.promoteSessionToWarm).toHaveBeenCalledWith( + 'local-session', + expect.any(Date), + ); + }); + it('collect() returns sessionId in result', async () => { const result = await service.collect('test-session-id'); expect(result.sessionId).toBe('test-session-id'); diff --git a/apps/gateway/src/preferences/system-override.service.spec.ts b/apps/gateway/src/preferences/system-override.service.spec.ts new file mode 100644 index 00000000..8d65080a --- /dev/null +++ b/apps/gateway/src/preferences/system-override.service.spec.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from 'vitest'; +import type { MosaicConfig } from '@mosaicstack/config'; +import { SystemOverrideService } from './system-override.service.js'; + +const localConfig = { queue: { type: 'local' } } as MosaicConfig; + +describe('SystemOverrideService local tier', () => { + it('keeps ephemeral overrides isolated by tenant and user scope', async () => { + const service = new SystemOverrideService(localConfig); + const firstScope = { tenantId: 'tenant-a', userId: 'user-a' }; + const secondScope = { tenantId: 'tenant-b', userId: 'user-b' }; + + await service.set('shared-session', 'first override', firstScope); + await service.set('shared-session', 'second override', secondScope); + + await expect(service.get('shared-session', firstScope)).resolves.toBe('first override'); + await expect(service.get('shared-session', secondScope)).resolves.toBe('second override'); + + await service.clear('shared-session', firstScope); + await expect(service.get('shared-session', firstScope)).resolves.toBeNull(); + await expect(service.get('shared-session', secondScope)).resolves.toBe('second override'); + }); +}); diff --git a/apps/gateway/src/queue/queue.service.spec.ts b/apps/gateway/src/queue/queue.service.spec.ts index 85f1e641..94e86a02 100644 --- a/apps/gateway/src/queue/queue.service.spec.ts +++ b/apps/gateway/src/queue/queue.service.spec.ts @@ -17,6 +17,7 @@ describe('QueueService local tier', () => { await expect( service.addRepeatableJob('mosaic-test', 'local-noop', {}, '* * * * *'), ).resolves.toBeUndefined(); + await expect(service.removeRepeatableJobs('mosaic-test', 'local-noop')).resolves.toBe(0); await expect(service.getHealthStatus()).resolves.toEqual({ queues: {}, healthy: true }); await expect(service.listJobs()).resolves.toEqual([]); await expect(service.retryJob('mosaic-test__1')).resolves.toEqual({ diff --git a/apps/gateway/src/queue/queue.service.ts b/apps/gateway/src/queue/queue.service.ts index a3a8162e..76f6268c 100644 --- a/apps/gateway/src/queue/queue.service.ts +++ b/apps/gateway/src/queue/queue.service.ts @@ -199,7 +199,14 @@ export class QueueService implements OnModuleInit, OnModuleDestroy { * safe retirement of previously registered system-wide jobs. */ async removeRepeatableJobs(queueName: string, jobName: string): Promise { + if (!this.enabled) { + this.logger.debug( + `Skipping repeatable-job removal for "${jobName}" on "${queueName}" (local tier — BullMQ disabled)`, + ); + return 0; + } const queue = this.getQueue(queueName); + if (!queue) return 0; const jobs = await queue.getRepeatableJobs(); const matchingJobs = jobs.filter((job) => job.name === jobName); await Promise.all(matchingJobs.map((job) => queue.removeRepeatableByKey(job.key)));