From 4d18794e74ac4f5bc6d63b757a2223ef4eafe29d Mon Sep 17 00:00:00 2001 From: mesalogo Date: Sun, 4 Oct 2026 16:21:55 +0800 Subject: [PATCH] perf: track execution duration by state and bound summary I/O --- build/run-write-io-perf.cjs | 78 +++ docs/architecture/runtime-storage.md | 26 +- docs/features/task-and-job/README.md | 6 +- docs/features/task-and-job/task-center-prd.md | 11 +- .../features/task-and-job/technical-design.md | 193 ++---- docs/review/execution-timing-io-2026-10-04.md | 67 ++ electron.vite.config.ts | 1 - src/main/assistant/assistant-database.test.ts | 24 + src/main/assistant/assistant-database.ts | 470 +++++--------- .../assistant-execution-stats.test.ts | 612 ++++++------------ .../assistant/execution-stats-reader.test.ts | 68 -- src/main/assistant/execution-stats-reader.ts | 52 -- src/main/execution-stats-worker.ts | 17 - src/main/ipc.test.ts | 150 ++++- src/main/ipc.ts | 77 ++- src/main/readonly-query-reader.test.ts | 12 +- src/preload/execution-stats.test.ts | 9 +- src/preload/index.ts | 5 + src/renderer/src/App.test.tsx | 38 +- src/renderer/src/App.tsx | 7 +- src/renderer/src/task-selectors.ts | 11 - src/renderer/src/task-store.test.ts | 5 - src/renderer/src/use-execution-stats.test.tsx | 351 ++-------- src/renderer/src/use-execution-stats.ts | 73 +-- src/shared/assistant-contracts.ts | 15 +- src/shared/contracts.ts | 1 + src/shared/ipc-channels.ts | 1 + tests/magic-notes-capture.electron.test.ts | 1 - tests/manual-compaction.electron.test.ts | 1 - tests/overlay-layering.electron.test.ts | 1 - tests/terminal-status-probe.electron.test.ts | 1 - 31 files changed, 987 insertions(+), 1397 deletions(-) create mode 100644 build/run-write-io-perf.cjs create mode 100644 docs/review/execution-timing-io-2026-10-04.md delete mode 100644 src/main/assistant/execution-stats-reader.test.ts delete mode 100644 src/main/assistant/execution-stats-reader.ts delete mode 100644 src/main/execution-stats-worker.ts diff --git a/build/run-write-io-perf.cjs b/build/run-write-io-perf.cjs new file mode 100644 index 00000000..33876723 --- /dev/null +++ b/build/run-write-io-perf.cjs @@ -0,0 +1,78 @@ +// Windows logical-I/O probe against an isolated copy of a quiescent data folder. +// GB_WPERF_DATA must contain assistant.sqlite. No source database is opened. +const { buildSync } = require('esbuild') +const { mkdtempSync, copyFileSync, cpSync, existsSync, rmSync } = require('node:fs') +const { tmpdir } = require('node:os') +const { join, resolve } = require('node:path') +const { randomUUID } = require('node:crypto') +const { performance } = require('node:perf_hooks') +const { execFileSync, spawnSync } = require('node:child_process') +const { deepStrictEqual } = require('node:assert') + +if (process.platform !== 'win32') throw new Error('This probe uses Windows process I/O counters') +if (!process.versions.electron) { + const result = spawnSync(require('electron'), [__filename], { + stdio: 'inherit', env: { ...process.env, ELECTRON_RUN_AS_NODE: '1' } + }) + process.exit(result.status ?? 1) +} +const io = () => JSON.parse(execFileSync('powershell.exe', ['-NoProfile', '-Command', + `Get-CimInstance Win32_Process -Filter 'ProcessId=${process.pid}' | Select-Object ReadTransferCount,WriteTransferCount | ConvertTo-Json -Compress` +], { encoding: 'utf8' })) +const source = process.env.GB_WPERF_DATA +if (!source || !existsSync(join(source, 'assistant.sqlite'))) throw new Error('Set GB_WPERF_DATA') +const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-write-io-')) +let database +try { + const path = join(directory, 'assistant.sqlite') + for (const suffix of ['', '-wal', '-shm']) { + if (existsSync(join(source, `assistant.sqlite${suffix}`))) { + copyFileSync(join(source, `assistant.sqlite${suffix}`), `${path}${suffix}`) + } + } + if (existsSync(join(source, 'notes'))) cpSync(join(source, 'notes'), join(directory, 'notes'), { recursive: true }) + const bundle = join(directory, 'database.cjs') + buildSync({ entryPoints: [resolve(__dirname, '../src/main/assistant/assistant-database.ts')], + outfile: bundle, bundle: true, platform: 'node', format: 'cjs', external: ['electron'], logLevel: 'error' }) + const { AssistantDatabase } = require(bundle) + database = new AssistantDatabase(path) + database.initialize(directory) + const raw = database.database + const largest = raw.prepare(`SELECT id, conversation_id, length(metadata_json) AS bytes + FROM messages ORDER BY length(metadata_json) DESC LIMIT 1`).get() + if (!largest) throw new Error('No messages') + const snapshot = database.getConversation(largest.conversation_id) + const message = snapshot.messages.find(item => item.id === largest.id) + // Summary reads must remain small even while a large history is streaming. + database.saveLocalConversations([{ header: { ...snapshot, messages: undefined }, + messages: [{ ...message, state: 'streaming' }] }]) + const taskId = randomUUID() + database.createTask({ id: taskId, projectId: snapshot.projectId, + conversationId: snapshot.id, title: 'I/O probe', instructions: '', workMode: 'ask', visible: false }) + database.startExecutionTiming(taskId) + raw.exec('PRAGMA wal_checkpoint(TRUNCATE)') + console.log(JSON.stringify({ schema: raw.prepare('PRAGMA user_version').get(), metadataBytes: largest.bytes })) + const results = [] + for (const [variant, query, count] of [ + ['read-conversation', () => database.getConversation(snapshot.id), 1], + ['read-summary', () => database.listConversationSummaries([]), 1], + ['conversation-timing', () => database.getExecutionStats({ conversationId: snapshot.id }), 20], + ...(snapshot.projectId ? [['project-timing', () => database.getExecutionStats({ projectId: snapshot.projectId }), 20]] : []) + ]) { + const before = io() + const start = performance.now() + for (let index = 0; index < count; index++) query() + const elapsedMs = Math.round(performance.now() - start) + const after = io() + results.push({ variant, calls: count, elapsedMs, + readMiB: Number(((after.ReadTransferCount - before.ReadTransferCount) / 1024 / 1024).toFixed(3)), + writeMiB: Number(((after.WriteTransferCount - before.WriteTransferCount) / 1024 / 1024).toFixed(3)) }) + } + database.endExecutionTiming(taskId) + deepStrictEqual(database.getConversation(snapshot.id).messages, + snapshot.messages.map(item => item.id === message.id ? { ...item, state: 'streaming' } : item)) + console.table(results) +} finally { + database?.close() + rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }) +} diff --git a/docs/architecture/runtime-storage.md b/docs/architecture/runtime-storage.md index cd1114c6..92657e9e 100644 --- a/docs/architecture/runtime-storage.md +++ b/docs/architecture/runtime-storage.md @@ -10,20 +10,11 @@ recovery. `SubagentProgressStorage` caches event encoding state, not task statis ## Runtime Read Design -Execution statistics use one lazily started worker with a read-only connection. -The worker performs the existing evidence query inside a short read transaction. -Main supplies the current reply lease IDs and receives only scalar summaries. -There is no worker pool or duplicate persisted summary. The worker is terminated -when its owning database closes; errors reject reads, without a synchronous Main -fallback. In-memory test databases use the synchronous query because independent -connections cannot access the same private in-memory database. - -Derived caches must specify scope, capacity, lifetime and invalidation. Statistics -keep at most eight scopes for 30 seconds. SQLite `data_version` detects other -connections' commits; `total_changes()` detects writes on the owning connection. -Reads inside an existing transaction clear and bypass the cache, so rolled-back -values cannot escape. Closing the connection clears its cache. Live leases bypass -the cache because elapsed time changes without a database write. +Execution duration is persisted on running-state transitions in Main. Snapshots +read small indexed timing rows; they require no historical evidence query, worker +or result cache. Main owns current segments independently of renderer visibility. +The [duration design](../features/task-and-job/technical-design.md) defines the +state boundaries, restart handling and incomplete history. Read transactions end before replies are posted. Long-lived read transactions would prevent WAL checkpoints from reclaiming old pages. Each reader has its own @@ -36,9 +27,10 @@ rules, without introducing a general query protocol before a second use case exi Hidden task panels use React `Activity` to preserve state while scheduling hidden render work below visible updates. Panel factories run inside that boundary. Terminal and browser panels retain their existing mounted effects and sessions. -Statistics refresh only when the task center and document are visible. Results -must remain scoped and late responses from a previous selection must be discarded. +Duration snapshots refresh on Main change notifications while the task center is +open. The renderer ticks locally for open segments and fetches fresh snapshots on +reopening. Results remain scoped; superseded responses are discarded. The [task statistics design](../features/task-and-job/technical-design.md) owns the -query contract and evidence rules. Tests and timings demonstrate only their fixture +snapshot contract and timing rules. Tests and timings demonstrate only their fixture conditions; they do not establish a latency guarantee for arbitrary retained history. diff --git a/docs/features/task-and-job/README.md b/docs/features/task-and-job/README.md index 703a0532..2b0c00a8 100644 --- a/docs/features/task-and-job/README.md +++ b/docs/features/task-and-job/README.md @@ -9,9 +9,9 @@ 3. [Scheduled Task PRD](./scheduled-task-prd.md)：时间或事件触发的 Task。 4. [Goal Task PRD](./goal-task-prd.md)：围绕可验证结果有界推进的 Task。 5. [Job 与 Subjob PRD](./job-and-subjob-prd.md)：Task 内部串行、并行和委派执行。 -6. [技术设计](./technical-design.md)：统计查询契约、请求证据与不完整数据边界，以及立即创建的事务和入队规则。 +6. [技术设计](./technical-design.md)：运行状态计时、快照订阅与不完整历史边界，以及立即创建的事务和入队规则。 -统计读取与共享存储、worker 生命周期和 Renderer 调度的通用约束见[运行时存储架构](../../architecture/runtime-storage.md)。 +共享存储、后台读取和 Renderer 调度的通用约束见[运行时存储架构](../../architecture/runtime-storage.md)。 正式功能范围固定为当前项目：顶部显示当前会话消息总数和 Agent 累计回复时间，下方列出本项目全部会话的任务。创建默认立即执行并关联当前会话，可选择本项目其他会话、新会话或定时周期；产品与验收规则见 Task Center PRD 和 Scheduled Task PRD。 @@ -20,7 +20,7 @@ [当前会话统计 Demo](./hierarchy-demo.html) 用于评审任务中心中的紧凑统计区：上方显示当前会话的消息数与 Agent 累计回复时长，下方管理当前项目全部会话的任务及用时。切换会话只更新会话统计，任务范围固定为当前项目，没有全局或所有项目入口。支持状态筛选、定位所属会话，以及暂停、继续和取消任务的模拟操作。新建任务支持填写内容、关联本项目已有或新建会话、立即或定时执行；仅在页面内模拟，刷新后恢复样例。 配套 `hierarchy-demo.css` 与 `hierarchy-demo.js` 仅服务于该静态原型；使用演示数据，未接入产品。 应用中的项目任务标题右侧仅保留 `＋` 新建按钮；展开列表后直接显示状态选项，默认选中“进行中”，收起后保留所选状态。静态 Demo 的筛选入口仍为旧交互。 -时间为固定样例，Demo 从每次回复的起止秒数计算。正式功能按请求有效区间累计，包含轮内工具和审批等待、不计轮间空闲；任务用时按 `schedule_runs` 关联请求汇总，不按父任务生命周期计算。缺失历史有提示，远程缺少可靠证据时不推算用时，数据来源与边界以技术设计为准。 +时间为固定样例，Demo 从每次回复的起止秒数计算。正式功能由 Main 累计回复请求处于 `running` 状态的时间，暂停、等待审批和轮间空闲不计时；任务用时汇总到所属任务卡。旧历史和中断时未闭合的区间标为不完整，不查询历史消息重建。远程请求只统计桌面观察到的运行区间，断连期间的用时未知；具体规则见技术设计。 先阅读统一领域模型。其他功能文档不得重新定义 Task、Conversation、Job、Run 或 Subagent。 若实现与文档出现冲突，应先修正统一模型，再同步功能 PRD。 diff --git a/docs/features/task-and-job/task-center-prd.md b/docs/features/task-and-job/task-center-prd.md index aa9b8428..07813350 100644 --- a/docs/features/task-and-job/task-center-prd.md +++ b/docs/features/task-and-job/task-center-prd.md @@ -64,10 +64,11 @@ FR-TC1：当前项目是唯一范围，列表只显示 `projectId` 匹配的 Tas ### 4.1 当前会话统计 -FR-TC2：顶部紧凑展示当前会话消息总数与 Agent 累计回复时间。累计时间按每轮请求的有效 -执行区间相加，包含轮内工具执行和审批等待，不计轮间空闲；不重复累加子专家用时。 -缺失历史或时间证据不完整时提示“部分历史用时不可用”，已知用时不能表示为完整总量。 -远程执行缺少可靠时间证据时显示不可用或不完整，不用事件接收时间虚构远程用时。 +FR-TC2：顶部紧凑展示当前会话消息总数与 Agent 累计回复时间。Main 累计每轮请求处于 +`running` 状态的时间，包含该状态下的工具执行；暂停、等待审批和轮间空闲不计时， +子专家用时不重复累加。界面根据 Main 快照本地计时，切换会话或隐藏面板不影响累计。 +旧历史和中断时未闭合的区间提示“部分历史用时不可用”，不扫描历史消息重建时间。 +远程请求统计桌面观察到的运行区间，断连期间的用时标为不完整。 没有当前会话时提示选择会话，项目任务列表仍保留。 项目名称放在当前会话统计卡片内部的顶部。 @@ -156,7 +157,7 @@ FR-TC3：任务累计执行时间按其历次执行请求汇总，周期触发 - [ ] 删除 Task 默认保留 Conversation 和既有消息。 - [x] Smart Heartbeat 不进入 Task Center。 - [ ] FR-TC1：仅显示当前项目任务，切换会话不缩小列表；无项目时显示选择项目提示。 -- [ ] FR-TC2：消息总数与累计回复时间随当前会话更新；轮内工具和审批计时，轮间空闲不计时，历史缺失及远程证据不足有提示。 +- [ ] FR-TC2：消息总数与累计回复时间随当前会话更新；仅运行状态计时，暂停、等待审批和轮间空闲不计时，旧历史与中断区间有不完整提示。 - [ ] FR-TC3：多次执行汇总到同一任务，不按父 Task 生命周期计算用时。 - [x] FR-TC4：标题右侧仅保留新建；展开列表直接显示状态选项，默认进行中，收起后保留所选条件。 - [ ] FR-TC5：创建默认立即执行和当前会话，可选择本项目其他会话、新会话及定时周期，空名称从内容生成。 diff --git a/docs/features/task-and-job/technical-design.md b/docs/features/task-and-job/technical-design.md index e76ecff4..dc6c87e5 100644 --- a/docs/features/task-and-job/technical-design.md +++ b/docs/features/task-and-job/technical-design.md @@ -1,147 +1,95 @@ -# Execution Statistics Query +# Execution Duration -## Runtime Read Path +## Running State -WAL、连接所有权、内存缓存和后台读取的通用规则见 -[运行时存储架构](../../architecture/runtime-storage.md)。 +Main owns reply timing independently of the visible conversation or task panel. +Acquiring a reply lease starts a segment after the request task has been created. +Manual context compression uses a maintenance lease and creates no clock. Nested +subagents and stable schedule parents are excluded; their work is already inside +the top-level reply interval. Background and channel requests use the same lease. -`tasks:execution-stats` 在 Main 校验来源、范围和当前回复租约，然后经 -`AssistantDatabase.getExecutionStatsAsync` 调用独立的 `execution-stats-worker`。 -worker 使用只读连接和短读事务执行下文查询，不运行数据库初始化或任务恢复。 -查询错误返回调用方，关闭数据库时终止 worker 并拒绝未完成请求。 +Task transitions away from `running` close the segment, including `paused`, +`queued`, `waiting_approval`, completion, failure and cancellation. Resuming a +leased request opens another segment. Repeated `running` notifications leave the +original segment intact. Abort closes timing immediately, even when runtime +cleanup is still pending. Releasing a lease prevents subsequent task updates from +restarting its clock. Concurrent replies contribute their durations independently. -worker 内按项目或会话范围缓存结果，最多 8 项、30 秒；每次读取先检查 SQLite 连接版本。 -活动回复绕过缓存，事务内读取清空并绕过缓存，关闭连接清空缓存。写入提交后的下一次读取 -会重新计算，覆盖本地消息、远程事件、任务删除和恢复。统计数据仍以查询快照时刻为准。 +Remote requests measure Main's observed running state. The existing committed +terminal checkpoint closes their timing; event persistence, replay deduplication +and Agent ACK order are unchanged. Disconnection/recovery gaps are incomplete. +This duration does not claim to measure autonomous execution while the desktop +is disconnected. Application exit closes observed detached remote intervals +without cancelling the remote operation and marks the remaining duration unknown. -Renderer 按会话及所属项目、项目分别保留内存统计快照，切回已查询范围时同步显示缓存， -同时后台刷新；首次打开未缓存范围仍等待查询，缓存不跨应用重启保留，也不在前端推算时长。 -保留可见时 5 秒刷新以及 revision、focus、visibility 触发刷新；失败的范围清除缓存并显示 -不可用，旧范围的迟到结果丢弃。重复缓存结果保留对象引用，更新通过 `startTransition` -提交，避免无变化的轮询反复更新 App 的任务时长映射。任务面板使用 React `Activity` -保留状态并降低隐藏时的渲染优先级；面板工厂在该边界内部调用。浏览器和终端沿用原有挂载 -与 effect 生命周期。 +## Storage + +Schema 59 adds `execution_timing`: one row per timed request with its project, +conversation, task-card owner, accumulated milliseconds, optional running start, +and incomplete flag. Scope indexes keep snapshots separate from large task and +message payloads. A schedule run copies the stable task association already +stored in `parent_task_id`; repeated runs aggregate under that card. Task deletion +cascades to its timing row. + +Only state transitions write timing. Tokens, tools, display ticks and snapshot +reads do not update it. Startup drops any unclosed segment and marks it incomplete, +retaining previously accumulated time. It never counts application downtime or +uses recovery timestamps as execution endpoints. + +Migration marks existing conversations and visible task cards incomplete without +reconstructing their past duration. Imported legacy conversation snapshots are +also incomplete. No timing query reads historical messages, tools or task events. +The same migration adds +`messages_summary_idx ON messages(conversation_id, sequence, role, created_at)` +for conversation summaries. It performs no history conversion or `VACUUM`. ## Contract `window.goodbuddy.tasks.getExecutionStats(input)` invokes `tasks:execution-stats`. -`ExecutionStatsInput` accepts exactly one UUID scope: `{ conversationId }` or -`{ projectId }`. Main validates the scope and trusted sender. The database returns: +Input accepts exactly one UUID scope: `{ conversationId }` or `{ projectId }`. +Main validates the scope and trusted sender, then reads the small timing rows +synchronously. The old evidence query, cache, reader and statistics worker are +removed. ```ts interface ExecutionStats { - durationMs: number - requestCount: number - incompleteRequestCount: number - activeRequestCount: number - asOf: number // epoch milliseconds at query time + durationMs: number // closed segments plus open segments through asOf + runningCount: number // open segments in this scope + incomplete: boolean + asOf: number // Main epoch milliseconds taskDurations: Array<{ id: string durationMs: number - incompleteRequestCount: number + runningCount: number + incomplete: boolean }> } ``` -The project scope sums request evidence across its conversations and returns card -totals in `taskDurations`. A scheduled execution uses `requestId = schedule_runs.id`; -join its `schedule_id` to the stable task's `schedule_id` to aggregate repeated runs. -Request creation also saves that proven relation in `parent_task_id`. Before plan -deletion cascades its run records, bind older requests that still lack this link. -Statistics can then use the retained link to a visible `origin = schedule` parent, -even after the parent's `schedule_id` has been cleared. A shared conversation alone -does not establish ownership. Other -identified requests retain their own task ID. Visible tasks with no executions -have a zero entry. Conversation queries return an empty `taskDurations` array. -Message counts used to detect missing history stay in Main; no bodies are returned. +Conversation snapshots have an empty `taskDurations` array. Historical request +counts are no longer part of this API. -## Evidence And Boundaries +`onExecutionStatsChanged(listener)` returns an unsubscribe function. Main sends +`tasks:execution-stats-changed` after timing transitions or history deletion; +synchronous changes are coalesced into one notification. The renderer subscribes +before fetching its current conversation and project snapshots and refreshes on +notifications. It discards superseded responses and clears failed scopes. -- Query all retained scoped task records, including hidden chat requests, without - the visible-task list's pagination. Require a persisted message `request_id` - link, a reply event (`text`, `reasoning`, `tool`, `done`, `error`, `subagent`, - `generated-image`, or `question`) associated by `task_events.task_id`, a schedule - run association, or a current reply lease. Local message snapshots do not persist - request links, so local replies normally use event membership and kind. Do not - parse each stream/tool payload merely to repeat its owning request ID. -- Runtime status events also prove a reply when `requestId` matches the owning - task, `type` is `status`, and `message` is a string. Maintenance status events - such as `{ status: 'running' }` do not. Status-only interrupted replies retain - their last evidenced interval just like text/tool replies. -- Exclude stable schedule parents (`schedule_id` is set) and expert tasks - (`origin = subagent`). Scheduled reply requests remain eligible. Nested expert - execution is already inside the parent's elapsed interval and is not added again. -- Maintenance-only records do not establish reply identity. This excludes manual - context compression, heartbeat summaries, and unexecuted suggestions. A record - with no reply identity evidence cannot be classified; it is not counted. -- Sum intervals from `started_at` or later `running` status events to terminal - events/statuses. Pause/queue closes an interval; a later run starts another. - Repeated approval/running transitions do not reset the start. Tools and approval - waits inside a reply are included; idle gaps between replies/runs are excluded. -- `interrupted` may be written at application restart. Count only through the - preceding evidence, mark the request incomplete, and never use its recovery - `completed_at` as an execution endpoint. Missing or invalid timing also marks - the request incomplete; unknown time is not replaced with zero-length certainty. -- A non-aborted reply lease confirms current execution: an open local interval - with a reliable start contributes through `asOf` and remains complete while active. - Manual context compression leases are excluded. - Without a live lease, stop at the latest evidence and mark an open interval - incomplete. Renderer must query again after `asOf`; it does not extrapolate. -- Remote-recoverable records, tasks in persisted SSH project execution spaces, - and events with remote operation provenance are - incomplete and contribute no duration: their receipt timestamps cannot establish - remote execution intervals, particularly after replay. No Runtime, Agent, or - remote persistence behavior changes are required by this read-only query. - The SSH project check applies before the first provenance event, including - channel/delegation requests with `remote_recoverable = false`, so an unknown - remote duration is not first reported as complete local execution time. -- Completed, failed, or running schedule runs with no request history count as - incomplete, including in their stable task card. Pending runs do not count. -- Former schedule parents are never execution clocks, including after deletion. - A removed plan without retained execution ownership is reported as incomplete, - not a complete zero; its former lifetime cannot reconstruct missing runs. -- Retained top-level assistant messages establish missing-history evidence. - Explicit missing request links count as incomplete; unlinked snapshots use a - count comparison within each conversation/task, after covering linked replies. - Missing schedule runs already counted above cover their messages, avoiding a - second missing count. Expert-task messages and nested subagent content are not - counted again. A wholly missing reply contributes zero known duration and one - incomplete request, not a complete zero. -- Exclude the first local assistant message only when it has no request/task link - and exactly matches a shipped Chinese/English default greeting. Persisted - greetings have no separate marker. Other assistant-only history and identical - text later in the conversation remain eligible; do not discard all messages - before the first user turn. -- These counts are a lower bound when local snapshots have no request links; - fully deleted history cannot be reconstructed. `requestCount` includes replies - inferred from missing-history evidence. Concurrent top-level requests may - overlap in wall-clock time; their durations are summed independently. +The UI adds `max(0, now - asOf) * runningCount` locally once per second while an +open segment exists, including the corresponding per-card calculation. Hidden +panels unsubscribe and fetch fresh snapshots when reopened. There is no five-second +statistics polling, message-count revision trigger, or tick-time IPC. Main timing +continues while the panel is hidden or another conversation is selected. -The query materializes scoped task evidence once, then returns only status and -terminal events. Only status payloads use JSON extraction. Request membership and -remote provenance inspect scalar event columns; last-event and pre-interruption -timestamps use `(task_id, id)` index seeks. This preserves a tool/delta event as -the last reliable boundary without loading its body. Grouped assistant-message -counts read task metadata from message rows; event rows are not scanned for that count. -It adds no tables, columns, migration, or persisted timing totals. Stable task -lifecycle timestamps are never duration endpoints. +## Validation -## Validation Coverage - -The focused execution-statistics tests cover hidden requests, scope isolation, -approval waits, repeated intervals, restart interruption, remote replay, missing -start times, and history beyond visible-task pagination. IPC tests cover sender -and scope validation; preload tests cover typed forwarding. Renderer integration -must supply `getExecutionStats` in its `DesktopApi.tasks` mocks. - -The large-payload regression inspects the production query plan, measures paired -conversation/project polling on 3,000 events with approximately 94 MiB of payload, -and replaces stream/tool bodies with invalid JSON to prove they are not parsed. -Run it with `npx vitest run src/main/assistant/assistant-execution-stats.test.ts -t seeks --reporter=verbose --disableConsoleIntercept` -to print timings. These are synthetic warm-query measurements, not a guarantee -for arbitrarily large histories. Scalar event filtering and per-message metadata -work still grow with retained history; if that becomes measurable in deployment, -prefer polling only during active execution and refreshing on history/task changes. +`assistant-execution-stats.test.ts` exercises real SQLite state transitions, +restart interruption, remote checkpoints/replay, schedule ownership, scope +isolation, deletion, schema migration and read-only snapshots. It verifies that +snapshot SQL never reads messages or task events. IPC tests cover trusted scope +validation and production lease completion, failure and abort; preload tests cover +the subscription bridge. Hook tests verify local ticking without polling, hidden +panels, stale responses, and failed reads. ## Custom Task Creation @@ -163,7 +111,8 @@ conversation queue pump. App does not follow creation with `runNow`. Execution uses the existing readiness and conversation serialization rules. A runtime failure belongs to the created task; it does not reject creation or create another task. Refresh failures after a successful creation are notifications, not a reason -to retry creation. No schema migration or preload changes are required. +to retry creation. No schema migration or preload changes are required for this +creation command. `schedule-creation.test.ts` verifies transaction rollback and timer non-duplication with real SQLite. IPC tests exercise that database through the production queue diff --git a/docs/review/execution-timing-io-2026-10-04.md b/docs/review/execution-timing-io-2026-10-04.md new file mode 100644 index 00000000..fc31c4c3 --- /dev/null +++ b/docs/review/execution-timing-io-2026-10-04.md @@ -0,0 +1,67 @@ +# Execution Timing and Summary I/O + +## Decision + +Duration means time spent in the running state, including model and tool work. +It is not a reconstruction of historical execution evidence. Main starts a +clock for a reply lease, pauses it when the task stops running, and closes it +when the lease ends. Maintenance requests and nested subagents do not open +another top-level clock. The renderer advances the displayed value locally. + +Waiting for an answer or approval does not accrue time. Multiple pending +questions resume the clock only after the last one resolves. Recovered waiting +requests remain paused during reconnection. Startup preserves closed intervals, +discards unclosed intervals, and marks them incomplete; old history is not +backfilled by scanning messages or events. + +Schema 59 stores timing rows and adds a covering index for conversation +summaries. The old historical aggregation, its caches, dedicated reader/worker, +worker build entries, and five-second statistics polling have been removed. +The preliminary materialized-CTE fix is superseded by this replacement. + +## Summary Reads + +Summary-only requests now read counts, first role, and latest message time from +`messages_summary_idx`, without loading message bodies or metadata. Full +histories require explicit detail IDs. IPC still includes recovered requests +whose pending questions must be restored. The renderer's existing retained +history requests and refresh/settlement rules are preserved. + +## Measurement + +Windows logical process I/O counters, not physical disk throughput. Measurements +used an isolated copy of `dist/GoodBuddy-windows-x64/data`; the source database +was not opened. Its largest message metadata was 4,283,685 characters. A large +message was marked streaming in the copy to exercise the former implicit-detail +path. Initialization, migration, and fixture preparation are outside samples. + +| Operation | Before | After | +| --- | --- | --- | +| Project duration | One read: 1,224 MiB read, 740 MiB written, 2.7-4 s | Twenty reads: about 8 KiB read, no measured writes, about 1 ms total | +| Pure conversation summaries | One read: about 423 MiB, about 0.4 s | One read: 0.535 MiB, no measured writes, 9 ms | + +These compare the old historical statistic with the intentionally simpler +state-based statistic. They do not establish that all live application I/O +comes from these paths. Requested full histories still incur their normal cost. +The installed running application was not replaced during this work. + +Reproduce on Windows with a quiescent source data directory: + +```powershell +$env:GB_WPERF_DATA = 'D:\my_git\goodbuddy\dist\GoodBuddy-windows-x64\data' +node build/run-write-io-perf.cjs +``` + +The script copies the database, upgrades the copy, probes production methods, +checks retained message contents, and deletes the temporary copy. + +## Verification + +- State-transition, interruption, recovery, question, and notification tests. +- Covering-query-plan and explicit full-history regression tests. +- Assistant database: 120 tests passed. +- Reader, persistence, timing hook, and preload: 22 tests passed. +- Focused App duration/focus/refresh/summary tests: 44 passed. +- Full App: 350 passed, 3 failed in project creation workflows (discard dialog, + create invocation, and create-error display). These remain unresolved. +- Production bundle passed; removed worker is absent from its entry points. diff --git a/electron.vite.config.ts b/electron.vite.config.ts index edc217d6..db336ab7 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -132,7 +132,6 @@ export default defineConfig({ input: { index: resolve('src/main/index.ts'), 'assistant-storage-worker': resolve('src/main/assistant-storage-worker.ts'), - 'execution-stats-worker': resolve('src/main/execution-stats-worker.ts'), 'readonly-query-worker': resolve('src/main/readonly-query-worker.ts'), 'document-parse-worker': resolve('src/main/document-parse-worker.ts'), 'wechat-sidecar': resolve( diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 290ab6be..ea91efdc 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -674,6 +674,30 @@ describe('AssistantDatabase', () => { } finally { database.close() } }) + it('reads summaries from a covering index without loading active message metadata', async () => { + const database = await createDatabase() + try { + const id = randomUUID() + const message = { id: randomUUID(), role: 'assistant' as const, state: 'streaming' as const, + content: 'Partial reply', reasoning: 'x'.repeat(1_000_000), createdAt: 7 } + database.saveLocalConversations([{ header: { id, title: 'Active', updatedAt: 7 }, messages: [message] }]) + database.createTask({ id: randomUUID(), conversationId: id, title: 'Active', instructions: '', workMode: 'ask', status: 'running' }) + const raw = (database as unknown as { database: DatabaseSync }).database + const prepare = vi.spyOn(raw, 'prepare') + const summary = database.listConversationSummaries([]).find(item => item.id === id)! + expect(summary.messages).toEqual([]) + expect(summary.messageSummary).toEqual({ count: 1, firstRole: 'assistant', latestMessageAt: 7 }) + const statements = prepare.mock.calls.map(([sql]) => sql) + prepare.mockRestore() + expect(statements.some(sql => sql.includes('metadata_json'))).toBe(false) + const aggregate = statements.find(sql => sql.includes('count(*) AS count'))! + const plan = raw.prepare(`EXPLAIN QUERY PLAN ${aggregate}`).all().map(row => String(row.detail)) + expect(plan.some(detail => /(?:SCAN|SEARCH) m USING COVERING INDEX messages_summary_idx/.test(detail))).toBe(true) + expect(plan.some(detail => /SEARCH f USING COVERING INDEX messages_summary_idx/.test(detail))).toBe(true) + expect(database.listConversationSummaries([id]).find(item => item.id === id)?.messages).toEqual([message]) + } finally { database.close() } + }) + it.each([0, 1, 100, 500, 999])('keeps listTasks(%i) recent history plus old visible live tasks', async (limit) => { const database = await createDatabase() vi.useFakeTimers() diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 06d08db9..c2da3d03 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -5,7 +5,6 @@ import { knowledgeReferenceKey } from '../../shared/knowledge-reference' import { appendConversationQuestionBlock } from '../../shared/conversation-question-blocks' import { DatabaseSync } from 'node:sqlite' import { Worker } from 'node:worker_threads' -import { ExecutionStatsReader } from './execution-stats-reader' import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader' import { ActivityHistoryRepository, migrateActivityHistoryOrder } from './activity-history-repository' import { statSync } from 'node:fs' @@ -139,7 +138,7 @@ import { SUBAGENT_PROGRESS_STORAGE_SCHEMA_VERSION } from './subagent-progress-storage' -export const ASSISTANT_DATABASE_SCHEMA_VERSION = 58 +export const ASSISTANT_DATABASE_SCHEMA_VERSION = 59 /** * Messages startup recovery may have to end (schema 58): still streaming, or * metadata naming an unfinished tool/subagent (pending, running, queued) or @@ -1983,12 +1982,102 @@ function deleteProjectRecords( } } +/** Only Main leases can open a clock. Task state changes cannot revive a released lease. */ +class ExecutionTiming { + private readonly active = new Set() + private readonly listeners = new Set<() => void>() + private notificationPending = false + + constructor(private readonly database: DatabaseSync) {} + + onChanged(listener: () => void): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + changed(): void { + if (this.notificationPending) return + this.notificationPending = true + queueMicrotask(() => { + this.notificationPending = false + for (const listener of this.listeners) listener() + }) + } + + start(taskId: string): void { + const task = this.database.prepare(`SELECT project_id, conversation_id, parent_task_id, + visible, origin, schedule_id, status FROM tasks WHERE id = ?`).get(taskId) + if (!task) throw new Error('Cannot time a request before its task exists') + if (task.origin === 'subagent' || task.schedule_id !== null || (task.origin === 'schedule' && task.visible === 1)) return + const result = this.database.prepare(`INSERT INTO execution_timing + (task_id, project_id, conversation_id, owner_id, running_since) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(task_id) DO UPDATE SET running_since = excluded.running_since + WHERE execution_timing.running_since IS NULL AND excluded.running_since IS NOT NULL`) + .run(taskId, task.project_id ?? null, task.conversation_id ?? null, + task.parent_task_id ?? (task.visible ? taskId : null), task.status === 'running' ? Date.now() : null) + this.active.add(taskId) + if (result.changes) this.changed() + } + + state(taskId: string, status: string): void { + if (!this.active.has(taskId)) return + const running = status === 'running' + const now = Date.now() + const result = running + ? this.database.prepare(`UPDATE execution_timing SET running_since = ? + WHERE task_id = ? AND running_since IS NULL`).run(now, taskId) + : this.database.prepare(`UPDATE execution_timing + SET duration_ms = duration_ms + MAX(0, ? - running_since), running_since = NULL + WHERE task_id = ? AND running_since IS NOT NULL`).run(now, taskId) + if (result.changes) this.changed() + } + + end(taskId: string, incomplete = false): void { + this.state(taskId, 'stopped') + this.active.delete(taskId) + if (incomplete && this.database.prepare(`UPDATE execution_timing SET incomplete = 1 + WHERE task_id = ? AND incomplete = 0`).run(taskId).changes) this.changed() + } + + snapshot(scope: ExecutionStatsInput): ExecutionStats { + const conversation = 'conversationId' in scope + const scopeId = conversation ? scope.conversationId : scope.projectId + const column = conversation ? 'conversation_id' : 'project_id' + const asOf = Date.now() + const result: ExecutionStats = { + durationMs: 0, runningCount: 0, asOf, + incomplete: Boolean(this.database.prepare(`SELECT 1 FROM conversations + WHERE ${conversation ? 'id' : 'project_id'} = ? AND timing_incomplete = 1 LIMIT 1`).get(scopeId)), + taskDurations: [] + } + const cards = new Map() + for (const row of this.database.prepare(`SELECT owner_id, duration_ms, running_since, incomplete + FROM execution_timing WHERE ${column} = ?`).iterate(scopeId)) { + const runningCount = row.running_since === null ? 0 : 1 + const durationMs = Number(row.duration_ms) + (runningCount ? Math.max(0, asOf - Number(row.running_since)) : 0) + const incomplete = row.incomplete === 1 + result.durationMs += durationMs + result.runningCount += runningCount + result.incomplete ||= incomplete + if (!conversation && row.owner_id) { + const id = String(row.owner_id) + const card = cards.get(id) ?? { id, durationMs: 0, runningCount: 0, incomplete: false } + card.durationMs += durationMs + card.runningCount += runningCount + card.incomplete ||= incomplete + cards.set(id, card) + } + } + result.taskDurations = [...cards.values()] + return result + } +} + export class AssistantDatabase { - private static readonly executionStatsCacheTtlMs = 30_000 - private static readonly executionStatsCacheLimit = 8 private database?: DatabaseSync private activityHistoryRepository?: { database: DatabaseSync; repository: ActivityHistoryRepository } - private executionStatsReader?: ExecutionStatsReader + private executionTiming?: ExecutionTiming private readonlyReader?: ReadonlyQueryReader private checkpointWorker?: Worker private readonlyWorkerPath?: string @@ -2000,10 +2089,6 @@ export class AssistantDatabase { private channelEventWrites = 0 private channelOutboxWrites = 0 private modelUsageNotificationQueued = false - private readonly executionStatsCache = new Map< - string, - { value: ExecutionStats; cachedAt: number; dataVersion: number; totalChanges: number } - >() constructor( private readonly databasePath: string, @@ -2253,6 +2338,10 @@ export class AssistantDatabase { const recoveredAt = new Date().toISOString() database.exec('BEGIN IMMEDIATE') try { + // A process interruption has no known end time. Retain only closed segments. + database.exec(`UPDATE execution_timing SET running_since = NULL, incomplete = 1 + WHERE running_since IS NOT NULL`) + this.executionTiming = new ExecutionTiming(database) database .prepare( `UPDATE schedule_runs @@ -2428,12 +2517,10 @@ export class AssistantDatabase { close(): void { this.stopWalCheckpointWorker() - this.executionStatsReader?.close() - this.executionStatsReader = undefined + this.executionTiming = undefined this.readonlyReader?.close() this.readonlyReader = undefined this.foldedSearchConnection = undefined - this.executionStatsCache.clear() this.activityHistoryRepository = undefined this.database?.close() this.database = undefined @@ -2498,6 +2585,7 @@ export class AssistantDatabase { this.pendingMagicNoteCleanup.clear() this.reconcileMagicNoteFiles() this.options.onMagicNotesChanged?.() + this.executionTiming?.changed() } listProjects(includeArchived = false): AssistantProject[] { @@ -3027,6 +3115,7 @@ export class AssistantDatabase { deleteProjectRecords(database, projectId) database.exec('COMMIT') + this.executionTiming?.changed() } catch (error) { database.exec('ROLLBACK') throw error @@ -3081,33 +3170,27 @@ export class AssistantDatabase { const database = this.requireDatabase() const conversations = this.listConversationRows() type MessageSummaryRow = { - conversation_id: string; count: number; streaming: number | null + conversation_id: string; count: number firstRole: ConversationMessage['role'] | null; latestMessageAt: string | null } - const activeIds = new Set( - (database.prepare(`SELECT conversation_id FROM (${activeVisibleTaskSelect})`).all() as - { conversation_id: string }[]).map(row => row.conversation_id) - ) // Batched reads (PERF-15): one aggregate for every summary and one ordered // scan for every detailed conversation, instead of 1-2 statements each. const summaries = new Map() if (detailIds) { for (const row of database.prepare( `SELECT m.conversation_id AS conversation_id, count(*) AS count, - max(m.state = 'streaming') AS streaming, max(m.created_at) AS latestMessageAt, - (SELECT f.role FROM messages f WHERE f.conversation_id = m.conversation_id + (SELECT f.role FROM messages AS f INDEXED BY messages_summary_idx WHERE f.conversation_id = m.conversation_id ORDER BY f.sequence LIMIT 1) AS firstRole - FROM messages m JOIN conversations c ON c.id = m.conversation_id + FROM messages AS m INDEXED BY messages_summary_idx JOIN conversations c ON c.id = m.conversation_id WHERE c.status = 'active' GROUP BY m.conversation_id` ).all() as MessageSummaryRow[]) summaries.set(row.conversation_id, row) } const summaryOf = (id: string): MessageSummaryRow => summaries.get(id) ?? - { conversation_id: id, count: 0, streaming: null, firstRole: null, latestMessageAt: null } + { conversation_id: id, count: 0, firstRole: null, latestMessageAt: null } const detailed = conversations.filter(conversation => - !detailIds || detailIds.has(conversation.id) || activeIds.has(conversation.id) || - Boolean(summaryOf(conversation.id).streaming)) + !detailIds || detailIds.has(conversation.id)) const messagesByConversation = new Map() if (detailed.length > 0) { for (const row of database.prepare( @@ -3315,8 +3398,8 @@ export class AssistantDatabase { (id, project_id, runtime_selection_json, knowledge_retrieval_mode, context_state_json, work_mode, title, branch_source_conversation_id, branch_source_title, status, - created_at, updated_at) - VALUES (?, ?, ?, ?, ?, 'ask', ?, ?, ?, 'active', ?, ?)` + created_at, updated_at, timing_incomplete) + VALUES (?, ?, ?, ?, ?, 'ask', ?, ?, ?, 'active', ?, ?, 1)` ) const insertMessage = database.prepare( `INSERT INTO messages @@ -3356,6 +3439,7 @@ export class AssistantDatabase { } } database.exec('COMMIT') + this.executionTiming?.changed() } catch (error) { database.exec('ROLLBACK') throw error @@ -3736,6 +3820,7 @@ export class AssistantDatabase { ) .run(conversationId).changes === 1 database.exec('COMMIT') + if (deleted) this.executionTiming?.changed() return deleted } catch (error) { database.exec('ROLLBACK') @@ -4989,54 +5074,21 @@ export class AssistantDatabase { return rows.map(toTask) } - getExecutionStats( - input: ExecutionStatsInput, - activeRequestIds: ReadonlySet = new Set() - ): ExecutionStats { - const scope = executionStatsInputSchema.parse(input) - const cacheKey = 'conversationId' in scope - ? `conversation:${scope.conversationId}` - : `project:${scope.projectId}` - const database = this.requireDatabase() - // Never retain values read from a transaction that its caller can roll back. - if (database.isTransaction || activeRequestIds.size > 0) { - this.executionStatsCache.clear() - return this.readExecutionStatsSnapshot(scope, activeRequestIds) - } - const dataVersion = Number((database.prepare('PRAGMA data_version').get() as { data_version: number }).data_version) - const totalChanges = Number((database.prepare('SELECT total_changes() AS total_changes').get() as { total_changes: number }).total_changes) - const cached = activeRequestIds.size === 0 - ? this.executionStatsCache.get(cacheKey) - : undefined - if (cached && cached.dataVersion === dataVersion && cached.totalChanges === totalChanges && - Date.now() - cached.cachedAt < AssistantDatabase.executionStatsCacheTtlMs) { - this.executionStatsCache.delete(cacheKey) - this.executionStatsCache.set(cacheKey, cached) - return structuredClone(cached.value) - } - const result = this.readExecutionStatsSnapshot(scope, activeRequestIds) - this.executionStatsCache.delete(cacheKey) - this.executionStatsCache.set(cacheKey, { - value: structuredClone(result), cachedAt: Date.now(), dataVersion, totalChanges - }) - while (this.executionStatsCache.size > AssistantDatabase.executionStatsCacheLimit) { - this.executionStatsCache.delete(this.executionStatsCache.keys().next().value!) - } - return result + getExecutionStats(input: ExecutionStatsInput): ExecutionStats { + this.requireDatabase() + return this.executionTiming!.snapshot(executionStatsInputSchema.parse(input)) } - getExecutionStatsAsync( - input: ExecutionStatsInput, - activeRequestIds: ReadonlySet, - workerPath: string - ): Promise { - this.requireDatabase() - const scope = executionStatsInputSchema.parse(input) - if (this.databasePath === ':memory:') { - return Promise.resolve(this.getExecutionStats(scope, activeRequestIds)) - } - this.executionStatsReader ??= new ExecutionStatsReader(this.databasePath, workerPath) - return this.executionStatsReader.read(scope, activeRequestIds) + onExecutionStatsChanged(listener: () => void): () => void { + return this.executionTiming!.onChanged(listener) + } + + startExecutionTiming(taskId: string): void { + this.executionTiming!.start(taskId) + } + + endExecutionTiming(taskId: string, incomplete = false): void { + this.executionTiming!.end(taskId, incomplete) } openReadOnly(): void { @@ -5044,6 +5096,13 @@ export class AssistantDatabase { this.database = new DatabaseSync(this.databasePath, { readOnly: true, timeout: 5_000 }) } + /** Worker-only connection: no startup recovery, schema migration or Main object sharing. */ + openSupervisionWorker(): void { + if (this.database) throw new Error('Database already open') + this.database = new DatabaseSync(this.databasePath, { timeout: 5_000 }) + this.database.exec('PRAGMA foreign_keys = ON; PRAGMA wal_autocheckpoint = 0') + } + /** * Opens a connection used only to checkpoint the WAL from a worker thread * (PERF-15). It never writes rows; see enableWalCheckpointWorker. @@ -5145,246 +5204,6 @@ export class AssistantDatabase { return this.readonlyQueryReader() } - private readExecutionStatsSnapshot( - scope: ExecutionStatsInput, - activeRequestIds: ReadonlySet - ): ExecutionStats { - const database = this.requireDatabase() - if (database.isTransaction) return this.computeExecutionStats(scope, activeRequestIds) - database.exec('BEGIN') - try { - const result = this.computeExecutionStats(scope, activeRequestIds) - database.exec('COMMIT') - return result - } catch (error) { - database.exec('ROLLBACK') - throw error - } - } - - private computeExecutionStats( - scope: ExecutionStatsInput, - activeRequestIds: ReadonlySet - ): ExecutionStats { - const database = this.requireDatabase() - const result: ExecutionStats = { - durationMs: 0, requestCount: 0, incompleteRequestCount: 0, - activeRequestCount: 0, asOf: Date.now(), taskDurations: [] - } - const scopeId = 'conversationId' in scope ? scope.conversationId : scope.projectId - const scopeColumn = 'conversationId' in scope ? 'conversation_id' : 'project_id' - const durations = new Map() - const observed = new Set() - const coverage = new Map() - const bucket = (conversationId: string | null, owner: string | null): string => - JSON.stringify([conversationId, owner]) - const add = (id: string | null, durationMs: number, incomplete: number): void => { - if (!id || !('projectId' in scope)) return - const entry = durations.get(id) ?? { id, durationMs: 0, incompleteRequestCount: 0 } - entry.durationMs += durationMs - entry.incompleteRequestCount += incomplete - durations.set(id, entry) - } - // Materialize task-level evidence once. Stream/tool payloads are never parsed; - // task_events_task_idx supplies request membership and boundary seeks. - const rows = database.prepare(` - WITH scoped AS MATERIALIZED ( - SELECT t.id, t.conversation_id, t.visible, t.started_at, t.remote_recoverable, - COALESCE(parent.id, retained.id) AS schedule_task_id, - space.kind = 'ssh' AS ssh_project, - EXISTS (SELECT 1 FROM messages m - WHERE m.conversation_id = t.conversation_id AND m.request_id = t.id) AS linked, - EXISTS (SELECT 1 FROM task_events proof WHERE proof.task_id = t.id - AND proof.kind IN ('text', 'reasoning', 'tool', 'done', 'error', - 'subagent', 'generated-image', 'question')) AS reply_event, - EXISTS (SELECT 1 FROM task_events remote WHERE remote.task_id = t.id - AND remote.remote_operation_id IS NOT NULL) AS remote_event, - (SELECT tail.created_at FROM task_events tail WHERE tail.task_id = t.id - ORDER BY tail.id DESC LIMIT 1) AS last_at - FROM tasks t - LEFT JOIN schedule_runs sr ON sr.id = t.id - LEFT JOIN tasks parent ON parent.schedule_id = sr.schedule_id - LEFT JOIN tasks retained ON retained.id = t.parent_task_id - AND retained.origin = 'schedule' AND retained.visible = 1 - LEFT JOIN project_execution_spaces space ON space.project_id = t.project_id - WHERE t.${scopeColumn} = ? - AND t.schedule_id IS NULL AND t.origin != 'subagent' - AND NOT (t.origin = 'schedule' AND t.visible = 1) - ) - SELECT scoped.*, e.kind, e.created_at, - CASE WHEN e.kind = 'status' - THEN json_extract(e.payload_json, '$.status') END AS event_status, - CASE WHEN e.kind = 'status' THEN - json_extract(e.payload_json, '$.requestId') = scoped.id - AND json_extract(e.payload_json, '$.type') = 'status' - AND json_type(e.payload_json, '$.message') = 'text' - END AS runtime_status, - CASE WHEN e.kind = 'status' THEN ( - SELECT previous.created_at FROM task_events previous - WHERE previous.task_id = scoped.id AND previous.id < e.id - ORDER BY previous.id DESC LIMIT 1 - ) END AS previous_at - FROM scoped LEFT JOIN task_events e ON e.task_id = scoped.id - AND e.kind IN ('status', 'done', 'error') - ORDER BY scoped.id, e.id - `).iterate(scopeId) - let task: { - id: string; proven: boolean; remote: boolean; incomplete: boolean - start: number | undefined; last: number | undefined; duration: number - conversationId: string | null; owner: string | null - lastEvidence: number - } | undefined - const finish = (): void => { - if (!task?.proven) return - result.requestCount++ - observed.add(task.id) - const key = bucket(task.conversationId, task.owner) - coverage.set(key, (coverage.get(key) ?? 0) + 1) - if (activeRequestIds.has(task.id)) result.activeRequestCount++ - if (task.remote) { - task.duration = 0 - task.incomplete = true - } else if (task.start !== undefined) { - if (Number.isFinite(task.lastEvidence) && task.lastEvidence >= task.start && task.lastEvidence <= result.asOf) { - task.last = Math.max(task.last ?? task.start, task.lastEvidence) - } - const live = activeRequestIds.has(task.id) && task.start <= result.asOf - task.duration += Math.max(0, (live ? result.asOf : task.last ?? task.start) - task.start) - task.incomplete ||= !live - } - result.durationMs += task.duration - if (task.incomplete) result.incompleteRequestCount++ - add(task.owner ?? task.id, task.duration, Number(task.incomplete)) - } - for (const raw of rows) { - const row = raw as unknown as { - id: string; started_at: string | null; remote_recoverable: number - linked: number; kind: string | null; created_at: string | null - reply_event: number; remote_event: number; last_at: string | null; previous_at: string | null - event_status: string | null; runtime_status: number | null; ssh_project: number | null - conversation_id: string | null; visible: number; schedule_task_id: string | null - } - if (task?.id !== row.id) { - finish() - const start = Date.parse(row.started_at ?? '') - task = { - id: row.id, proven: row.linked === 1 || row.reply_event === 1 || row.schedule_task_id !== null || activeRequestIds.has(row.id), - remote: row.remote_recoverable === 1 || row.remote_event === 1 || row.ssh_project === 1, - lastEvidence: Date.parse(row.last_at ?? ''), - conversationId: row.conversation_id, - owner: row.schedule_task_id ?? (row.visible === 1 ? row.id : null), - incomplete: !Number.isFinite(start), - start: Number.isFinite(start) ? start : undefined, - last: Number.isFinite(start) ? start : undefined, duration: 0 - } - } - const current = task! - current.proven ||= row.runtime_status === 1 - if (row.kind === null) continue - const status = row.event_status - // Recovery writes interrupted at restart, not when execution actually stopped. - if (status === 'interrupted') { - if (current.start !== undefined) { - const previous = Date.parse(row.previous_at ?? '') - if (Number.isFinite(previous) && previous >= current.start && previous <= result.asOf) { - current.last = Math.max(current.last ?? current.start, previous) - } - current.duration += Math.max(0, (current.last ?? current.start) - current.start) - } - current.start = undefined - current.last = undefined - current.incomplete = true - continue - } - const time = Date.parse(row.created_at ?? '') - if (!Number.isFinite(time) || time > result.asOf || (current.last !== undefined && time < current.last)) { - current.incomplete = true - continue - } - if (status === 'running' && current.start === undefined) current.start = time - const closes = row.kind === 'done' || row.kind === 'error' || - ['completed', 'failed', 'cancelled', 'paused', 'queued', 'idle'].includes(status ?? '') - if (closes && current.start !== undefined) { - current.duration += time - current.start - current.start = undefined - } - current.last = time - } - finish() - - // A schedule run has an independent request ID. Its stable task is never a clock. - for (const row of database.prepare(` - SELECT t.id, t.conversation_id, t.origin, t.schedule_id, - sr.id AS run_id, sr.status AS run_status - FROM tasks t LEFT JOIN schedule_runs sr ON sr.schedule_id = t.schedule_id - WHERE t.${scopeColumn} = ? AND t.visible = 1 AND t.origin != 'subagent' - ORDER BY t.id, sr.id - `).iterate(scopeId)) { - const id = row.id as string - add(id, 0, 0) - const key = bucket(row.conversation_id as string | null, id) - // A removed plan without retained execution links cannot establish a zero. - if (row.origin === 'schedule' && row.schedule_id === null && !coverage.has(key)) { - result.requestCount++ - result.incompleteRequestCount++ - add(id, 0, 1) - coverage.set(key, 1) - } - if (row.run_id && ['completed', 'failed', 'running'].includes(row.run_status as string) && !observed.has(row.run_id as string)) { - observed.add(row.run_id as string) - result.requestCount++ - result.incompleteRequestCount++ - add(id, 0, 1) - coverage.set(key, (coverage.get(key) ?? 0) + 1) - } - } - // Local snapshots lack request IDs. Compare top-level assistant rows per - // conversation/task, never nested subagent blocks or child-task messages. - const messages = database.prepare(` - SELECT m.conversation_id, m.request_id, - COALESCE(parent.id, retained.id, json_extract(m.metadata_json, '$.task.id'), - CASE WHEN mt.visible = 1 THEN mt.id END) AS owner, - COUNT(*) AS count - FROM messages m JOIN conversations c ON c.id = m.conversation_id - LEFT JOIN tasks mt ON mt.id = m.request_id - LEFT JOIN schedule_runs sr ON sr.id = m.request_id - LEFT JOIN tasks parent ON parent.schedule_id = sr.schedule_id - LEFT JOIN tasks retained ON retained.id = mt.parent_task_id - AND retained.origin = 'schedule' AND retained.visible = 1 - WHERE ${'conversationId' in scope ? 'm.conversation_id' : 'c.project_id'} = ? - AND m.role = 'assistant' - AND NOT (c.channel IS NULL AND m.sequence = 0 AND m.request_id IS NULL - AND json_extract(m.metadata_json, '$.task.id') IS NULL - AND m.content IN (?, ?)) - AND COALESCE(mt.origin, '') != 'subagent' - AND NOT EXISTS (SELECT 1 FROM tasks child WHERE child.origin = 'subagent' - AND child.id = json_extract(m.metadata_json, '$.task.id')) - GROUP BY m.conversation_id, owner, m.request_id - ORDER BY m.request_id IS NULL - `).all( - scopeId, - // Persisted greetings have no marker. Match only shipped default copy, - // not arbitrary assistant-only/imported history or localized prefixes. - '你好，我是 GoodBuddy。你可以直接向我提问、添加本地文件，或使用知识库整理和检索信息。需要我操作文件或调用工具时，请选择合适的 Agent Runtime 和工作模式。', - 'Hi, I’m GoodBuddy. Ask me a question, add local files, or use your knowledge base to organize and retrieve information. When you want me to operate on files or use tools, choose the appropriate Agent Runtime and work mode.' - ) - for (const row of messages) { - const owner = row.owner as string | null - const key = bucket(row.conversation_id as string, owner) - const available = coverage.get(key) ?? 0 - const linked = row.request_id as string | null - const count = linked ? 1 : row.count as number - const covered = linked ? Number(observed.has(linked)) : Math.min(count, available) - coverage.set(key, Math.max(0, available - covered)) - const missing = count - covered - result.requestCount += missing - result.incompleteRequestCount += missing - add(owner, 0, missing) - } - result.taskDurations = [...durations.values()] - return result - } - /** Activity history storage and queries (PERF-15); the SQL lives in the repository. */ activityHistory(): ActivityHistoryRepository { const database = this.requireDatabase() @@ -5986,6 +5805,7 @@ export class AssistantDatabase { `UPDATE conversations SET updated_at = ? WHERE id = ?` ) .run(now, owner.conversation_id) + this.executionTiming?.state(taskId, status) database.exec('COMMIT') } catch (error) { database.exec('ROLLBACK') @@ -6277,6 +6097,7 @@ export class AssistantDatabase { throw new Error('任务不存在') } this.appendTaskEvent(taskId, 'status', { status, error }) + this.executionTiming?.state(taskId, status) } resolveAssistantSuggestionTask( @@ -6860,6 +6681,7 @@ export class AssistantDatabase { if (taskUpdate.changes !== 1) { throw new Error('远程终态无法更新所属任务') } + this.executionTiming?.state(event.taskId, taskStatus) } private insertRemoteTaskEventOnce( @@ -12400,6 +12222,32 @@ export class AssistantDatabase { COMMIT; `) } + if (version.user_version < 59) { + database.exec('BEGIN IMMEDIATE') + try { + const columns = database.prepare('PRAGMA table_info(conversations)').all() + if (!columns.some(column => column.name === 'timing_incomplete')) { + database.exec(` + CREATE TABLE execution_timing ( + task_id TEXT PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE, + project_id TEXT, conversation_id TEXT, owner_id TEXT, + duration_ms INTEGER NOT NULL DEFAULT 0, + running_since INTEGER, + incomplete INTEGER NOT NULL DEFAULT 0 + ); + CREATE INDEX execution_timing_project ON execution_timing(project_id); + CREATE INDEX execution_timing_conversation ON execution_timing(conversation_id); + INSERT INTO execution_timing (task_id, project_id, conversation_id, owner_id, incomplete) + SELECT id, project_id, conversation_id, id, 1 FROM tasks WHERE visible = 1 AND origin != 'subagent'; + ALTER TABLE conversations ADD COLUMN timing_incomplete INTEGER NOT NULL DEFAULT 0; + UPDATE conversations SET timing_incomplete = 1; + CREATE INDEX conversations_timing_incomplete ON conversations(project_id) WHERE timing_incomplete = 1; + CREATE INDEX messages_summary_idx ON messages(conversation_id, sequence, role, created_at); + `) + } + database.exec('PRAGMA user_version = 59; COMMIT;') + } catch (error) { database.exec('ROLLBACK'); throw error } + } } supervisionExperiences(): SupervisionExperienceStore { diff --git a/src/main/assistant/assistant-execution-stats.test.ts b/src/main/assistant/assistant-execution-stats.test.ts index 128b1987..bde87f9b 100644 --- a/src/main/assistant/assistant-execution-stats.test.ts +++ b/src/main/assistant/assistant-execution-stats.test.ts @@ -1,445 +1,257 @@ import { randomUUID } from 'node:crypto' -import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { DatabaseSync } from 'node:sqlite' -import { performance } from 'node:perf_hooks' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { AssistantDatabase } from './assistant-database' -describe('execution duration evidence', () => { +describe('persisted running-state duration', () => { let directory: string + let path: string let database: AssistantDatabase let projectId: string let conversationId: string - const epoch = Date.parse('2026-09-17T00:00:00Z') - const at = (seconds: number): void => { vi.setSystemTime(epoch + seconds * 1000) } - const request = (overrides: Partial[0]> = {}): string => { - const id = randomUUID() - database.createTask({ id, projectId, conversationId, title: 'Reply', instructions: 'Reply', workMode: 'ask', visible: false, ...overrides }) + let now: number + const raw = (): DatabaseSync => (database as unknown as { database: DatabaseSync }).database + const stats = () => database.getExecutionStats({ conversationId }) + const task = (options: Partial[0]> = {}): string => { + const id = options.id ?? randomUUID() + database.createTask({ id, projectId, conversationId, title: 'Reply', instructions: 'Reply', + workMode: 'ask', visible: false, ...options }) return id } - const text = (id: string): void => database.appendTaskEvent(id, 'text', { requestId: id, type: 'text', delta: 'Reply' }) - beforeEach(async () => { - directory = await mkdtemp(join(tmpdir(), 'goodbuddy-execution-stats-')) - database = new AssistantDatabase(join(directory, 'assistant.sqlite')) + directory = await mkdtemp(join(tmpdir(), 'goodbuddy-timing-')) + path = join(directory, 'assistant.sqlite') + database = new AssistantDatabase(path) database.initialize(directory) projectId = database.listProjects()[0]!.id conversationId = randomUUID() - vi.useFakeTimers({ toFake: ['Date'] }) - at(0) - }) - afterEach(async () => { - database.close() - vi.useRealTimers() - await rm(directory, { recursive: true, force: true }) + now = Date.now() + vi.spyOn(Date, 'now').mockImplementation(() => now) + database.saveLocalConversations([{ header: { id: conversationId, projectId, title: 'Timing', updatedAt: now }, messages: [] }]) }) + afterEach(async () => { database.close(); vi.restoreAllMocks(); await rm(directory, { recursive: true, force: true }) }) - it('sums hidden replies without round idle, counts approval/tool time, and scopes projects', () => { - const first = request() - text(first) - at(2) - database.updateTaskStatus(first, 'waiting_approval') - at(8) - database.updateTaskStatus(first, 'running') - at(10) - database.updateTaskStatus(first, 'completed') - at(100) - const second = request() - text(second) - at(105) - database.updateTaskStatus(second, 'failed') - const anotherConversation = request({ conversationId: randomUUID() }) - text(anotherConversation) - at(108) - database.updateTaskStatus(anotherConversation, 'cancelled') - const unrelated = request({ projectId: undefined, conversationId: randomUUID() }) - text(unrelated) - at(120) - database.updateTaskStatus(unrelated, 'completed') - expect(database.listTasks()).toEqual([]) - expect(database.getExecutionStats({ conversationId })).toEqual({ - durationMs: 15000, requestCount: 2, incompleteRequestCount: 0, - activeRequestCount: 0, asOf: epoch + 120000, taskDurations: [] - }) - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 18000, requestCount: 3 }) - expect(database.getExecutionStats({ conversationId: randomUUID() })).toMatchObject({ durationMs: 0, requestCount: 0 }) - }) - - it('invalidates cached scopes for local messages, remote events and external deletion', () => { - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(0) - database.saveLocalConversations([{ - header: { id: conversationId, projectId, title: 'History', updatedAt: epoch }, - messages: [{ id: randomUUID(), role: 'assistant', content: 'Retained', state: 'complete', createdAt: epoch }] - }]) - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(1) - const id = request() - database.getExecutionStats({ projectId }) - database.appendRemoteTaskEventOnce({ taskId: id, bindingId: randomUUID(), operationId: randomUUID(), semanticSequence: '1', eventIndex: 0, kind: 'done', payload: { requestId: id, type: 'done' } }) - expect(database.getExecutionStats({ projectId }).incompleteRequestCount).toBe(1) - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - try { - raw.prepare('DELETE FROM messages WHERE conversation_id = ?').run(conversationId) - raw.prepare('DELETE FROM tasks WHERE id = ?').run(id) - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(0) - expect(database.getExecutionStats({ projectId }).requestCount).toBe(0) - } finally { raw.close() } - }) - - it('does not cache transaction snapshots across rollback, or serve a closed connection', () => { - const connection = (database as unknown as { database: DatabaseSync }).database - const id = request() - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(0) - connection.exec('BEGIN') - text(id) - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(1) - connection.exec('ROLLBACK') - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(0) - database.close() - expect(() => database.getExecutionStats({ conversationId })).toThrow() - database.initialize(directory) - expect(database.getExecutionStats({ conversationId }).requestCount).toBe(0) - }) - - it('bounds cache scopes, expires idle entries and isolates returned objects', () => { - const prepare = vi.spyOn(DatabaseSync.prototype, 'prepare') - const reads = (): number => prepare.mock.calls.filter(([sql]) => sql.includes('WITH scoped AS MATERIALIZED')).length - try { - const first = database.getExecutionStats({ projectId }) - first.requestCount = 999 - expect(database.getExecutionStats({ projectId }).requestCount).toBe(0) - expect(reads()).toBe(1) - for (let index = 0; index < 8; index++) database.getExecutionStats({ conversationId: randomUUID() }) - database.getExecutionStats({ projectId }) - expect(reads()).toBe(10) - at(31) - database.getExecutionStats({ projectId }) - expect(reads()).toBe(11) - } finally { prepare.mockRestore() } - }) - - it('excludes stable schedules, nested experts, and status-only maintenance but includes scheduled reply requests', () => { - const schedule = database.createSchedule({ projectId, title: 'Schedule', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - database.updateTaskStatus(schedule.taskId, 'running') - text(schedule.taskId) - const child = request({ origin: 'subagent', parentTaskId: schedule.taskId }) - text(child) - const compact = request({ title: 'Compressed context' }) - const actual = request({ conversationId: schedule.conversationId, parentTaskId: schedule.taskId }) - text(actual) - at(10) - for (const id of [schedule.taskId, child, compact, actual]) database.updateTaskStatus(id, 'completed') - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 10000, requestCount: 1, incompleteRequestCount: 0 }) - }) - - it('uses message request identity when a reply has no runtime events and does not use message timestamps', () => { - const id = request() - database.saveLocalConversations([{ - header: { id: conversationId, projectId, title: 'Chat', updatedAt: epoch }, - messages: [{ id: randomUUID(), role: 'assistant', content: '', state: 'error', createdAt: epoch - 86400000 }] - }]) - // Only persisted request links are evidence; local snapshot saves do not supply them. - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - raw.prepare('UPDATE messages SET request_id = ? WHERE conversation_id = ?').run(id, conversationId) - raw.close() - at(3) - database.updateTaskStatus(id, 'failed') - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 3000, requestCount: 1, incompleteRequestCount: 0 }) - }) - - it('keeps only evidenced time for open and restart-interrupted replies', () => { - const id = request() - at(5) - text(id) - at(100) - expect(database.getExecutionStats({ conversationId }, new Set([id]))).toMatchObject({ durationMs: 100000, requestCount: 1, incompleteRequestCount: 0, activeRequestCount: 1 }) - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 5000, incompleteRequestCount: 1, activeRequestCount: 0 }) - database.close() - at(86400) - database.initialize(directory) - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 5000, requestCount: 1, incompleteRequestCount: 1, activeRequestCount: 0 }) - // A later resumed segment cannot include the shutdown/restart gap. - at(86500) + it('accumulates only running segments across pause, approval, queue, resume and completion', () => { + const id = task({ visible: true }) + database.startExecutionTiming(id) + now += 2000 database.updateTaskStatus(id, 'running') - at(86510) - database.updateTaskStatus(id, 'completed') - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 15000, incompleteRequestCount: 1 }) - }) - - it('segments repeated runs and pauses without using the final completed_at span', () => { - const id = request() - text(id) - at(5) + now += 1000 database.updateTaskStatus(id, 'paused') - at(100) + now += 90_000 + expect(stats()).toMatchObject({ durationMs: 3000, runningCount: 0, incomplete: false }) database.updateTaskStatus(id, 'running') - at(107) - database.updateTaskStatus(id, 'completed') - at(200) + now += 4000 + database.updateTaskStatus(id, 'waiting_approval') + now += 60_000 + database.updateTaskStatus(id, 'queued') database.updateTaskStatus(id, 'running') - at(203) + now += 2000 + expect(stats()).toMatchObject({ durationMs: 9000, runningCount: 1 }) database.updateTaskStatus(id, 'completed') - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 15000, requestCount: 1, incompleteRequestCount: 0 }) + database.endExecutionTiming(id) + now += 60_000 + expect(stats()).toMatchObject({ durationMs: 9000, runningCount: 0 }) + expect(database.getExecutionStats({ projectId }).taskDurations).toEqual([ + { id, durationMs: 9000, runningCount: 0, incomplete: false } + ]) }) - it('does not substitute remote replay receipt time or unknown starts for actual execution time', () => { - const remote = request() - at(86400) - database.appendRemoteTaskEventOnce({ taskId: remote, bindingId: randomUUID(), operationId: randomUUID(), semanticSequence: '1', eventIndex: 0, kind: 'done', payload: { requestId: remote, type: 'done' } }) - database.updateTaskStatus(remote, 'completed') - const unknown = request() - text(unknown) - at(86410) - database.updateTaskStatus(unknown, 'completed') - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - raw.prepare('UPDATE tasks SET started_at = NULL WHERE id = ?').run(unknown) - raw.close() - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 0, requestCount: 2, incompleteRequestCount: 2 }) + it.each(['failed', 'cancelled', 'interrupted'] as const)('closes on %s and cannot resume after lease release', (status) => { + const id = task() + database.startExecutionTiming(id) + now += 3000 + database.updateTaskStatus(id, status) + database.endExecutionTiming(id) + now += 30_000 + database.updateTaskStatus(id, 'running') + expect(stats()).toMatchObject({ durationMs: 3000, runningCount: 0 }) }) - it('includes all retained requests beyond list pagination', () => { - for (let index = 0; index < 105; index++) { - at(index * 10) - const id = request() - text(id) - at(index * 10 + 1) - database.updateTaskStatus(id, 'completed') - } - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 105000, requestCount: 105, incompleteRequestCount: 0 }) + it.each(['waiting_approval', 'paused', 'queued'] as const)('registers a new %s lease without opening a segment until running', (status) => { + const id = task({ status }) + database.startExecutionTiming(id) + now += 60_000 + expect(stats()).toMatchObject({ durationMs: 0, runningCount: 0 }) + database.updateTaskStatus(id, 'running') + now += 3000 + database.endExecutionTiming(id) + expect(stats()).toMatchObject({ durationMs: 3000, runningCount: 0 }) }) - it('groups real schedule request IDs into stable cards across runs without parent lifecycle time', () => { - const schedule = database.createSchedule({ projectId, title: 'Schedule', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - const idle = database.createSchedule({ projectId, conversationId: schedule.conversationId, title: 'Idle', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch + 86400000).toISOString() }) - for (const start of [100, 1000]) { - at(start) - const queued = database.queueScheduleNow(schedule.id) - database.claimConversationQueueItem(schedule.conversationId) - database.completeConversationUserQueueItem(queued.id) - // Production agentRun uses scheduleRunId as requestId, without parentTaskId. - const id = queued.scheduleRunId! - request({ id, conversationId: schedule.conversationId }) - database.updateTaskStatus(schedule.taskId, 'running') - text(id) - at(start + 10) - database.updateTaskStatus(id, 'completed') - database.completeTaskScheduleRun(id) - } - const live = database.queueScheduleNow(schedule.id) - database.claimConversationQueueItem(schedule.conversationId) - request({ id: live.scheduleRunId!, conversationId: schedule.conversationId }) - text(live.scheduleRunId!) - at(1025) - const stats = database.getExecutionStats({ projectId }, new Set([live.scheduleRunId!])) - expect(stats).toMatchObject({ durationMs: 35000, requestCount: 3, incompleteRequestCount: 0, activeRequestCount: 1 }) - expect(stats.taskDurations).toEqual(expect.arrayContaining([ - { id: schedule.taskId, durationMs: 35000, incompleteRequestCount: 0 }, - { id: idle.taskId, durationMs: 0, incompleteRequestCount: 0 } - ])) - expect(stats.taskDurations.some(item => item.id === live.scheduleRunId)).toBe(false) + it('stops immediately on lease abort even if task cleanup is delayed', () => { + const id = task() + database.startExecutionTiming(id) + now += 5000 + database.endExecutionTiming(id) + now += 60_000 + database.updateTaskStatus(id, 'running') + database.updateTaskStatus(id, 'cancelled') + expect(stats()).toMatchObject({ durationMs: 5000, runningCount: 0 }) }) - it.each([false, true])('retains schedule execution totals after plan deletion (repair old link: %s)', (oldLink) => { - const schedule = database.createSchedule({ projectId, title: 'Removable plan', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - at(100) - const run = database.queueScheduleNow(schedule.id) - database.claimConversationQueueItem(schedule.conversationId) - const task = database.createTask({ id: run.scheduleRunId!, projectId, conversationId: schedule.conversationId, title: 'Run', instructions: 'Reply', workMode: 'ask', visible: false }) - expect(task.parentTaskId).toBe(schedule.taskId) - text(task.id) - at(110) - database.updateTaskStatus(task.id, 'completed') - database.completeTaskScheduleRun(task.id) - if (oldLink) { - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - raw.prepare('UPDATE tasks SET parent_task_id = NULL WHERE id = ?').run(task.id) - raw.close() - } - const before = database.getExecutionStats({ projectId }) - at(10000) - database.removeSchedule(schedule.id) - expect(database.getExecutionStats({ projectId })).toEqual({ ...before, asOf: epoch + 10000000 }) - expect(before.taskDurations).toEqual([{ id: schedule.taskId, durationMs: 10000, incompleteRequestCount: 0 }]) + it('does not write or read history on snapshots, tokens, or repeated running notifications', async () => { + const id = task() + database.startExecutionTiming(id) + await Promise.resolve() + const changed = vi.fn() + const unsubscribe = database.onExecutionStatsChanged(changed) + const before = raw().prepare('SELECT total_changes() AS n').get()!.n + const prepare = vi.spyOn(raw(), 'prepare') + for (let tick = 0; tick < 10; tick++) { now += 1000; stats(); database.getExecutionStats({ projectId }) } + expect(raw().prepare('SELECT total_changes() AS n').get()!.n).toBe(before) + expect(prepare.mock.calls.map(([sql]) => sql).join('\n')).not.toMatch(/\b(messages|task_events|tools)\b/) + database.appendTaskEvent(id, 'text', { requestId: id, type: 'text', delta: 'Token' }) + database.updateTaskStatus(id, 'running') + await Promise.resolve() + expect(changed).not.toHaveBeenCalled() + database.updateTaskStatus(id, 'paused') + await Promise.resolve() + expect(changed).toHaveBeenCalledOnce() + unsubscribe() + }) + + it('retains closed intervals and marks the open interval unknown after restart', () => { + const id = task() + database.startExecutionTiming(id) + now += 2000 + database.updateTaskStatus(id, 'paused') + database.updateTaskStatus(id, 'running') + now += 5000 database.close() + now += 86_400_000 database.initialize(directory) - expect(database.getExecutionStats({ projectId }).taskDurations).toEqual(before.taskDurations) + expect(stats()).toMatchObject({ durationMs: 2000, runningCount: 0, incomplete: true }) + database.startExecutionTiming(id) + database.updateTaskStatus(id, 'running') + now += 3000 + database.endExecutionTiming(id) + expect(stats()).toMatchObject({ durationMs: 5000, runningCount: 0, incomplete: true }) }) - it('does not claim a complete zero for an already removed plan with no retained execution relation', () => { - const schedule = database.createSchedule({ projectId, title: 'Old removed plan', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - database.removeSchedule(schedule.id) - expect(database.getExecutionStats({ projectId }).taskDurations).toEqual([{ id: schedule.taskId, durationMs: 0, incompleteRequestCount: 1 }]) - }) - - it('recognizes runtime status-only replies but not maintenance statuses at restart', () => { - const id = request() - const maintenance = request({ title: 'Maintenance' }) - at(5) - database.appendTaskEvent(id, 'status', { requestId: id, type: 'status', message: 'Processing request' }) - database.updateTaskStatus(maintenance, 'running') - database.close() - at(10000) - database.initialize(directory) - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 5000, requestCount: 1, incompleteRequestCount: 1 }) - }) - - it('marks SSH channel timing unknown before the first remote provenance and keeps that result on replay', () => { - const project = database.createSshProject({ - project: { name: 'SSH project', description: '', rootPath: '/srv/workspace', defaultWorkMode: 'execute', runtimeSelection: { provider: 'opencode' } }, - executionSpace: { kind: 'ssh', hostId: randomUUID(), remoteRootPath: '/srv/workspace' }, - assertCurrent: () => {} - }) - const conversation = database.getOrCreateRemoteConversation({ projectId: project.id, channel: 'wecom', accountId: 'account', externalConversationId: 'external', conversationType: 'direct', title: 'SSH channel', accountDisplay: 'Account' }) - const id = request({ projectId: project.id, conversationId: conversation.id, origin: 'delegation' }) - at(5) - database.appendTaskEvent(id, 'status', { requestId: id, type: 'status', message: 'Remote Agent is processing' }) - const expected = { durationMs: 0, requestCount: 1, incompleteRequestCount: 1, activeRequestCount: 1 } - expect(database.getExecutionStats({ projectId: project.id }, new Set([id]))).toMatchObject(expected) - at(100) - database.appendRemoteTaskEventOnce({ taskId: id, bindingId: randomUUID(), operationId: randomUUID(), semanticSequence: '1', eventIndex: 0, kind: 'text', payload: { requestId: id, type: 'text', delta: 'Reply' } }) - expect(database.getExecutionStats({ projectId: project.id }, new Set([id]))).toMatchObject(expected) - database.updateTaskStatus(id, 'completed') - expect(database.getExecutionStats({ conversationId: conversation.id })).toMatchObject({ ...expected, activeRequestCount: 0 }) - }) - - it('marks retained assistant replies without request history incomplete and ignores nested subagent content', () => { - const known = request() - text(known) - at(2) - database.updateTaskStatus(known, 'completed') - const child = request({ origin: 'subagent', parentTaskId: known }) - text(child) + it('excludes maintenance and nested tasks and sums concurrent replies across conversations', () => { + const maintenance = task() + const child = task({ origin: 'subagent' }) + database.startExecutionTiming(child) + const first = task() + database.startExecutionTiming(first) + const otherConversation = randomUUID() + const second = task({ conversationId: otherConversation }) + database.startExecutionTiming(second) + now += 3000 + database.updateTaskStatus(maintenance, 'completed') database.updateTaskStatus(child, 'completed') - const childMessage = randomUUID() - database.saveLocalConversations([{ - header: { id: conversationId, projectId, title: 'Retained', updatedAt: epoch }, - messages: [ - { id: randomUUID(), role: 'assistant', content: 'Known reply', state: 'complete', createdAt: epoch, subagents: [] }, - { id: randomUUID(), role: 'assistant', content: 'Older reply without a task', state: 'complete', createdAt: epoch }, - { id: childMessage, role: 'assistant', content: 'Child reply', state: 'complete', createdAt: epoch } - ] - }]) - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - raw.prepare('UPDATE messages SET request_id = ? WHERE id = ?').run(child, childMessage) - raw.close() - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 2000, requestCount: 2, incompleteRequestCount: 1 }) - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 2000, requestCount: 2, incompleteRequestCount: 1 }) + expect(stats()).toMatchObject({ durationMs: 3000, runningCount: 1 }) + expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 6000, runningCount: 2 }) }) - it('flags missing schedule execution once even when its assistant message is retained', () => { - const schedule = database.createSchedule({ projectId, title: 'Lost run', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - const queued = database.queueScheduleNow(schedule.id) - database.claimConversationQueueItem(schedule.conversationId) - database.completeScheduleRun(queued.scheduleRunId!, 'completed') - database.saveLocalConversations([{ - header: { id: schedule.conversationId, projectId, title: 'Lost run', updatedAt: epoch }, - messages: [{ id: randomUUID(), role: 'assistant', content: 'Retained result', state: 'complete', createdAt: epoch, task: { id: schedule.taskId, title: 'Lost run' } }] - }]) - expect(database.getExecutionStats({ projectId })).toMatchObject({ - durationMs: 0, requestCount: 1, incompleteRequestCount: 1, - taskDurations: [{ id: schedule.taskId, durationMs: 0, incompleteRequestCount: 1 }] + it.each(['running', 'waiting_approval'] as const)('recovers remote %s timing without counting known waiting or changing terminal checkpoints', (status) => { + const project = database.createSshProject({ + project: { name: 'Remote', description: '', rootPath: '/srv/project', defaultWorkMode: 'execute' }, + executionSpace: { kind: 'ssh', hostId: randomUUID(), remoteRootPath: '/srv/project' }, assertCurrent: () => {} + }) + const remoteConversationId = randomUUID() + database.saveLocalConversations([{ header: { id: remoteConversationId, projectId: project.id, title: 'Remote', updatedAt: now }, messages: [] }]) + const assistantMessageId = randomUUID() + const id = task({ projectId: project.id, conversationId: remoteConversationId, remoteRecovery: { + recoverable: true, currentUserMessageId: randomUUID(), currentAssistantMessageId: assistantMessageId + } }) + database.startExecutionTiming(id) + now += 2000 + database.updateTaskStatus(id, 'paused') + database.updateTaskStatus(id, status) + database.close() + now += 86_400_000 + database.initialize(directory) + database.startExecutionTiming(id) + if (status === 'waiting_approval') { + now += 60_000 + expect(database.getTask(id).status).toBe('waiting_approval') + expect(database.getExecutionStats({ conversationId: remoteConversationId })).toMatchObject({ + durationMs: 2000, runningCount: 0, incomplete: false + }) + database.startExecutionTiming(id) + now += 60_000 + expect(database.getExecutionStats({ conversationId: remoteConversationId }).durationMs).toBe(2000) + database.updateTaskStatus(id, 'running') + } + now += 3000 + const bindingId = randomUUID() + const operationId = randomUUID() + const append = (eventIndex: number, type: 'done' | 'remote-semantic-checkpoint') => database.appendRemoteConversationTaskEventOnce({ + taskId: id, conversationId: remoteConversationId, assistantMessageId, + bindingId, operationId, semanticSequence: '1', eventIndex, event: { type, requestId: id } + }) + expect(append(0, 'done')).toBe(true) + expect(database.getExecutionStats({ conversationId: remoteConversationId })).toMatchObject({ + durationMs: 5000, runningCount: 1, incomplete: status === 'running' + }) + expect(append(1, 'remote-semantic-checkpoint')).toBe(true) + now += 60_000 + expect(append(0, 'done')).toBe(false) + expect(append(1, 'remote-semantic-checkpoint')).toBe(false) + expect(database.getTask(id).status).toBe('completed') + expect(database.getExecutionStats({ conversationId: remoteConversationId })).toMatchObject({ + durationMs: 5000, runningCount: 0, incomplete: status === 'running' }) }) - it('cannot make a remote or missing-start request complete just because it has a live lease', () => { - const remote = request() - database.appendRemoteTaskEventOnce({ taskId: remote, bindingId: randomUUID(), operationId: randomUUID(), semanticSequence: '1', eventIndex: 0, kind: 'text', payload: { requestId: remote, type: 'text', delta: 'Remote' } }) - const unknown = request() - text(unknown) - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - raw.prepare('UPDATE tasks SET started_at = NULL WHERE id = ?').run(unknown) - raw.close() - at(100) - expect(database.getExecutionStats({ conversationId }, new Set([remote, unknown]))).toMatchObject({ durationMs: 0, incompleteRequestCount: 2, activeRequestCount: 2 }) - }) - - it('counts confirmed live replies before their first event without counting maintenance leases', () => { - const id = request() - request({ title: 'Context compression' }) - at(10) - expect(database.getExecutionStats({ conversationId }, new Set([id]))).toMatchObject({ durationMs: 10000, requestCount: 1, incompleteRequestCount: 0, activeRequestCount: 1 }) - }) - - it('reports wholly missing reply history and task attribution without using message timestamps', () => { - const schedule = database.createSchedule({ projectId, title: 'Old task', prompt: 'Reply', workMode: 'ask', recurrence: 'daily', nextRunAt: new Date(epoch).toISOString() }) - database.saveLocalConversations([{ - header: { id: schedule.conversationId, projectId, title: 'Old conversation', updatedAt: epoch }, - messages: [{ id: randomUUID(), role: 'assistant', content: 'Old reply', state: 'complete', createdAt: epoch - 100000, task: { id: schedule.taskId, title: 'Old task' } }] - }]) - expect(database.getExecutionStats({ conversationId: schedule.conversationId })).toMatchObject({ durationMs: 0, requestCount: 1, incompleteRequestCount: 1 }) - expect(database.getExecutionStats({ projectId }).taskDurations).toEqual([{ id: schedule.taskId, durationMs: 0, incompleteRequestCount: 1 }]) - }) - - it('does not let request coverage in another conversation conceal missing replies', () => { - const id = request() - text(id) - at(1) - database.updateTaskStatus(id, 'completed') - database.saveLocalConversations([{ - header: { id: randomUUID(), projectId, title: 'Other conversation', updatedAt: epoch }, - messages: [{ id: randomUUID(), role: 'assistant', content: 'Missing history', state: 'complete', createdAt: epoch }] - }]) - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 1000, requestCount: 2, incompleteRequestCount: 1 }) - }) - - it.each([ - ['zh-CN', '你好，我是 GoodBuddy。你可以直接向我提问、添加本地文件，或使用知识库整理和检索信息。需要我操作文件或调用工具时，请选择合适的 Agent Runtime 和工作模式。'], - ['en-US', 'Hi, I’m GoodBuddy. Ask me a question, add local files, or use your knowledge base to organize and retrieve information. When you want me to operate on files or use tools, choose the appropriate Agent Runtime and work mode.'] - ])('excludes the persisted default greeting but retains unknown reply history (%s)', async (locale, greeting) => { - expect(await readFile(join(process.cwd(), 'src/renderer/src/i18n/locales', locale, 'app.ts'), 'utf8')).toContain(greeting) - const header = { id: conversationId, projectId, title: 'New conversation', updatedAt: epoch } - const welcome = { id: randomUUID(), role: 'assistant' as const, content: greeting, state: 'complete' as const, createdAt: epoch } - database.saveLocalConversations([{ header, messages: [welcome] }]) - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 0, requestCount: 0, incompleteRequestCount: 0 }) - database.saveLocalConversations([{ header, messages: [welcome, - { id: randomUUID(), role: 'user', content: 'Question', state: 'complete', createdAt: epoch + 1 }, - { id: randomUUID(), role: 'assistant', content: greeting, state: 'complete', createdAt: epoch + 2 } - ] }]) - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 0, requestCount: 1, incompleteRequestCount: 1 }) - }) - - it('seeks interruption boundaries without parsing large tool/delta payloads on repeated scope queries', () => { - const id = request() - const raw = new DatabaseSync(join(directory, 'assistant.sqlite')) - const prepare = vi.spyOn(DatabaseSync.prototype, 'prepare') - try { - const insert = raw.prepare('INSERT INTO task_events (task_id, kind, payload_json, created_at) VALUES (?, ?, ?, ?)') - const payload = JSON.stringify({ requestId: id, output: 'x'.repeat(32768) }) - raw.exec('BEGIN') - for (let index = 0; index < 3000; index++) { - insert.run(id, index % 2 ? 'text' : 'tool', payload, new Date(epoch + 5000).toISOString()) - } - raw.exec('COMMIT') - at(100) - database.updateTaskStatus(id, 'interrupted') - const baseline = raw.prepare("SELECT SUM(length(json_extract(payload_json, '$.requestId'))) FROM task_events WHERE task_id = ?") - const baselineStart = performance.now() - baseline.get(id) - baseline.get(id) - const baselineMs = performance.now() - baselineStart - // One visible-panel tick reads both scopes; repeat to exercise warm polling. - const start = performance.now() - for (let iteration = 0; iteration < 5; iteration++) { - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 5000, requestCount: 1, incompleteRequestCount: 1 }) - expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 5000, requestCount: 1, incompleteRequestCount: 1 }) - } - const pairMs = (performance.now() - start) / 5 - const query = prepare.mock.calls.find(([sql]) => sql.includes('WITH scoped AS MATERIALIZED'))![0] - const plan = raw.prepare(`EXPLAIN QUERY PLAN ${query}`).all(projectId).map(row => row.detail as string) - for (const alias of ['proof', 'tail', 'previous']) { - expect(plan.some(detail => detail.includes(`SEARCH ${alias} USING INDEX task_events_task_idx`))).toBe(true) - } - // Filtering these through task_events_task_idx would read every payload row. - expect(plan.some(detail => detail.includes('SEARCH remote USING INDEX task_events_remote_task_idx'))).toBe(true) - expect(plan.some(detail => detail.includes('SEARCH e USING INDEX task_events_lifecycle_idx'))).toBe(true) - // Malformed bodies make accidental JSON extraction fail deterministically. - raw.prepare("UPDATE task_events SET payload_json = '{unparsed' WHERE task_id = ? AND kind IN ('text', 'tool')").run(id) - expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 5000, requestCount: 1, incompleteRequestCount: 1 }) - console.info(`execution stats: 3000 events, ${Math.round(payload.length * 3000 / 1024 / 1024)} MiB payload; old JSON extraction pair ${baselineMs.toFixed(1)} ms; scalar scope pair ${pairMs.toFixed(1)} ms`) - } finally { - prepare.mockRestore() - raw.close() + it('aggregates repeated schedule runs under the stable card without timing the parent', () => { + const schedule = database.createSchedule({ projectId, conversationId, title: 'Scheduled', prompt: 'Run', + workMode: 'ask', recurrence: 'once', nextRunAt: new Date(now + 60_000).toISOString() }) + database.startExecutionTiming(schedule.taskId) + for (let i = 0; i < 2; i++) { + const item = database.queueScheduleNow(schedule.id) + database.claimConversationQueueItem(conversationId, item.id) + task({ id: item.scheduleRunId!, parentTaskId: schedule.taskId }) + database.startExecutionTiming(item.scheduleRunId!) + now += 3000 + database.updateTaskStatus(item.scheduleRunId!, 'completed') + database.endExecutionTiming(item.scheduleRunId!) + database.completeTaskScheduleRun(item.scheduleRunId!) } + expect(database.getExecutionStats({ projectId }).taskDurations).toEqual([ + { id: schedule.taskId, durationMs: 6000, runningCount: 0, incomplete: false } + ]) + }) + + it('marks migrated history incomplete without reconstructing messages and adds the covering index', () => { + const id = task({ visible: true }) + database.updateTaskStatus(id, 'completed') + database.close() + const old = new DatabaseSync(path) + old.exec(`DROP TABLE execution_timing; + DROP INDEX conversations_timing_incomplete; + ALTER TABLE conversations DROP COLUMN timing_incomplete; + DROP INDEX messages_summary_idx; + PRAGMA user_version = 58;`) + old.close() + database.initialize(directory) + expect(stats()).toMatchObject({ durationMs: 0, runningCount: 0, incomplete: true }) + expect(database.getExecutionStats({ projectId }).taskDurations).toEqual([ + { id, durationMs: 0, runningCount: 0, incomplete: true } + ]) + expect(raw().prepare('PRAGMA index_info(messages_summary_idx)').all().map(row => row.name)) + .toEqual(['conversation_id', 'sequence', 'role', 'created_at']) + expect(raw().prepare('PRAGMA integrity_check').get()).toEqual({ integrity_check: 'ok' }) + }) + + it('deletes timing with its conversation and notifies snapshot subscribers', async () => { + const id = task() + database.startExecutionTiming(id) + now += 3000 + database.updateTaskStatus(id, 'completed') + database.endExecutionTiming(id) + await Promise.resolve() + const changed = vi.fn() + database.onExecutionStatsChanged(changed) + database.deleteLocalConversation(conversationId) + await Promise.resolve() + expect(changed).toHaveBeenCalledOnce() + expect(database.getExecutionStats({ projectId })).toMatchObject({ durationMs: 0, runningCount: 0 }) + expect(raw().prepare('SELECT count(*) AS n FROM execution_timing').get()!.n).toBe(0) }) }) diff --git a/src/main/assistant/execution-stats-reader.test.ts b/src/main/assistant/execution-stats-reader.test.ts deleted file mode 100644 index e4807e9f..00000000 --- a/src/main/assistant/execution-stats-reader.test.ts +++ /dev/null @@ -1,68 +0,0 @@ -// @vitest-environment node -import { randomUUID } from 'node:crypto' -import { mkdtemp, rm } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import { join, resolve } from 'node:path' -import { DatabaseSync } from 'node:sqlite' -import { build } from 'esbuild' -import { afterAll, beforeAll, expect, it } from 'vitest' -import { AssistantDatabase } from './assistant-database' -import { ExecutionStatsReader } from './execution-stats-reader' - -let directory: string -let workerPath: string -beforeAll(async () => { - directory = await mkdtemp(join(tmpdir(), 'goodbuddy-statistics-reader-')) - workerPath = process.env.GOODBUDDY_STATS_WORKER_PATH ?? join(directory, 'worker.cjs') - if (!process.env.GOODBUDDY_STATS_WORKER_PATH) { - await build({ entryPoints: [resolve('src/main/execution-stats-worker.ts')], outfile: workerPath, - bundle: true, platform: 'node', format: 'cjs', logLevel: 'silent' }) - } -}) -afterAll(async () => { await rm(directory, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) }) - -it('reads committed WAL evidence off-thread without recovering live tasks and refreshes external writes', async () => { - const path = join(directory, 'assistant.sqlite') - const database = new AssistantDatabase(path) - database.initialize(directory) - const projectId = database.listProjects()[0]!.id - const conversationId = randomUUID() - const id = randomUUID() - database.createTask({ id, projectId, conversationId, title: 'Live', instructions: 'Reply', workMode: 'ask', visible: false }) - database.appendTaskEvent(id, 'text', { requestId: id, type: 'text', delta: 'Reply' }) - const sql = new DatabaseSync(path) - try { - const status = database.getTask(id).status - const [conversation, project] = await Promise.all([ - database.getExecutionStatsAsync({ conversationId }, new Set([id]), workerPath), - database.getExecutionStatsAsync({ projectId }, new Set([id]), workerPath) - ]) - expect(conversation).toMatchObject({ requestCount: 1, activeRequestCount: 1, incompleteRequestCount: 0 }) - expect(project.requestCount).toBe(1) - expect(database.getTask(id).status).toBe(status) - database.updateTaskStatus(id, 'completed') - expect(await database.getExecutionStatsAsync({ conversationId }, new Set(), workerPath)).toMatchObject({ requestCount: 1, activeRequestCount: 0 }) - sql.exec('BEGIN IMMEDIATE') - sql.prepare('DELETE FROM tasks WHERE id = ?').run(id) - expect((await database.getExecutionStatsAsync({ conversationId }, new Set(), workerPath)).requestCount).toBe(1) - sql.exec('ROLLBACK') - expect((await database.getExecutionStatsAsync({ conversationId }, new Set(), workerPath)).requestCount).toBe(1) - sql.prepare('DELETE FROM tasks WHERE id = ?').run(id) - expect((await database.getExecutionStatsAsync({ conversationId }, new Set(), workerPath)).requestCount).toBe(0) - // A returned result must not leave a read transaction pinning the WAL. - expect(sql.prepare('PRAGMA wal_checkpoint(TRUNCATE)').get()?.busy).toBe(0) - const pending = database.getExecutionStatsAsync({ projectId }, new Set(), workerPath) - const rejection = expect(pending).rejects.toThrow('closed') - database.close() - await rejection - } finally { sql.close(); database.close() } -}) - -it('rejects worker startup errors and permits the next read to retry', async () => { - const reader = new ExecutionStatsReader(join(directory, 'missing.sqlite'), workerPath) - try { - for (let attempt = 0; attempt < 2; attempt++) { - await expect(reader.read({ projectId: randomUUID() }, new Set())).rejects.toThrow() - } - } finally { reader.close() } -}) diff --git a/src/main/assistant/execution-stats-reader.ts b/src/main/assistant/execution-stats-reader.ts deleted file mode 100644 index 5bcdd9a0..00000000 --- a/src/main/assistant/execution-stats-reader.ts +++ /dev/null @@ -1,52 +0,0 @@ -import { Worker } from 'node:worker_threads' -import type { ExecutionStats, ExecutionStatsInput } from '../../shared/assistant-contracts' - -export class ExecutionStatsReader { - private worker?: Worker - private nextId = 0 - private readonly pending = new Map void - reject: (error: Error) => void - }>() - - constructor(private readonly databasePath: string, private readonly workerPath: string) {} - - read(scope: ExecutionStatsInput, activeRequestIds: ReadonlySet): Promise { - if (!this.worker) { - const worker = new Worker(this.workerPath, { workerData: { databasePath: this.databasePath } }) - this.worker = worker - worker.on('message', (message: { id: number; result?: ExecutionStats; error?: string }) => { - const request = this.pending.get(message.id) - if (!request) return - this.pending.delete(message.id) - if (message.result) request.resolve(message.result) - else request.reject(new Error(message.error ?? 'Statistics query failed')) - }) - const failed = (error: Error): void => { - if (this.worker !== worker) return - this.worker = undefined - for (const request of this.pending.values()) request.reject(error) - this.pending.clear() - void worker.terminate() - } - worker.once('error', failed) - worker.once('exit', (code) => failed(new Error(`Statistics worker exited (${code})`))) - worker.unref() - } - // Bound outstanding work when a window repeatedly changes scope during a slow read. - if (this.pending.size >= 16) return Promise.reject(new Error('Statistics reader is busy')) - const id = ++this.nextId - return new Promise((resolve, reject) => { - this.pending.set(id, { resolve, reject }) - this.worker!.postMessage({ id, scope, activeRequestIds: [...activeRequestIds] }) - }) - } - - close(): void { - const worker = this.worker - this.worker = undefined - for (const request of this.pending.values()) request.reject(new Error('Statistics reader closed')) - this.pending.clear() - if (worker) void worker.terminate() - } -} diff --git a/src/main/execution-stats-worker.ts b/src/main/execution-stats-worker.ts deleted file mode 100644 index ca74c3ce..00000000 --- a/src/main/execution-stats-worker.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { parentPort, workerData } from 'node:worker_threads' -import type { ExecutionStatsInput } from '../shared/assistant-contracts' -import { AssistantDatabase } from './assistant/assistant-database' - -const database = new AssistantDatabase((workerData as { databasePath: string }).databasePath) -database.openReadOnly() -parentPort!.on('message', (message: { - id: number; scope: ExecutionStatsInput; activeRequestIds: string[] -}) => { - try { - const result = database.getExecutionStats(message.scope, new Set(message.activeRequestIds)) - parentPort!.postMessage({ id: message.id, result }) - } catch (error) { - parentPort!.postMessage({ id: message.id, error: error instanceof Error ? error.message : String(error) }) - } -}) -parentPort!.on('close', () => database.close()) diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index 56628ec1..88672a1d 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -49,11 +49,20 @@ import { CapabilityService } from './capabilities/capability-service' import { ObsidianService } from './obsidian' import { packageObsidianMcpVault } from './obsidian/package-mcpvault' import { - registerIpcHandlers, + registerIpcHandlers as registerProductionIpcHandlers, sendRemoteEnvironmentUpdateProgress, sendRemoteProjectSaveProgress } from './ipc' +// Most IPC tests use partial databases. Real database fixtures retain their timing implementation. +function registerIpcHandlers(...args: Parameters): ReturnType { + const database = args[7] + database.onExecutionStatsChanged ??= vi.fn(() => () => {}) + database.startExecutionTiming ??= vi.fn() + database.endExecutionTiming ??= vi.fn() + return registerProductionIpcHandlers(...args) +} + describe('embedding IPC boundary', () => { it('validates explicit IDs and keeps ZIP paths in the native dialog', async () => { const source = await readFile( @@ -4435,13 +4444,13 @@ describe('registerIpcHandlers token usage', () => { vi.clearAllMocks() }) - it('validates execution statistics scope and sender before reading evidence', async () => { - const summary = { durationMs: 5000, requestCount: 2, incompleteRequestCount: 1, activeRequestCount: 0, asOf: 10000, taskDurations: [{ id: 'task', durationMs: 5000, incompleteRequestCount: 1 }] } + it('validates execution statistics scope and sender before reading a timing snapshot', async () => { + const summary = { durationMs: 5000, runningCount: 0, incomplete: true, asOf: 10000, taskDurations: [{ id: 'task', durationMs: 5000, runningCount: 0, incomplete: true }] } const assistantDatabase = { queueDueSchedules: vi.fn(() => []), listConversationQueueItems: vi.fn(() => []), listPendingConversationQueueIds: vi.fn(() => []), - getExecutionStatsAsync: vi.fn(async () => summary) + getExecutionStats: vi.fn(() => summary) } const webContents = { on: vi.fn(), removeListener: vi.fn(), mainFrame: { url: 'file:///goodbuddy/index.html' }, @@ -4459,13 +4468,13 @@ describe('registerIpcHandlers token usage', () => { const event = { sender: webContents, senderFrame: webContents.mainFrame } const id = '00000000-0000-4000-8000-000000000301' for (const scope of [{ conversationId: id }, { projectId: id }]) { - await expect(handler(event, scope)).resolves.toEqual(summary) - expect(assistantDatabase.getExecutionStatsAsync).toHaveBeenLastCalledWith(scope, new Set(), expect.stringContaining('execution-stats-worker.js')) + expect(handler(event, scope)).toEqual(summary) + expect(assistantDatabase.getExecutionStats).toHaveBeenLastCalledWith(scope) } expect(() => handler(event, {})).toThrow() expect(() => handler(event, { projectId: id, conversationId: id })).toThrow() expect(() => handler({ sender: {}, senderFrame: webContents.mainFrame }, { projectId: id })).toThrow() - expect(assistantDatabase.getExecutionStatsAsync).toHaveBeenCalledTimes(2) + expect(assistantDatabase.getExecutionStats).toHaveBeenCalledTimes(2) } finally { await dispose() } @@ -5796,6 +5805,128 @@ describe('registerIpcHandlers agent terminal state', () => { knowledgeLibraryIds: [] }) + it.each(['completed', 'failed', 'cancelled'] as const)('persists reply duration through the production lease and %s path', async (status) => { + const database = new AssistantDatabase(':memory:') + database.initialize(process.cwd()) + const projectId = database.listProjects()[0]!.id + const conversationId = crypto.randomUUID() + const requestId = crypto.randomUUID() + let now = Date.now() + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now) + database.saveLocalConversations([{ header: { id: conversationId, projectId, title: 'Timing', updatedAt: now }, messages: [] }]) + let finish!: () => void + const gate = new Promise(resolve => { finish = resolve }) + let started!: () => void + const ready = new Promise(resolve => { started = resolve }) + const run = vi.fn(async function* (request: AgentExecutionRequest) { + started() + await gate + if (status !== 'completed') throw new Error('Stopped') + yield { type: 'done' as const, requestId: request.requestId } + }) + const harness = createHarness({ runtimeId: 'model', capability: 'chat', supportsToolExecution: true, run }, + undefined, 'always', undefined, false, undefined, undefined, undefined, false, + undefined, undefined, undefined, undefined, database) + try { + const event = trustedEvent(harness.webContents) + await harness.handler!(event, { requestId, conversationId, projectId, prompt: 'Run', workMode: 'ask' }) + await ready + now += 5000 + const read = () => electronMocks.handlers.get(ipcChannels.tasksExecutionStats)!(event, { conversationId }) + expect(read()).toMatchObject({ durationMs: 5000, runningCount: 1, incomplete: false }) + if (status === 'cancelled') { + harness.cancelHandler!(event, requestId) + now += 60_000 + expect(read()).toMatchObject({ durationMs: 5000, runningCount: 0 }) + } + finish() + await vi.waitFor(() => expect(database.getTask(requestId).status).toBe(status)) + now += 60_000 + expect(read()).toMatchObject({ durationMs: 5000, runningCount: 0 }) + expect(harness.webContents.send).toHaveBeenCalledWith(ipcChannels.tasksExecutionStatsChanged) + } finally { + finish() + await harness.dispose() + clock.mockRestore() + database.close() + } + }) + + it.each(['answer', 'runtime-event'] as const)('pauses ordinary question timing until the last %s resolution and retains failed answers', async (resolution) => { + const database = new AssistantDatabase(':memory:') + database.initialize(process.cwd()) + const projectId = database.listProjects()[0]!.id + const conversationId = crypto.randomUUID() + const requestId = crypto.randomUUID() + const questionIds = [crypto.randomUUID(), crypto.randomUUID()] + let now = Date.now() + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now) + database.saveLocalConversations([{ header: { id: conversationId, projectId, title: 'Questions', updatedAt: now }, messages: [] }]) + const deferred = () => { + let resolve!: () => void + const promise = new Promise(complete => { resolve = complete }) + return { promise, resolve } + } + const ready = deferred() + const finish = deferred() + const steps = questionIds.map(() => ({ release: deferred(), processed: deferred() })) + const respondToQuestion = vi.fn().mockRejectedValueOnce(new Error('Answer delivery failed')).mockResolvedValue(undefined) + const run = vi.fn(async function* () { + now += 2000 + for (const questionId of questionIds) yield { + type: 'question' as const, requestId, questionId, + questions: [{ header: 'Input', question: 'Continue?', options: [], custom: true, multiple: false }] + } + ready.resolve() + for (const [index, questionId] of questionIds.entries()) { + await steps[index]!.release.promise + if (resolution === 'runtime-event') yield { type: 'question-resolved' as const, requestId, questionId } + steps[index]!.processed.resolve() + } + await finish.promise + yield { type: 'done' as const, requestId } + }) + const harness = createHarness({ runtimeId: 'model', capability: 'chat', supportsToolExecution: true, run, respondToQuestion }, + undefined, 'always', undefined, false, undefined, undefined, undefined, false, + undefined, undefined, undefined, undefined, database) + const read = () => database.getExecutionStats({ conversationId }) + try { + const event = trustedEvent(harness.webContents) + await harness.handler!(event, { requestId, conversationId, projectId, prompt: 'Ask me', workMode: 'ask' }) + await ready.promise + expect(database.getTask(requestId).status).toBe('waiting_approval') + now += 60_000 + expect(read()).toMatchObject({ durationMs: 2000, runningCount: 0 }) + const answer = electronMocks.handlers.get(ipcChannels.agentQuestionRespond)! + await expect(answer(event, { questionId: questionIds[0], answers: [['Yes']] })).rejects.toThrow('Answer delivery failed') + now += 60_000 + expect(database.getTask(requestId).status).toBe('waiting_approval') + expect(read()).toMatchObject({ durationMs: 2000, runningCount: 0 }) + for (const [index, questionId] of questionIds.entries()) { + if (resolution === 'answer') await answer(event, { questionId, answers: index === 0 ? [['Yes']] : [] }) + steps[index]!.release.resolve() + await steps[index]!.processed.promise + if (index === 0) { + now += 60_000 + expect(database.getTask(requestId).status).toBe('waiting_approval') + expect(read()).toMatchObject({ durationMs: 2000, runningCount: 0 }) + } + } + expect(database.getTask(requestId).status).toBe('running') + now += 3000 + finish.resolve() + await vi.waitFor(() => expect(database.getTask(requestId).status).toBe('completed')) + expect(read()).toMatchObject({ durationMs: 5000, runningCount: 0 }) + if (resolution === 'answer') expect(respondToQuestion).toHaveBeenLastCalledWith(questionIds[1], undefined) + } finally { + for (const step of steps) step.release.resolve() + finish.resolve() + await harness.dispose() + clock.mockRestore() + database.close() + } + }) + it('binds the production image service before text dispatch and routes regenerate, Ask rejection, and deletion through IPC', async () => { const database = new AssistantDatabase(':memory:') database.initialize(process.cwd()) @@ -6721,11 +6852,14 @@ describe('registerIpcHandlers agent terminal state', () => { const database = new AssistantDatabase(':memory:') database.initialize(process.cwd()) const conversationId = crypto.randomUUID() + let timingNow = Date.now() + const timingClock = vi.spyOn(Date, 'now').mockImplementation(() => timingNow) database.saveLocalConversations([{ header: { id: conversationId, title: 'Immediate', updatedAt: Date.now(), workMode: 'ask' }, messages: [] }]) const run = vi.fn(async function* (request: AgentExecutionRequest) { + timingNow += 3000 if (status === 'failed') { yield { type: 'error' as const, requestId: request.requestId, message: 'Runtime failed' } return @@ -6773,8 +6907,10 @@ describe('registerIpcHandlers agent terminal state', () => { expect(run).toHaveBeenCalledOnce() expect(database.listConversationQueueItems()).toEqual([]) expect(database.listSchedules()).toHaveLength(1) + expect(database.getExecutionStats({ conversationId })).toMatchObject({ durationMs: 3000, runningCount: 0, incomplete: false }) } finally { await harness.dispose() + timingClock.mockRestore() database.close() } }) diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 27b1cedb..64e3cff9 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -21,7 +21,7 @@ import { stat } from 'node:fs/promises' import { randomUUID } from 'node:crypto' -import { basename, extname, join } from 'node:path' +import { basename, extname } from 'node:path' import { z } from 'zod' import { documentResourceInputSchema } from '../shared/document-result-contracts' import { maximumAttachmentsPerMessage } from '../shared/attachment-limits' @@ -1111,14 +1111,16 @@ export function registerIpcHandlers( } type ActiveRequestLease = { controller: AbortController - isReply: boolean conversationId: string detachOnApplicationExit: boolean teamMode?: boolean recoveredMessageId?: string - release(): void + release(incomplete?: boolean): void } const activeRequests = new Map() + const removeExecutionStatsListener = assistantDatabase.onExecutionStatsChanged(() => { + if (!window.isDestroyed()) window.webContents.send(ipcChannels.tasksExecutionStatsChanged) + }) const onRuntimeSettingsChanged = async (): Promise => { // The expert scheduler cancels its work on replacement. Its parent must // not synthesize partial old-generation results with the replacement model. @@ -1139,12 +1141,19 @@ export function registerIpcHandlers( controller: AbortController, isReply = true ): ActiveRequestLease => { + if (isReply) assistantDatabase.startExecutionTiming(requestId) + const stopTiming = (): void => { + if (isReply) assistantDatabase.endExecutionTiming(requestId) + } + controller.signal.addEventListener('abort', stopTiming, { once: true }) + if (controller.signal.aborted) stopTiming() const lease: ActiveRequestLease = { controller, - isReply, conversationId, detachOnApplicationExit: false, - release: (): void => { + release: (incomplete = false): void => { + if (isReply) assistantDatabase.endExecutionTiming(requestId, incomplete) + controller.signal.removeEventListener('abort', stopTiming) if (activeRequests.get(requestId) === lease) { activeRequests.delete(requestId) } @@ -1162,6 +1171,14 @@ export function registerIpcHandlers( string, { requestId: string; runtime: AgentRuntime; question: Extract } >() + const resumeAfterQuestions = (requestId: string): boolean => { + const lease = activeRequests.get(requestId) + if (!lease || lease.controller.signal.aborted || + [...pendingAgentQuestions.values()].some(pending => pending.requestId === requestId) || + assistantDatabase.getTask(requestId).status !== 'waiting_approval') return false + assistantDatabase.updateTaskStatus(requestId, 'running') + return true + } const supervisionModelPool = new SupervisionModelPool() let activeSshDirectoryBrowse: AbortController | undefined let shuttingDown = false @@ -1398,6 +1415,7 @@ export function registerIpcHandlers( preserveApplicationExitDetached && lease.detachOnApplicationExit ) { + assistantDatabase.endExecutionTiming(requestId, true) continue } lease.controller.abort(new Error(reason)) @@ -1702,9 +1720,7 @@ export function registerIpcHandlers( assistantDatabase.updateTaskStatus(task.taskId, 'waiting_approval') } else { pendingAgentQuestions.delete(rawEvent.questionId) - if (![...pendingAgentQuestions.values()].some(pending => pending.requestId === task.taskId)) { - assistantDatabase.updateTaskStatus(task.taskId, 'running') - } + resumeAfterQuestions(task.taskId) } publishConversationChange() continue @@ -1828,7 +1844,7 @@ export function registerIpcHandlers( for (const [id, pending] of pendingAgentQuestions) { if (pending.requestId === task.taskId) pendingAgentQuestions.delete(id) } - lease.release() + lease.release(!sawTerminal) publishConversationChange() assistantDatabase.completeTaskScheduleRun(task.taskId) } @@ -2159,11 +2175,6 @@ export function registerIpcHandlers( const runtimeConversationId = remoteContext?.conversationId ?? `${origin}:${schedule.id}` - const activeRequestLease = leaseActiveRequest( - requestId, - runtimeConversationId, - controller - ) if (input.origin !== 'delegation') { assistantDatabase.updateTaskStatus(taskId, 'running') } else { @@ -2178,6 +2189,11 @@ export function registerIpcHandlers( visible: false }) } + const activeRequestLease = leaseActiveRequest( + requestId, + runtimeConversationId, + controller + ) let output = '' let completed = false let backgroundQuestionError: Error | undefined @@ -4423,10 +4439,13 @@ export function registerIpcHandlers( runtime: selectedRuntime, question: publicEvent }) + assistantDatabase.updateTaskStatus(request.requestId, 'waiting_approval') + publishConversationChange() } if (publicEvent.type === 'question-resolved' && pendingAgentQuestions.get(publicEvent.questionId)?.requestId === request.requestId) { pendingAgentQuestions.delete(publicEvent.questionId) + if (resumeAfterQuestions(request.requestId)) publishConversationChange() } if (publicEvent.type === 'error') { runtimeErrorEvent = publicEvent @@ -4600,7 +4619,7 @@ export function registerIpcHandlers( } } knowledgeGateway?.revoke(request.knowledgeCapabilityToken) - activeRequestLease.release() + activeRequestLease.release(remoteRecoveryPending) if (remoteRecoveryPending && request.projectId) { startRemoteProjectRecovery(request.projectId) } @@ -4696,15 +4715,8 @@ export function registerIpcHandlers( })) }) pendingAgentQuestions.delete(response.questionId) - if (activeRequests.get(pending.requestId)?.recoveredMessageId) { - if ( - ![...pendingAgentQuestions.values()].some(value => value.requestId === pending.requestId) && - assistantDatabase.getTask(pending.requestId).status === 'waiting_approval' - ) { - assistantDatabase.updateTaskStatus(pending.requestId, 'running') - } - } - if (recorded) { + const resumed = resumeAfterQuestions(pending.requestId) + if (recorded || resumed) { publishConversationChange() } } @@ -6344,9 +6356,13 @@ export function registerIpcHandlers( registerHandler(ipcChannels.conversationsListSummaries, async (event, input: unknown) => { assertTrustedSender(event, window) const { detailIds } = conversationListRequestSchema.parse(input) - return projectConversationRequests(await assistantDatabase.listConversationSummariesAsync([ - ...detailIds, ...[...activeRequests.values()].map(request => request.conversationId) - ])) + // Recovery needs the message carrying pending questions; ordinary active + // replies already reach the renderer through their event stream. + return projectConversationRequests(await assistantDatabase.listConversationSummariesAsync([...new Set([ + ...detailIds, ...[...activeRequests.values()] + .filter(request => request.recoveredMessageId) + .map(request => request.conversationId) + ])])) }) registerHandler(ipcChannels.conversationsGet, async (event, input: unknown) => { assertTrustedSender(event, window) @@ -6840,11 +6856,7 @@ export function registerIpcHandlers( }) registerHandler(ipcChannels.tasksExecutionStats, (event, input: unknown) => { assertTrustedSender(event, window) - return assistantDatabase.getExecutionStatsAsync( - executionStatsInputSchema.parse(input), - new Set([...activeRequests].filter(([, lease]) => lease.isReply && !lease.controller.signal.aborted).map(([id]) => id)), - join(app.getAppPath(), 'out/main/execution-stats-worker.js') - ) + return assistantDatabase.getExecutionStats(executionStatsInputSchema.parse(input)) }) registerHandler(ipcChannels.tasksSetStatus, (event, input: unknown) => { assertTrustedSender(event, window) @@ -7480,6 +7492,7 @@ export function registerIpcHandlers( registerKnowledgeIpcHandlers(registerHandler, window, knowledgeService, settingsStore) return async () => { + removeExecutionStatsListener() disposeWindowIpc() shuttingDown = true if (nativeClientCoordinator) window.webContents.removeListener('destroyed', closeNativeClients) diff --git a/src/main/readonly-query-reader.test.ts b/src/main/readonly-query-reader.test.ts index 5651e6aa..ebaf5e78 100644 --- a/src/main/readonly-query-reader.test.ts +++ b/src/main/readonly-query-reader.test.ts @@ -187,8 +187,11 @@ describe('assistant readonly worker', () => { const summaries = database.listConversationSummaries([conversations[2]!.id]) const byId = new Map(summaries.map(item => [item.id, item])) expect(byId.get(conversations[2]!.id)).toEqual(database.getConversation(conversations[2]!.id)) - // A streaming conversation always carries its messages. - expect(byId.get(conversations[1]!.id)).toEqual(database.getConversation(conversations[1]!.id)) + // Streaming histories are loaded only when explicitly requested. + expect(byId.get(conversations[1]!.id)?.messages).toEqual([]) + expect(byId.get(conversations[1]!.id)?.messageSummary).toEqual({ count: 2, firstRole: 'user', latestMessageAt: 2 }) + expect(database.listConversationSummaries([conversations[1]!.id]) + .find(item => item.id === conversations[1]!.id)).toEqual(database.getConversation(conversations[1]!.id)) expect(byId.get(conversations[0]!.id)?.messageSummary).toEqual({ count: 0, firstRole: undefined, latestMessageAt: undefined }) expect(byId.get(conversations[3]!.id)?.messageSummary).toEqual({ count: 1, firstRole: 'assistant', latestMessageAt: 3 }) expect(byId.get(conversations[3]!.id)?.messages).toEqual([]) @@ -213,9 +216,10 @@ describe('assistant readonly worker', () => { const detailSets = [[], [conversations[3]!.id], conversations.map(item => item.id), [randomUUID()]] const expected = detailSets.map(ids => database.listConversationSummaries(ids)) const expectedFull = conversations.map(item => database.getConversation(item.id)) - // Summaries without details carry messageSummary; retained/streaming ones carry messages. + // Only explicitly retained histories carry messages, including during streaming. expect(expected[0]!.find(item => item.id === conversations[4]!.id)?.messageSummary).toEqual({ count: 4, firstRole: 'user', latestMessageAt: 43 }) - expect(expected[0]!.find(item => item.id === conversations[5]!.id)?.messages).toHaveLength(5) + expect(expected[0]!.find(item => item.id === conversations[5]!.id)?.messages).toHaveLength(0) + expect(expected[0]!.find(item => item.id === conversations[5]!.id)?.messageSummary?.count).toBe(5) expect(expected[1]!.find(item => item.id === conversations[3]!.id)).toEqual(expectedFull[3]) database.enableReadonlyWorker(workerPath) diff --git a/src/preload/execution-stats.test.ts b/src/preload/execution-stats.test.ts index 1160c74f..c74c6dab 100644 --- a/src/preload/execution-stats.test.ts +++ b/src/preload/execution-stats.test.ts @@ -12,10 +12,17 @@ vi.mock('electron', () => electron) it('reads execution statistics through the explicit typed bridge', async () => { await import('./index') const api = electron.contextBridge.exposeInMainWorld.mock.calls[0]![1] as DesktopApi - const summary = { durationMs: 1000, requestCount: 1, incompleteRequestCount: 0, activeRequestCount: 0, asOf: 2000, taskDurations: [{ id: 'task', durationMs: 1000, incompleteRequestCount: 0 }] } + const summary = { durationMs: 1000, runningCount: 0, incomplete: false, asOf: 2000, taskDurations: [{ id: 'task', durationMs: 1000, runningCount: 0, incomplete: false }] } electron.ipcRenderer.invoke.mockResolvedValue(summary) for (const scope of [{ conversationId: 'conversation' }, { projectId: 'project' }]) { expect(await api.tasks.getExecutionStats(scope)).toBe(summary) expect(electron.ipcRenderer.invoke).toHaveBeenLastCalledWith(ipcChannels.tasksExecutionStats, scope) } + const listener = vi.fn() + const unsubscribe = api.tasks.onExecutionStatsChanged(listener) + const handler = electron.ipcRenderer.on.mock.calls.find(([channel]) => channel === ipcChannels.tasksExecutionStatsChanged)![1] + handler() + expect(listener).toHaveBeenCalledOnce() + unsubscribe() + expect(electron.ipcRenderer.removeListener).toHaveBeenCalledWith(ipcChannels.tasksExecutionStatsChanged, handler) }) diff --git a/src/preload/index.ts b/src/preload/index.ts index fe761df7..fcd54ef8 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -1273,6 +1273,11 @@ const desktopApi: DesktopApi = { ipcRenderer.invoke(ipcChannels.tasksList) as Promise, getExecutionStats: (input) => ipcRenderer.invoke(ipcChannels.tasksExecutionStats, input) as Promise, + onExecutionStatsChanged: (listener) => { + const handler = (): void => listener() + ipcRenderer.on(ipcChannels.tasksExecutionStatsChanged, handler) + return () => { ipcRenderer.removeListener(ipcChannels.tasksExecutionStatsChanged, handler) } + }, setStatus: async ( taskId: string, status: 'completed' | 'cancelled' diff --git a/src/renderer/src/App.test.tsx b/src/renderer/src/App.test.tsx index 307b9a06..2d7be659 100644 --- a/src/renderer/src/App.test.tsx +++ b/src/renderer/src/App.test.tsx @@ -674,9 +674,10 @@ const api: DesktopApi & RuntimeNativeClientApi = { tasks: { list: vi.fn(async () => []), getExecutionStats: vi.fn(async () => ({ - durationMs: 0, requestCount: 0, incompleteRequestCount: 0, - activeRequestCount: 0, asOf: Date.now(), taskDurations: [], + durationMs: 0, runningCount: 0, incomplete: false, + asOf: Date.now(), taskDurations: [], })), + onExecutionStatsChanged: vi.fn(() => () => {}), setStatus: vi.fn(async () => {}), }, activityHistory: { @@ -1437,6 +1438,10 @@ describe("App", () => { }; }); vi.mocked(api.tasks.list).mockReset().mockResolvedValue([]); + vi.mocked(api.tasks.getExecutionStats).mockReset().mockImplementation(async () => ({ + durationMs: 0, runningCount: 0, incomplete: false, asOf: Date.now(), taskDurations: [], + })); + vi.mocked(api.tasks.onExecutionStatsChanged).mockReset().mockReturnValue(() => {}); vi.mocked(api.activityHistory.get).mockReset().mockResolvedValue({ records: [], legacyHistoryMayBeIncomplete: false, @@ -7241,6 +7246,35 @@ describe("App", () => { expect(runtimeButton).toBeEnabled(); }); + it("renders authoritative duration snapshots and refreshes them from timing notifications", async () => { + const conversationId = crypto.randomUUID(); + vi.mocked(api.conversations.list).mockResolvedValue([{ + id: conversationId, projectId, title: "Timed conversation", updatedAt: Date.now(), messages: [], + }]); + const summary = { durationMs: 2000, runningCount: 0, incomplete: true, asOf: Date.now(), taskDurations: [] }; + vi.mocked(api.tasks.getExecutionStats).mockResolvedValue(summary); + let changed: (() => void) | undefined; + const unsubscribe = vi.fn(); + vi.mocked(api.tasks.onExecutionStatsChanged).mockImplementation((listener) => { + changed = listener; + return unsubscribe; + }); + render(); + await waitFor(() => expect(api.conversationQueue.ready).toHaveBeenCalledWith(conversationId)); + fireEvent.click(screen.getByRole("button", { name: "切换助手工作栏" })); + fireEvent.click(screen.getByRole("tab", { name: "任务中心" })); + const statistics = await screen.findByLabelText("当前会话统计"); + expect(await within(statistics).findByText("00:00:02")).toBeVisible(); + expect(within(statistics).getByText("部分回复缺少计时记录，时长仅包含已知记录。")).toBeVisible(); + expect(api.tasks.getExecutionStats).toHaveBeenCalledWith({ conversationId }); + expect(api.tasks.getExecutionStats).toHaveBeenCalledWith({ projectId }); + vi.mocked(api.tasks.getExecutionStats).mockResolvedValue({ ...summary, durationMs: 65_000 }); + act(() => changed?.()); + expect(await within(statistics).findByText("00:01:05")).toBeVisible(); + fireEvent.click(screen.getByRole("button", { name: "切换助手工作栏" })); + await waitFor(() => expect(unsubscribe).toHaveBeenCalled()); + }); + it("associates live scheduled approvals using the dispatched task rather than another task in the conversation", async () => { const conversationId = crypto.randomUUID(); const taskId = crypto.randomUUID(); diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index e6eb82ff..a844fa60 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -173,7 +173,6 @@ import { useProductTasks, useSidebarArtifacts, useTaskActivityRows, - useTaskStatsRevision, useTaskStatusRows, } from "./task-selectors"; import { @@ -2797,20 +2796,18 @@ function App(): React.JSX.Element { // The supervisor page needs every task; App follows them only while it is shown. const heartbeatTasks = useAllTasks(taskStore, view !== "heartbeat"); const executionStatsEnabled = assistantSidebarOpen && assistantSidebarTab === "tasks"; - const executionStatsRevision = useTaskStatsRevision(taskStore, executionStatsEnabled); const statsConversation = activeConversation?.projectId === activeProjectId ? activeConversation : undefined; const statsMessageCount = statsConversation?.messageCount ?? 0; const executionStats = useExecutionStats( statsConversation?.id, activeProjectId || undefined, - `${executionStatsRevision}:${statsMessageCount}:${activeConversationIds.has(activeId)}`, executionStatsEnabled, ); const taskDurations = useMemo(() => new Map( executionStats.project?.taskDurations.map((task) => [task.id, { durationMs: task.durationMs, - incomplete: task.incompleteRequestCount > 0, + incomplete: task.incomplete, }]) ?? [], ), [executionStats.project]); const tasksByConversation = useMemo(() => { @@ -6717,7 +6714,7 @@ function App(): React.JSX.Element { ); const statsConversationId = statsConversation?.id; const statsDurationMs = executionStats.conversation?.durationMs; - const statsIncomplete = (executionStats.conversation?.incompleteRequestCount ?? 0) > 0; + const statsIncomplete = executionStats.conversation?.incomplete ?? false; const conversationStats = useMemo( () => statsConversationId && statsDurationMs !== undefined diff --git a/src/renderer/src/task-selectors.ts b/src/renderer/src/task-selectors.ts index edca0bd9..34e68ca3 100644 --- a/src/renderer/src/task-selectors.ts +++ b/src/renderer/src/task-selectors.ts @@ -109,17 +109,6 @@ export function useTaskActivityRows(store: TaskStore): TaskActivityRow[] { ); } -const selectStatsRevision = (tasks: AssistantTask[]): string => - tasks.map((task) => `${task.id}:${task.status}:${task.completedAt ?? ""}`).join("|"); -const selectNoRevision = (): string => ""; - -/** - * Changes whenever a task starts, finishes or changes status; constant while - * `enabled` is false, so nothing re-renders for statistics nobody sees. - */ -export function useTaskStatsRevision(store: TaskStore, enabled: boolean): string { - return useTaskSelector(store, enabled ? selectStatsRevision : selectNoRevision); -} /** Whether a task of this conversation is running or waiting for approval. */ export function useConversationHasBusyTask(store: TaskStore, conversationId: string): boolean { const select = useCallback( diff --git a/src/renderer/src/task-store.test.ts b/src/renderer/src/task-store.test.ts index 3a43922a..79a8a270 100644 --- a/src/renderer/src/task-store.test.ts +++ b/src/renderer/src/task-store.test.ts @@ -7,7 +7,6 @@ import { useProductTasks, useSidebarArtifacts, useTaskActivityRows, - useTaskStatsRevision, useTaskStatusRows, } from "./task-selectors"; import { createTaskStore, mergeArtifacts } from "./task-store"; @@ -103,7 +102,6 @@ describe("task selectors", () => { status: useTaskStatusRows(store), activity: useTaskActivityRows(store), busy: useConversationHasBusyTask(store, "c1"), - revision: useTaskStatsRevision(store, false), }; }); const first = result.current; @@ -124,13 +122,10 @@ describe("task selectors", () => { const { result } = renderHook(() => ({ busy: useConversationHasBusyTask(store, "c1"), status: useTaskStatusRows(store), - revision: useTaskStatsRevision(store, true), })); - const revision = result.current.revision; act(() => store.setTasks((current) => current.map((item) => ({ ...item, status: "completed" })))); expect(result.current.busy).toBe(false); expect(result.current.status[0]!.status).toBe("completed"); - expect(result.current.revision).not.toBe(revision); }); it("keep the artifact map and sidebar entries while artifacts are unchanged", () => { diff --git a/src/renderer/src/use-execution-stats.test.tsx b/src/renderer/src/use-execution-stats.test.tsx index ae7700d4..acecb451 100644 --- a/src/renderer/src/use-execution-stats.test.tsx +++ b/src/renderer/src/use-execution-stats.test.tsx @@ -1,299 +1,84 @@ import { act, cleanup, renderHook } from '@testing-library/react' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import type { ExecutionStats, ExecutionStatsInput } from '../../shared/assistant-contracts' +import type { ExecutionStats } from '../../shared/assistant-contracts' import { useExecutionStats } from './use-execution-stats' -function stats(durationMs: number): ExecutionStats { - return { - durationMs, requestCount: 1, incompleteRequestCount: 0, - activeRequestCount: 0, asOf: 1_000, taskDurations: [] - } -} - -function deferred() { - let resolve!: (value: ExecutionStats) => void - let reject!: (reason: Error) => void - const promise = new Promise((res, rej) => { - resolve = res - reject = rej - }) - return { promise, resolve, reject } -} - -const empty = { conversation: undefined, project: undefined } -const initialProps = { - conversationId: 'chat-a' as string | undefined, - projectId: 'project-a' as string | undefined, - revision: '1', enabled: true -} - -function renderStats() { - return renderHook(({ conversationId, projectId, revision, enabled }) => - useExecutionStats(conversationId, projectId, revision, enabled), { initialProps }) -} - -const getExecutionStats = vi.fn<(input: ExecutionStatsInput) => Promise>() - +const snapshot = (runningCount = 0, durationMs = 1000): ExecutionStats => ({ + durationMs, runningCount, incomplete: false, asOf: Date.now(), + taskDurations: [{ id: 'task', durationMs, runningCount, incomplete: false }] +}) +const getExecutionStats = vi.fn() +const listeners = new Set<() => void>() +const flush = async (): Promise => { await act(async () => { await Promise.resolve() }) } beforeEach(() => { vi.useFakeTimers() - vi.spyOn(document, 'visibilityState', 'get').mockReturnValue('visible') - getExecutionStats.mockReset().mockResolvedValue(stats(10)) - vi.stubGlobal('goodbuddy', { tasks: { getExecutionStats } }) + vi.setSystemTime(10_000) + getExecutionStats.mockReset().mockImplementation(async () => snapshot()) + vi.stubGlobal('goodbuddy', { tasks: { getExecutionStats, + onExecutionStatsChanged: (listener: () => void) => { + listeners.add(listener) + return () => listeners.delete(listener) + } + } }) }) +afterEach(() => { cleanup(); listeners.clear(); vi.useRealTimers(); vi.unstubAllGlobals() }) -afterEach(() => { - cleanup() - vi.useRealTimers() - vi.restoreAllMocks() - vi.unstubAllGlobals() -}) - -describe('useExecutionStats', () => { - it.each(['conversation', 'project'] as const)( - 'restores cached statistics immediately when returning to a %s', async (scope) => { - const { result, rerender } = renderStats() - await act(async () => {}) - const cached = result.current - getExecutionStats.mockResolvedValue(stats(20)) - rerender({ - ...initialProps, conversationId: 'chat-b', - projectId: scope === 'project' ? 'project-b' : 'project-a' - }) - await act(async () => {}) - const conversation = deferred() - const project = deferred() - getExecutionStats.mockReturnValueOnce(conversation.promise).mockReturnValueOnce(project.promise) - rerender(initialProps) - expect(result.current.conversation).toBe(cached.conversation) - expect(result.current.project).toEqual(scope === 'project' ? cached.project : stats(20)) - await act(async () => { - conversation.resolve(stats(30)) - project.resolve(stats(40)) - }) - expect(result.current).toEqual({ conversation: stats(30), project: stats(40) }) - } - ) - - it('retains snapshot references for identical cached polling results', async () => { - getExecutionStats.mockImplementation(async () => stats(10)) - const { result } = renderStats() - await act(async () => {}) - const before = result.current - await act(async () => { vi.advanceTimersByTime(5_000) }) - expect(result.current.conversation).toBe(before.conversation) - expect(result.current.project).toBe(before.project) - expect(getExecutionStats).toHaveBeenCalledTimes(4) - }) - it.each(['project', 'conversation'] as const)( - 'clears the previous %s selection and ignores its late polling result', async (scope) => { - const { result, rerender } = renderStats() - await act(async () => {}) - expect(result.current).toEqual({ conversation: stats(10), project: stats(10) }) - - const oldConversation = deferred() - const oldProject = deferred() - getExecutionStats.mockReturnValueOnce(oldConversation.promise).mockReturnValueOnce(oldProject.promise) - await act(async () => { vi.advanceTimersByTime(5_000) }) - - const newConversation = deferred() - const newProject = deferred() - getExecutionStats.mockReturnValueOnce(newConversation.promise).mockReturnValueOnce(newProject.promise) - rerender({ - ...initialProps, conversationId: 'chat-b', - projectId: scope === 'project' ? 'project-b' : 'project-a' - }) - expect(result.current).toEqual({ - conversation: undefined, project: scope === 'project' ? undefined : stats(10) - }) - expect(getExecutionStats.mock.calls.slice(-2)).toEqual([ - [{ conversationId: 'chat-b' }], - [{ projectId: scope === 'project' ? 'project-b' : 'project-a' }] - ]) - - await act(async () => { - newConversation.resolve(stats(20)) - newProject.resolve(stats(30)) - }) - await act(async () => { - oldConversation.resolve(stats(90)) - oldProject.resolve(stats(99)) - }) - expect(result.current).toEqual({ conversation: stats(20), project: stats(30) }) - } - ) - - it('does not resurrect cancelled results when switching back to the same IDs', async () => { - const oldConversation = deferred() - const oldProject = deferred() - getExecutionStats.mockReturnValueOnce(oldConversation.promise).mockReturnValueOnce(oldProject.promise) - const { result, rerender } = renderStats() - rerender({ ...initialProps, conversationId: 'chat-b', projectId: 'project-b' }) - await act(async () => {}) - const currentConversation = deferred() - const currentProject = deferred() - getExecutionStats.mockReturnValueOnce(currentConversation.promise).mockReturnValueOnce(currentProject.promise) - rerender(initialProps) - await act(async () => { - oldConversation.resolve(stats(90)) - oldProject.resolve(stats(99)) - }) - expect(result.current).toEqual(empty) - await act(async () => { - currentConversation.resolve(stats(20)) - currentProject.resolve(stats(30)) - }) - expect(result.current).toEqual({ conversation: stats(20), project: stats(30) }) - }) - - it('does not overlap polling, focus or visibility refreshes while either query is pending', async () => { - const conversation = deferred() - const project = deferred() - getExecutionStats.mockReturnValueOnce(conversation.promise).mockReturnValueOnce(project.promise) - const { result } = renderStats() - await act(async () => { - vi.advanceTimersByTime(15_000) - window.dispatchEvent(new Event('focus')) - document.dispatchEvent(new Event('visibilitychange')) - conversation.resolve(stats(20)) - }) - await act(async () => { - vi.advanceTimersByTime(5_000) - window.dispatchEvent(new Event('focus')) - }) +describe('state-driven duration snapshots', () => { + it('ticks locally for concurrent running segments without polling or writes', async () => { + getExecutionStats.mockImplementation(async () => snapshot(2)) + const { result } = renderHook(() => useExecutionStats('conversation', 'project', true)) + await flush() expect(getExecutionStats).toHaveBeenCalledTimes(2) - expect(result.current).toEqual(empty) - await act(async () => { project.resolve(stats(30)) }) - expect(result.current).toEqual({ conversation: stats(20), project: stats(30) }) - await act(async () => { vi.advanceTimersByTime(4_999) }) + act(() => { vi.advanceTimersByTime(6000) }) + expect(result.current.conversation?.durationMs).toBe(13_000) + expect(result.current.project?.taskDurations[0]?.durationMs).toBe(13_000) expect(getExecutionStats).toHaveBeenCalledTimes(2) - await act(async () => { vi.advanceTimersByTime(1) }) + getExecutionStats.mockImplementation(async () => snapshot(0, 13_000)) + act(() => { listeners.forEach((listener) => listener()) }) + await flush() + act(() => { vi.advanceTimersByTime(60_000) }) + expect(result.current.conversation?.durationMs).toBe(13_000) expect(getExecutionStats).toHaveBeenCalledTimes(4) }) - it('replaces the poller on revision changes and removes timers and listeners on unmount', async () => { - const conversation = deferred() - const project = deferred() - getExecutionStats.mockReturnValueOnce(conversation.promise).mockReturnValueOnce(project.promise) - const { result, rerender, unmount } = renderStats() - await act(async () => { vi.advanceTimersByTime(2_000) }) - rerender({ ...initialProps, revision: '2' }) - await act(async () => {}) - expect(vi.getTimerCount()).toBe(1) - await act(async () => { - conversation.resolve(stats(90)) - project.resolve(stats(99)) - vi.advanceTimersByTime(3_000) - }) - expect(result.current).toEqual({ conversation: stats(10), project: stats(10) }) - expect(getExecutionStats).toHaveBeenCalledTimes(4) - await act(async () => { vi.advanceTimersByTime(2_000) }) - expect(getExecutionStats).toHaveBeenCalledTimes(6) - - const lastConversation = deferred() - const lastProject = deferred() - getExecutionStats.mockReturnValueOnce(lastConversation.promise).mockReturnValueOnce(lastProject.promise) - await act(async () => { window.dispatchEvent(new Event('focus')) }) - unmount() - expect(vi.getTimerCount()).toBe(0) - await act(async () => { - lastConversation.resolve(stats(50)) - lastProject.resolve(stats(60)) - }) - await act(async () => { - vi.advanceTimersByTime(15_000) - window.dispatchEvent(new Event('focus')) - document.dispatchEvent(new Event('visibilitychange')) - }) - expect(getExecutionStats).toHaveBeenCalledTimes(8) - }) - - it.each(['conversation', 'project', 'both'] as const)( - 'clears stale statistics when %s queries fail and recovers on the next poll', async (failure) => { - const { result } = renderStats() - await act(async () => {}) - expect(result.current).toEqual({ conversation: stats(10), project: stats(10) }) - getExecutionStats.mockImplementation(async (input) => { - if (failure === 'both' || - (failure === 'conversation' && 'conversationId' in input) || - (failure === 'project' && 'projectId' in input)) { - throw new Error('stats unavailable') - } - return stats(20) - }) - await act(async () => { vi.advanceTimersByTime(5_000) }) - expect(result.current).toEqual({ - conversation: failure === 'project' ? stats(20) : undefined, - project: failure === 'conversation' ? stats(20) : undefined - }) - getExecutionStats.mockResolvedValue(stats(30)) - await act(async () => { vi.advanceTimersByTime(5_000) }) - expect(result.current).toEqual({ conversation: stats(30), project: stats(30) }) - } - ) - - it('does not query while hidden and resumes when visible', async () => { - const visibility = vi.spyOn(document, 'visibilityState', 'get').mockReturnValue('hidden') - const { result } = renderStats() - await act(async () => { - vi.advanceTimersByTime(15_000) - window.dispatchEvent(new Event('focus')) - document.dispatchEvent(new Event('visibilitychange')) - }) - expect(getExecutionStats).not.toHaveBeenCalled() - expect(result.current).toEqual(empty) - visibility.mockReturnValue('visible') - await act(async () => { document.dispatchEvent(new Event('visibilitychange')) }) - expect(getExecutionStats).toHaveBeenCalledTimes(2) - expect(result.current).toEqual({ conversation: stats(10), project: stats(10) }) - visibility.mockReturnValue('hidden') - await act(async () => { - vi.advanceTimersByTime(10_000) - window.dispatchEvent(new Event('focus')) - }) - expect(getExecutionStats).toHaveBeenCalledTimes(2) - visibility.mockReturnValue('visible') - await act(async () => { window.dispatchEvent(new Event('focus')) }) - expect(getExecutionStats).toHaveBeenCalledTimes(4) - }) - - it.each(['disabled', 'no-project'] as const)('stops querying and hides results when %s', async (mode) => { - const { result, rerender } = renderStats() - await act(async () => {}) - const conversation = deferred() - const project = deferred() - getExecutionStats.mockReturnValueOnce(conversation.promise).mockReturnValueOnce(project.promise) - await act(async () => { vi.advanceTimersByTime(5_000) }) - rerender({ - ...initialProps, enabled: mode !== 'disabled', - projectId: mode === 'no-project' ? undefined : initialProps.projectId - }) - expect(result.current).toEqual(empty) - expect(vi.getTimerCount()).toBe(0) - await act(async () => { - conversation.resolve(stats(90)) - project.reject(new Error('late failure')) - }) - await act(async () => { - vi.advanceTimersByTime(15_000) - window.dispatchEvent(new Event('focus')) - document.dispatchEvent(new Event('visibilitychange')) - }) - expect(getExecutionStats).toHaveBeenCalledTimes(4) - expect(result.current).toEqual(empty) - rerender(initialProps) - await act(async () => {}) - expect(getExecutionStats).toHaveBeenCalledTimes(6) - expect(result.current).toEqual({ conversation: stats(10), project: stats(10) }) - }) - - it('queries only the project when no conversation is selected', async () => { - const { result, rerender } = renderStats() - await act(async () => {}) - getExecutionStats.mockClear() - rerender({ ...initialProps, conversationId: undefined }) + it('unsubscribes while hidden and gets authoritative snapshots on reopening', async () => { + const { result, rerender, unmount } = renderHook(({ enabled }) => useExecutionStats('c', 'p', enabled), + { initialProps: { enabled: true } }) + await flush() + rerender({ enabled: false }) + expect(listeners.size).toBe(0) expect(result.current.conversation).toBeUndefined() - await act(async () => {}) - expect(getExecutionStats.mock.calls).toEqual([[{ projectId: 'project-a' }]]) - expect(result.current).toEqual({ conversation: undefined, project: stats(10) }) + act(() => { vi.advanceTimersByTime(60_000) }) + expect(getExecutionStats).toHaveBeenCalledTimes(2) + getExecutionStats.mockImplementation(async () => snapshot(1, 60_000)) + rerender({ enabled: true }) + await flush() + expect(result.current.conversation?.durationMs).toBe(60_000) + unmount() + expect(listeners.size).toBe(0) + }) + + it('discards old scopes and in-flight snapshots superseded by a state change', async () => { + const pending: Array<(value: ExecutionStats) => void> = [] + getExecutionStats.mockImplementation(() => new Promise((resolve) => pending.push(resolve))) + const { result, rerender } = renderHook(({ id }) => useExecutionStats(id, 'p', true), { initialProps: { id: 'old' } }) + rerender({ id: 'new' }) + act(() => { listeners.forEach((listener) => listener()) }) + await act(async () => { pending[4]!(snapshot(0, 9000)); pending[5]!(snapshot(0, 9000)) }) + await act(async () => { for (const resolve of pending.slice(0, 4)) resolve(snapshot(1, 1000)) }) + expect(result.current.conversation?.durationMs).toBe(9000) + expect(result.current.project?.runningCount).toBe(0) + }) + + it('does not tick idle snapshots and clears a failed scope', async () => { + const { result } = renderHook(() => useExecutionStats('c', 'p', true)) + await flush() + act(() => { vi.advanceTimersByTime(60_000) }) + expect(result.current.conversation?.durationMs).toBe(1000) + expect(getExecutionStats).toHaveBeenCalledTimes(2) + getExecutionStats.mockRejectedValue(new Error('unavailable')) + act(() => { listeners.forEach((listener) => listener()) }) + await flush() + expect(result.current).toEqual({ conversation: undefined, project: undefined }) }) }) diff --git a/src/renderer/src/use-execution-stats.ts b/src/renderer/src/use-execution-stats.ts index 3aa3c710..25703e51 100644 --- a/src/renderer/src/use-execution-stats.ts +++ b/src/renderer/src/use-execution-stats.ts @@ -1,61 +1,56 @@ import { startTransition, useEffect, useState } from 'react' import type { ExecutionStats } from '../../shared/assistant-contracts' -// Keep timing evidence in Main; never extrapolate a stopped or disconnected run. export function useExecutionStats( conversationId: string | undefined, projectId: string | undefined, - revision: string, enabled: boolean ): { conversation?: ExecutionStats; project?: ExecutionStats } { - const [snapshots, setSnapshots] = useState(() => new Map()) - const conversationKey = JSON.stringify([projectId, conversationId]) - const projectKey = JSON.stringify([projectId]) + const key = JSON.stringify([projectId, conversationId]) + const [snapshot, setSnapshot] = useState<{ + key: string; conversation?: ExecutionStats; project?: ExecutionStats + }>() + const [now, setNow] = useState(Date.now) useEffect(() => { if (!enabled || !projectId) return let cancelled = false - let pending = false + let revision = 0 const refresh = async (): Promise => { - if (pending || document.visibilityState === 'hidden') return - pending = true + const current = ++revision const [conversation, project] = await Promise.allSettled([ conversationId ? window.goodbuddy.tasks.getExecutionStats({ conversationId }) : Promise.resolve(undefined), window.goodbuddy.tasks.getExecutionStats({ projectId }) ]) - pending = false - if (!cancelled) startTransition(() => setSnapshots((current) => { - if (cancelled) return current - const next = new Map(current) - // IPC clones cached snapshots. Preserve references when their contents are unchanged - // so idle polling does not rerender App and rebuild task duration maps. - for (const [key, result] of [[conversationKey, conversation], [projectKey, project]] as const) { - const value = result.status === 'fulfilled' ? result.value : undefined - if (JSON.stringify(current.get(key)) === JSON.stringify(value)) continue - if (value) next.set(key, value) - else next.delete(key) - } - return next.size === current.size && - next.get(conversationKey) === current.get(conversationKey) && - next.get(projectKey) === current.get(projectKey) ? current : next - })) + if (cancelled || current !== revision) return + startTransition(() => { + if (cancelled || current !== revision) return + setNow(Date.now()) + setSnapshot({ key, + conversation: conversation.status === 'fulfilled' ? conversation.value : undefined, + project: project.status === 'fulfilled' ? project.value : undefined }) + }) } + const unsubscribe = window.goodbuddy.tasks.onExecutionStatsChanged(() => { void refresh() }) void refresh() - const interval = window.setInterval(() => { void refresh() }, 5_000) - const onFocus = (): void => { void refresh() } - window.addEventListener('focus', onFocus) - document.addEventListener('visibilitychange', onFocus) - return () => { - cancelled = true - window.clearInterval(interval) - window.removeEventListener('focus', onFocus) - document.removeEventListener('visibilitychange', onFocus) - } - }, [conversationId, projectId, conversationKey, projectKey, revision, enabled]) - return { - conversation: enabled && projectId && conversationId - ? snapshots.get(conversationKey) : undefined, - project: enabled && projectId ? snapshots.get(projectKey) : undefined + return () => { cancelled = true; unsubscribe() } + }, [conversationId, projectId, key, enabled]) + + const current = enabled && snapshot?.key === key ? snapshot : undefined + const running = Boolean(current?.conversation?.runningCount || current?.project?.runningCount) + useEffect(() => { + if (!running) return + // Display only: no IPC or persistence on ticks, including in background tabs. + const interval = window.setInterval(() => setNow(Date.now()), 1_000) + return () => window.clearInterval(interval) + }, [running]) + const display = (value: ExecutionStats | undefined): ExecutionStats | undefined => { + if (!value?.runningCount) return value + const elapsed = Math.max(0, now - value.asOf) + return { ...value, durationMs: value.durationMs + elapsed * value.runningCount, + taskDurations: value.taskDurations.map((task) => ({ ...task, + durationMs: task.durationMs + elapsed * task.runningCount })) } } + return { conversation: display(current?.conversation), project: display(current?.project) } } diff --git a/src/shared/assistant-contracts.ts b/src/shared/assistant-contracts.ts index 94bdd8ea..85ef18cd 100644 --- a/src/shared/assistant-contracts.ts +++ b/src/shared/assistant-contracts.ts @@ -14,17 +14,16 @@ export const executionStatsInputSchema = z.union([ ]) export type ExecutionStatsInput = z.infer export interface ExecutionStats { - /** Sum of evidenced top-level reply intervals, including tools and approvals. */ + /** Accumulated running time, including open segments through asOf. */ durationMs: number - requestCount: number - /** Requests/replies with missing timing evidence; confirmed live intervals are complete as of asOf. */ - incompleteRequestCount: number - /** Requests currently leased by Main; does not authorize renderer extrapolation. */ - activeRequestCount: number - /** Epoch milliseconds at query time. */ + /** Number of open segments; the UI adds (now - asOf) * runningCount. */ + runningCount: number + /** Pre-timing history or an interrupted segment has unknown duration. */ + incomplete: boolean + /** Main epoch milliseconds at snapshot time. */ asOf: number /** Project card totals; empty for conversation queries. */ - taskDurations: Array<{ id: string; durationMs: number; incompleteRequestCount: number }> + taskDurations: Array<{ id: string; durationMs: number; runningCount: number; incomplete: boolean }> } export const interactiveWorkModes = ['ask', 'execute'] as const export const workModeSchema = z.enum(interactiveWorkModes) diff --git a/src/shared/contracts.ts b/src/shared/contracts.ts index d8fa622b..7e6faf2b 100644 --- a/src/shared/contracts.ts +++ b/src/shared/contracts.ts @@ -2155,6 +2155,7 @@ export type DesktopApi = { tasks: { list: () => Promise getExecutionStats: (input: ExecutionStatsInput) => Promise + onExecutionStatsChanged: (listener: () => void) => () => void setStatus: ( taskId: string, status: Extract diff --git a/src/shared/ipc-channels.ts b/src/shared/ipc-channels.ts index 7789cddd..dd44ebc4 100644 --- a/src/shared/ipc-channels.ts +++ b/src/shared/ipc-channels.ts @@ -252,6 +252,7 @@ export const ipcChannels = { workspacePathOpen: 'workspace:path:open', tasksList: 'tasks:list', tasksExecutionStats: 'tasks:execution-stats', + tasksExecutionStatsChanged: 'tasks:execution-stats-changed', tasksSetStatus: 'tasks:set-status', activityHistoryGet: 'activity-history:get', activityHistoryReplace: 'activity-history:replace', diff --git a/tests/magic-notes-capture.electron.test.ts b/tests/magic-notes-capture.electron.test.ts index 0bd1d1ae..f1d7cd8b 100644 --- a/tests/magic-notes-capture.electron.test.ts +++ b/tests/magic-notes-capture.electron.test.ts @@ -42,7 +42,6 @@ it('captures messages and full history from App through real Electron preload, I } } ] }) await build({ entryPoints: ['src/preload/index.ts'], outfile: join(directory, 'preload.cjs'), bundle: true, platform: 'node', format: 'cjs', external: ['electron'] }) - await build({ entryPoints: ['src/main/execution-stats-worker.ts'], outfile: join(directory, 'out/main/execution-stats-worker.js'), bundle: true, platform: 'node', format: 'cjs' }) await server.listen() const bootstrap = join(directory, 'bootstrap.cjs') await writeFile(bootstrap, `import(${JSON.stringify(pathToFileURL(driver).href)}).catch(error => { console.error(error); require('electron').app.exit(1) })`) diff --git a/tests/manual-compaction.electron.test.ts b/tests/manual-compaction.electron.test.ts index 13cb1cba..c33d1896 100644 --- a/tests/manual-compaction.electron.test.ts +++ b/tests/manual-compaction.electron.test.ts @@ -34,7 +34,6 @@ it('compacts direct model context from App through preload, IPC, HTTP and SQLite } } ] }) await build({ entryPoints: ['src/preload/index.ts'], outfile: join(directory, 'preload.cjs'), bundle: true, platform: 'node', format: 'cjs', external: ['electron'] }) - await build({ entryPoints: ['src/main/execution-stats-worker.ts'], outfile: join(directory, 'out/main/execution-stats-worker.js'), bundle: true, platform: 'node', format: 'cjs' }) await server.listen() const bootstrap = join(directory, 'bootstrap.cjs') await writeFile(bootstrap, `import(${JSON.stringify(pathToFileURL(driver).href)}).catch(error => { console.error(error); require('electron').app.exit(1) })`) diff --git a/tests/overlay-layering.electron.test.ts b/tests/overlay-layering.electron.test.ts index 647dd29e..79a138b1 100644 --- a/tests/overlay-layering.electron.test.ts +++ b/tests/overlay-layering.electron.test.ts @@ -74,7 +74,6 @@ it.each([ } } ] }) await build({ entryPoints: ['src/preload/index.ts'], outfile: join(directory, 'preload.cjs'), bundle: true, platform: 'node', format: 'cjs', external: ['electron'] }) - await build({ entryPoints: ['src/main/execution-stats-worker.ts'], outfile: join(directory, 'out/main/execution-stats-worker.js'), bundle: true, platform: 'node', format: 'cjs' }) await server.listen() const env: NodeJS.ProcessEnv = { ...process.env, GB_MCP_DIRECTORY: directory, GB_MCP_URL: server.resolvedUrls!.local[0] + 'mcp.html' } delete env.ELECTRON_RUN_AS_NODE diff --git a/tests/terminal-status-probe.electron.test.ts b/tests/terminal-status-probe.electron.test.ts index 54127998..8bf53aa2 100644 --- a/tests/terminal-status-probe.electron.test.ts +++ b/tests/terminal-status-probe.electron.test.ts @@ -31,7 +31,6 @@ it('validates terminal UI through production model HTTP, IPC, persistence and re }) } } ] }) await build({ entryPoints: ['src/preload/index.ts'], outfile: join(directory, 'preload.cjs'), bundle: true, platform: 'node', format: 'cjs', external: ['electron'] }) - await build({ entryPoints: ['src/main/execution-stats-worker.ts'], outfile: join(directory, 'out/main/execution-stats-worker.js'), bundle: true, platform: 'node', format: 'cjs' }) await server.listen() const bootstrap = join(directory, 'bootstrap.cjs') await writeFile(bootstrap, `import(${JSON.stringify(pathToFileURL(driver).href)}).catch(e=>{console.error(e);require('electron').app.exit(1)})`)