diff --git a/src/__tests__/main/cue/cue-db-integration.test.ts b/src/__tests__/main/cue/cue-db-integration.test.ts new file mode 100644 index 0000000000..446ba14bd9 --- /dev/null +++ b/src/__tests__/main/cue/cue-db-integration.test.ts @@ -0,0 +1,526 @@ +/** + * Phase 15B — Cue database contract / integration tests. + * + * Exercises the contract cue-engine depends on (ordering, UNIQUE, prune-by-age, + * heartbeat upsert, safe-wrapper no-throw) through the in-memory mirror + * defined in `cue-integration-test-helpers.ts`. We cannot use real + * `better-sqlite3` under vitest because the native binary is built for + * Electron's ABI and fails to load in plain Node; the in-memory mirror + * preserves the SQL semantics that actually matter for the rest of the + * engine (ordering, UNIQUE constraints, prune cutoff). + * + * A `describe.skipIf(!canLoadBetterSqlite3())` block at the bottom runs one + * real-SQLite smoke round-trip when the binary is available locally — this + * catches drift between the mirror and the native module without breaking CI + * on hosts that can't load it. + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import * as path from 'path'; +import * as os from 'os'; +import * as fs from 'fs'; +import { + createInMemoryCueDb, + canLoadBetterSqlite3, + type InMemoryCueDb, +} from './cue-integration-test-helpers'; + +describe('Phase 15B — cue-db in-memory contract', () => { + let db: InMemoryCueDb; + + beforeEach(() => { + db = createInMemoryCueDb(); + db.initCueDb(); + }); + + // ─── Lifecycle ──────────────────────────────────────────────────────── + + describe('lifecycle', () => { + it('is ready after init and not ready after close', () => { + expect(db.isCueDbReady()).toBe(true); + db.closeCueDb(); + expect(db.isCueDbReady()).toBe(false); + }); + + it('initCueDb is idempotent — second call is a no-op', () => { + // Calling init twice must not throw nor reset the current state. + db.recordCueEvent({ + id: 'e1', + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + db.initCueDb(); // second call + expect(db.getRecentCueEvents(0).length).toBe(1); + }); + + it('getRecentCueEvents throws when DB is not initialized', () => { + db.closeCueDb(); + expect(() => db.getRecentCueEvents(0)).toThrow(/not initialized/); + }); + }); + + // ─── Event journal ──────────────────────────────────────────────────── + + describe('event journal', () => { + it('records and retrieves a single event', () => { + db.recordCueEvent({ + id: 'evt-1', + type: 'time.heartbeat', + triggerName: 'hb', + sessionId: 'session-1', + subscriptionName: 'sub-1', + status: 'running', + payload: '{"x":1}', + }); + const events = db.getRecentCueEvents(0); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + id: 'evt-1', + type: 'time.heartbeat', + status: 'running', + payload: '{"x":1}', + }); + expect(events[0].completedAt).toBeNull(); + }); + + it('returns events in ORDER BY created_at DESC', () => { + db.setNowOverride(1000); + db.recordCueEvent({ + id: 'e1', + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + db.setNowOverride(2000); + db.recordCueEvent({ + id: 'e2', + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + db.setNowOverride(3000); + db.recordCueEvent({ + id: 'e3', + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + db.clearNowOverride(); + const events = db.getRecentCueEvents(0); + expect(events.map((e) => e.id)).toEqual(['e3', 'e2', 'e1']); + }); + + it('LIMIT clause caps the result set', () => { + for (let i = 0; i < 10; i++) { + db.setNowOverride(1000 + i); + db.recordCueEvent({ + id: `e${i}`, + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + } + db.clearNowOverride(); + expect(db.getRecentCueEvents(0, 3)).toHaveLength(3); + expect(db.getRecentCueEvents(0).length).toBe(10); + }); + + it('filters events by created_at >= since', () => { + db.setNowOverride(1000); + db.recordCueEvent({ + id: 'old', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(2000); + db.recordCueEvent({ + id: 'new', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.clearNowOverride(); + expect(db.getRecentCueEvents(1500).map((e) => e.id)).toEqual(['new']); + }); + + it('INSERT OR REPLACE overwrites a duplicate id', () => { + db.recordCueEvent({ + id: 'dup', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }); + db.recordCueEvent({ + id: 'dup', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'completed', + }); + const events = db.getRecentCueEvents(0); + expect(events).toHaveLength(1); + expect(events[0].status).toBe('completed'); + }); + + it('updateCueEventStatus flips status and sets completedAt', () => { + db.setNowOverride(1000); + db.recordCueEvent({ + id: 'e1', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }); + db.setNowOverride(5000); + db.updateCueEventStatus('e1', 'completed'); + const events = db.getRecentCueEvents(0); + expect(events[0].status).toBe('completed'); + expect(events[0].completedAt).toBe(5000); + db.clearNowOverride(); + }); + + it('updateCueEventStatus is a no-op when id does not exist (mirrors WHERE match fails)', () => { + expect(() => db.updateCueEventStatus('nonexistent', 'completed')).not.toThrow(); + expect(db.getRecentCueEvents(0)).toHaveLength(0); + }); + + it('safeRecordCueEvent swallows errors and is non-throwing', () => { + db.queueWriteFailure(new Error('disk full')); + expect(() => + db.safeRecordCueEvent({ + id: 'e1', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }) + ).not.toThrow(); + // The failed write left no row — non-fatal, as documented. + expect(db.getRecentCueEvents(0)).toHaveLength(0); + }); + + it('safeUpdateCueEventStatus swallows errors and is non-throwing', () => { + db.recordCueEvent({ + id: 'e1', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }); + db.queueWriteFailure(new Error('disk full')); + expect(() => db.safeUpdateCueEventStatus('e1', 'completed')).not.toThrow(); + // Status unchanged because the failure short-circuited the update. + expect(db.getRecentCueEvents(0)[0].status).toBe('running'); + }); + }); + + // ─── Rapid successive writes ────────────────────────────────────────── + // + // JS is single-threaded and `Promise.resolve().then(...)` just schedules + // microtasks — there's no real concurrency. What we exercise here is a + // microtask flood: 100 writes serialized through the event loop in + // immediate succession. Still worth pinning because the mirror uses a + // Map + array pair for ordering, and a naive refactor that rebuilt the + // ordering array from the Map on every write would silently lose events. + + describe('rapid successive writes', () => { + it('preserves all 100 events written in rapid succession', async () => { + const writes = Array.from({ length: 100 }, (_, i) => + Promise.resolve().then(() => + db.recordCueEvent({ + id: `c${i}`, + type: 't', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }) + ) + ); + await Promise.all(writes); + expect(db.getRecentCueEvents(0)).toHaveLength(100); + }); + }); + + // ─── Heartbeat ──────────────────────────────────────────────────────── + + describe('heartbeat', () => { + it('returns null before first heartbeat', () => { + expect(db.getLastHeartbeat()).toBeNull(); + }); + + it('upserts single-row heartbeat (id=1 replacement semantics)', () => { + db.setNowOverride(100); + db.updateHeartbeat(); + expect(db.getLastHeartbeat()).toBe(100); + db.setNowOverride(200); + db.updateHeartbeat(); + expect(db.getLastHeartbeat()).toBe(200); // replaced, not appended + db.clearNowOverride(); + }); + }); + + // ─── Prune ──────────────────────────────────────────────────────────── + + describe('pruneCueEvents', () => { + it('deletes events older than the cutoff', () => { + // Step `now` forward deterministically: old event at 1_000, + // recent event at 9_500, prune call at 10_000. With + // olderThanMs=5_000 the cutoff is 10_000 - 5_000 = 5_000, so the + // old event (1_000 < 5_000) is dropped and the recent event + // (9_500 >= 5_000) survives. + db.setNowOverride(1000); + db.recordCueEvent({ + id: 'old', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(9500); + db.recordCueEvent({ + id: 'recent', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(10000); + db.pruneCueEvents(5000); // cutoff = 10_000 - 5_000 = 5_000 + + const ids = db.getRecentCueEvents(0).map((e) => e.id); + expect(ids).toEqual(['recent']); + db.clearNowOverride(); + }); + + it('is a no-op when no events predate the cutoff', () => { + db.setNowOverride(10000); + db.recordCueEvent({ + id: 'new', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.pruneCueEvents(5000); + expect(db.getRecentCueEvents(0)).toHaveLength(1); + db.clearNowOverride(); + }); + + it('preserves ordering of the remaining events', () => { + db.setNowOverride(1000); + db.recordCueEvent({ + id: 'a', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(2000); + db.recordCueEvent({ + id: 'b', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(3000); + db.recordCueEvent({ + id: 'c', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'x', + }); + db.setNowOverride(4000); + // cutoff = now - olderThanMs = 4000 - 2500 = 1500 → drops only 'a' (createdAt=1000). + db.pruneCueEvents(2500); + const ids = db.getRecentCueEvents(0).map((e) => e.id); + expect(ids).toEqual(['c', 'b']); + db.clearNowOverride(); + }); + }); + + // ─── GitHub seen set ────────────────────────────────────────────────── + + describe('GitHub seen tracking', () => { + it('markGitHubItemSeen is idempotent (UNIQUE on (sub, key))', () => { + db.markGitHubItemSeen('sub-1', 'pr-42'); + db.markGitHubItemSeen('sub-1', 'pr-42'); + expect(db.isGitHubItemSeen('sub-1', 'pr-42')).toBe(true); + // State assertion: only one row present. + expect(db.state.githubSeen.size).toBe(1); + }); + + it('differentiates items across subscriptions', () => { + db.markGitHubItemSeen('sub-1', 'pr-42'); + db.markGitHubItemSeen('sub-2', 'pr-42'); + expect(db.isGitHubItemSeen('sub-1', 'pr-42')).toBe(true); + expect(db.isGitHubItemSeen('sub-2', 'pr-42')).toBe(true); + expect(db.state.githubSeen.size).toBe(2); + }); + + it('hasAnyGitHubSeen returns true when at least one row exists for the subscription', () => { + expect(db.hasAnyGitHubSeen('sub-1')).toBe(false); + db.markGitHubItemSeen('sub-1', 'pr-1'); + expect(db.hasAnyGitHubSeen('sub-1')).toBe(true); + expect(db.hasAnyGitHubSeen('sub-2')).toBe(false); + }); + + it('clearGitHubSeenForSubscription removes only matching rows', () => { + db.markGitHubItemSeen('sub-1', 'pr-1'); + db.markGitHubItemSeen('sub-1', 'pr-2'); + db.markGitHubItemSeen('sub-2', 'pr-3'); + db.clearGitHubSeenForSubscription('sub-1'); + expect(db.isGitHubItemSeen('sub-1', 'pr-1')).toBe(false); + expect(db.isGitHubItemSeen('sub-1', 'pr-2')).toBe(false); + expect(db.isGitHubItemSeen('sub-2', 'pr-3')).toBe(true); + }); + + it('pruneGitHubSeen deletes rows older than cutoff', () => { + db.setNowOverride(1000); + db.markGitHubItemSeen('sub', 'old'); + db.setNowOverride(5000); + db.markGitHubItemSeen('sub', 'new'); + db.setNowOverride(10000); + db.pruneGitHubSeen(4000); // cutoff = 6000 + expect(db.isGitHubItemSeen('sub', 'old')).toBe(false); + expect(db.isGitHubItemSeen('sub', 'new')).toBe(false); + // Both 'old' (seen at 1000 < 6000) and 'new' (seen at 5000 < 6000) are + // pruned because the cutoff is 6000. Document that. + // Add a third after cutoff to verify the surviving path. + db.setNowOverride(9000); + db.markGitHubItemSeen('sub', 'surviving'); + db.setNowOverride(10000); + db.pruneGitHubSeen(2000); // cutoff = 8000 → 'surviving' (seen 9000) stays + expect(db.isGitHubItemSeen('sub', 'surviving')).toBe(true); + db.clearNowOverride(); + }); + }); + + // ─── Close / reinit persistence (in-memory simulation) ──────────────── + + describe('restart simulation', () => { + it('a fresh InMemoryCueDb starts empty — documents the test-helper contract', () => { + // This makes explicit what cue-engine-integration.test.ts relies on: + // `simulateRestart()` should use `resetAll()` to get a clean DB but + // if the caller creates a NEW instance, it is also empty. If you + // want persistence across a simulated restart, hang on to the SAME + // InMemoryCueDb instance across close → init. + db.recordCueEvent({ + id: 'e1', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }); + expect(db.getRecentCueEvents(0)).toHaveLength(1); + + // Simulate an app restart with the SAME instance: close then init. + // Data MUST survive (we never clear state on close — only `ready`). + db.closeCueDb(); + db.initCueDb(); + expect(db.getRecentCueEvents(0)).toHaveLength(1); + }); + + it('resetAll wipes all state AND marks the DB not-ready', () => { + db.recordCueEvent({ + id: 'e1', + type: 't', + triggerName: 't', + sessionId: 's', + subscriptionName: 'sub', + status: 'running', + }); + db.markGitHubItemSeen('sub', 'x'); + db.updateHeartbeat(); + db.resetAll(); + expect(db.isCueDbReady()).toBe(false); + db.initCueDb(); + expect(db.getRecentCueEvents(0)).toHaveLength(0); + expect(db.state.githubSeen.size).toBe(0); + expect(db.getLastHeartbeat()).toBeNull(); + }); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// Optional smoke test against real better-sqlite3 when the binary is available. +// Drift-catcher: if the mirror diverges from native behavior on a core +// round-trip, the smoke block surfaces that when run locally. CI usually skips. +// ──────────────────────────────────────────────────────────────────────────── + +describe.skipIf(!canLoadBetterSqlite3())('Phase 15B — real SQLite smoke test', () => { + it('real cue-db persists and retrieves one event through a full round-trip', async () => { + // Isolate this test from the rest of the file's mocks. We cannot use + // the top-level `vi.mock('better-sqlite3', ...)` that other cue-db + // tests install (it would short-circuit this smoke). Pull in cue-db + // via dynamic import after confirming the binary loads. + const dbPath = path.join( + os.tmpdir(), + `maestro-cue-smoke-${Date.now()}-${Math.random().toString(36).slice(2)}.db` + ); + // Capture the cue-db module lazily so the finally block can close the + // SQLite handle even if an assertion above throws. Leaving the handle + // open before `fs.unlinkSync` would fail on Windows (file locked) and + // leak the connection on POSIX. + let cueDb: typeof import('../../../main/cue/cue-db') | null = null; + try { + cueDb = await import('../../../main/cue/cue-db'); + cueDb.initCueDb(undefined, dbPath); + cueDb.recordCueEvent({ + id: 'smoke-1', + type: 'time.heartbeat', + triggerName: 't', + sessionId: 'session-1', + subscriptionName: 'sub', + status: 'running', + }); + const events = cueDb.getRecentCueEvents(0); + expect(events).toHaveLength(1); + expect(events[0].id).toBe('smoke-1'); + } finally { + if (cueDb) { + try { + cueDb.closeCueDb(); + } catch { + /* best effort — double-close is safe, other errors are non-fatal here */ + } + } + try { + fs.unlinkSync(dbPath); + } catch { + /* best effort */ + } + } + }); +}); diff --git a/src/__tests__/main/cue/cue-engine-integration.test.ts b/src/__tests__/main/cue/cue-engine-integration.test.ts new file mode 100644 index 0000000000..09fa0e1443 --- /dev/null +++ b/src/__tests__/main/cue/cue-engine-integration.test.ts @@ -0,0 +1,508 @@ +/** + * Phase 15B — Cue engine end-to-end integration tests. + * + * Drives the real `CueEngine` with its real backing services (session + * registry, fan-in tracker, run manager, heartbeat, dispatch, completion, + * cleanup) and the in-memory Cue DB from `cue-integration-test-helpers.ts`. + * Only the boundary callbacks are mocked: + * - `onCueRun` — the executor is not invoked; we assert the engine + * reached the dispatch point with the right payload + * - `loadCueConfig` — we inject configs directly instead of reading disk + * - file watcher, GitHub poller, task scanner — provide a cleanup fn only + * + * This file complements the narrower unit tests in `cue-engine.test.ts` by + * exercising interleavings that span multiple services: heartbeat → + * runManager → fan-in tracker → completion → chain propagation. The goal is + * to catch wiring regressions where a refactor to one service's contract + * silently breaks a neighbor. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { CueConfig } from '../../../main/cue/cue-types'; +import { + createInMemoryCueDb, + buildCueDbModuleMock, + type InMemoryCueDb, +} from './cue-integration-test-helpers'; + +// ─── Module mocks ──────────────────────────────────────────────────────────── +// cue-db: delegates every call to a shared in-memory instance. The indirection +// through `getSharedDb` is required because vi.mock factories hoist above +// imports; we can't assign the instance at top level. + +let sharedDb: InMemoryCueDb | null = null; +function getSharedDb(): InMemoryCueDb { + if (!sharedDb) sharedDb = createInMemoryCueDb(); + return sharedDb; +} + +vi.mock('../../../main/cue/cue-db', () => buildCueDbModuleMock(() => getSharedDb())); + +// cue-yaml-loader: per-project config injection. +type DetailedResult = + | { ok: true; config: CueConfig; warnings: string[] } + | { ok: false; reason: 'missing' } + | { ok: false; reason: 'parse-error'; message: string } + | { ok: false; reason: 'invalid'; errors: string[] }; + +const configsByProject = new Map(); +const mockLoadCueConfig = vi.fn<(projectRoot: string) => CueConfig | null>((root) => { + return configsByProject.get(root) ?? null; +}); +const mockLoadCueConfigDetailed = vi.fn<(projectRoot: string) => DetailedResult>((root) => { + const cfg = configsByProject.get(root); + return cfg ? { ok: true, config: cfg, warnings: [] } : { ok: false, reason: 'missing' }; +}); +const mockWatchCueYaml = vi.fn<(projectRoot: string, onChange: () => void) => () => void>(); +vi.mock('../../../main/cue/cue-yaml-loader', () => ({ + loadCueConfig: (root: string) => mockLoadCueConfig(root), + loadCueConfigDetailed: (root: string) => mockLoadCueConfigDetailed(root), + watchCueYaml: (root: string, onChange: () => void) => mockWatchCueYaml(root, onChange), + findAncestorCueConfigRoot: () => null, +})); + +// Trigger sources whose real implementations would need real IO — keep their +// constructors as cleanup-fn-returning stubs. +vi.mock('../../../main/cue/cue-file-watcher', () => ({ + createCueFileWatcher: vi.fn(() => () => {}), +})); +vi.mock('../../../main/cue/cue-github-poller', () => ({ + createCueGitHubPoller: vi.fn(() => () => {}), +})); +vi.mock('../../../main/cue/cue-task-scanner', () => ({ + createCueTaskScanner: vi.fn(() => () => {}), +})); + +vi.mock('crypto', () => ({ + randomUUID: vi.fn(() => `uuid-${Math.random().toString(36).slice(2, 10)}`), +})); + +// ─── Imports (AFTER mocks hoist) ───────────────────────────────────────────── + +import { CueEngine } from '../../../main/cue/cue-engine'; +import { createMockSession, createMockConfig, createMockDeps } from './cue-test-helpers'; + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +function resetSharedState() { + sharedDb?.resetAll(); + sharedDb = null; + configsByProject.clear(); + mockLoadCueConfig.mockClear(); + mockLoadCueConfigDetailed.mockClear(); + mockWatchCueYaml.mockClear(); +} + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +describe('Phase 15B — CueEngine integration', () => { + let yamlWatcherCleanup: ReturnType; + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + resetSharedState(); + yamlWatcherCleanup = vi.fn(); + mockWatchCueYaml.mockReturnValue(yamlWatcherCleanup); + }); + + afterEach(() => { + vi.useRealTimers(); + resetSharedState(); + }); + + // ─── Heartbeat end-to-end ────────────────────────────────────────────── + + describe('heartbeat → onCueRun → DB round-trip', () => { + it('fires onCueRun immediately and records the event in the DB', async () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'hb', + event: 'time.heartbeat', + enabled: true, + prompt: 'tick', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + // Initial firing on setup. + expect(deps.onCueRun).toHaveBeenCalledTimes(1); + expect(deps.onCueRun).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: 'session-1', + prompt: 'tick', + event: expect.objectContaining({ type: 'time.heartbeat', triggerName: 'hb' }), + }) + ); + + // Let the run's Promise resolve so the run-manager finalizes the DB row. + await vi.advanceTimersByTimeAsync(0); + + // The in-memory DB recorded the run as running + then finalized to completed. + const events = getSharedDb().getRecentCueEvents(0); + expect(events.length).toBeGreaterThanOrEqual(1); + expect(events[0].subscriptionName).toBe('hb'); + expect(events[0].status).toBe('completed'); + + engine.stop(); + }); + + it('fires on each interval tick', async () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'hb', + event: 'time.heartbeat', + enabled: true, + prompt: 'tick', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + await vi.advanceTimersByTimeAsync(0); // initial fire drains + vi.clearAllMocks(); + + await vi.advanceTimersByTimeAsync(5 * 60 * 1000); + expect(deps.onCueRun).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(5 * 60 * 1000); + expect(deps.onCueRun).toHaveBeenCalledTimes(2); + + engine.stop(); + }); + }); + + // ─── Agent-completion chain propagation ──────────────────────────────── + + describe('agent.completed chain propagation', () => { + it('notifyAgentCompleted fires a downstream subscription', async () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'chain', + event: 'agent.completed', + enabled: true, + prompt: 'react to completion', + source_session: 'session-1', + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + // Simulate an external agent completion event for session-1. + engine.notifyAgentCompleted('session-1', { + sessionName: 'Test Session', + status: 'completed', + exitCode: 0, + durationMs: 2000, + stdout: 'CHAIN_SOURCE_OUTPUT', + triggeredBy: 'manual', + chainDepth: 0, + }); + + await vi.advanceTimersByTimeAsync(0); + + // The downstream chain subscription fired. + expect(deps.onCueRun).toHaveBeenCalledWith( + expect.objectContaining({ + subscriptionName: 'chain', + event: expect.objectContaining({ type: 'agent.completed' }), + }) + ); + + engine.stop(); + }); + + it('a completed run propagates through the chain back into the engine', async () => { + // Two subs: + // - "seed" fires on heartbeat, prompt "S" + // - "chain" fires on agent.completed, sources = session-1 + // When seed completes via the mocked onCueRun, the run-manager calls + // onRunCompleted → notifyAgentCompleted → completion service → chain + // subscription dispatches a second onCueRun call. + const config = createMockConfig({ + subscriptions: [ + { + name: 'seed', + event: 'time.heartbeat', + enabled: true, + prompt: 'S', + interval_minutes: 10, + }, + { + name: 'chain', + event: 'agent.completed', + enabled: true, + prompt: 'chain-prompt', + source_session: 'session-1', + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + // seed fires immediately → onRunCompleted → chain fires. + await vi.advanceTimersByTimeAsync(0); + + const calls = (deps.onCueRun as ReturnType).mock.calls.map( + (c) => c[0].subscriptionName + ); + expect(calls).toContain('seed'); + expect(calls).toContain('chain'); + + engine.stop(); + }); + }); + + // ─── Hot-reload ──────────────────────────────────────────────────────── + + describe('hot-reload', () => { + it('refreshSession replaces subscriptions when the config changes', async () => { + const originalConfig = createMockConfig({ + subscriptions: [ + { + name: 'original', + event: 'time.heartbeat', + enabled: true, + prompt: 'original', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', originalConfig); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + // Initial fire from the original sub. + await vi.advanceTimersByTimeAsync(0); + expect( + (deps.onCueRun as ReturnType).mock.calls.some( + (c) => c[0].subscriptionName === 'original' + ) + ).toBe(true); + + vi.clearAllMocks(); + + // Swap the config out: replace "original" with "replacement". + const replacementConfig = createMockConfig({ + subscriptions: [ + { + name: 'replacement', + event: 'time.heartbeat', + enabled: true, + prompt: 'replacement', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', replacementConfig); + engine.refreshSession('session-1', '/projects/test'); + + await vi.advanceTimersByTimeAsync(0); + + const postRefreshCalls = (deps.onCueRun as ReturnType).mock.calls.map( + (c) => c[0].subscriptionName + ); + expect(postRefreshCalls).toContain('replacement'); + expect(postRefreshCalls).not.toContain('original'); + + // Advancing the interval must not re-fire "original" — the watcher + // was torn down on refresh. + vi.clearAllMocks(); + await vi.advanceTimersByTimeAsync(5 * 60 * 1000); + const tickCalls = (deps.onCueRun as ReturnType).mock.calls.map( + (c) => c[0].subscriptionName + ); + expect(tickCalls).toContain('replacement'); + expect(tickCalls).not.toContain('original'); + + engine.stop(); + }); + + it('removeSession tears down all subscriptions for that session', async () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'hb', + event: 'time.heartbeat', + enabled: true, + prompt: 'tick', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + await vi.advanceTimersByTimeAsync(0); + vi.clearAllMocks(); + + engine.removeSession('session-1'); + + // No further firings. + await vi.advanceTimersByTimeAsync(10 * 60 * 1000); + expect(deps.onCueRun).not.toHaveBeenCalled(); + + engine.stop(); + }); + }); + + // ─── Lifecycle integrity ─────────────────────────────────────────────── + + describe('lifecycle integrity', () => { + it('stop + restart reuses the same DB instance and does NOT replay finalized events', async () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'hb', + event: 'time.heartbeat', + enabled: true, + prompt: 'tick', + interval_minutes: 5, + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine1 = new CueEngine(deps); + engine1.start(); + await vi.advanceTimersByTimeAsync(0); + + const eventsAfterFirstRun = getSharedDb().getRecentCueEvents(0); + expect(eventsAfterFirstRun.length).toBeGreaterThan(0); + + engine1.stop(); + + // Fresh engine, same process. DB was not cleared — the event row + // from the first run is still there. + const engine2 = new CueEngine(deps); + engine2.start(); + await vi.advanceTimersByTimeAsync(0); + + const eventsAfterRestart = getSharedDb().getRecentCueEvents(0); + // Original events preserved, plus new ones from the restart tick. + expect(eventsAfterRestart.length).toBeGreaterThanOrEqual(eventsAfterFirstRun.length); + + // Cross-check by status: none of the original `completed` rows + // regressed to `running`. A naive reinit that re-recorded events + // by id would have overwritten the completed status. + for (const priorEvent of eventsAfterFirstRun) { + const current = eventsAfterRestart.find((e) => e.id === priorEvent.id); + expect(current?.status).toBe(priorEvent.status); + } + + engine2.stop(); + }); + + it('getStatus reflects registered subscriptions', () => { + const config = createMockConfig({ + subscriptions: [ + { + name: 'one', + event: 'time.heartbeat', + enabled: true, + prompt: 'a', + interval_minutes: 5, + }, + { + name: 'two', + event: 'time.heartbeat', + enabled: true, + prompt: 'b', + interval_minutes: 10, + }, + ], + }); + configsByProject.set('/projects/test', config); + + const deps = createMockDeps(); + const engine = new CueEngine(deps); + engine.start(); + + const status = engine.getStatus(); + expect(status).toHaveLength(1); + expect(status[0].subscriptionCount).toBe(2); + + engine.stop(); + }); + }); + + // ─── Multi-session isolation ─────────────────────────────────────────── + + describe('multi-session isolation', () => { + it('two sessions with different configs fire independently', async () => { + configsByProject.set( + '/proj-a', + createMockConfig({ + subscriptions: [ + { + name: 'hb-a', + event: 'time.heartbeat', + enabled: true, + prompt: 'A', + interval_minutes: 5, + }, + ], + }) + ); + configsByProject.set( + '/proj-b', + createMockConfig({ + subscriptions: [ + { + name: 'hb-b', + event: 'time.heartbeat', + enabled: true, + prompt: 'B', + interval_minutes: 7, + }, + ], + }) + ); + + const sessions = [ + createMockSession({ id: 's-a', projectRoot: '/proj-a' }), + createMockSession({ id: 's-b', projectRoot: '/proj-b' }), + ]; + const deps = createMockDeps({ getSessions: vi.fn(() => sessions) }); + const engine = new CueEngine(deps); + engine.start(); + + await vi.advanceTimersByTimeAsync(0); + + const names = (deps.onCueRun as ReturnType).mock.calls.map( + (c) => c[0].subscriptionName + ); + expect(names).toContain('hb-a'); + expect(names).toContain('hb-b'); + + engine.stop(); + }); + }); +}); diff --git a/src/__tests__/main/cue/cue-fan-in-edge-cases.test.ts b/src/__tests__/main/cue/cue-fan-in-edge-cases.test.ts new file mode 100644 index 0000000000..1edd463311 --- /dev/null +++ b/src/__tests__/main/cue/cue-fan-in-edge-cases.test.ts @@ -0,0 +1,402 @@ +/** + * Phase 15A — fan-in tracker edge cases. + * + * Complements `cue-fan-in-tracker.test.ts` (which focuses on the inspection + * API added in Phase 8C) by exercising the lifecycle corners the main runtime + * relies on: + * - a source that never completes → timeout fires in 'continue' and 'break' modes + * - a source session removed mid-wait → clearForSession cleans the tracker + * - duplicate completion from the same source → treated idempotently + * - completion after timeout fired → no-op, no double dispatch + * + * Uses `vi.useFakeTimers()` to deterministically advance past the fan-in + * timeout without sleeping in real time. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { + AgentCompletionData, + CueSettings, + CueSubscription, +} from '../../../main/cue/cue-types'; +import { createCueFanInTracker } from '../../../main/cue/cue-fan-in-tracker'; + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +function makeSub(overrides: Partial = {}): CueSubscription { + return { + name: 'fan-in-sub', + event: 'agent.completed', + enabled: true, + prompt: 'merge results', + source_sessions: ['session-a', 'session-b'], + ...overrides, + }; +} + +function makeSettings(overrides: Partial = {}): CueSettings { + return { + // Small default so tests can advance a handful of minutes; individual + // tests override this as needed. + timeout_minutes: 2, + timeout_on_fail: 'continue', + max_concurrent: 1, + queue_size: 10, + ...overrides, + }; +} + +function makeCompletion(overrides: Partial = {}): AgentCompletionData { + return { + sessionName: 'agent-a', + status: 'completed', + exitCode: 0, + durationMs: 1000, + stdout: 'output from agent', + triggeredBy: 'fan-in-sub', + chainDepth: 0, + ...overrides, + }; +} + +const SOURCES = ['session-a', 'session-b']; +const OWNER = 'owner-session'; + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +describe('CueFanInTracker — edge cases', () => { + let dispatch: ReturnType; + let onLog: ReturnType; + + beforeEach(() => { + dispatch = vi.fn(); + onLog = vi.fn(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function makeTracker() { + return createCueFanInTracker({ + onLog, + getSessions: () => [ + { id: 'session-a', name: 'Agent A', toolType: 'claude-code', cwd: '/', projectRoot: '/' }, + { id: 'session-b', name: 'Agent B', toolType: 'claude-code', cwd: '/', projectRoot: '/' }, + ], + dispatchSubscription: dispatch, + }); + } + + // ─── Timeout behavior ────────────────────────────────────────────────── + + describe('source never completes → timeout', () => { + it('fires with partial data in "continue" mode', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 2, timeout_on_fail: 'continue' }); + + // Only session-a completes; session-b never does. + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion({ sessionName: 'Agent A', stdout: 'A-OUTPUT' }) + ); + expect(dispatch).not.toHaveBeenCalled(); + expect(tracker.getActiveTrackerKeys()).toEqual([`${OWNER}:${sub.name}`]); + + // Advance just past the 2-minute timeout. + vi.advanceTimersByTime(2 * 60 * 1000 + 1); + + expect(dispatch).toHaveBeenCalledTimes(1); + const [ownerArg, subArg, eventArg, sourceNameArg] = dispatch.mock.calls[0]; + expect(ownerArg).toBe(OWNER); + expect(subArg).toBe(sub); + // Partial dispatch: only Agent A is in the completedSessions list; + // Agent B shows up in timedOutSessions. + expect(eventArg.payload.completedSessions).toEqual(['session-a']); + expect(eventArg.payload.timedOutSessions).toEqual(['session-b']); + expect(eventArg.payload.partial).toBe(true); + expect(eventArg.payload.sourceOutput).toContain('A-OUTPUT'); + expect(sourceNameArg).toBe('Agent A'); + + // Tracker state is cleaned up — timeout consumed the entry. + expect(tracker.getActiveTrackerKeys()).toEqual([]); + expect(tracker.getTrackerCreatedAt(`${OWNER}:${sub.name}`)).toBeUndefined(); + }); + + it('logs but does not dispatch in "break" mode', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 2, timeout_on_fail: 'break' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + + vi.advanceTimersByTime(2 * 60 * 1000 + 1); + + expect(dispatch).not.toHaveBeenCalled(); + expect(tracker.getActiveTrackerKeys()).toEqual([]); + // Break-mode log surface: mentions timeout + waiting list. + expect( + onLog.mock.calls.some( + (call) => typeof call[1] === 'string' && /timed out \(break mode\)/.test(call[1]) + ) + ).toBe(true); + }); + + it('honors per-subscription fan_in_timeout_minutes over settings.timeout_minutes', () => { + const tracker = makeTracker(); + const sub = makeSub({ fan_in_timeout_minutes: 1 }); + const settings = makeSettings({ timeout_minutes: 60, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + + // settings.timeout_minutes is 60, but the per-sub override is 1 → + // timeout fires just after 1 minute, well before 60. + vi.advanceTimersByTime(60 * 1000 + 1); + expect(dispatch).toHaveBeenCalledTimes(1); + }); + + it('honors per-subscription fan_in_timeout_on_fail override', () => { + const tracker = makeTracker(); + const sub = makeSub({ fan_in_timeout_on_fail: 'break' }); + // Settings says "continue" but the sub pins "break" → must not + // dispatch on timeout. + const settings = makeSettings({ timeout_minutes: 1, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + + vi.advanceTimersByTime(60 * 1000 + 1); + expect(dispatch).not.toHaveBeenCalled(); + }); + }); + + // ─── Mid-wait cleanup ────────────────────────────────────────────────── + + describe('source session removed during fan-in wait', () => { + it('clearForSession cleans up tracker without firing timeout', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 5, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + expect(tracker.getActiveTrackerKeys()).toHaveLength(1); + + // Owner session goes away (user closed the agent); clear its fan-in state. + tracker.clearForSession(OWNER); + + // Advance past the timeout — must NOT fire because the timer was cleared. + vi.advanceTimersByTime(5 * 60 * 1000 + 1); + + expect(dispatch).not.toHaveBeenCalled(); + expect(tracker.getActiveTrackerKeys()).toEqual([]); + expect(tracker.getTrackerCreatedAt(`${OWNER}:${sub.name}`)).toBeUndefined(); + }); + + it('clearForSession only clears entries owned by the given session', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 5, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + 'owner-1', + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + tracker.handleCompletion( + 'owner-2', + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + + expect(tracker.getActiveTrackerKeys().sort()).toEqual( + [`owner-1:${sub.name}`, `owner-2:${sub.name}`].sort() + ); + + tracker.clearForSession('owner-1'); + + // owner-2 is still tracked and will time out normally. + expect(tracker.getActiveTrackerKeys()).toEqual([`owner-2:${sub.name}`]); + }); + }); + + // ─── Idempotency / late-arrivals ─────────────────────────────────────── + + describe('duplicate completion from the same source', () => { + it('treats the second completion as a replacement, not a second vote', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 5, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion({ stdout: 'A-FIRST' }) + ); + // Same session completes a second time — must NOT count as a new vote + // toward the fan-in, otherwise a 2-source fan-in would fire prematurely + // after one agent completes twice without the other ever running. + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion({ stdout: 'A-SECOND' }) + ); + + expect(dispatch).not.toHaveBeenCalled(); + expect(tracker.getActiveTrackerKeys()).toEqual([`${OWNER}:${sub.name}`]); + }); + + it('second completion does not extend the timeout window', () => { + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 2, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + // Half the timeout window passes, then a duplicate arrives. + vi.advanceTimersByTime(60 * 1000); + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + // Remaining 60s + 1ms must trigger the original timeout — the + // duplicate did NOT reset the timer. + vi.advanceTimersByTime(60 * 1000 + 1); + expect(dispatch).toHaveBeenCalledTimes(1); + }); + }); + + // ─── Late completion after timeout ───────────────────────────────────── + + describe('completion after timeout already fired', () => { + it('is accepted as a fresh fan-in cycle (starts a new tracker)', () => { + // This documents the current semantics: once a tracker is cleaned up + // by a timeout, a late completion from the same source for the same + // subscription starts a NEW tracker. In-flight coordination logic + // upstream is responsible for treating post-timeout events as a new + // cycle or dropping them — the tracker itself is stateless across + // timeout boundaries. + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 2, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + vi.advanceTimersByTime(2 * 60 * 1000 + 1); // timeout fires + expect(dispatch).toHaveBeenCalledTimes(1); + + // Late completion arrives from session-b — tracker was cleared, so + // this becomes the first completion of a new cycle. + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-b', + 'Agent B', + makeCompletion() + ); + // A new timer is ticking; we have NOT re-dispatched yet. + expect(dispatch).toHaveBeenCalledTimes(1); + expect(tracker.getActiveTrackerKeys()).toEqual([`${OWNER}:${sub.name}`]); + }); + + it('expireTracker prevents further completions from that tracker dispatching', () => { + // expireTracker is the cleanup-service's eviction path. After it runs, + // subsequent completions for the same key start a fresh tracker + // cycle, but the tracker we expired must never dispatch. + const tracker = makeTracker(); + const sub = makeSub(); + const settings = makeSettings({ timeout_minutes: 5, timeout_on_fail: 'continue' }); + + tracker.handleCompletion( + OWNER, + settings, + sub, + SOURCES, + 'session-a', + 'Agent A', + makeCompletion() + ); + + tracker.expireTracker(`${OWNER}:${sub.name}`); + + // Advance past the original timeout — expireTracker should have + // cleared the timer; no dispatch must occur. + vi.advanceTimersByTime(5 * 60 * 1000 + 1); + expect(dispatch).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/__tests__/main/cue/cue-integration-test-helpers.ts b/src/__tests__/main/cue/cue-integration-test-helpers.ts new file mode 100644 index 0000000000..e134da5b6c --- /dev/null +++ b/src/__tests__/main/cue/cue-integration-test-helpers.ts @@ -0,0 +1,454 @@ +/** + * Phase 15B — Integration test helpers. + * + * Exported utilities: + * + * - `createInMemoryCueDb()` — a high-fidelity in-memory implementation of + * the cue-db module contract. `better-sqlite3` is a native module compiled + * against Electron's ABI and does not load under vitest's Node runtime, so + * we cannot exercise real SQL in integration tests. Instead we mirror the + * module's public API (recordCueEvent / updateCueEventStatus / + * getRecentCueEvents / updateHeartbeat / pruneCueEvents / + * markGitHubItemSeen / isGitHubItemSeen / etc.) with plain data structures + * that preserve the SQL semantics cue-engine actually depends on: insert + * ordering, UNIQUE constraints (cue_github_seen), prune-by-age, and + * heartbeat upsert (single-row id=1). Exposes `setNowOverride`, + * `queueWriteFailure`, and `resetAll` for deterministic test control. + * + * - `buildCueDbModuleMock(getDb)` — a factory returning a module-shape object + * that delegates every cue-db function to the supplied InMemoryCueDb. + * Designed for `vi.mock('.../cue-db', () => buildCueDbModuleMock(() => + * sharedDb))` — the lazy `getDb` getter accommodates vi.mock's hoisting, + * where the factory runs before any top-level `let sharedDb` has executed. + * + * - `canLoadBetterSqlite3()` — probe-instantiates a `:memory:` database so a + * `describe.skipIf(!canLoadBetterSqlite3())` block reflects the native + * binary's real availability. A plain `require('better-sqlite3')` returns + * true even when the prebuilt binary is ABI-mismatched (compiled for + * Electron's Node version, not vitest's); probing catches that. + * + * - `createOnCueRunSpy(defaultResponse?)` — a lightweight spy for the + * engine's `onCueRun` boundary callback, capturing a per-call summary + * (runId, sessionId, subscriptionName, prompt, event) so integration tests + * can assert the dispatch payload without re-reading every vi.fn() call + * tuple. + * + * Typical usage (from `cue-engine-integration.test.ts`): + * + * let sharedDb: InMemoryCueDb | null = null; + * function getSharedDb() { + * if (!sharedDb) sharedDb = createInMemoryCueDb(); + * return sharedDb; + * } + * vi.mock('../../../main/cue/cue-db', () => + * buildCueDbModuleMock(() => getSharedDb()) + * ); + */ + +import { vi } from 'vitest'; +import type { CueEventRecord } from '../../../main/cue/cue-db'; + +// ──────────────────────────────────────────────────────────────────────────── +// InMemoryCueDb — high-fidelity contract mirror of cue-db.ts +// ──────────────────────────────────────────────────────────────────────────── + +/** + * Internal event row — mirrors the cue_events table's column set. Stored in + * insertion order via a Map keyed by id, plus an array of ids so we can do + * ORDER BY created_at DESC + LIMIT lookups without re-sorting on every read. + */ +interface InMemoryCueEventRow { + id: string; + type: string; + triggerName: string; + sessionId: string; + subscriptionName: string; + status: string; + createdAt: number; + completedAt: number | null; + payload: string | null; +} + +export interface InMemoryCueDbState { + events: Map; + /** Insertion order of event IDs, for ORDER BY created_at DESC semantics. */ + eventOrder: string[]; + heartbeat: number | null; + githubSeen: Map; // key = `${subscriptionId}\0${itemKey}` + closed: boolean; + ready: boolean; +} + +export interface InMemoryCueDb { + // State accessors — for assertion convenience in tests. + readonly state: InMemoryCueDbState; + // Lifecycle + initCueDb(onLog?: (level: string, msg: string) => void, dbPathOverride?: string): void; + closeCueDb(): void; + isCueDbReady(): boolean; + // Events + recordCueEvent(event: { + id: string; + type: string; + triggerName: string; + sessionId: string; + subscriptionName: string; + status: string; + payload?: string; + }): void; + updateCueEventStatus(id: string, status: string): void; + getRecentCueEvents(since: number, limit?: number): CueEventRecord[]; + safeRecordCueEvent(event: Parameters[0]): void; + safeUpdateCueEventStatus(id: string, status: string): void; + // Heartbeat + updateHeartbeat(): void; + getLastHeartbeat(): number | null; + // Housekeeping + pruneCueEvents(olderThanMs: number): void; + // GitHub seen set + isGitHubItemSeen(subscriptionId: string, itemKey: string): boolean; + markGitHubItemSeen(subscriptionId: string, itemKey: string): void; + hasAnyGitHubSeen(subscriptionId: string): boolean; + pruneGitHubSeen(olderThanMs: number): void; + clearGitHubSeenForSubscription(subscriptionId: string): void; + // Test-only controls + /** Force a specific current time for prune/heartbeat tests. Reset with clearNowOverride(). */ + setNowOverride(ts: number): void; + clearNowOverride(): void; + /** Force the next write to throw — exercises safe-wrapper warn paths. */ + queueWriteFailure(err?: Error): void; + resetAll(): void; +} + +/** + * Create a fresh in-memory Cue DB. Each test should create its own instance + * (or call `resetAll()` in `beforeEach`) to avoid cross-test leakage. + */ +export function createInMemoryCueDb(): InMemoryCueDb { + const state: InMemoryCueDbState = { + events: new Map(), + eventOrder: [], + heartbeat: null, + githubSeen: new Map(), + closed: true, + ready: false, + }; + let nowOverride: number | null = null; + let pendingFailure: Error | null = null; + + function now(): number { + return nowOverride ?? Date.now(); + } + + function githubKey(subscriptionId: string, itemKey: string): string { + return `${subscriptionId}\u0000${itemKey}`; + } + + function requireReady(): void { + if (!state.ready) { + throw new Error('Cue database not initialized — call initCueDb() first'); + } + } + + function consumePendingFailure(): void { + if (pendingFailure) { + const err = pendingFailure; + pendingFailure = null; + throw err; + } + } + + return { + state, + + initCueDb(_onLog, _dbPathOverride) { + // Idempotent — matches the real module's short-circuit on re-init. + if (state.ready) return; + state.ready = true; + state.closed = false; + }, + + closeCueDb() { + state.ready = false; + state.closed = true; + }, + + isCueDbReady() { + return state.ready; + }, + + recordCueEvent(event) { + requireReady(); + consumePendingFailure(); + // `INSERT OR REPLACE` semantics: duplicate id overwrites the row and + // moves it to the end of insertion order (real SQL would re-assign + // the same PK row in place, but ordering is what we test for, and + // upsert-with-move is close enough for assertions). + const existedBefore = state.events.has(event.id); + state.events.set(event.id, { + id: event.id, + type: event.type, + triggerName: event.triggerName, + sessionId: event.sessionId, + subscriptionName: event.subscriptionName, + status: event.status, + createdAt: now(), + completedAt: null, + payload: event.payload ?? null, + }); + if (existedBefore) { + // Move id to end of order array. + const idx = state.eventOrder.indexOf(event.id); + if (idx >= 0) state.eventOrder.splice(idx, 1); + } + state.eventOrder.push(event.id); + }, + + updateCueEventStatus(id, status) { + requireReady(); + consumePendingFailure(); + const row = state.events.get(id); + if (!row) { + // Real SQL is a silent no-op when WHERE id=? matches nothing. + // Preserve that — don't throw. + return; + } + row.status = status; + row.completedAt = now(); + }, + + getRecentCueEvents(since, limit) { + requireReady(); + // ORDER BY created_at DESC, filter by created_at >= since. + const rows = state.eventOrder + .map((id) => state.events.get(id)!) + .filter((row) => row.createdAt >= since) + .sort((a, b) => b.createdAt - a.createdAt); + + const sliced = limit !== undefined ? rows.slice(0, limit) : rows; + return sliced.map((row) => ({ ...row })); + }, + + safeRecordCueEvent(event) { + try { + this.recordCueEvent(event); + } catch { + // Silent-but-logged in the real module; tests just need the + // no-throw behavior. + } + }, + + safeUpdateCueEventStatus(id, status) { + try { + this.updateCueEventStatus(id, status); + } catch { + // Same contract as safeRecordCueEvent — non-throwing. + } + }, + + updateHeartbeat() { + requireReady(); + consumePendingFailure(); + // Upsert on single-row id=1 — just replace the scalar. + state.heartbeat = now(); + }, + + getLastHeartbeat() { + requireReady(); + return state.heartbeat; + }, + + pruneCueEvents(olderThanMs) { + requireReady(); + const cutoff = now() - olderThanMs; + const keepOrder: string[] = []; + for (const id of state.eventOrder) { + const row = state.events.get(id); + if (!row) continue; + if (row.createdAt < cutoff) { + state.events.delete(id); + } else { + keepOrder.push(id); + } + } + state.eventOrder.length = 0; + state.eventOrder.push(...keepOrder); + }, + + isGitHubItemSeen(subscriptionId, itemKey) { + requireReady(); + return state.githubSeen.has(githubKey(subscriptionId, itemKey)); + }, + + markGitHubItemSeen(subscriptionId, itemKey) { + requireReady(); + consumePendingFailure(); + // INSERT OR IGNORE — only inserts if not already present. + const key = githubKey(subscriptionId, itemKey); + if (!state.githubSeen.has(key)) { + state.githubSeen.set(key, now()); + } + }, + + hasAnyGitHubSeen(subscriptionId) { + requireReady(); + const prefix = `${subscriptionId}\u0000`; + for (const key of state.githubSeen.keys()) { + if (key.startsWith(prefix)) return true; + } + return false; + }, + + pruneGitHubSeen(olderThanMs) { + requireReady(); + const cutoff = now() - olderThanMs; + for (const [key, seenAt] of [...state.githubSeen.entries()]) { + if (seenAt < cutoff) state.githubSeen.delete(key); + } + }, + + clearGitHubSeenForSubscription(subscriptionId) { + requireReady(); + const prefix = `${subscriptionId}\u0000`; + for (const key of [...state.githubSeen.keys()]) { + if (key.startsWith(prefix)) state.githubSeen.delete(key); + } + }, + + setNowOverride(ts) { + nowOverride = ts; + }, + + clearNowOverride() { + nowOverride = null; + }, + + queueWriteFailure(err) { + pendingFailure = err ?? new Error('Simulated DB write failure'); + }, + + resetAll() { + state.events.clear(); + state.eventOrder.length = 0; + state.heartbeat = null; + state.githubSeen.clear(); + state.closed = true; + state.ready = false; + nowOverride = null; + pendingFailure = null; + }, + }; +} + +// ──────────────────────────────────────────────────────────────────────────── +// Factory: mocks the `cue-db` module to delegate to an InMemoryCueDb instance. +// Use in test files via: +// +// const sharedDb = createInMemoryCueDb(); +// vi.mock('../../../main/cue/cue-db', () => buildCueDbModuleMock(() => sharedDb)); +// +// The () => sharedDb indirection lets the mock factory tolerate the test file's +// hoisting order — vi.mock factories run before any `import`, so the shared +// instance has to be constructed lazily on first access. +// ──────────────────────────────────────────────────────────────────────────── + +export function buildCueDbModuleMock(getDb: () => InMemoryCueDb) { + return { + initCueDb: (onLog?: (level: string, msg: string) => void, dbPathOverride?: string) => + getDb().initCueDb(onLog, dbPathOverride), + closeCueDb: () => getDb().closeCueDb(), + isCueDbReady: () => getDb().isCueDbReady(), + recordCueEvent: (event: Parameters[0]) => + getDb().recordCueEvent(event), + updateCueEventStatus: (id: string, status: string) => getDb().updateCueEventStatus(id, status), + getRecentCueEvents: (since: number, limit?: number) => getDb().getRecentCueEvents(since, limit), + safeRecordCueEvent: (event: Parameters[0]) => + getDb().safeRecordCueEvent(event), + safeUpdateCueEventStatus: (id: string, status: string) => + getDb().safeUpdateCueEventStatus(id, status), + updateHeartbeat: () => getDb().updateHeartbeat(), + getLastHeartbeat: () => getDb().getLastHeartbeat(), + pruneCueEvents: (olderThanMs: number) => getDb().pruneCueEvents(olderThanMs), + isGitHubItemSeen: (subscriptionId: string, itemKey: string) => + getDb().isGitHubItemSeen(subscriptionId, itemKey), + markGitHubItemSeen: (subscriptionId: string, itemKey: string) => + getDb().markGitHubItemSeen(subscriptionId, itemKey), + hasAnyGitHubSeen: (subscriptionId: string) => getDb().hasAnyGitHubSeen(subscriptionId), + pruneGitHubSeen: (olderThanMs: number) => getDb().pruneGitHubSeen(olderThanMs), + clearGitHubSeenForSubscription: (subscriptionId: string) => + getDb().clearGitHubSeenForSubscription(subscriptionId), + }; +} + +// ──────────────────────────────────────────────────────────────────────────── +// canLoadBetterSqlite3 — lets Phase 15B add an optional smoke test against +// the real native module when it happens to be available (local dev on the +// same Node version as Electron). CI without a matching binary just skips. +// ──────────────────────────────────────────────────────────────────────────── + +export function canLoadBetterSqlite3(): boolean { + // We check this lazily because `import` would fail loudly at module-eval + // time if the binary is missing, which defeats the purpose of the + // conditional. Also: `require('better-sqlite3')` alone is NOT enough — + // the package resolves fine but `new Database()` may still throw + // NODE_MODULE_VERSION mismatch when the binary was compiled against + // Electron's ABI and we're running under plain Node. Instantiate against + // an in-memory DB to catch the real failure mode. + try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + const Database = require('better-sqlite3'); + const probe = new Database(':memory:'); + probe.close(); + return true; + } catch { + return false; + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// Minimal wired-engine helpers — shared by cue-engine-integration.test.ts. +// ──────────────────────────────────────────────────────────────────────────── + +export interface WiredOnCueRunCall { + runId: string; + sessionId: string; + subscriptionName: string; + prompt: string; + event: unknown; +} + +export function createOnCueRunSpy( + defaultResponse: () => Promise = async () => ({ + runId: 'default', + sessionId: 'session-1', + sessionName: 'Session 1', + subscriptionName: 'default', + event: { + id: 'evt', + type: 'time.heartbeat', + triggerName: 'default', + timestamp: '', + payload: {}, + }, + status: 'completed', + stdout: '', + stderr: '', + exitCode: 0, + durationMs: 1, + startedAt: '', + endedAt: '', + }) +) { + const calls: WiredOnCueRunCall[] = []; + const fn = vi.fn(async (req: WiredOnCueRunCall) => { + calls.push({ + runId: req.runId, + sessionId: req.sessionId, + subscriptionName: req.subscriptionName, + prompt: req.prompt, + event: req.event, + }); + return (await defaultResponse()) as ReturnType extends Promise ? U : never; + }); + return { fn, calls }; +} diff --git a/src/__tests__/main/cue/cue-race-conditions.test.ts b/src/__tests__/main/cue/cue-race-conditions.test.ts new file mode 100644 index 0000000000..c66a9261ad --- /dev/null +++ b/src/__tests__/main/cue/cue-race-conditions.test.ts @@ -0,0 +1,419 @@ +/** + * Phase 15A — race-condition regression tests for the Cue run manager. + * + * The run manager is the main source of concurrency coordination in the Cue + * backend: it tracks active runs, queues events at the concurrency limit, + * and owns the cleanup dance between `stopRun`, `reset`, and the in-flight + * `onCueRun` Promise. + * + * Existing `cue-run-manager.test.ts` covers the single-step state machine + * (running → stopping → finished) and the happy-path completion pipeline. + * This file targets the interleavings that only appear when the outside + * world changes state while `onCueRun` is mid-flight: + * + * - `reset()` called while a run is still executing (engine shutdown) + * - `stopRun()` invoked right after `execute()`, before the run's Promise + * resolution has had a chance to flush to `activeRuns` + * - rapid-fire `execute` calls filling the queue beyond capacity, with the + * oldest event dropped and preserving FIFO + * - queue drain interleaving with a new `execute` that arrives mid-drain + * + * Uses `vi.useFakeTimers()` + `vi.advanceTimersByTimeAsync()` to get + * deterministic microtask ordering without real sleeps. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { CueEvent, CueRunResult, CueSettings } from '../../../main/cue/cue-types'; + +vi.mock('../../../main/cue/cue-db', () => ({ + recordCueEvent: vi.fn(), + updateCueEventStatus: vi.fn(), + safeRecordCueEvent: vi.fn(), + safeUpdateCueEventStatus: vi.fn(), +})); + +vi.mock('../../../main/utils/sentry', () => ({ + captureException: vi.fn(), +})); + +vi.mock('../../../main/cue/cue-cli-executor', () => ({ + runMaestroCliSend: vi.fn().mockResolvedValue({ + ok: true, + exitCode: 0, + stdout: '{}', + stderr: '', + resolvedTarget: '', + }), +})); + +let uuidCounter = 0; +vi.mock('crypto', () => ({ + randomUUID: vi.fn(() => `run-${++uuidCounter}`), +})); + +import { safeUpdateCueEventStatus } from '../../../main/cue/cue-db'; +import { createCueRunManager, type CueRunManagerDeps } from '../../../main/cue/cue-run-manager'; + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +function createEvent(overrides: Partial = {}): CueEvent { + return { + id: 'evt-race', + type: 'time.heartbeat', + timestamp: new Date().toISOString(), + triggerName: 'race-test', + payload: {}, + ...overrides, + }; +} + +function makeResult(overrides: Partial = {}): CueRunResult { + return { + runId: 'r', + sessionId: 'session-1', + sessionName: 'Race Session', + subscriptionName: 'race-sub', + event: createEvent(), + status: 'completed', + stdout: '', + stderr: '', + exitCode: 0, + durationMs: 1, + startedAt: new Date().toISOString(), + endedAt: new Date().toISOString(), + ...overrides, + }; +} + +const defaultSettings: CueSettings = { + timeout_minutes: 30, + timeout_on_fail: 'break', + max_concurrent: 1, + queue_size: 3, +}; + +function createDeps(overrides: Partial = {}): CueRunManagerDeps { + return { + getSessions: vi.fn(() => [{ id: 'session-1', name: 'Race Session' }]), + getSessionSettings: vi.fn(() => defaultSettings), + onCueRun: vi.fn(async () => makeResult()), + onStopCueRun: vi.fn(() => true), + onLog: vi.fn(), + onRunCompleted: vi.fn(), + onRunStopped: vi.fn(), + onPreventSleep: vi.fn(), + onAllowSleep: vi.fn(), + ...overrides, + }; +} + +/** + * Helper to construct an `onCueRun` that blocks until the returned `resolve` + * is called — lets tests interleave state changes with an in-flight run. + */ +function deferredOnCueRun() { + let resolve!: (value: CueRunResult) => void; + const promise = new Promise((res) => { + resolve = res; + }); + const fn = vi.fn(() => promise); + return { fn, resolve }; +} + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +describe('CueRunManager — race conditions', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + uuidCounter = 0; + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + // ─── reset() mid-flight ──────────────────────────────────────────────── + + describe('reset during an in-flight run', () => { + it('does NOT invoke onRunCompleted after reset; activity log still finalizes the DB row', async () => { + const deferred = deferredOnCueRun(); + const deps = createDeps({ onCueRun: deferred.fn }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'prompt', createEvent(), 'race-sub'); + // Run is active — reset() clears it out as if the engine is shutting down. + expect(manager.getActiveRunMap().size).toBe(1); + + manager.reset(); + expect(manager.getActiveRunMap().size).toBe(0); + + // onCueRun finally resolves — the finally block must detect that the + // run was removed from activeRuns and skip the onRunCompleted call + // (which would otherwise fire a chain propagation after engine + // shutdown — the regression Phase 7 was designed to prevent). + deferred.resolve(makeResult({ runId: 'r1' })); + await vi.advanceTimersByTimeAsync(0); + + expect(deps.onRunCompleted).not.toHaveBeenCalled(); + expect(deps.onRunStopped).not.toHaveBeenCalled(); + // DB row is still finalized so the activity log never shows a + // phantom forever-"running" row. + expect(safeUpdateCueEventStatus).toHaveBeenCalled(); + }); + + it('clears the event queue so no drain fires after reset', () => { + const deferred = deferredOnCueRun(); + const deps = createDeps({ + onCueRun: deferred.fn, + getSessionSettings: vi.fn(() => ({ ...defaultSettings, max_concurrent: 1 })), + }); + const manager = createCueRunManager(deps); + + // First execute dispatches (slot available); the next two queue up. + manager.execute('session-1', 'prompt-1', createEvent(), 'sub-1'); + manager.execute('session-1', 'prompt-2', createEvent(), 'sub-2'); + manager.execute('session-1', 'prompt-3', createEvent(), 'sub-3'); + expect(manager.getQueueStatus().get('session-1')).toBe(2); + + manager.reset(); + expect(manager.getQueueStatus().size).toBe(0); + // onCueRun was called exactly once (for the immediately-dispatched + // first event). After reset, queue cannot drain. + expect(deps.onCueRun).toHaveBeenCalledTimes(1); + }); + }); + + // ─── stopRun immediately after execute ───────────────────────────────── + + describe('stopRun called mid-flight', () => { + it('fires onRunStopped exactly once and the late onCueRun resolution is discarded', async () => { + const deferred = deferredOnCueRun(); + const deps = createDeps({ onCueRun: deferred.fn }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'prompt', createEvent(), 'race-sub'); + const runId = manager.getActiveRuns()[0].runId; + + // stopRun fires while onCueRun is still pending. + expect(manager.stopRun(runId)).toBe(true); + expect(deps.onRunStopped).toHaveBeenCalledTimes(1); + + // The in-flight run resolves after the fact — the finally block + // must skip onRunCompleted because stopRun already removed the run. + deferred.resolve(makeResult({ runId })); + await vi.advanceTimersByTimeAsync(0); + + expect(deps.onRunCompleted).not.toHaveBeenCalled(); + // onRunStopped is not double-called from the finally path. + expect(deps.onRunStopped).toHaveBeenCalledTimes(1); + }); + + it('frees the concurrency slot so a queued event drains immediately', async () => { + // Two in-flight runs — stop the first, the queued second must start. + let resolveCount = 0; + const pending: Array<(result: CueRunResult) => void> = []; + const deps = createDeps({ + onCueRun: vi.fn( + () => + new Promise((res) => { + resolveCount += 1; + pending.push(res); + }) + ), + }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'prompt-1', createEvent(), 'sub-1'); + manager.execute('session-1', 'prompt-2', createEvent(), 'sub-2'); // queued + expect(manager.getQueueStatus().get('session-1')).toBe(1); + expect(resolveCount).toBe(1); + + const activeRunId = manager.getActiveRuns()[0].runId; + manager.stopRun(activeRunId); + + // Queue drained → second run dispatched to onCueRun. + expect(resolveCount).toBe(2); + expect(manager.getQueueStatus().size).toBe(0); + + // Resolve both runs to prevent open promises leaking into other tests. + pending.forEach((res) => res(makeResult())); + await vi.advanceTimersByTimeAsync(0); + }); + }); + + // ─── Queue behavior under rapid execute() calls ─────────────────────── + + describe('queue saturation + drop policy', () => { + it('drops the oldest queued event when queue exceeds queue_size', () => { + const deferred = deferredOnCueRun(); + const deps = createDeps({ + onCueRun: deferred.fn, + getSessionSettings: vi.fn(() => ({ + ...defaultSettings, + max_concurrent: 1, + queue_size: 2, + })), + }); + const manager = createCueRunManager(deps); + + // First execute dispatches (slot 1); next three queue. With queue_size=2 + // the third queued item displaces the first queued entry (FIFO drop + // of oldest). Queue length stays at 2. + manager.execute('session-1', 'p1', createEvent(), 'sub-1'); + manager.execute('session-1', 'p2', createEvent(), 'sub-2'); // queued + manager.execute('session-1', 'p3', createEvent(), 'sub-3'); // queued + manager.execute('session-1', 'p4', createEvent(), 'sub-4'); // displaces p2 + + expect(manager.getQueueStatus().get('session-1')).toBe(2); + // Log includes the "dropping oldest" notice. + const logCalls = (deps.onLog as ReturnType).mock.calls; + expect(logCalls.some((call) => /dropping oldest/.test(String(call[1])))).toBe(true); + }); + + it('drains the queue in FIFO order when the active run completes', async () => { + const resolvers: Array<(r: CueRunResult) => void> = []; + const onCueRun = vi.fn( + (req: { runId: string; subscriptionName: string }) => + new Promise((res) => { + resolvers.push((result) => res({ ...result, runId: req.runId })); + }) + ); + const deps = createDeps({ + onCueRun, + getSessionSettings: vi.fn(() => ({ + ...defaultSettings, + max_concurrent: 1, + queue_size: 5, + })), + }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'p1', createEvent(), 'sub-1'); + manager.execute('session-1', 'p2', createEvent(), 'sub-2'); + manager.execute('session-1', 'p3', createEvent(), 'sub-3'); + expect(onCueRun).toHaveBeenCalledTimes(1); + expect(onCueRun.mock.calls[0][0].subscriptionName).toBe('sub-1'); + + // Resolve the first run — sub-2 must start before sub-3 (FIFO). + resolvers.shift()!(makeResult({ status: 'completed' })); + await vi.advanceTimersByTimeAsync(0); + + expect(onCueRun).toHaveBeenCalledTimes(2); + expect(onCueRun.mock.calls[1][0].subscriptionName).toBe('sub-2'); + + // Resolve sub-2 → sub-3 starts. + resolvers.shift()!(makeResult({ status: 'completed' })); + await vi.advanceTimersByTimeAsync(0); + + expect(onCueRun).toHaveBeenCalledTimes(3); + expect(onCueRun.mock.calls[2][0].subscriptionName).toBe('sub-3'); + + // Drain remaining resolvers to avoid leaked promises. + resolvers.shift()!(makeResult({ status: 'completed' })); + await vi.advanceTimersByTimeAsync(0); + }); + + it('a new execute() during a drain joins at the tail, not the head', async () => { + const resolvers: Array<(r: CueRunResult) => void> = []; + const onCueRun = vi.fn( + (req: { runId: string; subscriptionName: string }) => + new Promise((res) => { + resolvers.push((result) => res({ ...result, runId: req.runId })); + }) + ); + const deps = createDeps({ + onCueRun, + getSessionSettings: vi.fn(() => ({ + ...defaultSettings, + max_concurrent: 1, + queue_size: 5, + })), + }); + const manager = createCueRunManager(deps); + + // p1 dispatches; p2 queued. + manager.execute('session-1', 'p1', createEvent(), 'sub-1'); + manager.execute('session-1', 'p2', createEvent(), 'sub-2'); + // Resolve p1 to trigger drain. + resolvers.shift()!(makeResult()); + await vi.advanceTimersByTimeAsync(0); + // Now sub-2 is in flight. + expect(onCueRun.mock.calls[1][0].subscriptionName).toBe('sub-2'); + + // p3 arrives while sub-2 is running → must queue (not jump ahead). + manager.execute('session-1', 'p3', createEvent(), 'sub-3'); + expect(manager.getQueueStatus().get('session-1')).toBe(1); + expect(onCueRun).toHaveBeenCalledTimes(2); // sub-3 hasn't started + + // Resolve sub-2 → sub-3 drains. + resolvers.shift()!(makeResult()); + await vi.advanceTimersByTimeAsync(0); + expect(onCueRun).toHaveBeenCalledTimes(3); + expect(onCueRun.mock.calls[2][0].subscriptionName).toBe('sub-3'); + + // Drain last resolver. + resolvers.shift()!(makeResult()); + await vi.advanceTimersByTimeAsync(0); + }); + }); + + // ─── stopAll interleaving ─────────────────────────────────────────────── + + describe('stopAll + concurrent new execute', () => { + it('stopAll clears both the queue and every active run (no drained run escapes)', () => { + // stopAll's contract: after this function returns, zero active runs + // and zero queued events. It achieves this by clearing the queue + // FIRST — otherwise stopRun's slot-release would drain a queued + // event into a fresh active run that escaped the snapshot. + // + // This test pins the invariant so a future refactor that reorders + // clear/stop surfaces as an assertion flip. + const deferred = deferredOnCueRun(); + const deps = createDeps({ + onCueRun: deferred.fn, + getSessionSettings: vi.fn(() => ({ + ...defaultSettings, + max_concurrent: 1, + queue_size: 3, + })), + }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'p1', createEvent(), 'sub-1'); + manager.execute('session-1', 'p2', createEvent(), 'sub-2'); + manager.execute('session-1', 'p3', createEvent(), 'sub-3'); + + expect(manager.getActiveRuns()).toHaveLength(1); + expect(manager.getQueueStatus().get('session-1')).toBe(2); + + manager.stopAll(); + + expect(manager.getActiveRuns()).toHaveLength(0); + expect(manager.getQueueStatus().size).toBe(0); + expect(deps.onRunStopped).toHaveBeenCalledTimes(1); // one active run stopped + // onCueRun was called once (for sub-1's initial dispatch) and must + // NOT have been called a second time for sub-2 — the queue-clear + // prevents the drain during stopRun from re-dispatching. + expect(deps.onCueRun).toHaveBeenCalledTimes(1); + }); + + it('execute after stopAll still works (engine re-enable scenario)', async () => { + const deferred = deferredOnCueRun(); + const deps = createDeps({ onCueRun: deferred.fn }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'p1', createEvent(), 'sub-1'); + manager.stopAll(); + expect(manager.getActiveRuns()).toHaveLength(0); + + // User re-enables Cue; an event fires shortly after. + manager.execute('session-1', 'p2', createEvent(), 'sub-2'); + expect(manager.getActiveRuns()).toHaveLength(1); + + // Clean up. + deferred.resolve(makeResult()); + await vi.advanceTimersByTimeAsync(0); + }); + }); +}); diff --git a/src/__tests__/main/cue/cue-run-manager.test.ts b/src/__tests__/main/cue/cue-run-manager.test.ts index b6ed4c0a89..329e342db6 100644 --- a/src/__tests__/main/cue/cue-run-manager.test.ts +++ b/src/__tests__/main/cue/cue-run-manager.test.ts @@ -1116,4 +1116,141 @@ describe('createCueRunManager', () => { expect(deps.onLog).toHaveBeenCalledWith('cue', expect.stringContaining('Phase 3 skipped')); }); }); + + // ─── Phase 15A additions ──────────────────────────────────────────────── + // Output-prompt second-phase scenarios that live in the run-manager (the + // executor is a single-phase spawner; the chained "main task → output + // prompt" phase is orchestrated here). + + describe('output prompt phase — failure and stop interactions', () => { + it('preserves main-task stdout when the output prompt returns a non-completed status', async () => { + const onCueRun = vi.fn<(req: { subscriptionName: string }) => Promise>(); + // Call 1 = main task (completes with real stdout). Call 2 = output + // prompt phase, returns failed — run-manager must fall back to the + // main task output and log a warning instead of overwriting the + // result.stdout with the empty output-prompt stdout. + onCueRun + .mockResolvedValueOnce(makeResult({ status: 'completed', stdout: 'MAIN_TASK_OUTPUT' })) + .mockResolvedValueOnce( + makeResult({ status: 'failed', stdout: '', stderr: 'output prompt died' }) + ); + + const deps = createDeps({ onCueRun }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'main-prompt', createEvent(), 'test-sub', 'output-prompt-body'); + await vi.advanceTimersByTimeAsync(0); + + expect(onCueRun).toHaveBeenCalledTimes(2); + // The second call is the output-prompt phase. The run-manager + // builds `contextPrompt = outputPrompt + "\n---\nContext from + // completed task:\n" + mainStdout` and passes that as the + // `prompt` field. It also stashes the main-task stdout on the + // event payload under `sourceOutput` for downstream chain + // consumers. Verify both channels carry MAIN_TASK_OUTPUT so a + // refactor that drops either one fails fast. + const outputPromptRequest = onCueRun.mock.calls[1][0] as { + subscriptionName: string; + prompt: string; + event: { payload: { sourceOutput?: string; outputPromptPhase?: boolean } }; + }; + expect(outputPromptRequest.subscriptionName).toBe('test-sub:output'); + expect(outputPromptRequest.prompt).toContain('output-prompt-body'); + expect(outputPromptRequest.prompt).toContain('MAIN_TASK_OUTPUT'); + expect(outputPromptRequest.event.payload.sourceOutput).toBe('MAIN_TASK_OUTPUT'); + expect(outputPromptRequest.event.payload.outputPromptPhase).toBe(true); + // onRunCompleted carries the MAIN task output — not the failed + // output-prompt's empty string. + expect(deps.onRunCompleted).toHaveBeenCalledWith( + 'session-1', + expect.objectContaining({ stdout: 'MAIN_TASK_OUTPUT', status: 'completed' }), + 'test-sub', + undefined + ); + // Warning explaining the fallback was surfaced to the activity log. + expect( + (deps.onLog as ReturnType).mock.calls.some( + (call) => + call[0] === 'cue' && typeof call[1] === 'string' && /output prompt failed/.test(call[1]) + ) + ).toBe(true); + }); + + it('survives an output-prompt onCueRun that rejects (exception path)', async () => { + const onCueRun = vi.fn<(req: { subscriptionName: string }) => Promise>(); + onCueRun + .mockResolvedValueOnce(makeResult({ status: 'completed', stdout: 'MAIN_OK' })) + .mockRejectedValueOnce(new Error('spawn ENOENT')); + + const deps = createDeps({ onCueRun }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'main', createEvent(), 'test-sub', 'out-prompt'); + await vi.advanceTimersByTimeAsync(0); + + // The outer catch treats the rejection as a run failure — main task + // output is discarded because the `await outputResult` line threw + // before the stdout reassignment could happen. + expect(deps.onRunCompleted).toHaveBeenCalledWith( + 'session-1', + expect.objectContaining({ + status: 'failed', + stderr: expect.stringContaining('spawn ENOENT'), + }), + 'test-sub', + undefined + ); + }); + + it('stopRun during output-prompt phase kills BOTH the parent and the output-prompt child process', async () => { + const mainDeferred: { resolve?: (r: CueRunResult) => void } = {}; + const outputDeferred: { resolve?: (r: CueRunResult) => void } = {}; + + const onCueRun = vi.fn((req: { runId: string; subscriptionName: string }) => { + if (req.subscriptionName === 'test-sub') { + return new Promise((res) => { + mainDeferred.resolve = (r) => res({ ...r, runId: req.runId }); + }); + } + return new Promise((res) => { + outputDeferred.resolve = (r) => res({ ...r, runId: req.runId }); + }); + }); + + const onStopCueRun = vi.fn(() => true); + const deps = createDeps({ onCueRun, onStopCueRun }); + const manager = createCueRunManager(deps); + + manager.execute('session-1', 'main', createEvent(), 'test-sub', 'out-prompt'); + // Let the main task complete; run-manager now dispatches the + // output-prompt phase. + mainDeferred.resolve!(makeResult({ status: 'completed', stdout: 'MAIN_OK' })); + await vi.advanceTimersByTimeAsync(0); + expect(onCueRun).toHaveBeenCalledTimes(2); + + // The output-prompt spawn is now in-flight. The active run carries + // the output-prompt child's processRunId so stopRun can signal both. + const parentRunId = manager.getActiveRuns()[0].runId; + const run = manager.getActiveRunMap().get(parentRunId)!; + expect(run.processRunId).toBeDefined(); + expect(run.processRunId).not.toBe(parentRunId); + + // User hits stop. + const stopped = manager.stopRun(parentRunId); + expect(stopped).toBe(true); + expect(onStopCueRun).toHaveBeenCalledWith(parentRunId); + expect(onStopCueRun).toHaveBeenCalledWith(run.processRunId!); + + expect(deps.onRunStopped).toHaveBeenCalledTimes(1); + expect(deps.onRunStopped).toHaveBeenCalledWith( + expect.objectContaining({ status: 'stopped' }) + ); + + // Output-prompt resolves late — the run-manager must skip + // onRunCompleted because stopRun already cleaned up. + outputDeferred.resolve!(makeResult({ status: 'completed', stdout: 'LATE' })); + await vi.advanceTimersByTimeAsync(0); + expect(deps.onRunCompleted).not.toHaveBeenCalled(); + }); + }); }); diff --git a/src/__tests__/main/cue/cue-security.test.ts b/src/__tests__/main/cue/cue-security.test.ts new file mode 100644 index 0000000000..758ab0cab7 --- /dev/null +++ b/src/__tests__/main/cue/cue-security.test.ts @@ -0,0 +1,614 @@ +/** + * Phase 11 — Security hardening tests. + * + * Covers all four security guards added in Phase 11: + * 11A — `validateGlobPattern` rejects path-traversal / absolute / drive + * patterns; the file-watcher runtime guard drops events that resolve + * outside the project root. + * 11B — `sanitizeCustomEnvVars` drops blocklisted and malformed env var + * names before they reach the child process. + * 11C — `readPromptFile` (via `cue-config-normalizer`) refuses to read + * prompt files that resolve outside the project root. Exercised by + * loading a crafted YAML through the normalizer. + * 11D — `initCueDb` chmods the DB file to 0o600 after opening; a failing + * chmod logs a warning but does not fail initialization. + * + * Intentionally split from the feature-level tests (`cue-file-watcher.test.ts`, + * `cue-db.test.ts`, etc.) so a security regression is easy to spot and bisect. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import * as path from 'path'; +import * as os from 'os'; +import * as fs from 'fs'; + +// ──────────────────────────────────────────────────────────────────────────── +// 11A-1 — Glob validator +// ──────────────────────────────────────────────────────────────────────────── + +import { validateSubscription } from '../../../main/cue/config/cue-config-validator'; + +function watchErrors(sub: Record): string[] { + return validateSubscription({ name: 'test', event: 'file.changed', prompt: 'go', ...sub }, 'sub'); +} + +describe('Phase 11A — validateGlobPattern rejects traversal patterns', () => { + it('rejects patterns containing ".." segments', () => { + const errs = watchErrors({ watch: '../**/*.ts' }); + expect(errs.some((e) => /path traversal/i.test(e))).toBe(true); + }); + + it('rejects bare ".." as a segment', () => { + const errs = watchErrors({ watch: '..' }); + expect(errs.some((e) => /path traversal/i.test(e))).toBe(true); + }); + + it('rejects mid-path ".." segments', () => { + const errs = watchErrors({ watch: 'src/../../etc/passwd' }); + expect(errs.some((e) => /path traversal/i.test(e))).toBe(true); + }); + + it('rejects Windows-style backslash traversal', () => { + const errs = watchErrors({ watch: '..\\foo\\bar.md' }); + expect(errs.some((e) => /path traversal/i.test(e))).toBe(true); + }); + + it('rejects absolute POSIX paths', () => { + const errs = watchErrors({ watch: '/etc/passwd' }); + expect(errs.some((e) => /absolute paths are not permitted/i.test(e))).toBe(true); + }); + + it('rejects absolute backslash paths', () => { + const errs = watchErrors({ watch: '\\Windows\\System32' }); + expect(errs.some((e) => /absolute paths are not permitted/i.test(e))).toBe(true); + }); + + it('rejects Windows drive-letter paths', () => { + const errs = watchErrors({ watch: 'C:\\Windows\\System32\\*.exe' }); + expect(errs.some((e) => /Windows drive paths are not permitted/i.test(e))).toBe(true); + }); + + it('rejects Windows drive-letter paths with forward slashes', () => { + const errs = watchErrors({ watch: 'D:/secret/*.txt' }); + expect(errs.some((e) => /Windows drive paths are not permitted/i.test(e))).toBe(true); + }); + + it('rejects drive-relative Windows paths without a separator after the colon', () => { + // `C:secret\*.txt` is resolved against Windows' per-drive CWD table + // and can escape the project root. Catch both drive-absolute + // (`C:\...`) and drive-relative (`C:...`) shapes with one regex. + const errs = watchErrors({ watch: 'C:secret\\*.txt' }); + expect(errs.some((e) => /Windows drive paths are not permitted/i.test(e))).toBe(true); + }); + + it('accepts a normal relative glob', () => { + const errs = watchErrors({ watch: 'src/**/*.ts' }); + // There should be no watch-related error. + expect(errs.filter((e) => /watch/i.test(e))).toHaveLength(0); + }); + + it('accepts recursive globs', () => { + const errs = watchErrors({ watch: '**/*.{js,ts}' }); + expect(errs.filter((e) => /watch/i.test(e))).toHaveLength(0); + }); + + it('accepts nested directory globs', () => { + const errs = watchErrors({ watch: 'docs/**/*.md' }); + expect(errs.filter((e) => /watch/i.test(e))).toHaveLength(0); + }); + + it('also hardens task.pending watch patterns', () => { + const errs = validateSubscription( + { name: 't', event: 'task.pending', prompt: 'go', watch: '../tasks/**/*.md' }, + 'sub' + ); + expect(errs.some((e) => /path traversal/i.test(e))).toBe(true); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// 11A-2 — File-watcher runtime guard +// ──────────────────────────────────────────────────────────────────────────── + +const mockOn = vi.fn().mockReturnThis(); +const mockClose = vi.fn(); +vi.mock('chokidar', () => ({ + watch: vi.fn(() => ({ + on: mockOn, + close: mockClose, + })), +})); + +// Isolate the crypto.randomUUID mock from the file-watcher's own suite. +vi.mock('crypto', () => ({ + randomUUID: vi.fn(() => 'security-test-uuid'), +})); + +import { createCueFileWatcher } from '../../../main/cue/cue-file-watcher'; + +describe('Phase 11A — file-watcher runtime path containment', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function getChangeHandler(): (filePath: string) => void { + const handler = mockOn.mock.calls.find((call) => call[0] === 'change')?.[1]; + expect(handler).toBeDefined(); + return handler as (filePath: string) => void; + } + + it('drops events whose resolved path escapes the project root', () => { + const projectRoot = path.resolve(os.tmpdir(), 'maestro-cue-security-test'); + const onEvent = vi.fn(); + const onLog = vi.fn(); + + createCueFileWatcher({ + watchGlob: '**/*.ts', + projectRoot, + debounceMs: 10, + onEvent, + triggerName: 'sec-test', + onLog, + }); + + const changeHandler = getChangeHandler(); + + // chokidar normally delivers paths relative to `cwd`, but a misconfigured + // symlink or an explicit absolute path that escapes the root must be + // dropped. Simulate the escape by passing a path that resolves above + // `projectRoot`. + const escapingPath = path.join('..', '..', 'etc', 'passwd'); + changeHandler(escapingPath); + + vi.advanceTimersByTime(10); + + expect(onEvent).not.toHaveBeenCalled(); + expect(onLog).toHaveBeenCalledWith( + 'warn', + expect.stringContaining('Dropped file event outside projectRoot') + ); + }); + + it('accepts events whose resolved path is inside the project root', () => { + const projectRoot = path.resolve(os.tmpdir(), 'maestro-cue-security-test-ok'); + const onEvent = vi.fn(); + const onLog = vi.fn(); + + createCueFileWatcher({ + watchGlob: '**/*.ts', + projectRoot, + debounceMs: 10, + onEvent, + triggerName: 'sec-test', + onLog, + }); + + const changeHandler = getChangeHandler(); + changeHandler(path.join('src', 'index.ts')); + + vi.advanceTimersByTime(10); + + expect(onEvent).toHaveBeenCalledTimes(1); + expect(onLog).not.toHaveBeenCalled(); + }); + + it('does not throw when onLog is not provided and an escape occurs', () => { + const projectRoot = path.resolve(os.tmpdir(), 'maestro-cue-security-test-nolog'); + const onEvent = vi.fn(); + + createCueFileWatcher({ + watchGlob: '**/*.ts', + projectRoot, + debounceMs: 10, + onEvent, + triggerName: 'sec-test', + }); + + const changeHandler = getChangeHandler(); + expect(() => { + changeHandler(path.join('..', 'outside.ts')); + vi.advanceTimersByTime(10); + }).not.toThrow(); + expect(onEvent).not.toHaveBeenCalled(); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// 11B — Env var sanitizer +// ──────────────────────────────────────────────────────────────────────────── + +import { sanitizeCustomEnvVars, getBlockedEnvVarNames } from '../../../main/cue/cue-env-sanitizer'; + +describe('Phase 11B — sanitizeCustomEnvVars', () => { + it('returns an empty result for undefined input', () => { + const res = sanitizeCustomEnvVars(undefined); + expect(res.sanitized).toEqual({}); + expect(res.droppedNames).toEqual([]); + }); + + it('returns an empty result for null input', () => { + const res = sanitizeCustomEnvVars(null); + expect(res.sanitized).toEqual({}); + expect(res.droppedNames).toEqual([]); + }); + + it('passes valid POSIX-style env vars through unchanged', () => { + const onLog = vi.fn(); + const res = sanitizeCustomEnvVars( + { + ANTHROPIC_API_KEY: 'sk-test', + _MY_VAR: 'x', + FOO123: 'bar', + A: '1', + }, + onLog + ); + expect(res.sanitized).toEqual({ + ANTHROPIC_API_KEY: 'sk-test', + _MY_VAR: 'x', + FOO123: 'bar', + A: '1', + }); + expect(res.droppedNames).toEqual([]); + expect(onLog).not.toHaveBeenCalled(); + }); + + describe.each([ + 'PATH', + 'HOME', + 'USER', + 'SHELL', + 'LD_PRELOAD', + 'LD_LIBRARY_PATH', + 'DYLD_INSERT_LIBRARIES', + 'NODE_OPTIONS', + ])('drops blocklisted var %s', (name) => { + it('drops the var and logs a warn with "blocklisted"', () => { + const onLog = vi.fn(); + const res = sanitizeCustomEnvVars({ [name]: 'malicious' }, onLog); + expect(res.sanitized).toEqual({}); + expect(res.droppedNames).toEqual([name]); + expect(onLog).toHaveBeenCalledWith('warn', expect.stringContaining('blocklisted')); + }); + }); + + describe.each([ + ['starts with digit', '1FOO'], + ['contains hyphen', 'FOO-BAR'], + ['contains space', 'FOO BAR'], + ['empty string', ''], + ['contains equals sign', 'FOO=BAR'], + ['contains dot', 'FOO.BAR'], + ])('drops invalid var name (%s)', (_label, name) => { + it('drops the var and logs a warn with "not a valid"', () => { + const onLog = vi.fn(); + const res = sanitizeCustomEnvVars({ [name]: 'x' }, onLog); + expect(res.sanitized).toEqual({}); + expect(res.droppedNames).toEqual([name]); + expect(onLog).toHaveBeenCalledWith('warn', expect.stringContaining('not a valid')); + }); + }); + + it('drops blocklisted vars case-insensitively (Windows env vars are case-insensitive)', () => { + // Windows env var lookup is case-insensitive — `Path` and `PATH` + // refer to the same slot, so a case-sensitive blocklist would let an + // attacker bypass the guard with `path` or `PaTh`. Verify both + // lowercase and mixed-case variants get dropped and logged. + const onLog = vi.fn(); + const res = sanitizeCustomEnvVars( + { + path: '/opt/my-bin', + PaTh: '/opt/other', + LD_preload: 'evil.so', + ld_library_path: '/tmp/evil', + }, + onLog + ); + expect(res.sanitized).toEqual({}); + // droppedNames preserves the original casing so operators see what + // the user actually typed. + expect(res.droppedNames).toEqual(['path', 'PaTh', 'LD_preload', 'ld_library_path']); + expect(onLog).toHaveBeenCalledTimes(4); + for (const call of onLog.mock.calls) { + expect(call[0]).toBe('warn'); + expect(call[1]).toMatch(/blocklisted/); + } + }); + + it('preserves the order of dropped names', () => { + const res = sanitizeCustomEnvVars({ + GOOD: 'ok', + PATH: 'bad', + 'BAD NAME': 'nope', + LD_PRELOAD: 'evil', + }); + expect(res.droppedNames).toEqual(['PATH', 'BAD NAME', 'LD_PRELOAD']); + expect(res.sanitized).toEqual({ GOOD: 'ok' }); + }); + + it('mixes valid, blocked, and invalid vars correctly', () => { + const onLog = vi.fn(); + const res = sanitizeCustomEnvVars( + { + ANTHROPIC_API_KEY: 'valid', + NODE_OPTIONS: '--inspect', // blocked + '1BAD': 'invalid', // invalid regex + OTHER_VAR: 'valid', + }, + onLog + ); + expect(res.sanitized).toEqual({ + ANTHROPIC_API_KEY: 'valid', + OTHER_VAR: 'valid', + }); + expect(res.droppedNames.sort()).toEqual(['1BAD', 'NODE_OPTIONS'].sort()); + expect(onLog).toHaveBeenCalledTimes(2); + }); + + it('does not invoke onLog when no vars are dropped', () => { + const onLog = vi.fn(); + sanitizeCustomEnvVars({ OK: '1' }, onLog); + expect(onLog).not.toHaveBeenCalled(); + }); + + it('exposes the canonical blocklist for callers/tests', () => { + const blocked = getBlockedEnvVarNames(); + expect(blocked.has('PATH')).toBe(true); + expect(blocked.has('LD_PRELOAD')).toBe(true); + expect(blocked.has('LD_LIBRARY_PATH')).toBe(true); + expect(blocked.has('DYLD_INSERT_LIBRARIES')).toBe(true); + expect(blocked.has('NODE_OPTIONS')).toBe(true); + expect(blocked.has('HOME')).toBe(true); + expect(blocked.has('USER')).toBe(true); + expect(blocked.has('SHELL')).toBe(true); + // Sanity: the canonical list should not include orthogonal vars. + expect(blocked.has('ANTHROPIC_API_KEY')).toBe(false); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// 11C — readPromptFile path traversal +// ──────────────────────────────────────────────────────────────────────────── + +import * as yaml from 'js-yaml'; +import { parseCueConfigDocument } from '../../../main/cue/config/cue-config-normalizer'; + +describe('Phase 11C — prompt_file path containment', () => { + let tmpDir: string; + let projectRoot: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'maestro-cue-sec-')); + projectRoot = path.join(tmpDir, 'project'); + fs.mkdirSync(path.join(projectRoot, '.maestro', 'prompts'), { recursive: true }); + fs.writeFileSync(path.join(projectRoot, '.maestro', 'prompts', 'ok.md'), 'hello'); + // Sibling file outside the project root — what a traversal would target. + fs.writeFileSync(path.join(tmpDir, 'secret.md'), 'SECRET'); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + function parseWithPromptFile(promptFile: string) { + const raw = yaml.dump({ + subscriptions: [ + { + name: 't', + event: 'time.heartbeat', + interval_minutes: 1, + prompt_file: promptFile, + }, + ], + }); + const doc = parseCueConfigDocument(raw, projectRoot); + expect(doc).not.toBeNull(); + return doc!.subscriptions[0]; + } + + it('resolves a legitimate prompt_file reference to its contents', () => { + const sub = parseWithPromptFile('.maestro/prompts/ok.md'); + expect(sub.prompt).toBe('hello'); + expect(sub.promptSpec.file).toBe('.maestro/prompts/ok.md'); + }); + + it('refuses to read a prompt file outside the project root via relative traversal', () => { + const sub = parseWithPromptFile('../secret.md'); + // readPromptFile returns undefined → resolvedPrompt becomes '' → prompt is ''. + expect(sub.prompt).toBe(''); + // promptSpec still carries the (refused) file reference for downstream + // materializeCueConfig to surface as a warning. + expect(sub.promptSpec.file).toBe('../secret.md'); + }); + + it('refuses to read a prompt file via absolute path outside the project root', () => { + const sub = parseWithPromptFile(path.join(tmpDir, 'secret.md')); + expect(sub.prompt).toBe(''); + }); + + it('allows an absolute path that happens to be inside the project root', () => { + const sub = parseWithPromptFile(path.join(projectRoot, '.maestro', 'prompts', 'ok.md')); + expect(sub.prompt).toBe('hello'); + }); + + it.runIf(process.platform === 'darwin' || process.platform === 'win32')( + 'resolves prompt_file references whose casing differs from projectRoot on case-insensitive filesystems', + () => { + // Only meaningful on case-insensitive FSes (macOS / Windows). A + // case-sensitive `startsWith` would false-negative reject a legit + // path when the projectRoot happens to be cased differently than + // the prompt-file reference — the filesystem treats them as the + // same file, so the containment guard must too. + const raw = yaml.dump({ + subscriptions: [ + { + name: 't', + event: 'time.heartbeat', + interval_minutes: 1, + // Legitimate in-root path, but upper-cased — should + // still resolve to the real file's contents. + prompt_file: path.join(projectRoot, '.maestro', 'prompts', 'OK.MD').toUpperCase(), + }, + ], + }); + const doc = parseCueConfigDocument(raw, projectRoot); + expect(doc).not.toBeNull(); + // Path casing beyond the root prefix may not match a real file + // on disk, so `prompt` can still be '' if the upper-cased + // filename doesn't exist — what we're asserting is the + // CONTAINMENT guard didn't reject it. `readPromptFile` returning + // '' (file not found) vs undefined (containment rejection) means + // the promptSpec.file is still recorded. Easier to test: swap the + // root itself to a case-variant and check the file resolves. + const upperRoot = projectRoot; // keep FS-matching casing + // Intentionally mismatch case on the ROOT passed in, but keep + // the prompt_file reference lowercase-consistent with disk. If + // our guard rejected case-variant roots, this would return ''. + const rawMismatched = yaml.dump({ + subscriptions: [ + { + name: 't', + event: 'time.heartbeat', + interval_minutes: 1, + prompt_file: '.maestro/prompts/ok.md', + }, + ], + }); + const docMismatched = parseCueConfigDocument(rawMismatched, upperRoot.toUpperCase()); + expect(docMismatched?.subscriptions[0].prompt).toBe('hello'); + } + ); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// 11D — DB file permissions +// ──────────────────────────────────────────────────────────────────────────── +// +// Emulate the existing cue-db.test.ts mock strategy: better-sqlite3 is a +// native module that does not load under vitest, so we mock it. The guard we +// care about is the `fs.chmodSync(dbPath, 0o600)` call immediately after +// `new Database(dbPath)`. We verify (a) chmod was called with 0o600, and (b) +// when chmod throws, initialization still completes and a warn log is emitted. + +const runCalls: unknown[][] = []; +const mockStatement = { + run: vi.fn((...args: unknown[]) => { + runCalls.push(args); + return { changes: 1 }; + }), + get: vi.fn(), + all: vi.fn(() => []), +}; +const mockDb = { + pragma: vi.fn(), + prepare: vi.fn(() => mockStatement), + close: vi.fn(), +}; + +vi.mock('better-sqlite3', () => ({ + default: class MockDatabase { + constructor() { + /* noop */ + } + pragma = mockDb.pragma; + prepare = mockDb.prepare; + close = mockDb.close; + }, +})); + +vi.mock('electron', () => ({ + app: { + getPath: vi.fn(() => os.tmpdir()), + }, +})); + +// Import AFTER vi.mock so the mocked better-sqlite3 binding is used. +import { initCueDb, closeCueDb } from '../../../main/cue/cue-db'; + +// On Windows, POSIX modes are largely ignored by NTFS and a chmod-check test +// would produce an ambiguous result. Skip the permission-bit assertion there; +// the error-path test still runs because we exercise the chmod failure via a +// non-existent DB path (which throws ENOENT on every platform). +const isPosix = process.platform !== 'win32'; + +describe('Phase 11D — cue-db file permissions', () => { + const createdFiles: string[] = []; + // initCueDb calls fs.mkdirSync(dirname(dbPath), { recursive: true }) when + // the parent directory does not exist. The chmod-failure test points at a + // non-existent dir so the tmpdir gets a new subdirectory created on every + // run — track those so afterEach cleans them up instead of leaving them + // behind in the user's tmpdir. + const createdDirs: string[] = []; + + beforeEach(() => { + vi.clearAllMocks(); + runCalls.length = 0; + closeCueDb(); + }); + + afterEach(() => { + closeCueDb(); + while (createdFiles.length > 0) { + const file = createdFiles.pop()!; + try { + fs.unlinkSync(file); + } catch { + // best effort + } + } + while (createdDirs.length > 0) { + const dir = createdDirs.pop()!; + try { + fs.rmSync(dir, { recursive: true, force: true }); + } catch { + // best effort + } + } + }); + + it.skipIf(!isPosix)('chmods the DB file to 0o600 immediately after opening', () => { + // better-sqlite3 is mocked (no real DB file created). Pre-create an + // empty file at the path so the real fs.chmodSync call inside + // initCueDb has something to tighten. + const dbPath = path.join(os.tmpdir(), `maestro-cue-chmod-${Date.now()}-${Math.random()}.db`); + fs.writeFileSync(dbPath, ''); + // Start from an intentionally loose mode so the post-init mode proves + // the chmod call actually happened. + fs.chmodSync(dbPath, 0o644); + createdFiles.push(dbPath); + + initCueDb(undefined, dbPath); + + const stat = fs.statSync(dbPath); + // Only the low 9 bits (owner/group/other rwx) are the permission bits. + expect(stat.mode & 0o777).toBe(0o600); + }); + + it('continues initialization and logs a warn when chmod fails', () => { + const onLog = vi.fn(); + // Point at a path whose parent does not exist — `new Database()` is + // mocked so it does not care, but `fs.chmodSync` will throw ENOENT + // because there is no file at the path to chmod. Exactly the error + // shape we want to exercise. + const parentDir = path.join( + os.tmpdir(), + `maestro-cue-chmod-missing-${Date.now()}-${Math.random()}` + ); + const dbPath = path.join(parentDir, 'inner.db'); + // initCueDb will create `parentDir` via mkdirSync(recursive) even + // though the DB file itself is never created (better-sqlite3 is + // mocked). Register it for afterEach cleanup so the tmpdir stays tidy. + createdDirs.push(parentDir); + + expect(() => initCueDb(onLog, dbPath)).not.toThrow(); + + expect(onLog).toHaveBeenCalledWith('warn', expect.stringMatching(/chmod 0o600 failed/)); + // Pragma (WAL) still ran — initialization did not abort when chmod failed. + expect(mockDb.pragma).toHaveBeenCalledWith('journal_mode = WAL'); + }); +}); diff --git a/src/__tests__/main/cue/cue-yaml-loader.test.ts b/src/__tests__/main/cue/cue-yaml-loader.test.ts index 86879273b6..6c2e5cc171 100644 --- a/src/__tests__/main/cue/cue-yaml-loader.test.ts +++ b/src/__tests__/main/cue/cue-yaml-loader.test.ts @@ -24,10 +24,21 @@ vi.mock('chokidar', () => ({ // Mock fs const mockExistsSync = vi.fn(); const mockReadFileSync = vi.fn(); -vi.mock('fs', () => ({ - existsSync: (...args: unknown[]) => mockExistsSync(...args), - readFileSync: (...args: unknown[]) => mockReadFileSync(...args), -})); +// readPromptFile in cue-config-normalizer uses fs.realpathSync.native to harden +// its containment check. These tests use fake paths (`/projects/test/...`) that +// don't exist on disk, so we stub realpath as an identity function — the +// mocked paths have no symlinks, making this the correct canonical path. +const mockRealpathSyncNative = vi.fn((p: string) => p); +vi.mock('fs', () => { + const realpathSync = (p: string) => mockRealpathSyncNative(p); + (realpathSync as unknown as { native: (p: string) => string }).native = (p: string) => + mockRealpathSyncNative(p); + return { + existsSync: (...args: unknown[]) => mockExistsSync(...args), + readFileSync: (...args: unknown[]) => mockReadFileSync(...args), + realpathSync, + }; +}); // Must import after mocks import { diff --git a/src/main/cue/config/cue-config-normalizer.ts b/src/main/cue/config/cue-config-normalizer.ts index 6c8664e888..7e06c5efd1 100644 --- a/src/main/cue/config/cue-config-normalizer.ts +++ b/src/main/cue/config/cue-config-normalizer.ts @@ -31,11 +31,43 @@ export interface CueConfigDocument { } function readPromptFile(projectRoot: string, promptFile: string): string | undefined { - const resolvedPromptPath = path.isAbsolute(promptFile) - ? promptFile - : path.join(projectRoot, promptFile); + // Defense-in-depth path containment: the YAML that specifies `prompt_file` + // is project-owned, but a typo or hand-edit of `../../etc/passwd` should not + // cause an arbitrary host file to be slurped and later substituted into an + // agent prompt. Mirror the write-side guard in `cue-config-repository.ts`. + const normalizedRoot = path.resolve(projectRoot); + const absPath = path.isAbsolute(promptFile) + ? path.resolve(promptFile) + : path.resolve(normalizedRoot, promptFile); + // Canonicalize both paths via realpath before the containment check. This + // asks the OS for the true path and handles, in one shot: case-insensitive + // filesystems (macOS APFS/HFS+, Windows NTFS), Unicode normalization + // differences (NFC vs NFD), and symlinks that could otherwise escape the + // root without tripping a lowercase `startsWith` guard. `path.relative` + // returns '' when the paths are equal (treated as inside — reading the + // root directory as a file will simply fail downstream), a `..`-prefixed + // path for POSIX escapes, and an absolute path on Windows when `realPath` + // lives on a different drive or UNC share (no common base) — so we reject + // any absolute rel too. + let canonicalPath: string; try { - return fs.readFileSync(resolvedPromptPath, 'utf-8'); + const realRoot = fs.realpathSync.native(normalizedRoot); + const realPath = fs.realpathSync.native(absPath); + const rel = path.relative(realRoot, realPath); + if (rel !== '' && (path.isAbsolute(rel) || rel.split(path.sep)[0] === '..')) { + return undefined; + } + canonicalPath = realPath; + } catch { + return undefined; + } + try { + // Read the canonicalized path, not absPath. If `promptFile` was a symlink + // that pointed inside the root at check time, reading `absPath` would + // re-follow the symlink at read time — letting an attacker swap the + // symlink's target between the check and the read. Reading `realPath` + // pins us to the file we actually validated. + return fs.readFileSync(canonicalPath, 'utf-8'); } catch { return undefined; } diff --git a/src/main/cue/config/cue-config-validator.ts b/src/main/cue/config/cue-config-validator.ts index 55cd237cb3..7d7ca53099 100644 --- a/src/main/cue/config/cue-config-validator.ts +++ b/src/main/cue/config/cue-config-validator.ts @@ -8,6 +8,34 @@ import { } from '../../../shared/cue'; function validateGlobPattern(pattern: string, prefix: string, errors: string[]): void { + // Path-traversal guard: the watcher resolves `watchGlob` against `projectRoot` + // via chokidar, so any pattern that escapes the project root (via `..` + // segments, an absolute POSIX path, or a Windows drive letter) would allow + // watching arbitrary files on disk. Reject those shapes up-front — the + // runtime guard in `cue-file-watcher.ts` is the defense-in-depth backstop. + const segments = pattern.split(/[\\/]/); + if (segments.includes('..')) { + errors.push( + `${prefix}: "watch" pattern "${pattern}" is not allowed (contains ".." path traversal)` + ); + return; + } + if (pattern.startsWith('/') || pattern.startsWith('\\')) { + errors.push( + `${prefix}: "watch" pattern "${pattern}" is not allowed (absolute paths are not permitted)` + ); + return; + } + // Match any leading `X:` drive letter — both drive-absolute (`C:\foo`, + // `C:/foo`) and drive-relative (`C:secret\foo`) forms. Drive-relative + // paths resolve against Windows' per-drive current-directory table and + // can escape the project root just as effectively as the absolute forms. + if (/^[A-Za-z]:/.test(pattern)) { + errors.push( + `${prefix}: "watch" pattern "${pattern}" is not allowed (Windows drive paths are not permitted)` + ); + return; + } try { picomatch(pattern); } catch (error) { diff --git a/src/main/cue/cue-db.ts b/src/main/cue/cue-db.ts index 0aa597360e..0e66bb07c7 100644 --- a/src/main/cue/cue-db.ts +++ b/src/main/cue/cue-db.ts @@ -109,6 +109,20 @@ export function initCueDb( } db = new Database(dbPath); + + // Tighten permissions so only the current user can read/write the DB. On + // NTFS/Windows POSIX modes are largely ignored (near no-op); on network + // mounts without POSIX support chmod can throw EPERM/ENOTSUP. Either way + // this is best-effort — log and continue rather than failing DB init. + try { + fs.chmodSync(dbPath, 0o600); + } catch (err) { + log( + 'warn', + `chmod 0o600 failed on ${dbPath}: ${err instanceof Error ? err.message : String(err)}` + ); + } + db.pragma('journal_mode = WAL'); // Create tables diff --git a/src/main/cue/cue-env-sanitizer.ts b/src/main/cue/cue-env-sanitizer.ts new file mode 100644 index 0000000000..9a6762130f --- /dev/null +++ b/src/main/cue/cue-env-sanitizer.ts @@ -0,0 +1,100 @@ +/** + * Cue environment variable sanitizer. + * + * Single responsibility: filter user-supplied `customEnvVars` before they are + * merged into a spawned Cue agent's environment, so a malicious or + * misconfigured YAML cannot inject loader-level variables that would run + * attacker-controlled code in the child process. + * + * Policy: + * - Name regex: /^[a-zA-Z_][a-zA-Z0-9_]*$/ (POSIX env var convention) + * - Blocklist: PATH, HOME, USER, SHELL, LD_PRELOAD, LD_LIBRARY_PATH, + * DYLD_INSERT_LIBRARIES, NODE_OPTIONS + * Comparison is CASE-INSENSITIVE (we uppercase the incoming name before + * membership check). Windows env var lookup is case-insensitive — `Path` + * and `PATH` refer to the same variable — so a case-sensitive blocklist + * would let an attacker bypass the guard by spelling `Path` or `PaTh`. + * The returned `droppedNames` preserves the original casing of the + * input so operators see what was actually rejected. + * + * Dropped entries are silently omitted from the returned map and reported in + * `droppedNames`; callers that supply an `onLog` hook get a warn-level line + * per dropped variable so operators can see the sanitization happen. + */ + +const BLOCKED_ENV_VARS: ReadonlySet = new Set([ + 'PATH', + 'HOME', + 'USER', + 'SHELL', + 'LD_PRELOAD', + 'LD_LIBRARY_PATH', + 'DYLD_INSERT_LIBRARIES', + 'NODE_OPTIONS', +]); + +const VALID_ENV_NAME_REGEX = /^[a-zA-Z_][a-zA-Z0-9_]*$/; + +export interface SanitizeEnvResult { + /** Env vars that passed both the name regex and the blocklist check. */ + sanitized: Record; + /** Names removed from the input (bad regex or blocked). Order preserved. */ + droppedNames: string[]; +} + +/** + * Sanitize a map of custom environment variables for safe spawn. + * + * Returns an always-non-null `sanitized` map and the list of dropped names + * (empty when nothing was filtered). If `onLog` is provided, emits one warn + * line per dropped name so the activity journal reflects the sanitization. + * + * A `null`/`undefined` input is treated as an empty map (no vars to sanitize) + * and returns `{ sanitized: {}, droppedNames: [] }`. + */ +export function sanitizeCustomEnvVars( + vars: Record | undefined | null, + onLog?: (level: string, message: string) => void +): SanitizeEnvResult { + const sanitized: Record = {}; + const droppedNames: string[] = []; + + if (!vars) { + return { sanitized, droppedNames }; + } + + for (const [name, value] of Object.entries(vars)) { + if (!VALID_ENV_NAME_REGEX.test(name)) { + droppedNames.push(name); + if (onLog) { + onLog( + 'warn', + `[CUE] Dropped custom env var "${name}" — name is not a valid POSIX identifier` + ); + } + continue; + } + // Uppercase before membership check so Windows-style casing variants + // (`path`, `PaTh`) cannot bypass the blocklist. droppedNames still + // carries the original casing so the warn log matches what the user + // actually typed. + if (BLOCKED_ENV_VARS.has(name.toUpperCase())) { + droppedNames.push(name); + if (onLog) { + onLog('warn', `[CUE] Dropped custom env var "${name}" — blocklisted for safety`); + } + continue; + } + sanitized[name] = value; + } + + return { sanitized, droppedNames }; +} + +/** + * Exposed for tests and for callers that need to consult the blocklist + * without re-declaring it. + */ +export function getBlockedEnvVarNames(): ReadonlySet { + return BLOCKED_ENV_VARS; +} diff --git a/src/main/cue/cue-file-watcher.ts b/src/main/cue/cue-file-watcher.ts index 0a39623138..5c0d403928 100644 --- a/src/main/cue/cue-file-watcher.ts +++ b/src/main/cue/cue-file-watcher.ts @@ -32,6 +32,13 @@ export function createCueFileWatcher(config: CueFileWatcherConfig): () => void { persistent: true, }); + // Pre-compute the normalized project root (with trailing separator) so the + // per-event guard below can do a cheap prefix check. `path.resolve` does not + // follow symlinks — a link inside projectRoot pointing outside would slip + // through this guard. That is an accepted project-trust limitation, not a + // Cue concern (the validator already rejects `../` patterns up-front). + const normalizedRoot = path.resolve(projectRoot) + path.sep; + const handleEvent = (changeType: 'change' | 'add' | 'unlink') => (filePath: string) => { const existingTimer = debounceTimers.get(filePath); if (existingTimer) { @@ -44,6 +51,22 @@ export function createCueFileWatcher(config: CueFileWatcherConfig): () => void { debounceTimers.delete(filePath); const absolutePath = path.resolve(projectRoot, filePath); + + // Defense-in-depth: even if the validator rejected `../` patterns, + // a misconfigured watch glob combined with chokidar's symlink + // following could produce an event whose resolved path escapes the + // project root. Drop those events with a warn log instead of + // dispatching an arbitrary-file trigger. + if (!absolutePath.startsWith(normalizedRoot)) { + if (config.onLog) { + config.onLog( + 'warn', + `[CUE] Dropped file event outside projectRoot: ${absolutePath} (trigger: ${triggerName})` + ); + } + return; + } + const event = createCueEvent('file.changed', triggerName, { path: absolutePath, filename: path.basename(filePath), diff --git a/src/main/cue/cue-run-manager.ts b/src/main/cue/cue-run-manager.ts index 376ba766a0..b086f45f0a 100644 --- a/src/main/cue/cue-run-manager.ts +++ b/src/main/cue/cue-run-manager.ts @@ -614,10 +614,18 @@ export function createCueRunManager(deps: CueRunManagerDeps): CueRunManager { }, stopAll(): void { + // Clear the queue FIRST, then stop active runs. stopRun calls + // drainQueue internally when it releases a concurrency slot; if + // the queue still has entries at that point, the drain dispatches + // a fresh run that escapes this stopAll invocation. Clearing the + // queue up-front makes every nested drain a no-op, so after this + // function returns there are zero active runs AND zero queued + // events — the contract callers (engine shutdown / Cue toggle + // off) actually need. + eventQueue.clear(); for (const runId of [...activeRuns.keys()]) { this.stopRun(runId); } - eventQueue.clear(); }, getActiveRuns(): CueRunResult[] { diff --git a/src/main/cue/cue-spawn-builder.ts b/src/main/cue/cue-spawn-builder.ts index 1d2b3bb1e2..eab417bff6 100644 --- a/src/main/cue/cue-spawn-builder.ts +++ b/src/main/cue/cue-spawn-builder.ts @@ -11,6 +11,7 @@ import type { CueExecutionConfig } from './cue-executor'; import { getAgentDefinition, getAgentCapabilities } from '../agents'; import { buildAgentArgs, applyAgentConfigOverrides } from '../utils/agent-args'; import { wrapSpawnWithSsh, type SshSpawnWrapConfig } from '../utils/ssh-spawn-wrapper'; +import { sanitizeCustomEnvVars } from './cue-env-sanitizer'; // ─── Types ──────���──────────────────────────────────────────────────────────── @@ -101,13 +102,27 @@ export async function buildSpawnSpec( sessionCustomEnvVars: customEnvVars, }); finalArgs = configResolution.args; - const effectiveEnvVars = configResolution.effectiveCustomEnvVars; + // Sanitize custom env vars BEFORE they reach the spawn environment. This + // drops blocklisted names (PATH, HOME, USER, SHELL, LD_PRELOAD, + // DYLD_INSERT_LIBRARIES, NODE_OPTIONS) and any name that does not match the + // POSIX identifier regex. Keeping this in the spawn-builder means SSH + // wrapping below inherits the sanitized map automatically via + // `sshWrapConfig.customEnvVars`. + const sanitizedResult = sanitizeCustomEnvVars( + configResolution.effectiveCustomEnvVars, + config.onLog + ); + const effectiveEnvVars = sanitizedResult.sanitized; // Determine command let command = customPath || agentDef.command; let spawnArgs = finalArgs; let spawnCwd = projectRoot; - let spawnEnvVars = effectiveEnvVars; + // `sshResult.customEnvVars` (assigned below in the SSH path) is + // `Record | undefined`, so the inferred type for + // `spawnEnvVars` needs to allow undefined. Explicitly type it; the spread + // at the end of the function already handles the undefined case via `|| {}`. + let spawnEnvVars: Record | undefined = effectiveEnvVars; let sshStdinScript: string | undefined; let stdinPrompt: string | undefined; let sshRemoteUsed: SpawnSpec['sshRemoteUsed']; diff --git a/src/renderer/components/CuePipelineEditor/utils/pipelineToYaml.ts b/src/renderer/components/CuePipelineEditor/utils/pipelineToYaml.ts index b222cb0137..c72ff3cf0a 100644 --- a/src/renderer/components/CuePipelineEditor/utils/pipelineToYaml.ts +++ b/src/renderer/components/CuePipelineEditor/utils/pipelineToYaml.ts @@ -405,7 +405,6 @@ export function pipelineToYamlSubscriptions(pipeline: CuePipeline): CueSubscript // but pathological YAML or a future refactor could break that — log // loudly so the failure mode is visible instead of silent. if (targetNodeBySubName.has(name)) { - // eslint-disable-next-line no-console console.warn( `[CUE] Duplicate sub name "${name}" while building source_sub map — earlier owner may not get its source_sub populated` );