| 1 | import { ipcMain } from 'electron' |
| 2 | import crypto from 'crypto' |
| 3 | import log from 'electron-log/main.js' |
| 4 | import type { GenerateStartPayload } from '@shared/generation' |
| 5 | import type { IpcContext } from '../ipc/context' |
| 6 | import { executeDeckAllPageEditGeneration } from '../generation/edit-deck-allpage-flow' |
| 7 | import { resolveEditContext } from '../generation/edit-flow' |
| 8 | import { createEmitAssistantMessage } from '../generation/generation-utils' |
| 9 | import { createGenerationContext, normalizeGeneratePayload } from '../generation/context' |
| 10 | import type { EditContext } from '../generation/types' |
| 11 | import { isCancellationMessage, normalizeRestoredSessionStatus } from '../generation/status-utils' |
| 12 | import { JobCoordinator, sessionLockKey, type JobLease } from '../agent-runtime' |
| 13 | import { settleEditJobFailure, settleEditJobSuccess } from './edit-job-finalization' |
| 14 | |
| 15 | type ActiveDeckEditJob = { |
| 16 | sessionId: string |
| 17 | runId: string |
| 18 | lease: JobLease |
| 19 | context: EditContext |
| 20 | } |
| 21 | |
| 22 | type DeckEditRunSnapshot = { |
| 23 | sessionId: string |
| 24 | runId: string | null |
| 25 | status: 'idle' | 'queued' | 'running' | 'completed' | 'failed' | 'cancelled' |
| 26 | hasActiveRun: boolean |
| 27 | progress: number |
| 28 | totalPages: number |
| 29 | completedPageCount: number |
| 30 | failedPageCount: number |
| 31 | events: never[] |
| 32 | error: string | null |
| 33 | startedAt: number | null |
| 34 | updatedAt: number | null |
| 35 | kind: 'deck-edit' |
| 36 | retryPayload?: GenerateStartPayload |
| 37 | } |
| 38 | |
| 39 | const buildDeckEditRetryPayload = ( |
| 40 | input: ReturnType<typeof normalizeGeneratePayload>, |
| 41 | modelConfigId?: string |
| 42 | ): GenerateStartPayload => ({ |
| 43 | sessionId: input.sessionId, |
| 44 | modelConfigId: modelConfigId || input.modelConfigId, |
| 45 | userMessage: input.rawUserMessage, |
| 46 | type: 'page', |
| 47 | chatType: 'main', |
| 48 | selectPageIds: input.selectPageIds, |
| 49 | imagePaths: input.rawImagePaths, |
| 50 | videoPaths: input.rawVideoPaths, |
| 51 | docPaths: input.rawDocPaths |
| 52 | }) |
| 53 | |
| 54 | const parseDeckEditRetryPayload = ( |
| 55 | metadata: string | null, |
| 56 | sessionId: string |
| 57 | ): GenerateStartPayload | undefined => { |
| 58 | if (!metadata) return undefined |
| 59 | try { |
| 60 | const parsed = JSON.parse(metadata) as { retryPayload?: unknown } |
| 61 | const input = normalizeGeneratePayload({ |
| 62 | ...(parsed.retryPayload && typeof parsed.retryPayload === 'object' |
| 63 | ? parsed.retryPayload |
| 64 | : {}), |
| 65 | sessionId, |
| 66 | type: 'page', |
| 67 | chatType: 'main' |
| 68 | }) |
| 69 | if (!input.rawUserMessage.trim()) return undefined |
| 70 | return buildDeckEditRetryPayload(input, input.modelConfigId) |
| 71 | } catch { |
| 72 | return undefined |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | export class DeckEditJobService { |
| 77 | private activeJobs = new Map<string, ActiveDeckEditJob>() |
| 78 | private reservedJobIds = new Map<string, string>() |
| 79 | |
| 80 | constructor(private ctx: IpcContext, private coordinator: JobCoordinator) {} |
| 81 | |
| 82 | async start(event: Electron.IpcMainInvokeEvent, payload: unknown): Promise<{ |
| 83 | success: boolean |
| 84 | runId?: string |
| 85 | alreadyRunning?: boolean |
| 86 | }> { |
| 87 | const input = normalizeGeneratePayload(payload) |
| 88 | if (!input.sessionId) throw new Error('sessionId 不能为空') |
| 89 | if (input.requestedType !== 'page' || input.chatType !== 'main') { |
| 90 | throw new Error('deck-edit:start 仅支持主会话批量编辑') |
| 91 | } |
| 92 | |
| 93 | const reservation = await this.coordinator.reserve({ |
| 94 | jobId: crypto.randomUUID(), |
| 95 | domain: 'edit', |
| 96 | owner: { kind: 'session', id: input.sessionId }, |
| 97 | claims: { write: [sessionLockKey(input.sessionId)] }, |
| 98 | wait: 'fail' |
| 99 | }) |
| 100 | if (reservation.status === 'busy') { |
| 101 | return { success: true, runId: reservation.conflictingJobId, alreadyRunning: true } |
| 102 | } |
| 103 | const lease = reservation.lease |
| 104 | this.reservedJobIds.set(input.sessionId, lease.jobId) |
| 105 | let context: EditContext | null = null |
| 106 | let jobCreated = false |
| 107 | try { |
| 108 | const editContext = await resolveEditContext(createGenerationContext(this.ctx), event, payload, { |
| 109 | runId: lease.jobId, |
| 110 | abortSignal: lease.signal |
| 111 | }) |
| 112 | context = editContext |
| 113 | if (lease.signal.aborted) throw new Error('生成已取消') |
| 114 | if (editContext.runId !== lease.jobId) { |
| 115 | throw new Error('批量编辑 runId 与 JobCoordinator lease 不一致') |
| 116 | } |
| 117 | |
| 118 | const totalPages = Math.max(1, input.selectPageIds.length || editContext.totalPages) |
| 119 | const retryPayload = buildDeckEditRetryPayload(input, editContext.modelConfigId) |
| 120 | await this.ctx.db.createGenerationRunWithSessionJob({ |
| 121 | run: { |
| 122 | id: editContext.runId, |
| 123 | sessionId: editContext.sessionId, |
| 124 | mode: 'edit', |
| 125 | totalPages, |
| 126 | modelConfigId: editContext.modelConfigId, |
| 127 | metadata: { |
| 128 | jobType: 'deck-edit', |
| 129 | editScope: 'deck', |
| 130 | selectPageIds: input.selectPageIds, |
| 131 | modelConfigId: editContext.modelConfigId, |
| 132 | modelConfigName: editContext.modelConfigName, |
| 133 | provider: editContext.provider, |
| 134 | model: editContext.model, |
| 135 | retryPayload |
| 136 | } |
| 137 | }, |
| 138 | job: { |
| 139 | id: editContext.runId, |
| 140 | sessionId: editContext.sessionId, |
| 141 | kind: 'deck-edit', |
| 142 | status: 'active', |
| 143 | previousSessionStatus: normalizeRestoredSessionStatus(editContext.previousSessionStatus), |
| 144 | totalPages |
| 145 | } |
| 146 | }) |
| 147 | jobCreated = true |
| 148 | if (lease.signal.aborted) throw new Error('生成已取消') |
| 149 | editContext.skipGenerationRunCreation = true |
| 150 | this.ctx.beginSessionRunState({ |
| 151 | sessionId: editContext.sessionId, |
| 152 | runId: editContext.runId, |
| 153 | mode: 'edit', |
| 154 | kind: 'deck-edit', |
| 155 | activityKind: 'deck-edit', |
| 156 | totalPages, |
| 157 | previousSessionStatus: editContext.previousSessionStatus, |
| 158 | status: 'running' |
| 159 | }) |
| 160 | |
| 161 | const job: ActiveDeckEditJob = { |
| 162 | sessionId: editContext.sessionId, |
| 163 | runId: editContext.runId, |
| 164 | lease, |
| 165 | context: editContext |
| 166 | } |
| 167 | this.activeJobs.set(editContext.sessionId, job) |
| 168 | void this.run(job) |
| 169 | return { success: true, runId: editContext.runId } |
| 170 | } catch (error) { |
| 171 | try { |
| 172 | if (context) { |
| 173 | const message = error instanceof Error ? error.message : String(error || '') |
| 174 | await settleEditJobFailure({ |
| 175 | ctx: this.ctx, |
| 176 | context, |
| 177 | error, |
| 178 | cancelled: lease.signal.aborted || isCancellationMessage(message), |
| 179 | hasPersistedJob: jobCreated, |
| 180 | logPrefix: '[deck-edit:job]' |
| 181 | }) |
| 182 | } |
| 183 | } finally { |
| 184 | lease.release() |
| 185 | this.reservedJobIds.delete(input.sessionId) |
| 186 | if (context) this.ctx.agentManager.removeSession(context.sessionId) |
| 187 | } |
| 188 | throw error |
| 189 | } |
| 190 | } |
| 191 | |
| 192 | async retry(event: Electron.IpcMainInvokeEvent, payload: unknown): Promise<{ |
| 193 | success: boolean |
| 194 | runId?: string |
| 195 | alreadyRunning?: boolean |
| 196 | failedPageCount: number |
| 197 | }> { |
| 198 | const record = payload && typeof payload === 'object' ? (payload as Record<string, unknown>) : {} |
| 199 | const sessionId = typeof record.sessionId === 'string' ? record.sessionId.trim() : '' |
| 200 | const failedRunId = |
| 201 | typeof record.failedRunId === 'string' ? record.failedRunId.trim() || undefined : undefined |
| 202 | const userMessage = typeof record.userMessage === 'string' ? record.userMessage.trim() : '' |
| 203 | if (!sessionId) throw new Error('sessionId 不能为空') |
| 204 | if (!userMessage) throw new Error('重试编辑失败：缺少原始编辑指令') |
| 205 | |
| 206 | const failedPages = failedRunId |
| 207 | ? await this.getFailedPagesForRun(sessionId, failedRunId) |
| 208 | : await this.ctx.db.listLatestFailedGenerationPages(sessionId) |
| 209 | let failedPageIds = Array.from( |
| 210 | new Set(failedPages.map((page) => page.page_id).filter((pageId) => pageId.length > 0)) |
| 211 | ) |
| 212 | if (failedPageIds.length === 0) { |
| 213 | failedPageIds = normalizeGeneratePayload(record).selectPageIds |
| 214 | } |
| 215 | if (failedPageIds.length === 0) { |
| 216 | failedPageIds = (await this.ctx.db.listSessionPages(sessionId)).map((page) => page.file_slug) |
| 217 | } |
| 218 | if (failedPageIds.length === 0) return { success: true, failedPageCount: 0 } |
| 219 | |
| 220 | const result = await this.start(event, { |
| 221 | ...record, |
| 222 | sessionId, |
| 223 | userMessage, |
| 224 | type: 'page', |
| 225 | chatType: 'main', |
| 226 | selectPageIds: failedPageIds, |
| 227 | persistUserMessage: false |
| 228 | }) |
| 229 | return { ...result, failedPageCount: 0 } |
| 230 | } |
| 231 | |
| 232 | async cancel(sessionId: string): Promise<boolean> { |
| 233 | const job = this.activeJobs.get(sessionId) |
| 234 | if (!job) { |
| 235 | const jobId = this.reservedJobIds.get(sessionId) |
| 236 | return jobId ? this.coordinator.cancel(jobId) : false |
| 237 | } |
| 238 | return this.coordinator.cancel(job.lease.jobId) |
| 239 | } |
| 240 | |
| 241 | async getState(sessionId: string): Promise<DeckEditRunSnapshot> { |
| 242 | const activeState = this.ctx.sessionRunStates.get(sessionId) |
| 243 | if (activeState?.activityKind === 'deck-edit') { |
| 244 | const run = await this.ctx.db.getGenerationRun(activeState.runId) |
| 245 | return { |
| 246 | sessionId, |
| 247 | runId: activeState.runId, |
| 248 | status: activeState.status, |
| 249 | hasActiveRun: activeState.status === 'queued' || activeState.status === 'running', |
| 250 | progress: activeState.progress, |
| 251 | totalPages: activeState.totalPages, |
| 252 | completedPageCount: activeState.completedPageKeys.length, |
| 253 | failedPageCount: activeState.failedPageKeys.length, |
| 254 | events: [], |
| 255 | error: activeState.error, |
| 256 | startedAt: activeState.startedAt, |
| 257 | updatedAt: activeState.updatedAt, |
| 258 | kind: 'deck-edit', |
| 259 | retryPayload: parseDeckEditRetryPayload(run?.metadata || null, sessionId) |
| 260 | } |
| 261 | } |
| 262 | |
| 263 | const job = await this.ctx.db.getLatestSessionJob(sessionId, ['deck-edit']) |
| 264 | const run = job ? await this.ctx.db.getGenerationRun(job.id) : undefined |
| 265 | const retryPayload = parseDeckEditRetryPayload(run?.metadata || null, sessionId) |
| 266 | if (job?.status === 'active') { |
| 267 | return { |
| 268 | sessionId, |
| 269 | runId: job.id, |
| 270 | status: 'running', |
| 271 | hasActiveRun: true, |
| 272 | progress: 0, |
| 273 | totalPages: job.total_pages || 1, |
| 274 | completedPageCount: 0, |
| 275 | failedPageCount: 0, |
| 276 | events: [], |
| 277 | error: null, |
| 278 | startedAt: job.activated_at || job.created_at, |
| 279 | updatedAt: job.updated_at, |
| 280 | kind: 'deck-edit', |
| 281 | retryPayload |
| 282 | } |
| 283 | } |
| 284 | |
| 285 | const generationPages = run ? await this.ctx.db.listGenerationPages(run.id) : [] |
| 286 | const completedPageCount = generationPages.filter((page) => page.status === 'completed').length |
| 287 | const persistedFailedPageCount = generationPages.filter((page) => page.status === 'failed').length |
| 288 | const userCancelled = job?.status === 'aborted' && job.abort_reason === 'cancelled' |
| 289 | const interrupted = job?.status === 'aborted' && !userCancelled |
| 290 | const failed = run?.status === 'failed' || run?.status === 'partial' || interrupted |
| 291 | const failedPageCount = failed |
| 292 | ? persistedFailedPageCount || Math.max(1, job?.total_pages || run?.total_pages || 1) |
| 293 | : 0 |
| 294 | |
| 295 | return { |
| 296 | sessionId, |
| 297 | runId: job?.id || null, |
| 298 | status: userCancelled |
| 299 | ? 'cancelled' |
| 300 | : failed |
| 301 | ? 'failed' |
| 302 | : run?.status === 'completed' |
| 303 | ? 'completed' |
| 304 | : 'idle', |
| 305 | hasActiveRun: false, |
| 306 | progress: run?.status === 'completed' || run?.status === 'partial' ? 100 : 0, |
| 307 | totalPages: job?.total_pages || 1, |
| 308 | completedPageCount, |
| 309 | failedPageCount, |
| 310 | events: [], |
| 311 | error: run?.error || job?.abort_reason || null, |
| 312 | startedAt: job?.created_at || null, |
| 313 | updatedAt: job?.updated_at || null, |
| 314 | kind: 'deck-edit', |
| 315 | retryPayload: failed && !userCancelled ? retryPayload : undefined |
| 316 | } |
| 317 | } |
| 318 | |
| 319 | async listActive(): Promise<DeckEditRunSnapshot[]> { |
| 320 | const jobs = await this.ctx.db.listActiveSessionJobs(['deck-edit']) |
| 321 | return Promise.all(jobs.map((job) => this.getState(job.session_id))) |
| 322 | } |
| 323 | |
| 324 | async abortInterruptedJobs(reason: string): Promise<void> { |
| 325 | const jobs = await this.ctx.db.listActiveSessionJobs(['deck-edit']) |
| 326 | for (const job of jobs) { |
| 327 | if (this.activeJobs.has(job.session_id)) continue |
| 328 | await this.ctx.db.updateSessionJobStatus(job.id, 'aborted', { abortReason: reason }) |
| 329 | await this.ctx.db.updateGenerationRunStatus(job.id, 'failed', reason) |
| 330 | await this.ctx.db.updateSessionStatus( |
| 331 | job.session_id, |
| 332 | normalizeRestoredSessionStatus(job.previous_session_status) |
| 333 | ) |
| 334 | } |
| 335 | } |
| 336 | |
| 337 | private async run(job: ActiveDeckEditJob): Promise<void> { |
| 338 | const emitAssistant = createEmitAssistantMessage(this.ctx.db, this.ctx.emitGenerateChunk) |
| 339 | try { |
| 340 | await executeDeckAllPageEditGeneration( |
| 341 | createGenerationContext(this.ctx), |
| 342 | emitAssistant, |
| 343 | job.context |
| 344 | ) |
| 345 | await settleEditJobSuccess({ ctx: this.ctx, context: job.context }) |
| 346 | } catch (error) { |
| 347 | const message = error instanceof Error ? error.message : String(error || '') |
| 348 | const cancelled = job.lease.signal.aborted || isCancellationMessage(message) |
| 349 | await settleEditJobFailure({ |
| 350 | ctx: this.ctx, |
| 351 | context: job.context, |
| 352 | error, |
| 353 | cancelled, |
| 354 | hasPersistedJob: true, |
| 355 | logPrefix: '[deck-edit:job]' |
| 356 | }) |
| 357 | } finally { |
| 358 | this.ctx.agentManager.removeSession(job.sessionId) |
| 359 | this.activeJobs.delete(job.sessionId) |
| 360 | this.reservedJobIds.delete(job.sessionId) |
| 361 | job.lease.release() |
| 362 | } |
| 363 | } |
| 364 | |
| 365 | private async getFailedPagesForRun(sessionId: string, runId: string) { |
| 366 | const run = await this.ctx.db.getGenerationRun(runId) |
| 367 | if (!run || run.session_id !== sessionId) { |
| 368 | throw new Error('重试失败：原失败任务不存在或不属于当前 Session') |
| 369 | } |
| 370 | return (await this.ctx.db.listGenerationPages(runId)).filter((page) => page.status === 'failed') |
| 371 | } |
| 372 | } |
| 373 | |
| 374 | export function registerDeckEditJobHandlers( |
| 375 | ctx: IpcContext, |
| 376 | coordinator: JobCoordinator |
| 377 | ): DeckEditJobService { |
| 378 | const service = new DeckEditJobService(ctx, coordinator) |
| 379 | const interruptedReady = service.abortInterruptedJobs('应用退出导致主会话编辑中断，可重新发起').catch((error) => { |
| 380 | log.warn('[deck-edit:job] failed to abort interrupted jobs', { |
| 381 | message: error instanceof Error ? error.message : String(error) |
| 382 | }) |
| 383 | }) |
| 384 | |
| 385 | ipcMain.handle('deck-edit:start', async (event, payload) => { |
| 386 | await interruptedReady |
| 387 | return service.start(event, payload) |
| 388 | }) |
| 389 | ipcMain.handle('deck-edit:cancel', async (_event, rawSessionId) => { |
| 390 | await interruptedReady |
| 391 | const sessionId = typeof rawSessionId === 'string' ? rawSessionId.trim() : '' |
| 392 | return { success: sessionId ? await service.cancel(sessionId) : true } |
| 393 | }) |
| 394 | ipcMain.handle('deck-edit:state', async (_event, rawSessionId) => { |
| 395 | await interruptedReady |
| 396 | const sessionId = typeof rawSessionId === 'string' ? rawSessionId.trim() : '' |
| 397 | if (!sessionId) throw new Error('sessionId 不能为空') |
| 398 | return service.getState(sessionId) |
| 399 | }) |
| 400 | ipcMain.handle('deck-edit:listActive', async () => { |
| 401 | await interruptedReady |
| 402 | return service.listActive() |
| 403 | }) |
| 404 | return service |
| 405 | } |
| 406 |