fix: resolve issue #96

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<string, unknown> | 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<Record<string, 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<string, Set<string>>();
+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<AutomationRun> = 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<void> {
@@ -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<string>();
+      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<void>((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;

apps/electron/src/main/kernelHost.ts

@@ -290,7 +290,7 @@ export class AgentKernelHost {
   private readonly automationRules: AutomationRuleRepository;
   private readonly automationRuns: AutomationRunRepository;
   private automationTimer: ReturnType<typeof setInterval> | null = null;
-  private readonly lastAutomationRunByRule = new Map<string, string>();
+  private readonly claimedAutomationOccurrenceByRule = new Map<string, string>();
   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 {

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<void>((resolve) => { this.signalRead = resolve })
+  private readonly released = new Promise<void>((resolve) => { this.releaseRead = resolve })
+
+  constructor(root: string, private readonly pauseFirstRead = false) {
+    super(root)
+  }
+
+  resume(): void { this.releaseRead() }
+
+  async read<T>(file: string, _fallback: T): Promise<T> {
+    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<void> {
+        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),