From 29cf4e7c362819229ed5273827583b83129ab134 Mon Sep 17 00:00:00 2001 From: Jackp0t <16182712+jackkp0t@user.noreply.gitee.com> Date: Sat, 12 Sep 2026 20:57:36 +0800 Subject: [PATCH 1/2] fix: resolve issue #96 --- apps/electron/src/main/kernelHost.test.ts | 121 +++++++++++++++--- apps/electron/src/main/kernelHost.ts | 9 +- .../src/automation/rules-repository.test.ts | 12 ++ .../shared/src/automation/rules-repository.ts | 25 +++- 4 files changed, 146 insertions(+), 21 deletions(-) diff --git a/apps/electron/src/main/kernelHost.test.ts b/apps/electron/src/main/kernelHost.test.ts index 1123e319..9ea38d4c 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 2b4bed19..3264c3e2 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 80543b07..8e7834b1 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 f1a24ccf..1992eb91 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 { From 6e294e8a00adc23ec66806b0afd083db4ca9bf6a Mon Sep 17 00:00:00 2001 From: Jackp0t <16182712+jackkp0t@user.noreply.gitee.com> Date: Wed, 21 Oct 2026 17:47:11 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=B8=B2=E8=A1=8C=E5=8C=96=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=8C=96=20claim=20=E4=B8=8E=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E7=9A=84=E8=AF=BB=E6=94=B9=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/automation/rules-repository.test.ts | 88 +++++++++++++++++++ .../shared/src/automation/rules-repository.ts | 53 ++++++++--- 2 files changed, 127 insertions(+), 14 deletions(-) diff --git a/packages/shared/src/automation/rules-repository.test.ts b/packages/shared/src/automation/rules-repository.test.ts index 322a9ed3..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)) diff --git a/packages/shared/src/automation/rules-repository.ts b/packages/shared/src/automation/rules-repository.ts index 1992eb91..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' /** @@ -75,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) { @@ -85,12 +89,31 @@ 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, { ...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 }) + }) } /** @@ -99,17 +122,19 @@ export class AutomationRunRepository { * 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 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 }) - return true } async list(): Promise {