diff --git a/apps/electron/src/main/kernelHost.test.ts b/apps/electron/src/main/kernelHost.test.ts index 1123e31..9ea38d4 100644 --- a/apps/electron/src/main/kernelHost.test.ts +++ b/apps/electron/src/main/kernelHost.test.ts @@ -1,11 +1,38 @@ import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test'; import { join } from 'node:path'; -import type { AgentEvent } from '@finagent/core'; +import type { AgentEvent, AutomationRule, AutomationRun } from '@finagent/core'; let lastKernelOptions: Record | null = null; let lastMarketData: FakeMarketDataService | null = null; let forwardedEvents: unknown[] = []; 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(); +}; class FakeMarketDataService { quoteSymbols: string[] = []; @@ -224,13 +251,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, @@ -238,17 +274,8 @@ mock.module('@finagent/shared', () => ({ summary: '', quiet: { count: 0, message: '' }, }), - runAutomation: async () => ({ - id: 'run', - ruleId: 'rule', - ranAt: 0, - evaluated: 0, - materialChanges: 0, - analyzed: 0, - notified: false, - failures: [], - }), - runDue: () => [], + runAutomation: () => runAutomationMock(), + runDue: () => scheduledRules, DEFAULT_BRIEF_HOUR: 16.5, THESIS_REVIEW_DAY: 0, THESIS_REVIEW_HOUR: 9, @@ -337,6 +364,14 @@ beforeEach(() => { lastKernelOptions = null; lastMarketData = null; forwardedEvents = []; + scheduledRules = []; + recordedAutomationRuns = []; + scheduledOccurrences.clear(); + automationExecutions = 0; + runAutomationMock = async () => { + automationExecutions += 1; + return automationRun(); + }; }); afterEach(() => { @@ -457,4 +492,60 @@ 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(); + }); }); diff --git a/apps/electron/src/main/kernelHost.ts b/apps/electron/src/main/kernelHost.ts index 2b4bed1..3264c3e 100644 --- a/apps/electron/src/main/kernelHost.ts +++ b/apps/electron/src/main/kernelHost.ts @@ -271,7 +271,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; @@ -2332,13 +2332,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 80543b0..8e7834b 100644 --- a/packages/shared/src/automation/rules-repository.test.ts +++ b/packages/shared/src/automation/rules-repository.test.ts @@ -100,4 +100,16 @@ describe('AutomationRunRepository', () => { expect((await repo.listByRule('r2')).map((r) => r.id)).toEqual(['run-2']) 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) + }) }) diff --git a/packages/shared/src/automation/rules-repository.ts b/packages/shared/src/automation/rules-repository.ts index f1a24cc..1992eb9 100644 --- a/packages/shared/src/automation/rules-repository.ts +++ b/packages/shared/src/automation/rules-repository.ts @@ -7,7 +7,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 +19,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 { @@ -88,7 +90,26 @@ export class AutomationRunRepository { 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.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 { + 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 {