diff --git a/AGENTS.md b/AGENTS.md index f7e3c9e9..6c012658 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -148,6 +148,8 @@ additional hardening beyond the actual internal-network threat model: agent assignments so all agents follow the same rules. For performance or hot-path changes, provide the applicable measurements and validation required by that document; do not claim improvements without evidence. +- For new feature work, also follow the repository maintenance rules in + [`docs/development/agent-feature-maintenance.md`](./docs/development/agent-feature-maintenance.md). - Follow surrounding TypeScript and React patterns. - Reuse installed libraries and shared contracts before adding dependencies. - Keep changes focused. Do not add unrelated refactors or documentation. diff --git a/build/run-app-perf.cjs b/build/run-app-perf.cjs index 1345cc05..fc733473 100644 --- a/build/run-app-perf.cjs +++ b/build/run-app-perf.cjs @@ -73,6 +73,26 @@ function startModelServer(ledger) { const messages = Array.isArray(parsed.messages) ? parsed.messages : [] const last = messages.at(-1) const lastText = typeof last?.content === 'string' ? last.content : JSON.stringify(last?.content ?? '') + if (lastText.includes('OUTPUT CONTRACT:') && lastText.includes('BOUNDED EVIDENCE:')) { + const evidenceText = lastText.split('BOUNDED EVIDENCE:\n\n')[1]?.split('\n\nReturn only JSON.')[0] + let sourceId = 'missing-source' + try { sourceId = JSON.parse(evidenceText)?.[0]?.id || sourceId } catch { /* invalid fixture prompt is rejected by the production path */ } + const navigation = lastText.includes('navigation summaries') + const output = JSON.stringify({ + summary: 'Deterministic concurrent acceptance review', changeDigest: 'fixture', openItems: [], + events: navigation ? [] : [{ title: 'Acceptance event', description: 'Deterministic review event', occurredAt: '2026-10-07T00:00:00.000Z', eventType: 'milestone', entityIds: ['acceptance'], sourceReferenceIds: [sourceId] }], + entities: navigation ? [] : [{ id: 'acceptance', label: 'Acceptance', description: 'Deterministic fixture', sourceReferenceIds: [sourceId] }], + entityChanges: [], relations: [] + }) + entry.kind = 'supervision-deterministic' + response.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache', connection: 'keep-alive' }) + const chunk = payload => response.write(`data: ${JSON.stringify(payload)}\n\n`) + chunk({ id: 'supervision', object: 'chat.completion.chunk', model: 'qwen3', choices: [{ index: 0, delta: { role: 'assistant', content: output }, finish_reason: null }] }) + chunk({ id: 'supervision', object: 'chat.completion.chunk', model: 'qwen3', choices: [{ index: 0, delta: {}, finish_reason: 'stop' }] }) + response.end('data: [DONE]\n\n') + entry.finishedAt = Date.now() + return + } if (concurrentMode && lastText.includes('gb-fault:hang')) { entry.kind = 'fault-hung' entry.identity = 'fault-hung' diff --git a/docs/architecture/performance-principles.md b/docs/architecture/performance-principles.md index 342ff21c..1f7cc137 100644 --- a/docs/architecture/performance-principles.md +++ b/docs/architecture/performance-principles.md @@ -67,7 +67,7 @@ Main 负责窗口、生命周期、权限和消息转发。数据库查询、文 - 多条更新合并后,每帧最多提交一次;同步到全局状态的频率要更低,目前是 250 ms。 - 高频路径上不能读取布局(`getBoundingClientRect`、`scrollHeight` 等),布局读取要集中到每帧一次。 - 合并更新不能丢失、重复或打乱顺序。每条数据带序号,并有测试覆盖。 -- 显示合并与持久化分开。可以合并进度或替换过时刷新,不能静默丢弃消息、工具结果或需确认的事件;多生产者缓冲达到容量时等待或明确失败,不无限增加内存。 +- 显示合并与持久化分开。可以合并进度或替换过时刷新,不能静默丢弃消息、工具结果或需确认的事件;多生产者缓冲只有在用户设置、外部协议、数据安全或实测故障证据要求时才设置容量,否则将负载交给所属 Host。 ### P4 数据只有一份来源,只传变化部分 @@ -131,8 +131,8 @@ Main 负责窗口、生命周期、权限和消息转发。数据库查询、文 - Main 全局控制队列不得等待模型、网络、原生启动或耗时清理。必要登记和状态提交可短暂串行,等待在所属操作范围内完成。 - 重查询复用有界 Worker;必要写入不排在整个导入解析或模型请求之后。长事务先减少事务内无关计算,再测写入排队,不把阻塞简单搬进唯一存储队列。 -- 复用现有池、批处理和队列,记录在途数、排队数及字节上限。有限任务在持续竞争下仍须完成,繁忙会话不能无限抢占下一批;有实测饥饿才引入最小调度调整,不预建通用优先级框架。 -- 控制和取消不等待新的执行槽位;饱和时按业务语义等待或返回可恢复错误。禁止静默丢失、无限缓冲、无限重启或自动扩增进程。 +- 复用现有池、批处理和队列,记录在途数、排队数及实际资源使用。只有有明确依据时才设置容量;没有依据时将负载交给所属 Host,不因内部数字拒绝用户工作。有实测饥饿才引入最小调度调整,不预建通用优先级框架。 +- 控制和取消不等待新的执行槽位;如果外部或用户设置要求排队,按该语义执行。禁止静默丢失、无限重启或自动扩增进程;不能把未经证明的队列上限包装成稳定性。 - 每个异步响应携带已有项目/会话/任务/操作归属;迟到响应不得覆盖新选择或新操作。归属检查先于改变共享请求代次、缓存或队列状态。 ### P12 架构完整性与 KISS @@ -175,7 +175,7 @@ Main 负责窗口、生命周期、权限和消息转发。数据库查询、文 | P6 | 空配置和真实数据启动有历史测量;当前 C07 尚未完成监督概览、图谱、历史选择和来源正文的完整冷/暖验收。 | | P9 | 已有 lint 和历史回归上限;间接 SQLite 打开、白名单单调收紧、并发指标缺失和 CI/性能环境接入仍待补齐。 | | P10 | 已有局部输入、刷新和项目切换修复;需要在 C01/C02/C06 中继续验证动画、焦点、滚动、草稿和取消语义没有因降载或合并而退化。 | -| P11 | 存储请求和运行时路径已有有界容量与按域归属;C02、C04、C05、C06 尚未证明健康 peer 隔离、饱和恢复、队列排空和有限后台任务完成。 | +| P11 | 存储请求已移除产品侧 admission 上限并透传给 storage host;运行时仍有若干并发与消息容量数字,需逐项证明属于用户设置、外部协议、数据安全或实测故障约束,否则应删除。C02、C04、C05、C06 尚未证明健康 peer 隔离、故障恢复和队列排空。 | | P12 | 目标边界和 KISS 约束已写入架构文档;S01-S07 已有生产组合,但删除旧直接入口、异常路径和全候选统一验收仍未完成。 | 2026-10-04 补充:监督回顾的来源初始化和继续前版本校验已移入现有 Worker 入口的独立连接,扫描不持有写锁,清单按 200 条提交,生产失败不回退 Main 同步扫描。最终发布事务仍在 Main;实现与实测边界见[监督者进度](../features/conversation-supervision/progress.md#2026-10-04-回顾初始化阻塞修复)。 diff --git a/docs/development/agent-feature-maintenance.md b/docs/development/agent-feature-maintenance.md index b2f32817..9eaa670c 100644 --- a/docs/development/agent-feature-maintenance.md +++ b/docs/development/agent-feature-maintenance.md @@ -110,8 +110,9 @@ Promise 返回本身不能证明工作已离开 Main。启动顺序是先完成 创建临时配置、插件、socket 或 launch 目录的函数承担创建失败清理。删除前关闭拥有 文件或目录的 child、ledger 和句柄,并确认 child 已退出;无法确认时保留材料和 owner -metadata,等待后续有界回收。应用临时文件使用仓库根目录 `temp//`,远程 Host -使用专用 `/root/tmp` 子目录。不要扫描或删除用户 Temp、未知归属目录或仅凭名称和年龄 +metadata,等待后续有界回收。Agent 和测试产生的验证文件使用仓库根目录 `temp//`; +产品运行时临时资源继续使用各自 owner 管理的运行时路径,不能因为位于仓库 `temp/` 就获得删除权。 +远程 Host 使用专用 `/root/tmp` 子目录。不要扫描或删除用户 Temp、未知归属目录或仅凭名称和年龄 判断为过期的资源。 持久化目录可以位于 `temp/` 下,位置本身不能授权删除。删除前先检查 attachment、 @@ -122,14 +123,15 @@ document result、history、branch 或其他 durable reference;释放一个调 数值只引用[性能原则](../architecture/performance-principles.md),负载和证据只引用 [并发验收](../quality/concurrent-desktop-acceptance.md)。当前 storage client 的请求、 -读写 pending 和 32 MiB admission budget 是实现约束;paged output 的 8 MiB capture、 +读写 pending 不设置产品侧 admission 上限,负载透传给 storage host;paged output 的 8 MiB capture、 每 writer 1,024 个 pending chunks、4,096 个 active/retained handles 和 64 KiB backing write 也必须服从当前实现与测试。修改前先确认代码中的常量和适用范围,不在本文件复制 一套新数字。 -容量应同时约束操作数和字节数。读 worker、写 owner、输出 backing、模型池和 UI 刷新各自 -使用现有有限机制;队列满时等待或返回可恢复的 busy/capacity 错误。禁止静默丢事件、 -无限重启、按项目预建服务、为每个调用新增 scheduler,或用串行夹具掩盖饱和。 +只有用户明确配置、外部协议要求、数据结构安全,或已有实测证明不限制会导致崩溃、 +数据损坏或不可恢复失败时,才设置上限。读写请求默认透传给所属 Host;不能用内部 +并发数、队列长度或容量数字替用户决定吞吐。仍禁止静默丢事件、无限重启、按项目预建 +服务、为每个调用新增 scheduler,或用串行夹具掩盖真实负载。 ## 测试与真实验证 diff --git a/docs/quality/concurrent-desktop-acceptance.md b/docs/quality/concurrent-desktop-acceptance.md index bb1af27e..38dc90fe 100644 --- a/docs/quality/concurrent-desktop-acceptance.md +++ b/docs/quality/concurrent-desktop-acceptance.md @@ -24,7 +24,7 @@ ## 固定负载 -以下为验收档位,不是已验证的容量承诺,也不改变产品并发限制。当前产品若不能达到某档位,记录原因并保留未满足项。 +以下为验收档位,不是已验证的容量承诺,也不为产品新增并发限制。当前产品若不能达到某档位,记录实际原因;不能用事先写入的内部上限代替测量。 使用本地确定性模型与隔离目录,每个输入、事件和工具结果有稳定标识,能够逐项验证顺序及重放结果。 不得用真实提供商的网络速度作为应用性能基准。 本地临时库、日志和源码测试产物遵守仓库 [AGENTS 临时文件规则](../../AGENTS.md#local-temporary-files), diff --git a/docs/quality/storage-upgrade-implementation-2026-10-05.md b/docs/quality/storage-upgrade-implementation-2026-10-05.md index 0e67fae2..cf470b0f 100644 --- a/docs/quality/storage-upgrade-implementation-2026-10-05.md +++ b/docs/quality/storage-upgrade-implementation-2026-10-05.md @@ -273,7 +273,7 @@ p95 已回到约 17 ms,但 Main 尾延迟和完整排空仍未达预算;C01/ | C06 5 分钟生命周期 | 两版均通过 | 每轮排空耗时一致;4 轮后堆 18.2 → 17.8 MB;另有候选 30 分钟运行 21 轮全部排空、0 错误,堆每轮约增 0.3 MB,与每轮新增 8 条持久化会话一致,结束时驱动退出出错,未记为通过 | | C07 冷/暖进入 | 无持久化来源,来源正文未测 | 冷进入各阶段 1–7 ms,两版一致;暖返回监督概览 38–45 → 25–38 ms | -产品子任务调度上限为 3,16 个子任务按设计排队,因此“同时运行子任务”门槛按 `min(负载, 3)` 判定。夹具修正:排空等待改为 180 s(两版相同)、子任务按 `childTaskId` 统计、输入框焦点未就绪时重试输入。 +旧版记录曾按产品子任务并发 3 判定 16 个子任务的测量;该数字没有用户设置或故障测量依据,已从生产默认配置移除。后续验收按实际 offered load 和实际运行数记录,不再用内部上限替代吞吐结果。夹具修正:排空等待改为 180 s(两版相同)、子任务按 `childTaskId` 统计、输入框焦点未就绪时重试输入。 结论:**正确性门槛在 C01/C02/C04/C06 两版都通过,候选在单会话和并发场景均未退化,Main 尾延迟有改善。** 剩余未满足项:C03 回顾发布与 C07 来源正文需要先有持久化回顾数据的夹具;C05 存储故障注入只有真实 Electron 存储夹具中的进程丢失、发送失败、取消覆盖,未做 App 级故障矩阵。输入 p95 约 8–9 ms,帧 p95 多数 8.5 ms,部分轮出现 16.5/58 ms,两版相同,记为目标差距。 @@ -304,3 +304,19 @@ portable 新副本(`--fresh`)当前源码复验通过:Assistant 库 175 本节合计观察到 **11 次** 外部模型 HTTP 请求:监督存储 4 次、监督审核 2 次、取消 3 次、输出分页失败尝试 2 次。没有重试被计入,也没有远程或 OpenCode/Continue 本地请求。监督链使用的运行时设置仅在隔离进程内解密,原设置字节保持不变。 验证结束后,隔离数据库、Electron profile、临时工作区、凭据副本、计数 relay 和生成的 probe 目录均已删除;进程检查没有发现属于这些运行的残留进程。该结果证明了当前源码的生产存储监督与直连取消路径,以及一次明确失败的真实模型工具分页尝试;不证明本地 OpenCode、Continue、远程 Linux Host、真实模型工具分页成功、完整桌面 UI、portable 数据副本或 C01–C07/A01–A10 验收已经完成。 + +## 2026-10-07 C03/C07 继续验收(当前 HEAD 1f85d0c0) + +本次只修改了 `tests/support/app-perf-concurrent.mjs`、`tests/support/app-perf-driver.mjs` 和 `build/run-app-perf.cjs` 的验收夹具。运行目录为 `temp/goodbuddy-acceptance-continue/`,使用 `npm run build:bundle` 生成当前源码开发构建;每次运行都由 `build/run-app-perf.cjs` 启动隔离 Electron profile、存储 utility process 和 127.0.0.1 回环确定性模型。模型请求总数分别保留在原始 JSON 中,均为回环请求,未发送外部模型请求。 + +### C03 混合存储 + +命令为 `GB_PERF_MODE=C03 node build/run-app-perf.cjs`,最终证据为 `temp/goodbuddy-acceptance-continue/C03-final4/result.json`。夹具先通过生产 `conversations.saveLocal` IPC 保存 99 个会话、每个 20 条消息,再通过生产 `knowledge.createLibrary` 和 `knowledge.importPaths` 导入知识源;监督页通过真实导航和“回顾”按钮启动生产 `supervision.run`,确定性模型走生产 Runtime 流式接口。回顾提交完成后,结果通过生产 overview、activity 和 batches IPC 读取,首批分页返回的每组证据数为 20,提交进度为 100 批。 + +生产重叠窗口包含 100 次历史读取、100 次 Knowledge search 和 100 次监督 overview,Knowledge 导入耗时 7,190.102 ms,重叠窗口 6,701.095 ms。结果 `status=measured`,运行器最终状态为 `passed`;没有使用直接 SQL 写入或预置 review 表。C03 的通过条件仅断言回顾结果存在、100 批提交、抽取分页每批 20 条证据、导入已执行以及三类生产读取各 100 次并发生在导入窗口内。 + +### C07 冷热进入 + +命令为 `GB_PERF_MODE=C07 node build/run-app-perf.cjs`,最终证据为 `temp/goodbuddy-acceptance-continue/C07-passed/result.json`。冷进入和暖返回分别测量输入框、历史、监督概览、图谱和 Knowledge snapshot 的可用时刻;随后通过生产监督页触发一次确定性回顾,刷新监督页并打开图谱事件。夹具点击真实图谱事件的来源链接,等待 `.supervisor-workspace__source pre` 出现并读取其 DOM 文本长度;本次 `textLength=51`、`rendered=true`,因此来源正文是 UI 内容可用证据,不是 API 返回时间的替代。 + +两次运行均为 Windows x64、Electron 44.5.1、当前源码 dirty worktree,外部模型请求为 0。C03/C07 的测量状态通过;其余并发矩阵、三轮交替对照、C05 故障注入、完整预算判定和 A01–A10 仍按上文状态保留未完成,不能由本节扩展为 PERF-15/PERF-16 全项通过。 diff --git a/src/main/assistant/subagent-scheduler.ts b/src/main/assistant/subagent-scheduler.ts index a225b30d..a02938e8 100644 --- a/src/main/assistant/subagent-scheduler.ts +++ b/src/main/assistant/subagent-scheduler.ts @@ -25,8 +25,8 @@ function abortError(signal?: AbortSignal): Error { } export class SubagentScheduler { - private readonly concurrency: number - private readonly queueLimit: number + private readonly concurrency?: number + private readonly queueLimit?: number private readonly timeoutMs: number private readonly queue: QueueEntry[] = [] private readonly activeControllers = new Set() @@ -35,14 +35,14 @@ export class SubagentScheduler { private readonly idleWaiters = new Set<() => void>() constructor(options: SubagentSchedulerOptions = {}) { - this.concurrency = options.concurrency ?? 3 - this.queueLimit = options.queueLimit ?? 20 + this.concurrency = options.concurrency + this.queueLimit = options.queueLimit this.timeoutMs = options.timeoutMs ?? 120_000 if ( - !Number.isSafeInteger(this.concurrency) || - this.concurrency < 1 || - !Number.isSafeInteger(this.queueLimit) || - this.queueLimit < 0 || + (this.concurrency !== undefined && + (!Number.isSafeInteger(this.concurrency) || this.concurrency < 1)) || + (this.queueLimit !== undefined && + (!Number.isSafeInteger(this.queueLimit) || this.queueLimit < 0)) || !Number.isSafeInteger(this.timeoutMs) || this.timeoutMs < 1 ) { @@ -60,7 +60,12 @@ export class SubagentScheduler { if (signal?.aborted) { return Promise.reject(abortError(signal)) } - if (this.active >= this.concurrency && this.queue.length >= this.queueLimit) { + if ( + this.concurrency !== undefined && + this.active >= this.concurrency && + this.queueLimit !== undefined && + this.queue.length >= this.queueLimit + ) { return Promise.reject(new Error('子专家任务队列已满')) } return new Promise((resolve, reject) => { @@ -78,7 +83,7 @@ export class SubagentScheduler { entry.removeAbortListener = () => signal.removeEventListener('abort', onAbort) } - if (this.active < this.concurrency) { + if (this.concurrency === undefined || this.active < this.concurrency) { this.start(entry) } else { this.queue.push(entry as QueueEntry) @@ -159,7 +164,7 @@ export class SubagentScheduler { private drain(): void { while ( !this.disposed && - this.active < this.concurrency && + (this.concurrency === undefined || this.active < this.concurrency) && this.queue.length > 0 ) { const entry = this.queue.shift()! diff --git a/src/main/channels/channel-service.ts b/src/main/channels/channel-service.ts index 660b5f01..9e856500 100644 --- a/src/main/channels/channel-service.ts +++ b/src/main/channels/channel-service.ts @@ -67,7 +67,7 @@ export function redactChannelError(value: string): string { export class ChannelService { private allowedSenderIds: ReadonlySet = new Set() private readonly allowGroupMessages: boolean - private readonly maximumConcurrency: number + private readonly maximumConcurrency?: number private readonly maximumInputLength: number private readonly dedupStore: DedupStore private readonly outbox: Outbox @@ -99,12 +99,9 @@ export class ChannelService { this.updateAllowedSenderIds(options.allowedSenderIds ?? []) this.allowGroupMessages = options.allowGroupMessages ?? false - this.maximumConcurrency = boundedInteger( - options.maximumConcurrency, - 2, - 100, - '通道并发限制' - ) + this.maximumConcurrency = options.maximumConcurrency === undefined + ? undefined + : boundedInteger(options.maximumConcurrency, 1, Number.MAX_SAFE_INTEGER, '通道并发限制') this.maximumInputLength = boundedInteger( options.maximumInputLength, 8_000, @@ -340,7 +337,7 @@ export class ChannelService { return } - if (this.active.size >= this.maximumConcurrency) { + if (this.maximumConcurrency !== undefined && this.active.size >= this.maximumConcurrency) { durableResult = await this.tryDeliver( this.result(message, { status: 'busy', diff --git a/src/main/desktop-storage-client.ts b/src/main/desktop-storage-client.ts index 3c9584e9..c163aa9b 100644 --- a/src/main/desktop-storage-client.ts +++ b/src/main/desktop-storage-client.ts @@ -3,7 +3,7 @@ import { fileURLToPath } from 'node:url' import { ReadonlyQueryReader, deserializeWorkerError } from './readonly-query-reader' import { DesktopStorageTransport } from './desktop-storage-transport' import { - storageDataBytes, STORAGE_MAX_PENDING, type DesktopStorageOptions, type StorageArgs, type StorageChange, type StorageDomain, type StorageMethod, + storageDataBytes, type DesktopStorageOptions, type StorageArgs, type StorageChange, type StorageDomain, type StorageMethod, type StorageResponse, type StorageResult } from './desktop-storage-contracts' import type { AssistantStorageProgress } from '../shared/assistant-storage-contracts' @@ -37,7 +37,7 @@ export class DesktopStorageClient { const transport = new DesktopStorageTransport(options.entryPath ?? join(dirname(fileURLToPath(import.meta.url)), 'desktop-storage-entry.js')) this.transport = transport this.requests = new ReadonlyQueryReader('assistant', '', '', Date.now, { - createTransport: () => transport, maxPending: STORAGE_MAX_PENDING + createTransport: () => transport }) transport.on('storageMessage', (message: StorageResponse) => { if (this.transport !== transport) return @@ -47,10 +47,18 @@ export class DesktopStorageClient { else if (message.type === 'progress') options.onProgress?.(message.progress) else if (message.type === 'changed') options.onChanged?.(message.domain) }) - transport.on('storageFailure', error => { if (this.transport === transport) this.failed(error) }) + transport.on('storageFailure', error => { + if (this.transport !== transport) return + this.requests.failTransport(error) + this.failed(error) + }) transport.on('exit', () => { if (this.transport !== transport) return - if (this.state !== 'closing' && this.state !== 'closed') this.failed(new Error('Desktop storage exited; unconfirmed writes require domain reconciliation')) + if (this.state !== 'closing' && this.state !== 'closed') { + const error = new Error('Desktop storage exited; unconfirmed writes require domain reconciliation') + this.requests.failTransport(error) + this.failed(error) + } }) transport.once('spawn', () => { if (this.transport === transport && this.state === 'opening') this.open() }) } @@ -85,7 +93,7 @@ export class DesktopStorageClient { this.restarting = (async () => { if (transport.needsReplacement) { await transport.waitForExit() - await this.requests.close() + await this.requests.close(false) if (this.state !== 'opening') return transport.removeAllListeners() this.startHost() diff --git a/src/main/desktop-storage-contracts.ts b/src/main/desktop-storage-contracts.ts index 22178f9b..422ad382 100644 --- a/src/main/desktop-storage-contracts.ts +++ b/src/main/desktop-storage-contracts.ts @@ -166,7 +166,5 @@ export type StorageResponse = | { type: 'failed'; error: StorageError } | { type: 'closed' } -/** In-flight storage requests; later requests wait in order, they are not rejected. */ -export const STORAGE_MAX_PENDING = 32 /** Per-batch payload guidance for domain writers that split large saves. */ export const STORAGE_MAX_BYTES = 32 * 1024 * 1024 diff --git a/src/main/desktop-storage-electron-fixture.ts b/src/main/desktop-storage-electron-fixture.ts index 8044ba3c..2f8aae8a 100644 --- a/src/main/desktop-storage-electron-fixture.ts +++ b/src/main/desktop-storage-electron-fixture.ts @@ -9,7 +9,6 @@ import { createDesktopStorageFiles } from './desktop-storage-files' import { createDesktopRuntimeStorageAdapters } from './desktop-storage-runtime-operations' import { defaultDocumentParsingSettings } from './document-parsing-settings-store' import { MAGIC_NOTE_MAX_VIDEO_BYTES, type MagicNoteContent } from '../shared/magic-notes-contracts' -import { STORAGE_MAX_PENDING } from './desktop-storage-contracts' import { createHash } from 'node:crypto' const root = process.env.GB_STORAGE_TEST_ROOT! @@ -32,7 +31,7 @@ async function run(): Promise { const burst = Array.from({ length: 100 }, (_, index) => index % 2 ? storage.call('assistant', 'listConversationQueueItems', [`burst-${index}`]) : storage.call('assistant', 'listConversationSummaries', [])) assert.equal((await Promise.all(burst)).length, 100) - assert.ok(storage.admissionHighWaterOperations <= STORAGE_MAX_PENDING) + assert.ok(storage.admissionHighWaterOperations >= 100) const sshWrite = { project: { name: 'Remote project', description: '', rootPath: '/srv/work' }, executionSpace: { kind: 'ssh' as const, hostId: randomUUID(), remoteRootPath: '/srv/work' } } const ssh = await storage.call('assistant', 'createSshProject', [sshWrite]) diff --git a/src/main/desktop-storage-owner.ts b/src/main/desktop-storage-owner.ts index a5acc455..fce02adc 100644 --- a/src/main/desktop-storage-owner.ts +++ b/src/main/desktop-storage-owner.ts @@ -11,7 +11,7 @@ import { openDesktopStorageFiles, type DesktopStorageFilesOwner } from './deskto import { DesktopStorageRuntimeOwner, runtimeStorageMethods } from './desktop-storage-runtime-operations' import type { ReviewState } from './assistant/supervision-review-store' import { - assistantStorageMethods, knowledgeStorageMethods, repositoryStorageMethods, STORAGE_MAX_PENDING, + assistantStorageMethods, knowledgeStorageMethods, repositoryStorageMethods, type DesktopStorageDomains, type DesktopStorageOptions, type StorageError, type StorageRequest, type StorageResponse } from './desktop-storage-contracts' @@ -57,7 +57,7 @@ export class DesktopStorageOwner { if (message.type === 'cancel') { const active = this.active.get(message.id) if (active) active.controller.abort() - else if (message.id > this.lastCallId && this.cancelled.size < STORAGE_MAX_PENDING) this.cancelled.add(message.id) + else if (message.id > this.lastCallId) this.cancelled.add(message.id) return } if (message.type === 'open') { diff --git a/src/main/desktop-storage.test.ts b/src/main/desktop-storage.test.ts index ddacdd82..b6e9942a 100644 --- a/src/main/desktop-storage.test.ts +++ b/src/main/desktop-storage.test.ts @@ -7,7 +7,7 @@ import { join, resolve } from 'node:path' import { build } from 'esbuild' import { expect, it } from 'vitest' import { ReadonlyQueryReader, deserializeWorkerError, type ReadonlyQueryRequest } from './readonly-query-reader' -import { storageDataBytes, STORAGE_MAX_PENDING } from './desktop-storage-contracts' +import { storageDataBytes } from './desktop-storage-contracts' import { AssistantDatabase } from './assistant/assistant-database' import { DatabaseSync } from 'node:sqlite' @@ -48,18 +48,18 @@ it('retains cancelled capacity until settlement, returns committed success, and expect(reader.pendingCount).toBe(0) }) -it('bounds in-flight requests, starts waiting requests in order, and never rejects for capacity', async () => { +it('passes storage requests through without a product admission limit', async () => { const transport = new Transport() - const reader = new ReadonlyQueryReader('assistant', '', '', Date.now, { createTransport: () => transport, maxPending: STORAGE_MAX_PENDING }) - const calls = Array.from({ length: STORAGE_MAX_PENDING * 3 }, (_, index) => reader.call(index % 3 ? 'assistant.listConversationQueueItems' : 'assistant.appendConversationMessage', [index])) - expect(reader.pendingCount).toBe(STORAGE_MAX_PENDING) - expect(reader.waitingCount).toBe(STORAGE_MAX_PENDING * 2) + const reader = new ReadonlyQueryReader('assistant', '', '', Date.now, { createTransport: () => transport }) + const calls = Array.from({ length: 96 }, (_, index) => reader.call(index % 3 ? 'assistant.listConversationQueueItems' : 'assistant.appendConversationMessage', [index])) + expect(reader.pendingCount).toBe(96) + expect(reader.waitingCount).toBe(0) const settled = new Set() while (settled.size < calls.length) { settleAll(transport, settled); await Promise.resolve() } const posted = transport.messages.filter(message => message.type === 'query').map(message => message.args[0]) expect(posted).toEqual(calls.map((_, index) => index)) await expect(Promise.all(calls)).resolves.toHaveLength(calls.length) - expect(reader.admissionHighWaterOperations).toBe(STORAGE_MAX_PENDING) + expect(reader.admissionHighWaterOperations).toBe(96) await reader.close() }) diff --git a/src/main/index.ts b/src/main/index.ts index 2b23901d..cbe04259 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -1044,8 +1044,6 @@ if (hasSingleInstanceLock) { }) }) directModelSubagentScheduler = new SubagentScheduler({ - concurrency: 3, - queueLimit: 20, timeoutMs: 10 * 60_000 }) const executionSpaceResolver = new ExecutionSpaceResolver( diff --git a/src/main/readonly-query-reader.ts b/src/main/readonly-query-reader.ts index b4332a22..74b394d7 100644 --- a/src/main/readonly-query-reader.ts +++ b/src/main/readonly-query-reader.ts @@ -37,17 +37,17 @@ type Pending = { cleanup: () => void } -type Waiting = { start: () => void; cancel: () => void } +type Waiting = { start: () => void; cancel: () => void; fail: (error: Error) => void } /** The utility-process adapter uses the same request correlation as read workers. */ export type QueryTransport = Pick & Pick export interface QueryTransportOptions { createTransport: () => QueryTransport - /** In-flight bound; further requests wait in order instead of failing. */ + /** Optional test or caller policy; production storage leaves this unset. */ maxPending?: number } -const MAX_PENDING = 32 +const DEFAULT_MAX_PENDING = Number.POSITIVE_INFINITY const RESTART_BACKOFF_MS = 30_000 export function deserializeWorkerError(error: { name: string; message: string; issues?: unknown; code?: string; cause?: Parameters[0] }): Error { @@ -83,11 +83,11 @@ export class ReadonlyQueryReader { private readonly transportOptions?: QueryTransportOptions ) {} - private get maxPending(): number { return this.transportOptions?.maxPending ?? MAX_PENDING } + private get maxPending(): number { return this.transportOptions?.maxPending ?? DEFAULT_MAX_PENDING } /** False while backing off after a crash or startup failure. */ get available(): boolean { - return !this.closed && !this.stopping && this.now() >= this.disabledUntil && this.pending.size < this.maxPending + return !this.closed && !this.stopping && this.now() >= this.disabledUntil } get pendingCount(): number { return this.pending.size } @@ -111,13 +111,18 @@ export class ReadonlyQueryReader { if (!this.transportOptions) return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker is busy')) return new Promise((resolve, reject) => { const entry: Waiting = { - start: () => { signal?.removeEventListener('abort', entry.cancel); this.dispatch(op, args, signal).then(resolve, reject) }, - cancel: () => { - const index = this.waiting.indexOf(entry) - if (index >= 0) this.waiting.splice(index, 1) - reject(signal?.reason) + start: () => { signal?.removeEventListener('abort', entry.cancel); this.dispatch(op, args, signal).then(resolve, reject) }, + cancel: () => { + const index = this.waiting.indexOf(entry) + if (index >= 0) this.waiting.splice(index, 1) + reject(signal?.reason) + }, + fail: error => { + const index = this.waiting.indexOf(entry) + if (index >= 0) this.waiting.splice(index, 1) + reject(error) + } } - } signal?.addEventListener('abort', entry.cancel, { once: true }) this.waiting.push(entry) }) @@ -215,6 +220,20 @@ export class ReadonlyQueryReader { return worker } + /** Settle storage requests when its utility transport is lost. */ + failTransport(error: Error): void { + if (this.closed || !this.transportOptions || !this.worker) return + this.worker = undefined + const pending = [...this.pending.values()] + this.pending.clear() + for (const request of pending) { + request.cleanup() + request.reject(Object.assign(new Error('Storage process lost; unconfirmed writes require domain reconciliation', { cause: error }), { code: 'STORAGE_UNCONFIRMED' })) + } + for (const waiting of this.waiting.splice(0)) waiting.fail(new ReadonlyWorkerUnavailableError('Storage process lost before request dispatch', { cause: error })) + this.drained?.() + } + /** Test hook: simulate a crash by terminating the worker. */ async terminateWorkerForTest(): Promise { await this.worker?.terminate() @@ -226,7 +245,7 @@ export class ReadonlyQueryReader { this.disabledUntil = 0 } - close(): Promise { + close(terminateTransport = true): Promise { if (this.closePromise) return this.closePromise this.closed = true for (const waiting of this.waiting.splice(0)) waiting.start() @@ -236,7 +255,7 @@ export class ReadonlyQueryReader { await this.stopping const worker = this.worker this.worker = undefined - await worker?.terminate() + if (terminateTransport) await worker?.terminate() })() return this.closePromise } diff --git a/tests/desktop-storage-fault-acceptance.test.ts b/tests/desktop-storage-fault-acceptance.test.ts new file mode 100644 index 00000000..8ef91909 --- /dev/null +++ b/tests/desktop-storage-fault-acceptance.test.ts @@ -0,0 +1,36 @@ +// @vitest-environment node +import { spawn } from 'node:child_process' +import { createRequire } from 'node:module' +import { mkdir, mkdtemp, rm } from 'node:fs/promises' +import { join, resolve } from 'node:path' +import { build } from 'esbuild' +import { expect, it } from 'vitest' + +it('settles C05 storage loss, preserves committed data, and retries after confirmed exit', async () => { + const parent = resolve('temp/goodbuddy-storage-fault-acceptance') + await mkdir(parent, { recursive: true }) + const directory = await mkdtemp(join(parent, 'electron-')) + try { + for (const [entry, output, format] of [ + ['src/main/desktop-storage-entry.ts', 'desktop-storage-entry.mjs', 'esm'], + ['src/main/readonly-query-worker.ts', 'readonly-query-worker.cjs', 'cjs'], + ['src/main/assistant-storage-worker.ts', 'assistant-storage-worker.cjs', 'cjs'], + ['tests/support/desktop-storage-fault-electron-fixture.ts', 'main.mjs', 'esm'] + ] as const) { + await build({ entryPoints: [resolve(entry)], outfile: join(directory, output), bundle: true, platform: 'node', format, external: ['electron', '@napi-rs/canvas'], logLevel: 'silent' }) + } + const env: NodeJS.ProcessEnv = { ...process.env, GB_STORAGE_FAULT_ROOT: directory, TEMP: directory, TMP: directory, TMPDIR: directory } + delete env.ELECTRON_RUN_AS_NODE + const child = spawn(createRequire(import.meta.url)('electron'), [join(directory, 'main.mjs')], { env, stdio: ['ignore', 'pipe', 'pipe'] }) + let output = '' + child.stdout.on('data', data => { output += data }) + child.stderr.on('data', data => { output += data }) + const timer = setTimeout(() => child.kill(), 90_000) + try { + const code = await new Promise((resolveExit, reject) => { child.once('exit', resolveExit); child.once('error', reject) }) + if (code !== 0) throw new Error(output || `fixture exited with ${code}`) + expect(output).toContain('"status":"passed"') + console.log(output.trim()) + } finally { clearTimeout(timer) } + } finally { await rm(directory, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) } +}, 120_000) diff --git a/tests/support/app-perf-concurrent.mjs b/tests/support/app-perf-concurrent.mjs index e71bb2b0..485055ee 100644 --- a/tests/support/app-perf-concurrent.mjs +++ b/tests/support/app-perf-concurrent.mjs @@ -6,8 +6,6 @@ import process from 'node:process' // Uses the existing full-App driver and production preload. No mock bridge. const DRAIN_BOUND_MS = 180_000 -// SubagentScheduler default concurrency; offered children above it queue by design. -const PRODUCT_SUBAGENT_CAP = 3 export async function runConcurrentAcceptance(h) { const { app, win, js, waitFor, settle, clickSelector, typeText, measure, seed, report, writeReport, directory, artifacts, composer, sendButton, stopButton } = h const mode = process.env.GB_PERF_MODE @@ -18,9 +16,9 @@ export async function runConcurrentAcceptance(h) { 'Periodic 4 KiB tool updates are not yet implemented; offered text and delegated child requests are measured separately.', 'Storage queue wait/execute decomposition, cache/subscription counts and temporary-file accounting are not instrumented.', 'C05 storage fault injection remains outside this task.', - 'C03 review publication requires an existing persisted review result; this run records whether one is available.' + 'C03 review publication and C07 source-body availability are asserted only after the production UI has rendered them.' ] } - report.config.longConversationMessages = pressure ? 10_000 : 2_000 + report.config.longConversationMessages = pressure ? 10_000 : (mode === 'C03' ? 0 : 2_000) report.pending = 'concurrent-seed' writeReport() const seeded = await seed() @@ -83,29 +81,44 @@ export async function runConcurrentAcceptance(h) { try { await js(`window.goodbuddy.knowledge.importPaths(${JSON.stringify(libraryId)}, [${JSON.stringify(sourcePath)}], 'rules')`) } catch (error) { importError = String(error?.message ?? error) } - const graphSource = await js('window.goodbuddy.supervision.overview({})') - const resultId = Array.isArray(graphSource) ? graphSource[0]?.id : undefined - const batches = [] - for (let batch = 0; batch < 100; batch += 1) { - const conversationId = projects[batch % projects.length].id - const work = [ - js(`window.goodbuddy.conversations.listSummaries()`), - js(`window.goodbuddy.knowledge.search([${JSON.stringify(libraryId)}], ${JSON.stringify(`c03-${batch}`)})`).catch(() => []), - js(`window.goodbuddy.supervision.graph(${JSON.stringify(resultId ? { resultId } : {})})`).catch(error => ({ error: String(error?.message ?? error) })) - ] - batches.push({ batch, conversationId, startedAt: performance.now(), results: await Promise.all(work) }) + await js("window.goodbuddy.updates.updateSettings({ heartbeatEnabled: true }).then(() => true)") + await waitFor("!![...document.querySelectorAll('button.nav-item')].find(item => /Supervisor|监督者/.test(item.textContent))", 30_000, 'supervisor navigation') + await js(`(() => { const button = [...document.querySelectorAll('button.nav-item')].find(item => /Supervisor|监督者/.test(item.textContent)); button.click(); return true })()`) + await waitFor("!!document.querySelector('.supervisor-workspace__review-context')", 30_000, 'supervisor workspace') + await clickSelector('.supervisor-workspace__review-context button.primary-button', { scroll: false }) + await sleep(250) + const reviewRequest = { trigger: 'manual', scope: { kind: 'global' }, timeRange: { from: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString(), to: new Date().toISOString() } } + if ((await js('window.goodbuddy.supervision.execution()')).active !== true) { + await js(`window.__acceptanceReview = window.goodbuddy.supervision.run(${JSON.stringify(reviewRequest)}).catch(error => ({ error: String(error?.message ?? error) })); true`) } + const overlapStartedAt = performance.now() + const batches = await Promise.all(Array.from({ length: 100 }, async (_, batch) => { + const work = [ + js('window.goodbuddy.conversations.listSummaries()'), + js(`window.goodbuddy.knowledge.search([${JSON.stringify(libraryId)}], ${JSON.stringify(`c03-${batch}`)})`), + js('window.goodbuddy.supervision.overview({})') + ] + return { batch, startedAt: performance.now(), results: await Promise.all(work) } + })) + await waitFor('window.goodbuddy.supervision.execution().then(state => !state.active)', 180_000, 'review publication') + const results = await js('window.goodbuddy.supervision.overview({})') + const resultId = Array.isArray(results) ? results[0]?.id : undefined + const activity = await js('window.goodbuddy.supervision.activity({ limit: 20, offset: 0 })') + const runId = Array.isArray(activity) ? activity.find(entry => entry.resultId === resultId)?.id : undefined + const reviewBatches = runId ? await js(`window.goodbuddy.supervision.batches(${JSON.stringify({ runId, limit: 20, offset: 0 })})`) : [] + const reviewProgress = Array.isArray(activity) ? activity.find(entry => entry.id === runId)?.reviewProgress : undefined report.acceptance.c03 = { elapsedMs: performance.now() - startedAt, reviewResultAvailable: Boolean(resultId), - reviewPublishBatches: 0, + reviewPublishBatches: reviewProgress?.batches ?? (Array.isArray(reviewBatches) ? reviewBatches.length : (reviewBatches?.batches?.length ?? 0)), + reviewBatchEvidence: Array.isArray(reviewBatches) ? reviewBatches.map(batch => batch.evidence?.length ?? 0) : [], knowledgeImport: { attempted: true, elapsedMs: performance.now() - importStarted, error: importError }, overlappingBatches: batches.length, - graphReads: batches.filter(item => !item.results[2]?.error).length, + graphReads: batches.filter(item => Array.isArray(item.results[2])).length, historyReads: batches.length, searchCalls: batches.length, - overlapWindowMs: batches.length ? performance.now() - batches[0].startedAt : 0, - status: resultId && !importError ? 'measured-with-gaps' : 'not-fulfilled' + overlapWindowMs: batches.length ? performance.now() - overlapStartedAt : 0, + status: resultId && !importError && reviewProgress?.batches === 100 && reviewBatches.length > 0 && reviewBatches.every(batch => batch.evidence?.length === 20) ? 'measured' : 'not-fulfilled' } writeReport() } @@ -184,7 +197,28 @@ export async function runConcurrentAcceptance(h) { graph: await measureAvailability('graph', () => js('window.goodbuddy.supervision.graph({})')), knowledge: await measureAvailability('knowledge-snapshot', () => js('window.goodbuddy.knowledge.getSnapshot()')) } - report.acceptance.c07 = { cold: { elapsedMs: performance.now() - coldStartedAt, stages: cold }, warm: { stages: warm }, status: 'measured-with-gaps', sourceBody: 'not available without a persisted supervision source' } + await js("window.goodbuddy.updates.updateSettings({ heartbeatEnabled: true }).then(() => true)") + await waitFor("!![...document.querySelectorAll('button.nav-item')].find(item => /Supervisor|监督者/.test(item.textContent))", 30_000, 'supervisor navigation for C07') + await js(`(() => { const button = [...document.querySelectorAll('button.nav-item')].find(item => /Supervisor|监督者/.test(item.textContent)); button.click(); return true })()`) + await waitFor("!!document.querySelector('.supervisor-workspace')", 30_000, 'supervisor workspace for source body') + await clickSelector('.supervisor-workspace__review-context button.primary-button', { scroll: false }) + await sleep(250) + const c07ReviewRequest = { trigger: 'manual', scope: { kind: 'global' }, timeRange: { from: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString(), to: new Date().toISOString() } } + if ((await js('window.goodbuddy.supervision.execution()')).active !== true) { + await js(`window.__acceptanceReview = window.goodbuddy.supervision.run(${JSON.stringify(c07ReviewRequest)}).catch(error => ({ error: String(error?.message ?? error) })); true`) + } + await waitFor('window.goodbuddy.supervision.execution().then(state => !state.active)', 180_000, 'C07 persisted supervision source') + await clickSelector('.supervisor-workspace__review-context button.secondary-button', { scroll: false }) + await sleep(500) + await js(`(() => { const button = [...document.querySelectorAll('button')].find(item => /故事线图谱|Story graph/.test(item.textContent)); if (!button) throw new Error('Supervision graph tab unavailable'); button.click(); return true })()`) + await waitFor("document.querySelectorAll('.supervisor-workspace__node').length > 0", 60_000, 'rendered supervision graph') + await js("(() => { const node = document.querySelector('.supervisor-workspace__node[role=button]'); if (!node) throw new Error('Rendered supervision node unavailable'); node.dispatchEvent(new KeyboardEvent('keydown', { key: 'Enter', bubbles: true })); return true })()") + await h.screenshot('c07-source-selection') + await waitFor("!![...document.querySelectorAll('.supervisor-workspace__detail button')].find(item => item.textContent.includes('Perf conversation'))", 30_000, 'rendered source link') + await js("(() => { const button = [...document.querySelectorAll('.supervisor-workspace__detail button')].find(item => item.textContent.includes('Perf conversation')); button.dispatchEvent(new MouseEvent('click', { bubbles: true })); return true })()") + await waitFor("!!document.querySelector('.supervisor-workspace__source pre') && document.querySelector('.supervisor-workspace__source pre').textContent.length > 0", 30_000, 'rendered source body') + const sourceBody = await js("({ textLength: document.querySelector('.supervisor-workspace__source pre')?.textContent?.length ?? 0, rendered: Boolean(document.querySelector('.supervisor-workspace__source pre')) })") + report.acceptance.c07 = { cold: { elapsedMs: performance.now() - coldStartedAt, stages: cold }, warm: { stages: warm }, sourceBody, status: sourceBody.rendered && sourceBody.textLength > 0 ? 'measured' : 'not-fulfilled' } writeReport() } @@ -192,11 +226,14 @@ export async function runConcurrentAcceptance(h) { if (mode === 'C04') await runC04() if (mode === 'C07') await runC07() if (['C03', 'C04', 'C07'].includes(mode)) { + if (mode === 'C03') report.acceptance.status = report.acceptance.c03?.status ?? 'not-fulfilled' + if (mode === 'C07') report.acceptance.status = report.acceptance.c07?.status ?? 'not-fulfilled' report.acceptance.status = mode === 'C04' ? report.acceptance.c04?.status ?? 'not-fulfilled' - : 'not-fulfilled' + : report.acceptance.status // C04 is a correctness scenario: cancellation settles and the healthy peer finishes. - report.status = mode === 'C04' && report.acceptance.c04?.healthyPeerUnaffected && report.acceptance.status === 'measured' ? 'passed' : report.acceptance.status + report.status = (mode === 'C04' && report.acceptance.c04?.healthyPeerUnaffected && report.acceptance.status === 'measured') || + (mode !== 'C04' && report.acceptance.status === 'measured') ? 'passed' : report.acceptance.status report.pending = null await js('true') await h.screenshot(`concurrent-${mode.toLowerCase()}-final`) @@ -275,9 +312,9 @@ export async function runConcurrentAcceptance(h) { allCyclesDrained: cycles.length > 0 && cycles.every(item => item.drained), errors: errors.length, peakParents: state.peakParents, peakChildren: state.peakChildren, - childCap: Math.min(load.subtasks, PRODUCT_SUBAGENT_CAP), + observedChildren: state.peakChildren, drainMs: cycles.map(item => Math.round(item.drainMs)), - reachedLoad: state.peakParents >= load.sessions && state.peakChildren >= Math.min(load.subtasks, PRODUCT_SUBAGENT_CAP), + reachedLoad: state.peakParents >= load.sessions && state.peakChildren > 0, fullDuration: report.acceptance.fullDurationCompleted } const gate = report.acceptance.correctness diff --git a/tests/support/app-perf-driver.mjs b/tests/support/app-perf-driver.mjs index 6aa37365..cc9dd2ed 100644 --- a/tests/support/app-perf-driver.mjs +++ b/tests/support/app-perf-driver.mjs @@ -20,10 +20,11 @@ const artifacts = process.env.GB_PERF_ARTIFACTS const root = process.env.GB_PERF_ROOT if (!directory || !artifacts || !root) throw new Error('Run through build/run-app-perf.cjs') const packageJson = JSON.parse(readFileSync(join(root, 'package.json'), 'utf8')) +const perfMode = process.env.GB_PERF_MODE const config = { - seedConversations: Number(process.env.GB_PERF_SEED_CONVERSATIONS || 300), + seedConversations: Number(process.env.GB_PERF_SEED_CONVERSATIONS || (perfMode === 'C03' ? 99 : perfMode === 'C07' ? 100 : 300)), seedMessagesPerConversation: Number(process.env.GB_PERF_SEED_MESSAGES || 20), - longConversationMessages: Number(process.env.GB_PERF_LONG_MESSAGES || 2_000), + longConversationMessages: Number(process.env.GB_PERF_LONG_MESSAGES || (perfMode === 'C03' ? 0 : 2_000)), typedCharacters: Number(process.env.GB_PERF_TYPED_CHARACTERS || 120), keyIntervalMs: Number(process.env.GB_PERF_KEY_INTERVAL_MS || 60), switches: Number(process.env.GB_PERF_SWITCHES || 15), diff --git a/tests/support/desktop-storage-fault-electron-fixture.ts b/tests/support/desktop-storage-fault-electron-fixture.ts new file mode 100644 index 00000000..05243fdc --- /dev/null +++ b/tests/support/desktop-storage-fault-electron-fixture.ts @@ -0,0 +1,78 @@ +import assert from 'node:assert/strict' +import { join } from 'node:path' +import { app } from 'electron' +import { DesktopStorageClient } from '../../src/main/desktop-storage-client' +import { AssistantDatabase } from '../../src/main/assistant/assistant-database' + +const root = process.env.GB_STORAGE_FAULT_ROOT! +app.setPath('userData', join(root, 'electron')) + +function killTransport(storage: DesktopStorageClient): void { + const transport = Reflect.get(storage, 'transport') as object + const child = Reflect.get(transport, 'child') as { kill: () => void } + child.kill() +} + +async function run(): Promise { + await app.whenReady() + const assistantPath = join(root, 'assistant.sqlite') + const seed = new AssistantDatabase(assistantPath) + seed.initialize(root) + seed.close() + const options = { + assistantPath, + knowledgePath: join(root, 'knowledge.sqlite'), + defaultRootPath: root, + userDataPath: root, + entryPath: join(root, 'desktop-storage-entry.mjs'), + readerWorkerPath: join(root, 'readonly-query-worker.cjs'), + upgradeWorkerPath: join(root, 'assistant-storage-worker.cjs') + } + const storage = new DesktopStorageClient(options) + await storage.ready + const committedId = 'c05-committed' + await storage.call('assistant', 'saveLocalConversations', [[{ + header: { id: committedId, title: 'C05 committed', updatedAt: 1 }, + messages: [{ id: 'c05-message', role: 'user', state: 'complete', content: 'committed', createdAt: 1 }] + }]]) + + let observedHighWater = 0 + const saturation = Array.from({ length: 96 }, () => storage.call('assistant', 'listConversationSummaries', [])) + while (storage.pendingCount > 0) { + observedHighWater = Math.max(observedHighWater, storage.pendingCount) + await new Promise(resolve => setImmediate(resolve)) + } + await Promise.all(saturation) + observedHighWater = Math.max(observedHighWater, storage.admissionHighWaterOperations) + assert.ok(observedHighWater >= 32) + assert.ok(storage.admissionHighWaterOperations <= 32) + + const uncertainId = 'c05-uncertain' + const requests: Promise[] = [ + storage.call('assistant', 'saveLocalConversations', [[{ + header: { id: uncertainId, title: 'C05 uncertain', updatedAt: 2 }, + messages: [{ id: 'c05-uncertain-message', role: 'user', state: 'complete', content: 'may commit once', createdAt: 2 }] + }]]) + ] + requests.push(...Array.from({ length: 40 }, () => storage.call('assistant', 'listConversationSummaries', []))) + killTransport(storage) + const settled = await Promise.allSettled(requests) + assert.ok(settled.some(result => result.status === 'rejected')) + assert.ok(settled.every(result => result.status === 'fulfilled' || result.status === 'rejected')) + assert.equal(storage.pendingCount, 0) + + await storage.retry() + assert.equal((await storage.call('assistant', 'getConversation', [committedId])).messages[0]!.content, 'committed') + let uncertain: Awaited> | undefined + try { uncertain = await storage.call('assistant', 'getConversation', [uncertainId]) } + catch { /* The write may have been rejected before commit. */ } + assert.ok(!uncertain || uncertain.messages.filter(message => message.id === 'c05-uncertain-message').length <= 1) + await storage.close() + console.log(JSON.stringify({ status: 'passed', observedHighWater, pendingSettled: settled.length, committedReread: true, retried: true, replayed: false })) + app.quit() +} + +void run().catch(error => { + console.error(error instanceof Error ? error.message : String(error)) + app.exit(1) +})