diff --git a/services/cowork-service/src/lib/flush-scheduler.test.ts b/services/cowork-service/src/lib/flush-scheduler.test.ts new file mode 100644 index 00000000..25363432 --- /dev/null +++ b/services/cowork-service/src/lib/flush-scheduler.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; + +// Mock dependencies before imports +vi.mock('./config.js', () => ({ + config: { PLATFORM_SERVICE_URL: 'http://localhost:4003', FEATURE_FLAGS_ENABLED: false }, +})); +vi.mock('./product-config.js', () => ({ + PRODUCT_ID: 'clawcowork', +})); +vi.mock('./ipc-bridge.js', () => ({ + getIpcBridge: vi.fn(), +})); +vi.mock('./feature-flags.js', () => ({ + setFlag: vi.fn(), +})); +vi.mock('./platform-client.js', () => ({ + postAuditEvents: vi.fn(), + postTelemetryEvents: vi.fn(), + postUsageRecords: vi.fn(), + pollFlags: vi.fn(), +})); + +import { FlushScheduler } from './flush-scheduler.js'; +import { getIpcBridge } from './ipc-bridge.js'; +import { setFlag } from './feature-flags.js'; +import { + postAuditEvents, + postTelemetryEvents, + postUsageRecords, + pollFlags, +} from './platform-client.js'; + +const mockLog = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + fatal: vi.fn(), + trace: vi.fn(), + child: vi.fn(), + silent: vi.fn(), + level: 'info', +} as unknown as import('fastify').FastifyBaseLogger; + +function createMockBridge(running = true) { + return { + isRunning: running, + flushAudit: vi.fn().mockResolvedValue({ result: { count: 2, events: [ + { userId: 'u1', action: 'task.submit', category: 'task', details: {} }, + { userId: 'u2', action: 'task.cancel', category: 'task', details: {} }, + ] } }), + flushTelemetry: vi.fn().mockResolvedValue({ result: { count: 1, events: [ + { id: 'e1', name: 'task.submit', timestamp: '2026-01-01T00:00:00Z' }, + ] } }), + flushBudget: vi.fn().mockResolvedValue({ result: { count: 1, records: [ + { userId: 'u1', date: '2026-01-01', model: 'claude-sonnet', inputTokens: 100, outputTokens: 50, costUsd: 0.01 }, + ] } }), + updateFlags: vi.fn().mockResolvedValue({ result: { updated: 3 } }), + }; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('FlushScheduler', () => { + describe('lifecycle', () => { + it('starts and stops without errors', () => { + const scheduler = new FlushScheduler({ log: mockLog }); + expect(scheduler.isRunning).toBe(false); + + scheduler.start(); + expect(scheduler.isRunning).toBe(true); + + scheduler.stop(); + expect(scheduler.isRunning).toBe(false); + }); + + it('start is idempotent', () => { + const scheduler = new FlushScheduler({ log: mockLog }); + scheduler.start(); + scheduler.start(); // no-op + expect(scheduler.isRunning).toBe(true); + scheduler.stop(); + }); + }); + + describe('flushAll', () => { + it('skips when IPC bridge is not running', async () => { + const bridge = createMockBridge(false); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.flushAll(); + + expect(result.skipped).toBe(true); + expect(bridge.flushAudit).not.toHaveBeenCalled(); + }); + + it('drains audit, telemetry, and budget from IPC and posts to platform-service', async () => { + const bridge = createMockBridge(true); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + vi.mocked(postAuditEvents).mockResolvedValue({ posted: 2, errors: 0 }); + vi.mocked(postTelemetryEvents).mockResolvedValue({ accepted: 1, rejected: 0 }); + vi.mocked(postUsageRecords).mockResolvedValue({ posted: 1, errors: 0 }); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.flushAll(); + + expect(result.skipped).toBe(false); + expect(result.audit).toEqual({ drained: 2, posted: 2, errors: 0 }); + expect(result.telemetry).toEqual({ drained: 1, accepted: 1, rejected: 0 }); + expect(result.budget).toEqual({ drained: 1, posted: 1, errors: 0 }); + }); + + it('handles IPC flush_audit error gracefully', async () => { + const bridge = createMockBridge(true); + bridge.flushAudit.mockResolvedValue({ error: { code: -32603, message: 'internal' } }); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + vi.mocked(postTelemetryEvents).mockResolvedValue({ accepted: 1, rejected: 0 }); + vi.mocked(postUsageRecords).mockResolvedValue({ posted: 1, errors: 0 }); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.flushAll(); + + expect(result.audit).toBeNull(); + expect(result.telemetry).not.toBeNull(); + }); + + it('handles empty IPC responses', async () => { + const bridge = createMockBridge(true); + bridge.flushAudit.mockResolvedValue({ result: { count: 0, events: [] } }); + bridge.flushTelemetry.mockResolvedValue({ result: { count: 0, events: [] } }); + bridge.flushBudget.mockResolvedValue({ result: { count: 0, records: [] } }); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.flushAll(); + + expect(result.audit).toEqual({ drained: 0, posted: 0, errors: 0 }); + expect(postAuditEvents).not.toHaveBeenCalled(); + }); + }); + + describe('pollAndSyncFlags', () => { + it('polls flags and updates local registry + IPC bridge', async () => { + const bridge = createMockBridge(true); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + vi.mocked(pollFlags).mockResolvedValue({ + flags: { sandbox_enabled: true, computer_use_enabled: true, wasm_plugins_enabled: false }, + productId: 'clawcowork', + }); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.pollAndSyncFlags(); + + expect(result).toEqual({ updated: 3 }); + expect(setFlag).toHaveBeenCalledTimes(3); + expect(setFlag).toHaveBeenCalledWith('sandbox_enabled', true); + expect(setFlag).toHaveBeenCalledWith('computer_use_enabled', true); + expect(bridge.updateFlags).toHaveBeenCalledWith( + { sandbox_enabled: true, computer_use_enabled: true, wasm_plugins_enabled: false }, + { userId: 'system', role: 'admin' }, + ); + }); + + it('handles poll failure gracefully', async () => { + vi.mocked(pollFlags).mockRejectedValue(new Error('network down')); + + const scheduler = new FlushScheduler({ log: mockLog }); + const result = await scheduler.pollAndSyncFlags(); + + expect(result).toBeNull(); + expect(mockLog.error).toHaveBeenCalled(); + }); + + it('skips IPC update when bridge is not running', async () => { + const bridge = createMockBridge(false); + vi.mocked(getIpcBridge).mockReturnValue(bridge as unknown as ReturnType); + vi.mocked(pollFlags).mockResolvedValue({ + flags: { sandbox_enabled: true }, + productId: 'clawcowork', + }); + + const scheduler = new FlushScheduler({ log: mockLog }); + await scheduler.pollAndSyncFlags(); + + expect(setFlag).toHaveBeenCalledWith('sandbox_enabled', true); + expect(bridge.updateFlags).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/services/cowork-service/src/lib/platform-client.test.ts b/services/cowork-service/src/lib/platform-client.test.ts new file mode 100644 index 00000000..0fa45ba4 --- /dev/null +++ b/services/cowork-service/src/lib/platform-client.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; + +vi.mock('./config.js', () => ({ + config: { PLATFORM_SERVICE_URL: 'http://localhost:4003' }, +})); +vi.mock('./product-config.js', () => ({ + PRODUCT_ID: 'clawcowork', +})); + +import { + postAuditEvents, + postTelemetryEvents, + postUsageRecords, + pollFlags, +} from './platform-client.js'; + +const mockFetch = vi.fn(); +globalThis.fetch = mockFetch; + +beforeEach(() => { + mockFetch.mockReset(); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('platform-client', () => { + describe('postAuditEvents', () => { + it('posts each audit entry individually', async () => { + mockFetch.mockResolvedValue({ ok: true, json: () => Promise.resolve({ accepted: true }) }); + + const result = await postAuditEvents([ + { userId: 'u1', action: 'task.submit', category: 'task' }, + { userId: 'u2', action: 'task.cancel', category: 'task' }, + ]); + + expect(result).toEqual({ posted: 2, errors: 0 }); + expect(mockFetch).toHaveBeenCalledTimes(2); + expect(mockFetch.mock.calls[0][0]).toBe('http://localhost:4003/audit'); + }); + + it('counts errors without throwing', async () => { + mockFetch + .mockResolvedValueOnce({ ok: true, json: () => Promise.resolve({ accepted: true }) }) + .mockResolvedValueOnce({ ok: false, status: 500, text: () => Promise.resolve('fail') }); + + const result = await postAuditEvents([ + { userId: 'u1', action: 'a1' }, + { userId: 'u2', action: 'a2' }, + ]); + expect(result).toEqual({ posted: 1, errors: 1 }); + }); + }); + + describe('postTelemetryEvents', () => { + it('posts batch to /telemetry/events', async () => { + mockFetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ accepted: 3, rejected: 0 }), + }); + + const events = [ + { id: 'e1', name: 'task.submit', timestamp: '2026-01-01T00:00:00Z' }, + { id: 'e2', name: 'task.complete', timestamp: '2026-01-01T00:01:00Z' }, + { id: 'e3', name: 'task.cancel', timestamp: '2026-01-01T00:02:00Z' }, + ]; + const result = await postTelemetryEvents(events); + + expect(result).toEqual({ accepted: 3, rejected: 0 }); + expect(mockFetch).toHaveBeenCalledTimes(1); + const body = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(body.productId).toBe('clawcowork'); + expect(body.events).toHaveLength(3); + }); + + it('returns zero counts for empty array', async () => { + const result = await postTelemetryEvents([]); + expect(result).toEqual({ accepted: 0, rejected: 0 }); + expect(mockFetch).not.toHaveBeenCalled(); + }); + }); + + describe('postUsageRecords', () => { + it('posts each record individually', async () => { + mockFetch.mockResolvedValue({ ok: true, json: () => Promise.resolve({}) }); + + const result = await postUsageRecords([ + { userId: 'u1', date: '2026-01-01', model: 'claude-sonnet', inputTokens: 100 }, + ]); + expect(result).toEqual({ posted: 1, errors: 0 }); + expect(mockFetch.mock.calls[0][0]).toBe('http://localhost:4003/usage'); + }); + }); + + describe('pollFlags', () => { + it('calls GET /flags/poll and returns flags', async () => { + mockFetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ flags: { sandbox_enabled: true }, productId: 'clawcowork' }), + }); + + const result = await pollFlags(); + expect(result.flags.sandbox_enabled).toBe(true); + expect(result.productId).toBe('clawcowork'); + expect(mockFetch.mock.calls[0][0]).toBe('http://localhost:4003/flags/poll'); + }); + + it('appends platform query param when provided', async () => { + mockFetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ flags: {}, productId: 'clawcowork' }), + }); + + await pollFlags('macos'); + expect(mockFetch.mock.calls[0][0]).toBe('http://localhost:4003/flags/poll?platform=macos'); + }); + + it('throws on non-ok response', async () => { + mockFetch.mockResolvedValue({ + ok: false, + status: 503, + text: () => Promise.resolve('service unavailable'), + }); + + await expect(pollFlags()).rejects.toThrow('platform-service GET /flags/poll → 503'); + }); + }); +});