返回 oh-my-ppt
edit-deck-batch-flow.ts
根目录 / src / main / generation / edit-deck-batch-flow.ts
1 import fs from 'fs'
2 import pLimit from 'p-limit'
3 import log from 'electron-log/main.js'
4 import type { GenerateChunkEvent } from '@shared/generation'
5 import type { EditedPageDescriptor, InvalidEditedPage } from './generation-utils'
6 import { isCancellationMessage } from './status-utils'
7
8 export const BATCH_EDIT_CHUNK_SIZE = 2
9 export const BATCH_EDIT_LAUNCH_STAGGER_MS = 100
10 export const BATCH_EDIT_HEARTBEAT_INTERVAL_MS = 15_000
11
12 export type DeckEditBatchPageRef = {
13 id: string
14 pageNumber: number
15 title: string
16 pageId: string
17 htmlPath: string
18 }
19
20 type FileSnapshot = {
21 exists: boolean
22 content: string
23 }
24
25 type DeckEditBatchSnapshot = {
26 indexPath: string
27 indexFile: FileSnapshot
28 pages: Map<string, FileSnapshot>
29 }
30
31 export type DeckEditCompletedBatch = {
32 status: 'completed'
33 pageId: string
34 changedPages: EditedPageDescriptor[]
35 retryCount: number
36 }
37
38 export type DeckEditFailedBatch = {
39 status: 'failed'
40 pageId: string
41 reason: string
42 retryCount: number
43 }
44
45 export type DeckEditBatchResult = DeckEditCompletedBatch | DeckEditFailedBatch
46
47 export class DeckEditIndexMutationError extends Error {
48 constructor() {
49 super('主会话 deck 编辑不允许修改 index.html,本次检测到壳层变更并已恢复。')
50 this.name = 'DeckEditIndexMutationError'
51 }
52 }
53
54 class DeckEditNoChangeError extends Error {
55 constructor() {
56 super('当前页面编辑没有检测到落盘变化。')
57 this.name = 'DeckEditNoChangeError'
58 }
59 }
60
61 class DeckEditPageValidationError extends Error {
62 constructor(readonly invalidPages: InvalidEditedPage[]) {
63 super(
64 invalidPages
65 .map((item) => `${item.page.pageId}(${item.page.title}):${item.reason}`)
66 .join(';')
67 )
68 this.name = 'DeckEditPageValidationError'
69 }
70 }
71
72 type RunPageAttemptArgs = {
73 pageId: string
74 pageNumber: number
75 userMessage: string
76 isRetry: boolean
77 emit: (chunk: GenerateChunkEvent) => void
78 }
79
80 export type ExecuteDeckEditBatchFlowArgs = {
81 pageRefs: DeckEditBatchPageRef[]
82 indexPath: string
83 originalUserMessage: string
84 runId: string
85 appLocale: 'zh' | 'en'
86 signal?: AbortSignal
87 launchStaggerMs?: number
88 heartbeatIntervalMs?: number
89 emit: (chunk: GenerateChunkEvent) => void
90 runPageAttempt: (args: RunPageAttemptArgs) => Promise<void>
91 validateChangedPages: (pages: EditedPageDescriptor[]) => InvalidEditedPage[]
92 buildRetryMessage: (args: {
93 baseMessage: string
94 error: unknown
95 kind: 'no_change' | 'validation' | 'agent'
96 }) => string | null
97 onPageCompleted?: (result: DeckEditCompletedBatch) => Promise<void>
98 onPageFailed?: (result: DeckEditFailedBatch) => Promise<void>
99 }
100
101 export function buildDeckEditPageUserMessage(args: {
102 originalUserMessage: string
103 pageId: string
104 }): string {
105 return [
106 args.originalUserMessage,
107 '',
108 'Page edit context:',
109 `- Edit ONLY this page: ${args.pageId}.`,
110 '- You may read other pages for visual reference, but you must not write them.'
111 ].join('\n')
112 }
113
114 const readFileSnapshot = async (filePath: string): Promise<FileSnapshot> => {
115 if (!fs.existsSync(filePath)) return { exists: false, content: '' }
116 return {
117 exists: true,
118 content: await fs.promises.readFile(filePath, 'utf-8')
119 }
120 }
121
122 const captureSnapshot = async (
123 pageRefs: DeckEditBatchPageRef[],
124 indexPath: string
125 ): Promise<DeckEditBatchSnapshot> => {
126 const pages = new Map<string, FileSnapshot>()
127 const pageSnapshots = await Promise.all(
128 pageRefs.map(async (page) => ({
129 pageId: page.pageId,
130 file: await readFileSnapshot(page.htmlPath)
131 }))
132 )
133 for (const page of pageSnapshots) pages.set(page.pageId, page.file)
134 return {
135 indexPath,
136 indexFile: await readFileSnapshot(indexPath),
137 pages
138 }
139 }
140
141 const restoreFileSnapshot = async (filePath: string, snapshot: FileSnapshot): Promise<void> => {
142 if (snapshot.exists) {
143 await fs.promises.writeFile(filePath, snapshot.content, 'utf-8')
144 return
145 }
146 await fs.promises.rm(filePath, { force: true })
147 }
148
149 const restoreSnapshot = async (
150 snapshot: DeckEditBatchSnapshot,
151 pageRefs: DeckEditBatchPageRef[]
152 ): Promise<void> => {
153 await Promise.all(
154 pageRefs.map((page) =>
155 restoreFileSnapshot(
156 page.htmlPath,
157 snapshot.pages.get(page.pageId) || { exists: false, content: '' }
158 )
159 )
160 )
161 await restoreFileSnapshot(snapshot.indexPath, snapshot.indexFile)
162 }
163
164 const restorePageSnapshots = async (
165 snapshot: DeckEditBatchSnapshot,
166 pageRefs: DeckEditBatchPageRef[]
167 ): Promise<void> => {
168 await Promise.all(
169 pageRefs.map((page) =>
170 restoreFileSnapshot(
171 page.htmlPath,
172 snapshot.pages.get(page.pageId) || { exists: false, content: '' }
173 )
174 )
175 )
176 }
177
178 const hasIndexChanged = async (snapshot: DeckEditBatchSnapshot): Promise<boolean> => {
179 const current = await readFileSnapshot(snapshot.indexPath)
180 return (
181 current.exists !== snapshot.indexFile.exists || current.content !== snapshot.indexFile.content
182 )
183 }
184
185 const readChangedPages = async (
186 snapshot: DeckEditBatchSnapshot,
187 pageRefs: DeckEditBatchPageRef[]
188 ): Promise<EditedPageDescriptor[]> => {
189 const changedPages: EditedPageDescriptor[] = []
190 for (const page of pageRefs) {
191 if (!fs.existsSync(page.htmlPath)) continue
192 const html = await fs.promises.readFile(page.htmlPath, 'utf-8')
193 const before = snapshot.pages.get(page.pageId)
194 if (before?.exists && before.content === html) continue
195 changedPages.push({ ...page, html })
196 }
197 return changedPages
198 }
199
200 const errorMessage = (error: unknown): string =>
201 error instanceof Error && error.message.length > 0 ? error.message : String(error || '未知错误')
202
203 const isCancellationError = (error: unknown, signal?: AbortSignal): boolean =>
204 Boolean(signal?.aborted) || isCancellationMessage(errorMessage(error))
205
206 const sleep = (ms: number, signal?: AbortSignal): Promise<void> =>
207 new Promise((resolve, reject) => {
208 if (signal?.aborted) {
209 reject(new Error('生成已取消'))
210 return
211 }
212 let timer: ReturnType<typeof setTimeout> | undefined
213 const onAbort = (): void => {
214 cleanup()
215 reject(new Error('生成已取消'))
216 }
217 const cleanup = (): void => {
218 if (timer) clearTimeout(timer)
219 signal?.removeEventListener('abort', onAbort)
220 }
221 timer = setTimeout(() => {
222 cleanup()
223 resolve()
224 }, ms)
225 signal?.addEventListener('abort', onAbort, { once: true })
226 })
227
228 const computeGlobalProgress = (totalPages: number, pageProgress: Map<string, number>): number => {
229 const total = Math.max(1, totalPages)
230 let sum = 0
231 for (const progress of pageProgress.values()) {
232 sum += Math.max(0, Math.min(100, progress))
233 }
234 return Math.round(10 + (sum / total) * 0.8)
235 }
236
237 const remapChunk = (
238 chunk: GenerateChunkEvent,
239 args: {
240 totalPages: number
241 pageNumber: number
242 pageId: string
243 appLocale?: 'zh' | 'en'
244 pageProgress: Map<string, number>
245 lastProgress: { value: number }
246 }
247 ): GenerateChunkEvent => {
248 if (!('progress' in chunk.payload) || typeof chunk.payload.progress !== 'number') return chunk
249 const previousPageProgress = args.pageProgress.get(args.pageId) || 0
250 args.pageProgress.set(args.pageId, Math.max(previousPageProgress, chunk.payload.progress))
251 const progress = Math.max(
252 args.lastProgress.value,
253 computeGlobalProgress(args.totalPages, args.pageProgress)
254 )
255 args.lastProgress.value = progress
256 const reportsCompletedStep =
257 chunk.type === 'llm_status' && /完成|complete|done/i.test(chunk.payload.label)
258 return {
259 ...chunk,
260 payload: {
261 ...chunk.payload,
262 label:
263 reportsCompletedStep
264 ? args.appLocale === 'en'
265 ? `P${args.pageNumber} step completed, validating page`
266 : `P${args.pageNumber} 当前步骤完成,正在校验页面`
267 : chunk.type === 'llm_status' &&
268 /理解|分析|规划|准备|启动|生成|编辑|完成|understand|analyz|plan|prepar|start|generat|edit|complet/i.test(
269 chunk.payload.label
270 )
271 ? args.appLocale === 'en'
272 ? `Editing P${args.pageNumber}`
273 : `正在编辑 P${args.pageNumber}`
274 : chunk.payload.label,
275 progress,
276 currentPage: args.pageNumber,
277 totalPages: args.totalPages
278 }
279 } as GenerateChunkEvent
280 }
281
282 export async function executeDeckEditBatchFlow(
283 args: ExecuteDeckEditBatchFlowArgs
284 ): Promise<DeckEditBatchResult[]> {
285 const batchStartedAt = Date.now()
286 const totalPages = args.pageRefs.length
287 const launchStaggerMs = Math.max(
288 0,
289 Math.floor(args.launchStaggerMs ?? BATCH_EDIT_LAUNCH_STAGGER_MS)
290 )
291 const heartbeatIntervalMs = Math.max(
292 0,
293 Math.floor(args.heartbeatIntervalMs ?? BATCH_EDIT_HEARTBEAT_INTERVAL_MS)
294 )
295 const operationSnapshot = await captureSnapshot(args.pageRefs, args.indexPath)
296 const results: DeckEditBatchResult[] = []
297 const lastProgress = { value: 10 }
298 const pageProgress = new Map<string, number>()
299 const limit = pLimit(BATCH_EDIT_CHUNK_SIZE)
300 const queuedAtByPageId = new Map(args.pageRefs.map((page) => [page.pageId, Date.now()]))
301 let fatalError: unknown = null
302
303 log.info('[deck-edit:batch] started', {
304 runId: args.runId,
305 totalPages,
306 concurrency: BATCH_EDIT_CHUNK_SIZE,
307 launchStaggerMs,
308 heartbeatIntervalMs
309 })
310 for (const page of args.pageRefs) {
311 log.info('[deck-edit:page] queued', {
312 runId: args.runId,
313 pageId: page.pageId,
314 pageNumber: page.pageNumber,
315 title: page.title
316 })
317 }
318
319 const emitPageProgress = (
320 page: DeckEditBatchPageRef,
321 label: string,
322 detail?: string
323 ): void => {
324 const progress = Math.max(lastProgress.value, computeGlobalProgress(totalPages, pageProgress))
325 lastProgress.value = progress
326 args.emit({
327 type: 'llm_status',
328 payload: {
329 runId: args.runId,
330 stage: 'editing',
331 label,
332 detail,
333 progress,
334 currentPage: page.pageNumber,
335 totalPages
336 }
337 })
338 }
339
340 const runPageWorker = async (
341 page: DeckEditBatchPageRef,
342 pageIndex: number
343 ): Promise<DeckEditBatchResult> => {
344 const workerStartedAt = Date.now()
345 log.info('[deck-edit:page] worker started', {
346 runId: args.runId,
347 pageId: page.pageId,
348 pageNumber: page.pageNumber,
349 queueWaitMs: workerStartedAt - (queuedAtByPageId.get(page.pageId) || workerStartedAt)
350 })
351 if (fatalError) throw fatalError
352 if (args.signal?.aborted) throw new Error('生成已取消')
353 const queueStaggerIndex = pageIndex % BATCH_EDIT_CHUNK_SIZE
354 const pageSnapshot: DeckEditBatchSnapshot = {
355 indexPath: operationSnapshot.indexPath,
356 indexFile: operationSnapshot.indexFile,
357 pages: new Map([
358 [page.pageId, operationSnapshot.pages.get(page.pageId) || { exists: false, content: '' }]
359 ])
360 }
361 const baseMessage = buildDeckEditPageUserMessage({
362 originalUserMessage: args.originalUserMessage,
363 pageId: page.pageId
364 })
365 let attemptMessage = baseMessage
366 let retryUsed = false
367
368 if (queueStaggerIndex > 0 && launchStaggerMs > 0) {
369 log.info('[deck-edit:page] launch stagger', {
370 runId: args.runId,
371 pageId: page.pageId,
372 pageNumber: page.pageNumber,
373 delayMs: queueStaggerIndex * launchStaggerMs
374 })
375 await sleep(queueStaggerIndex * launchStaggerMs, args.signal)
376 }
377
378 while (true) {
379 if (fatalError) throw fatalError
380 try {
381 const attemptStartedAt = Date.now()
382 let lastActivityAt = attemptStartedAt
383 let activityCount = 0
384 let silenceReported = false
385 const attempt = retryUsed ? 2 : 1
386 log.info('[deck-edit:page] attempt started', {
387 runId: args.runId,
388 pageId: page.pageId,
389 pageNumber: page.pageNumber,
390 attempt,
391 isRetry: retryUsed
392 })
393 const heartbeat =
394 heartbeatIntervalMs > 0
395 ? setInterval(() => {
396 const now = Date.now()
397 if (now - lastActivityAt < heartbeatIntervalMs) return
398 const silentForMs = now - lastActivityAt
399 silenceReported = true
400 log.warn('[deck-edit:page] model response silent', {
401 runId: args.runId,
402 pageId: page.pageId,
403 pageNumber: page.pageNumber,
404 attempt,
405 elapsedMs: now - attemptStartedAt,
406 silentForMs,
407 activityCount
408 })
409 lastActivityAt = now
410 }, heartbeatIntervalMs)
411 : undefined
412 try {
413 await args.runPageAttempt({
414 pageId: page.pageId,
415 pageNumber: page.pageNumber,
416 userMessage: attemptMessage,
417 isRetry: retryUsed,
418 emit: (chunk) => {
419 const now = Date.now()
420 activityCount += 1
421 if (activityCount === 1) {
422 log.info('[deck-edit:page] first agent activity', {
423 runId: args.runId,
424 pageId: page.pageId,
425 pageNumber: page.pageNumber,
426 attempt,
427 elapsedMs: now - attemptStartedAt,
428 eventType: chunk.type
429 })
430 } else if (silenceReported) {
431 log.info('[deck-edit:page] agent activity resumed', {
432 runId: args.runId,
433 pageId: page.pageId,
434 pageNumber: page.pageNumber,
435 attempt,
436 elapsedMs: now - attemptStartedAt,
437 silentForMs: now - lastActivityAt,
438 eventType: chunk.type
439 })
440 silenceReported = false
441 }
442 lastActivityAt = now
443 args.emit(
444 remapChunk(chunk, {
445 totalPages,
446 pageNumber: page.pageNumber,
447 pageId: page.pageId,
448 appLocale: args.appLocale,
449 pageProgress,
450 lastProgress
451 })
452 )
453 }
454 })
455 } finally {
456 if (heartbeat) clearInterval(heartbeat)
457 }
458 log.info('[deck-edit:page] agent attempt returned', {
459 runId: args.runId,
460 pageId: page.pageId,
461 pageNumber: page.pageNumber,
462 attempt,
463 elapsedMs: Date.now() - attemptStartedAt,
464 activityCount
465 })
466 if (args.signal?.aborted) throw new Error('生成已取消')
467 if (await hasIndexChanged(operationSnapshot)) throw new DeckEditIndexMutationError()
468 const changedPages = await readChangedPages(pageSnapshot, [page])
469 if (changedPages.length === 0) throw new DeckEditNoChangeError()
470 const invalidPages = args.validateChangedPages(changedPages)
471 if (invalidPages.length > 0) throw new DeckEditPageValidationError(invalidPages)
472 pageProgress.set(page.pageId, 100)
473 log.info('[deck-edit:page] completed', {
474 runId: args.runId,
475 pageId: page.pageId,
476 pageNumber: page.pageNumber,
477 attempt,
478 workerElapsedMs: Date.now() - workerStartedAt,
479 changedPageCount: changedPages.length
480 })
481 emitPageProgress(
482 page,
483 retryUsed
484 ? args.appLocale === 'en'
485 ? `P${page.pageNumber} retry succeeded`
486 : `P${page.pageNumber} 重试成功`
487 : args.appLocale === 'en'
488 ? `P${page.pageNumber} editing completed`
489 : `P${page.pageNumber} 编辑完成`
490 )
491 return {
492 status: 'completed',
493 pageId: page.pageId,
494 changedPages,
495 retryCount: retryUsed ? 1 : 0
496 }
497 } catch (error) {
498 const reason = errorMessage(error)
499 const errorName = error instanceof Error ? error.name : 'UnknownError'
500 if (
501 isCancellationError(error, args.signal) ||
502 error instanceof DeckEditIndexMutationError
503 ) {
504 log.warn('[deck-edit:page] fatal attempt error', {
505 runId: args.runId,
506 pageId: page.pageId,
507 pageNumber: page.pageNumber,
508 attempt: retryUsed ? 2 : 1,
509 errorName,
510 reason
511 })
512 fatalError = error
513 throw error
514 }
515 if (await hasIndexChanged(operationSnapshot)) {
516 const indexMutationError = new DeckEditIndexMutationError()
517 fatalError = indexMutationError
518 throw indexMutationError
519 }
520 await restorePageSnapshots(pageSnapshot, [page])
521 const retryMessage = !retryUsed
522 ? args.buildRetryMessage({
523 baseMessage,
524 error,
525 kind:
526 error instanceof DeckEditNoChangeError
527 ? 'no_change'
528 : error instanceof DeckEditPageValidationError
529 ? 'validation'
530 : 'agent'
531 })
532 : null
533 if (retryMessage) {
534 log.warn('[deck-edit:page] attempt failed; retry scheduled', {
535 runId: args.runId,
536 pageId: page.pageId,
537 pageNumber: page.pageNumber,
538 attempt: 1,
539 errorName,
540 reason
541 })
542 emitPageProgress(
543 page,
544 args.appLocale === 'en'
545 ? `P${page.pageNumber} first attempt failed, preparing to retry`
546 : `P${page.pageNumber} 首次处理失败,准备重试`,
547 errorMessage(error)
548 )
549 retryUsed = true
550 attemptMessage = retryMessage
551 if (launchStaggerMs > 0) {
552 await sleep((queueStaggerIndex + 1) * launchStaggerMs, args.signal)
553 }
554 continue
555 }
556 const failResult: DeckEditFailedBatch = {
557 status: 'failed',
558 pageId: page.pageId,
559 reason,
560 retryCount: retryUsed ? 1 : 0
561 }
562 log.error('[deck-edit:page] failed', {
563 runId: args.runId,
564 pageId: page.pageId,
565 pageNumber: page.pageNumber,
566 attempt: retryUsed ? 2 : 1,
567 workerElapsedMs: Date.now() - workerStartedAt,
568 errorName,
569 reason,
570 stack: error instanceof Error ? error.stack : undefined
571 })
572 pageProgress.set(page.pageId, 100)
573 emitPageProgress(
574 page,
575 args.appLocale === 'en'
576 ? `P${page.pageNumber} editing failed`
577 : `P${page.pageNumber} 编辑失败`,
578 failResult.reason
579 )
580 return failResult
581 }
582 }
583 }
584
585 try {
586 const settled = await Promise.allSettled(
587 args.pageRefs.map((page, pageIndex) => limit(() => runPageWorker(page, pageIndex)))
588 )
589 const rejected = settled.find((item) => item.status === 'rejected')
590 if (rejected?.status === 'rejected') throw rejected.reason
591 if (await hasIndexChanged(operationSnapshot)) throw new DeckEditIndexMutationError()
592
593 for (const item of settled) {
594 results.push((item as PromiseFulfilledResult<DeckEditBatchResult>).value)
595 }
596 log.info('[deck-edit:batch] workers settled', {
597 runId: args.runId,
598 elapsedMs: Date.now() - batchStartedAt,
599 completedPageCount: results.filter((item) => item.status === 'completed').length,
600 failedPageCount: results.filter((item) => item.status === 'failed').length
601 })
602 } catch (error) {
603 log.error('[deck-edit:batch] aborted; restoring snapshot', {
604 runId: args.runId,
605 elapsedMs: Date.now() - batchStartedAt,
606 reason: errorMessage(error),
607 stack: error instanceof Error ? error.stack : undefined
608 })
609 await restoreSnapshot(operationSnapshot, args.pageRefs)
610 throw error
611 }
612
613 // Publish only after every worker has settled and the final global invariant check passed.
614 // Callback failures are persistence failures: do not roll back generated files or retry the model.
615 for (const result of results) {
616 if (result.status === 'completed') {
617 await args.onPageCompleted?.(result)
618 } else {
619 await args.onPageFailed?.(result)
620 }
621 }
622
623 log.info('[deck-edit:batch] completed', {
624 runId: args.runId,
625 elapsedMs: Date.now() - batchStartedAt,
626 completedPageCount: results.filter((item) => item.status === 'completed').length,
627 failedPageCount: results.filter((item) => item.status === 'failed').length
628 })
629
630 return results
631 }
632
632 lines TYPESCRIPT