diff --git a/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.md b/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.md new file mode 100644 index 000000000..5fffef3f9 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.md @@ -0,0 +1,42 @@ +# Coalesce overlapping local history refreshes + +Status: implemented +Translation: current +PR: [#563](https://github.com/LodyAI/Lody/pull/563) + +[中文](2026-09-10-coalesce-local-history-refresh.zh.md) + +## Abstract + +Opening the same imported conversation in overlapping renderer lifecycles could start the same +local history refresh twice, while the CLI rejected the second request as an error. Local project +history operations now share the result of an equivalent in-flight request and serialize distinct +requests for the same project. The coordinator is process-local, so it does not claim coordination +between separate CLI processes. + +## Decision and scope + +The CLI service owns coordination because it is the shared boundary for catalog sync, selected +session import, and conflict resolution. A request identity includes the provider, workspace, +machine, local project, operation, root path, and operation target. Import targets are treated as a +set, so equivalent selections in a different order reuse the same result. + +Distinct operations and targets retain the existing single-writer behavior by waiting for the +project's current tail instead of surfacing a synthetic “already running” error. The original +operation result or failure still reaches every caller that requested it. A settled operation is +removed before callers continue, and a failed operation does not poison the queue, so a later +refresh can run normally. + +This supplements the imported-history guarantees in +[One history writer before windowed readers](../architecture/2026-09-07-single-history-writer.md). +It does not change replay comparison, catalog persistence, conflict resolution policy, or the +cross-process concurrency limit recorded there. + +## Evidence and limits + +Deterministic service tests use deferred promises to show that equivalent imports from separate +service consumers execute once and return the same result, coalesced failures reach every caller, +different targets execute in order, and fresh requests run after both failure and successful +completion. CLI typechecking and the focused service suite validate the process-local contract. The +tests model separate renderer consumers at the shared service boundary; they do not reproduce an +installed renderer's remount timing or exercise filesystem and provider I/O. diff --git a/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.zh.md b/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.zh.md new file mode 100644 index 000000000..22d86e4b7 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-09-10-coalesce-local-history-refresh.zh.md @@ -0,0 +1,36 @@ +# 合并重叠的本地历史刷新 + +Status: implemented +Translation: current +PR: [#563](https://github.com/LodyAI/Lody/pull/563) + +[English](2026-09-10-coalesce-local-history-refresh.md) + +## 摘要 + +在重叠的 renderer 生命周期中打开同一个已导入会话时，可能重复启动相同的本地历史刷新，而 +CLI 会把第二个请求作为错误拒绝。本地项目历史操作现在会让等价的进行中请求共享同一结果， +并把同一项目的不同请求串行执行。协调器只存在于单个进程内，因此不声称能协调不同 CLI 进程。 + +## 决策与范围 + +协调由 CLI service 负责，因为它是目录同步、选定会话导入与冲突解决的共同边界。请求身份由 +provider、workspace、machine、本地项目、操作、根路径和操作目标共同组成。导入目标按集合 +处理，因此顺序不同但内容等价的选择会复用同一结果。 + +不同操作和目标会等待项目当前队尾，以保留既有单写入行为，而不再抛出合成的“already running” +错误。原操作的结果或失败仍会传给所有请求它的调用方。操作结束后会在调用方继续之前 +从协调器中移除；失败也不会污染队列，因此之后的刷新仍可正常执行。 + +本决定补充 +[窗口化读取前保持单一历史写入者](../architecture/2026-09-07-single-history-writer.zh.md) +中的已导入历史保证。它不改变 replay 比较、目录持久化、冲突解决策略或该文档记录的跨进程并发 +限制。 + +## 证据与限制 + +确定性 service 测试使用 deferred promise，证明来自不同 service 消费者的等价导入只执行一次 +并返回同一结果、合并请求的失败会到达每个调用方、不同目标按顺序执行，以及失败或成功完成后 +可以发起新的请求。CLI 类型检查和定向 service 测试验证了进程内契约。这些测试在共享 service +边界模拟不同 renderer 消费者，但没有复现已安装 renderer 的真实 remount 时序，也没有执行文件 +系统或 provider I/O。 diff --git a/apps/cli/src/lib/local-project-history-sync-service.ts b/apps/cli/src/lib/local-project-history-sync-service.ts index 5b7856102..9fc49a980 100644 --- a/apps/cli/src/lib/local-project-history-sync-service.ts +++ b/apps/cli/src/lib/local-project-history-sync-service.ts @@ -46,7 +46,54 @@ import { import { formatErrorMessage } from '@/utils/format-error'; import type { Logger } from '@/utils/logger'; -const syncLeases = new Set(); +type HistorySyncCoordinator = { + tail: Promise; + requests: Map>; +}; + +// History refreshes are initiated from more than one renderer surface. Keep one +// coordinator per local project so equivalent requests can share their result, +// while distinct operations still preserve the service's single-writer order. +const historySyncCoordinators = new Map(); + +function withHistorySyncCoordination( + coordinatorKey: string, + requestKey: string, + operation: () => Promise +): Promise { + let coordinator = historySyncCoordinators.get(coordinatorKey); + if (!coordinator) { + coordinator = { tail: Promise.resolve(), requests: new Map() }; + historySyncCoordinators.set(coordinatorKey, coordinator); + } + + const existing = coordinator.requests.get(requestKey); + if (existing) { + // A request key includes every operation input, so this promise has the + // same result type as the caller that originally registered it. + return existing as Promise; + } + + const current = coordinator.tail.then(operation); + const settled = current.then( + () => undefined, + () => undefined + ); + coordinator.requests.set(requestKey, current); + coordinator.tail = settled; + + const cleanup = () => { + if (coordinator.requests.get(requestKey) === current) { + coordinator.requests.delete(requestKey); + } + if (coordinator.requests.size === 0 && coordinator.tail === settled) { + historySyncCoordinators.delete(coordinatorKey); + } + }; + void current.then(cleanup, cleanup); + + return current; +} // In-process serializer for machineRoomId-scoped catalog writes. History rows // are stored in machine Flock localProject entries, but each provider still does @@ -625,20 +672,13 @@ export class LocalProjectHistorySyncService { localProjectId: LocalProjectId; rootPath: string; }): Promise { - const leaseKey = + const coordinatorKey = `${this.providerKey}:${this.context.workspaceId}:` + `${this.context.machineId}:${args.localProjectId}`; - if (syncLeases.has(leaseKey)) { - throw new Error( - `${getProviderLabel(this.provider)} history sync is already running for this local project` - ); - } - syncLeases.add(leaseKey); - try { - return await this.syncLocalProjectInner(args); - } finally { - syncLeases.delete(leaseKey); - } + const requestKey = stableJson(['sync', args.rootPath]); + return withHistorySyncCoordination(coordinatorKey, requestKey, () => + this.syncLocalProjectInner(args) + ); } private async syncLocalProjectInner(args: { @@ -658,20 +698,17 @@ export class LocalProjectHistorySyncService { rootPath: string; acpSessionIds: string[]; }): Promise { - const leaseKey = + const coordinatorKey = `${this.providerKey}:${this.context.workspaceId}:` + `${this.context.machineId}:${args.localProjectId}`; - if (syncLeases.has(leaseKey)) { - throw new Error( - `${getProviderLabel(this.provider)} history sync is already running for this local project` - ); - } - syncLeases.add(leaseKey); - try { - return await this.importLocalProjectSessionsInner(args); - } finally { - syncLeases.delete(leaseKey); - } + const requestKey = stableJson([ + 'import', + args.rootPath, + [...new Set(args.acpSessionIds)].sort(), + ]); + return withHistorySyncCoordination(coordinatorKey, requestKey, () => + this.importLocalProjectSessionsInner(args) + ); } async resolveHistoryConflict(args: { @@ -680,20 +717,13 @@ export class LocalProjectHistorySyncService { sessionId: SessionId; acpSessionId: string; }): Promise { - const leaseKey = + const coordinatorKey = `${this.providerKey}:${this.context.workspaceId}:` + `${this.context.machineId}:${args.localProjectId}`; - if (syncLeases.has(leaseKey)) { - throw new Error( - `${getProviderLabel(this.provider)} history sync is already running for this local project` - ); - } - syncLeases.add(leaseKey); - try { - return await this.resolveHistoryConflictInner(args); - } finally { - syncLeases.delete(leaseKey); - } + const requestKey = stableJson(['resolve', args.rootPath, args.sessionId, args.acpSessionId]); + return withHistorySyncCoordination(coordinatorKey, requestKey, () => + this.resolveHistoryConflictInner(args) + ); } private async importLocalProjectSessionsInner(args: { diff --git a/apps/cli/tests/local-project-history-sync-service.test.ts b/apps/cli/tests/local-project-history-sync-service.test.ts index f65f16d71..bcbbd04eb 100644 --- a/apps/cli/tests/local-project-history-sync-service.test.ts +++ b/apps/cli/tests/local-project-history-sync-service.test.ts @@ -4,6 +4,8 @@ import type { ACPSessionId, ExternalAcpHistorySyncMeta, LocalProjectHistoryCatalogItem, + LocalProjectHistoryCatalogResult, + LocalProjectHistoryImportResult, LocalProjectId, MachineId, SessionHistoryInput, @@ -25,6 +27,54 @@ const machineId = 'machine-1' as MachineId; const localProjectId = 'project-1' as LocalProjectId; const provider = { cliType: 'builtin', agentType: 'codex' } as const; +function deferred() { + let settle: { resolve: (value: T) => void; reject: (reason: unknown) => void } | undefined; + const promise = new Promise((promiseResolve, promiseReject) => { + settle = { resolve: promiseResolve, reject: promiseReject }; + }); + if (!settle) { + throw new Error('Promise executor did not initialize synchronously'); + } + return { promise, ...settle }; +} + +function catalogResult(marker: number): LocalProjectHistoryCatalogResult { + return { listed: marker, lastListedAt: marker, sessions: [] }; +} + +function importResult(marker: number): LocalProjectHistoryImportResult { + return { + summary: { + listed: marker, + imported: marker, + refreshed: 0, + skipped: 0, + conflicted: 0, + failed: 0, + failures: [], + }, + catalog: catalogResult(marker), + }; +} + +function historySyncService() { + return new LocalProjectHistorySyncService( + {} as never, + { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + } as never, + { + workspaceId: 'workspace-1' as never, + machineId, + userId: 'user-1', + }, + provider + ); +} + function externalHistory(overrides: Partial = {}) { return { provider: { cliType: 'builtin', agentType: 'codex' }, @@ -91,6 +141,142 @@ function materializedReplay( }; } +describe('local project history request coordination', () => { + const importArgs = (acpSessionIds: string[]) => ({ + localProjectId, + rootPath: '/tmp/project-1', + acpSessionIds, + }); + + it('coalesces overlapping imports for the same session set', async () => { + const firstService = historySyncService(); + const secondService = historySyncService(); + const gate = deferred(); + const starts: string[][] = []; + ( + firstService as unknown as { + importLocalProjectSessionsInner: typeof firstService.importLocalProjectSessions; + } + ).importLocalProjectSessionsInner = async (args) => { + starts.push(args.acpSessionIds); + return gate.promise; + }; + ( + secondService as unknown as { + importLocalProjectSessionsInner: typeof secondService.importLocalProjectSessions; + } + ).importLocalProjectSessionsInner = async () => { + throw new Error('equivalent request started twice'); + }; + + const first = firstService.importLocalProjectSessions(importArgs(['acp-2', 'acp-1'])); + const second = secondService.importLocalProjectSessions( + importArgs(['acp-1', 'acp-2', 'acp-1']) + ); + await Promise.resolve(); + + expect(starts).toEqual([['acp-2', 'acp-1']]); + const expected = importResult(2); + gate.resolve(expected); + const [firstResult, secondResult] = await Promise.all([first, second]); + expect(firstResult).toBe(expected); + expect(secondResult).toBe(expected); + }); + + it('propagates a coalesced failure to every consumer', async () => { + const firstService = historySyncService(); + const secondService = historySyncService(); + const gate = deferred(); + ( + firstService as unknown as { + syncLocalProjectInner: typeof firstService.syncLocalProject; + } + ).syncLocalProjectInner = async () => gate.promise; + ( + secondService as unknown as { + syncLocalProjectInner: typeof secondService.syncLocalProject; + } + ).syncLocalProjectInner = async () => { + throw new Error('equivalent request started twice'); + }; + + const args = { localProjectId, rootPath: '/tmp/project-1' }; + const first = firstService.syncLocalProject(args); + const second = secondService.syncLocalProject(args); + const resultsPromise = Promise.allSettled([first, second]); + const expected = new Error('catalog unavailable'); + gate.reject(expected); + + const results = await resultsPromise; + expect(results).toEqual([ + { status: 'rejected', reason: expected }, + { status: 'rejected', reason: expected }, + ]); + }); + + it('serializes different requests for the same local project', async () => { + const service = historySyncService(); + const firstGate = deferred(); + const secondGate = deferred(); + const events: string[] = []; + ( + service as unknown as { + importLocalProjectSessionsInner: typeof service.importLocalProjectSessions; + } + ).importLocalProjectSessionsInner = async (args) => { + const sessionId = args.acpSessionIds[0] ?? 'missing'; + events.push(`start:${sessionId}`); + const result = sessionId === 'acp-1' ? await firstGate.promise : await secondGate.promise; + events.push(`finish:${sessionId}`); + return result; + }; + + const first = service.importLocalProjectSessions(importArgs(['acp-1'])); + const second = service.importLocalProjectSessions(importArgs(['acp-2'])); + await Promise.resolve(); + expect(events).toEqual(['start:acp-1']); + + firstGate.resolve(importResult(1)); + await first; + await Promise.resolve(); + expect(events).toEqual(['start:acp-1', 'finish:acp-1', 'start:acp-2']); + + secondGate.resolve(importResult(2)); + await second; + expect(events).toEqual(['start:acp-1', 'finish:acp-1', 'start:acp-2', 'finish:acp-2']); + }); + + it('allows a fresh request after completion or failure', async () => { + const service = historySyncService(); + const outcomes: Array = [ + new Error('catalog unavailable'), + catalogResult(2), + catalogResult(3), + ]; + ( + service as unknown as { + syncLocalProjectInner: typeof service.syncLocalProject; + } + ).syncLocalProjectInner = async () => { + const outcome = outcomes.shift(); + if (outcome instanceof Error) throw outcome; + if (!outcome) throw new Error('unexpected extra refresh'); + return outcome; + }; + + await expect( + service.syncLocalProject({ localProjectId, rootPath: '/tmp/project-1' }) + ).rejects.toThrow('catalog unavailable'); + await expect( + service.syncLocalProject({ localProjectId, rootPath: '/tmp/project-1' }) + ).resolves.toEqual(catalogResult(2)); + await expect( + service.syncLocalProject({ localProjectId, rootPath: '/tmp/project-1' }) + ).resolves.toEqual(catalogResult(3)); + expect(outcomes).toEqual([]); + }); +}); + describe('decideHistoryRefresh', () => { it('skips when replay digest is unchanged', () => { expect(