diff --git a/apps/electron/src/main/kernelHost.test.ts b/apps/electron/src/main/kernelHost.test.ts index d870b0ca..56ffab6a 100644 --- a/apps/electron/src/main/kernelHost.test.ts +++ b/apps/electron/src/main/kernelHost.test.ts @@ -1,6 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test'; import { join } from 'node:path'; -import type { AgentEvent, AutomationRule } from '@finagent/core'; +import type { AgentEvent, AutomationRule, AutomationRun } from '@finagent/core'; let lastKernelOptions: Record | null = null; let lastMarketData: FakeMarketDataService | null = null; @@ -11,6 +11,33 @@ let forwardedEvents: unknown[] = []; /** Evaluation runs handed to EvaluationStore.addRun (observability path). */ const savedEvalRuns: Array> = []; const routerFetchers = { getQuote: async () => ({ symbol: 'AAPL.US' }) }; +const scheduledRule: AutomationRule = { + id: 'scheduled-rule', + type: 'watchlist-daily-review', + enabled: true, + notify: 'material-only', + createdAt: 1, +}; +let scheduledRules: AutomationRule[] = []; +let recordedAutomationRuns: AutomationRun[] = []; +const scheduledOccurrences = new Map>(); +let automationExecutions = 0; +function automationRun(): AutomationRun { + return { + id: `run-${automationExecutions}`, + ruleId: 'scheduled-rule', + ranAt: Date.now(), + evaluated: 0, + materialChanges: 0, + analyzed: 0, + notified: false, + failures: [], + }; +} +let runAutomationMock: () => Promise = async () => { + automationExecutions += 1; + return automationRun(); +}; /** Poll until `predicate` holds; the settle path is fire-and-forget (`void`). */ async function waitFor(predicate: () => boolean, timeoutMs = 1000): Promise { @@ -255,13 +282,22 @@ mock.module('@finagent/shared', () => ({ strategyPerformance = async () => []; }, AutomationRuleRepository: class { - list = async () => []; + list = async () => scheduledRules; save = async () => undefined; remove = async () => undefined; }, AutomationRunRepository: class { - list = async () => []; - record = async () => undefined; + list = async () => recordedAutomationRuns; + record = async (run: AutomationRun) => { + recordedAutomationRuns = [run, ...recordedAutomationRuns.filter((existing) => existing.id !== run.id)]; + }; + claimScheduledOccurrence = async (ruleId: string, occurrence: string) => { + const occurrences = scheduledOccurrences.get(ruleId) ?? new Set(); + if (occurrences.has(occurrence)) return false; + occurrences.add(occurrence); + scheduledOccurrences.set(ruleId, occurrences); + return true; + }; }, buildBrief: () => ({ generatedAt: 0, @@ -271,18 +307,9 @@ mock.module('@finagent/shared', () => ({ }), runAutomation: async (_rule: unknown, context: unknown) => { lastAutomationContext = context; - return { - id: 'run', - ruleId: 'rule', - ranAt: 0, - evaluated: 0, - materialChanges: 0, - analyzed: 0, - notified: false, - failures: [], - }; + return runAutomationMock(); }, - runDue: () => [], + runDue: () => scheduledRules, DEFAULT_BRIEF_HOUR: 16.5, THESIS_REVIEW_DAY: 0, THESIS_REVIEW_HOUR: 9, @@ -375,6 +402,14 @@ beforeEach(() => { lastAutomationContext = null; lastCompareSymbols = null; forwardedEvents = []; + scheduledRules = []; + recordedAutomationRuns = []; + scheduledOccurrences.clear(); + automationExecutions = 0; + runAutomationMock = async () => { + automationExecutions += 1; + return automationRun(); + }; savedEvalRuns.length = 0; }); @@ -664,6 +699,62 @@ describe('AgentKernelHost', () => { host.dispose(); }); + it('does not repeat a scheduled occurrence after the host is recreated', async () => { + scheduledRules = [scheduledRule]; + const firstHost = new AgentKernelHost(); + + await firstHost['tickAutomations'](); + firstHost.dispose(); + + const recreatedHost = new AgentKernelHost(); + await recreatedHost['tickAutomations'](); + + expect(automationExecutions).toBe(1); + expect(recordedAutomationRuns).toHaveLength(1); + recreatedHost.dispose(); + }); + + it('claims a scheduled occurrence before execution to prevent overlap', async () => { + scheduledRules = [scheduledRule]; + let resolveRun: ((run: AutomationRun) => void) | undefined; + let signalStarted: (() => void) | undefined; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + runAutomationMock = () => { + automationExecutions += 1; + signalStarted?.(); + return new Promise((resolve) => { + resolveRun = resolve; + }); + }; + const host = new AgentKernelHost(); + const firstTick = host['tickAutomations'](); + await started; + + await host['tickAutomations'](); + expect(automationExecutions).toBe(1); + + resolveRun?.(automationRun()); + await firstTick; + host.dispose(); + }); + + it('does not retry a failed scheduled occurrence on the next tick', async () => { + scheduledRules = [scheduledRule]; + runAutomationMock = async () => { + automationExecutions += 1; + throw new Error('execution failed'); + }; + const host = new AgentKernelHost(); + + await host['tickAutomations'](); + await host['tickAutomations'](); + + expect(automationExecutions).toBe(1); + host.dispose(); + }); + it('records a user-cancelled run as cancelled, not failed', async () => { const originalSubscribe = fakeRuns.subscribe; let subscriber: ((event: AgentEvent) => void) | null = null; diff --git a/apps/electron/src/main/kernelHost.ts b/apps/electron/src/main/kernelHost.ts index 13bbb8a9..eeaec188 100644 --- a/apps/electron/src/main/kernelHost.ts +++ b/apps/electron/src/main/kernelHost.ts @@ -290,7 +290,7 @@ export class AgentKernelHost { private readonly automationRules: AutomationRuleRepository; private readonly automationRuns: AutomationRunRepository; private automationTimer: ReturnType | null = null; - private readonly lastAutomationRunByRule = new Map(); + private readonly claimedAutomationOccurrenceByRule = new Map(); private readonly thesisRepository: ThesisRepository; private readonly thesisService: ThesisService; private readonly alertRepository: AlertRuleRepository; @@ -2434,13 +2434,14 @@ export class AgentKernelHost { const now = Date.now(); const todayKey = new Date(now).toDateString(); for (const rule of runDue(rules, now)) { - if (this.lastAutomationRunByRule.get(rule.id) === todayKey) continue; + if (this.claimedAutomationOccurrenceByRule.get(rule.id) === todayKey) continue; + this.claimedAutomationOccurrenceByRule.set(rule.id, todayKey); try { + if (!(await this.automationRuns.claimScheduledOccurrence(rule.id, todayKey))) continue; const run = await this.executeAutomation(rule); await this.automationRuns.record(run); - this.lastAutomationRunByRule.set(rule.id, todayKey); } catch { - // A failing rule never blocks the other rules. + // The occurrence remains claimed; retry on its next scheduled date. } } } catch { diff --git a/packages/shared/src/automation/rules-repository.test.ts b/packages/shared/src/automation/rules-repository.test.ts index 24eff3b1..fe732f68 100644 --- a/packages/shared/src/automation/rules-repository.test.ts +++ b/packages/shared/src/automation/rules-repository.test.ts @@ -1,4 +1,5 @@ import { mkdtemp, rm } from 'node:fs/promises' +import { readFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'bun:test' @@ -42,6 +43,33 @@ function run(id: string, ruleId: string, ranAt: number): AutomationRun { } } +// Freeze a real on-disk snapshot before the first mutation writes it back. +// Synchronous snapshot reads make overlapping operations deterministic: an +// unprotected second mutation sees the stale file before the first is released. +class PausedSnapshotStore extends JsonFileStore { + private paused = false + private signalRead!: () => void + private releaseRead!: () => void + readonly snapshotReady = new Promise((resolve) => { this.signalRead = resolve }) + private readonly released = new Promise((resolve) => { this.releaseRead = resolve }) + + constructor(root: string, private readonly pauseFirstRead = false) { + super(root) + } + + resume(): void { this.releaseRead() } + + async read(file: string, _fallback: T): Promise { + const snapshot = JSON.parse(readFileSync(this.resolve(file), 'utf8')) as T + if (this.pauseFirstRead && !this.paused) { + this.paused = true + this.signalRead() + await this.released + } + return snapshot + } +} + describe('AutomationRuleRepository', () => { it('round-trips rules across repository instances (restart)', async () => { await new AutomationRuleRepository(store).save(rule('r1', { strategyId: 'value' })) @@ -81,6 +109,66 @@ describe('AutomationRuleRepository', () => { }) describe('AutomationRunRepository', () => { + for (const firstOperation of ['record', 'claim'] as const) { + it(`preserves runs and claims across instances when ${firstOperation} reads first`, async () => { + await store.write('automation-runs.json', { runs: [] }) + const pausedStore = new PausedSnapshotStore(dir, true) + const first = new AutomationRunRepository(pausedStore) + const second = new AutomationRunRepository(new PausedSnapshotStore(join(dir, '.'))) + const occurrence = 'Mon Jan 01 2024' + const firstMutation = firstOperation === 'record' + ? first.record(run('run-a', 'r1', 1)) + : first.claimScheduledOccurrence('r2', occurrence) + await pausedStore.snapshotReady + const secondMutation = firstOperation === 'record' + ? second.claimScheduledOccurrence('r2', occurrence) + : second.record(run('run-a', 'r1', 1)) + pausedStore.resume() + await Promise.all([firstMutation, secondMutation]) + + const fresh = new AutomationRunRepository(new JsonFileStore(dir)) + expect(await fresh.list()).toEqual([run('run-a', 'r1', 1)]) + expect(await fresh.claimScheduledOccurrence('r2', occurrence)).toBe(false) + }) + } + + it('allows only one concurrent claim for the same occurrence across instances', async () => { + await store.write('automation-runs.json', { runs: [] }) + const pausedStore = new PausedSnapshotStore(dir, true) + const first = new AutomationRunRepository(pausedStore) + const second = new AutomationRunRepository(new PausedSnapshotStore(dir)) + const firstClaim = first.claimScheduledOccurrence('r1', 'Mon Jan 01 2024') + await pausedStore.snapshotReady + const secondClaim = second.claimScheduledOccurrence('r1', 'Mon Jan 01 2024') + pausedStore.resume() + expect(await Promise.all([firstClaim, secondClaim])).toEqual([true, false]) + const fresh = new AutomationRunRepository(new JsonFileStore(dir)) + expect(await fresh.claimScheduledOccurrence('r1', 'Mon Jan 01 2024')).toBe(false) + }) + + it('does not block a queued record after a claim write fails', async () => { + class FailOnceStore extends JsonFileStore { + private failed = false + async write(file: string, data: unknown): Promise { + if (!this.failed) { + this.failed = true + throw new Error('simulated write failure') + } + await super.write(file, data) + } + } + const repo = new AutomationRunRepository(new FailOnceStore(dir)) + const results = await Promise.allSettled([ + repo.claimScheduledOccurrence('r1', 'Mon Jan 01 2024'), + repo.record(run('run-a', 'r1', 1)), + ]) + expect(results[0].status).toBe('rejected') + expect(results[1].status).toBe('fulfilled') + const fresh = new AutomationRunRepository(new JsonFileStore(dir)) + expect(await fresh.list()).toEqual([run('run-a', 'r1', 1)]) + expect(await fresh.claimScheduledOccurrence('r1', 'Mon Jan 01 2024')).toBe(true) + }) + it('records runs and lists them newest-first across restarts', async () => { const repo = new AutomationRunRepository(store) await repo.record(run('run-old', 'r1', 1_700_000_000_000)) @@ -101,6 +189,18 @@ describe('AutomationRunRepository', () => { expect(await repo.listByRule('r3')).toEqual([]) }) + it('claims scheduled occurrences across repository instances', async () => { + const repo = new AutomationRunRepository(store) + expect(await repo.claimScheduledOccurrence('r1', 'Mon Jan 01 2024')).toBe(true) + await repo.record(run('run-1', 'r1', 1_704_067_200_000)) + expect(await repo.claimScheduledOccurrence('r1', 'Mon Jan 01 2024')).toBe(false) + + const fresh = new AutomationRunRepository(new JsonFileStore(dir)) + expect(await fresh.claimScheduledOccurrence('r1', 'Mon Jan 01 2024')).toBe(false) + expect(await fresh.claimScheduledOccurrence('r1', 'Tue Jan 02 2024')).toBe(true) + expect(await fresh.claimScheduledOccurrence('r2', 'Mon Jan 01 2024')).toBe(true) + }) + it('persists the no-material outcome and captured portfolio scope across restarts', async () => { const noMaterialRun: AutomationRun = { ...run('run-quiet', 'portfolio-rule', 1_700_000_000_000), diff --git a/packages/shared/src/automation/rules-repository.ts b/packages/shared/src/automation/rules-repository.ts index f1a24ccf..eda490df 100644 --- a/packages/shared/src/automation/rules-repository.ts +++ b/packages/shared/src/automation/rules-repository.ts @@ -1,4 +1,5 @@ import type { AutomationRule, AutomationRun } from '@finagent/core' +import { resolve } from 'node:path' import type { JsonFileStore } from '../storage/json-file-store.ts' /** @@ -7,7 +8,7 @@ import type { JsonFileStore } from '../storage/json-file-store.ts' * research-diff / outcome repository pattern: * * automations.json — AutomationRule[] (five fixed rules; no cron UI) - * automation-runs.json — { runs: AutomationRun[] } newest first + * automation-runs.json — completed runs plus claimed scheduled occurrences * * The repository is deliberately thin: seeding the five default rules is the * kernel host's job (first run), the UI only toggles `enabled` / runs a rule, @@ -19,6 +20,8 @@ const RUNS_FILE = 'automation-runs.json' interface RunsFile { runs: AutomationRun[] + /** Scheduled occurrences claimed before execution, grouped by rule id. */ + scheduledOccurrences?: Record } export class AutomationRuleRepository { @@ -73,6 +76,9 @@ export class AutomationRuleRepository { } export class AutomationRunRepository { + // Share the entire read-modify-write queue across instances for each file. + // JsonFileStore's atomic rename cannot prevent stale snapshot overwrites. + private static readonly mutations = new Map>() private readonly store: JsonFileStore constructor(store: JsonFileStore) { @@ -83,12 +89,52 @@ export class AutomationRunRepository { return this.store.read(RUNS_FILE, { runs: [] }) } + private async mutate(operation: () => Promise): Promise { + const path = resolve(this.store.resolve(RUNS_FILE)) + const key = process.platform === 'win32' ? path.toLowerCase() : path + const previous = AutomationRunRepository.mutations.get(key) ?? Promise.resolve() + const result = previous.then(operation) + // Preserve this caller's error without poisoning subsequent mutations. + const settled = result.then(() => undefined, () => undefined) + AutomationRunRepository.mutations.set(key, settled) + try { + return await result + } finally { + if (AutomationRunRepository.mutations.get(key) === settled) { + AutomationRunRepository.mutations.delete(key) + } + } + } + /** Append a run (upsert by id); the list stays sorted newest-first. */ async record(run: AutomationRun): Promise { - const file = await this.read() - const next = [run, ...file.runs.filter((existing) => existing.id !== run.id)] - next.sort((a, b) => b.ranAt - a.ranAt || a.id.localeCompare(b.id)) - await this.store.write(RUNS_FILE, { runs: next }) + await this.mutate(async () => { + const file = await this.read() + const next = [run, ...file.runs.filter((existing) => existing.id !== run.id)] + next.sort((a, b) => b.ranAt - a.ranAt || a.id.localeCompare(b.id)) + await this.store.write(RUNS_FILE, { ...file, runs: next }) + }) + } + + /** + * Persistently claim one scheduled occurrence. A claim is made before the + * rule starts, so another scheduler tick (or a restarted host) cannot run + * the same occurrence again. + */ + async claimScheduledOccurrence(ruleId: string, occurrence: string): Promise { + return this.mutate(async () => { + const file = await this.read() + const occurrences = file.scheduledOccurrences?.[ruleId] ?? [] + if (occurrences.includes(occurrence)) return false + await this.store.write(RUNS_FILE, { + ...file, + scheduledOccurrences: { + ...file.scheduledOccurrences, + [ruleId]: [...occurrences, occurrence], + }, + }) + return true + }) } async list(): Promise {