From dc582b7acfb6785e5e4516ec2c65457204b5ece6 Mon Sep 17 00:00:00 2001 From: mesalogo Date: Fri, 9 Oct 2026 17:05:26 +0800 Subject: [PATCH] fix(supervisor): re-review changed sources instead of failing the run --- .../supervision-review-store.test.ts | 7 +- .../assistant/supervision-review-store.ts | 149 +++++++++++------- src/main/assistant/supervision-review.test.ts | 78 +++++++-- src/main/assistant/supervision-worker.test.ts | 13 +- src/renderer/src/SupervisionReviewStages.tsx | 1 + src/renderer/src/SupervisorActivity.tsx | 1 + .../src/i18n/locales/en-US/heartbeat.ts | 1 + .../src/i18n/locales/zh-CN/heartbeat.ts | 1 + src/shared/supervision-review-contracts.ts | 2 + 9 files changed, 172 insertions(+), 81 deletions(-) diff --git a/src/main/assistant/supervision-review-store.test.ts b/src/main/assistant/supervision-review-store.test.ts index 09ff2498..ca1899c0 100644 --- a/src/main/assistant/supervision-review-store.test.ts +++ b/src/main/assistant/supervision-review-store.test.ts @@ -70,9 +70,10 @@ it('scans read-only, transfers at most 200 rows, yields without a transaction, a expect(wrote).toBe(true) expect(f.store.progress(f.runId).sources).toBe(450) expect(f.store.load(f.runId).initializing).toBeUndefined() - // Frozen scan revisions must not silently advance to the concurrent write. - await expect(f.store.resumeWithReader(f.runId, f.reader)).rejects.toThrow('source changed') - expect(f.store.load(f.runId).restartRequired).toBe(true) + // A concurrent write is detected on resume and that one source is re-frozen, not the whole run failed. + await f.store.resumeWithReader(f.runId, f.reader) + expect(f.store.progress(f.runId)).toMatchObject({ sources: 450, revisedSources: 1 }) + expect(f.store.load(f.runId).restartRequired).toBeUndefined() await expect(f.worker.call('reviewManifestPage', [f.runId, 0])).rejects.toThrow('lost its snapshot') }) diff --git a/src/main/assistant/supervision-review-store.ts b/src/main/assistant/supervision-review-store.ts index 74d1f3d4..a49753e4 100644 --- a/src/main/assistant/supervision-review-store.ts +++ b/src/main/assistant/supervision-review-store.ts @@ -10,7 +10,14 @@ import { setImmediate as yieldTurn } from 'node:timers/promises' export const TIMELINE_CHECKPOINT_SCOPE = 'timeline' export type ReviewConfiguration = SupervisionReviewSettings & { timeoutSeconds: number; concurrency: number; version: 1 } -export type ReviewState = { request: SupervisionRunRequest; config: ReviewConfiguration; initializing?: boolean; restartRequired?: boolean; omittedSources?: number; phase?: SupervisionReviewProgress['phase']; stories?: SupervisionReviewProgress['stories'] } +export type ReviewState = { request: SupervisionRunRequest; config: ReviewConfiguration; initializing?: boolean; restartRequired?: boolean; omittedSources?: number; revisedSources?: number; phase?: SupervisionReviewProgress['phase']; stories?: SupervisionReviewProgress['stories'] } + +/** + * A source that keeps changing while it is reviewed (for example an assistant reply + * that is still being rewritten) is re-frozen at most this many times per run in total; + * after that further changed sources are omitted so the run always terminates. + */ +export const MAX_REVISED_REVIEW_SOURCES = 200 export class ReviewSourcesOmitted extends Error {} @@ -235,26 +242,19 @@ export class SupervisionReviewStore { unfinished(request: SupervisionRunRequest): string | undefined { return this.db.prepare(`SELECT s.id FROM supervision_runs s JOIN supervision_review_runs r ON r.run_id = s.id WHERE s.trigger = 'heartbeat' AND s.scope_json = ? AND s.status IN ('failed', 'running') - AND (COALESCE(json_extract(r.state_json, '$.restartRequired'), 0) = 0 OR EXISTS ( - SELECT 1 FROM supervision_review_sources source WHERE source.run_id = s.id AND source.conversation_id != '' - AND NOT EXISTS (SELECT 1 FROM conversations c WHERE c.id = source.conversation_id) - AND instr(s.error, '(' || source.source || ');') > 0)) ORDER BY s.created_at LIMIT 1`).get(JSON.stringify(request.scope))?.id as string | undefined } resume(runId: string, signal?: AbortSignal): void { const state = this.prepareResume(runId) if (state.initializing) { this.initializeSources(runId, state, signal); return } - // Existing sources must still match their frozen versions. + // Sources changed since they were frozen are re-frozen and reprocessed; removed ones are omitted. let after = '' for (;;) { const rows = this.resumePage(runId, after) if (!rows.length) break const changed = this.changedSource(rows, signal) - if (changed) { - if (this.omitDeletedConversations(runId)) return this.resume(runId, signal) - this.sourceChanged(runId, changed) - } + if (changed) { this.reconcileSource(runId, changed); after = changed; continue } after = rows.at(-1)!.source } } @@ -263,7 +263,8 @@ export class SupervisionReviewStore { const status = this.db.prepare('SELECT status FROM supervision_runs WHERE id = ?').get(runId)?.status if (status === 'cancelled') throw new Error('SUPERVISION_REVIEW_CANCELLED: 此回顾已取消，不能继续。请开始新的回顾。') this.omitDeletedConversations(runId) - if (this.load(runId).restartRequired) throw new Error('Review source changed or was removed; start a new review') + // Runs failed by an earlier release on a changed source are recoverable now. + this.db.prepare("UPDATE supervision_review_runs SET state_json = json_remove(state_json, '$.restartRequired') WHERE run_id = ?").run(runId) const changed = this.db.prepare(`UPDATE supervision_runs SET status = 'running', error = NULL, completed_at = NULL WHERE id = ? AND status IN ('paused', 'failed', 'running')`).run(runId) if (!changed.changes) throw new Error('SUPERVISION_REVIEW_NOT_RESUMABLE: 此回顾已完成或不存在，请开始新的回顾。') @@ -286,10 +287,8 @@ export class SupervisionReviewStore { if (!rows.length) break const changed = await reader.changedSource(rows, signal) this.checkRunning(runId, signal) - if (changed) { - if (this.omitDeletedConversations(runId)) { after = ''; continue } - this.sourceChanged(runId, changed) - } + // The read snapshot may lag the writer; reconciliation re-reads the source here. + if (changed) { this.reconcileSource(runId, changed); after = changed; await yieldTurn(); continue } after = rows.at(-1)!.source await yieldTurn() } @@ -349,8 +348,9 @@ export class SupervisionReviewStore { for (const metadata of rows) { const current = this.currentSource(metadata.source, metadata.processed_offset, Math.min(remaining, 8000)) if (!current || current.current_revision !== metadata.revision) { - if (this.omitDeletedConversations(runId)) return this.chunk(runId, projectId, conversationId, config) - this.sourceChanged(runId, metadata.source) + // Re-freeze (or omit) the changed source, then rebuild the batch from current offsets. + this.reconcileSource(runId, metadata.source) + return this.chunk(runId, projectId, conversationId, config) } const row = { ...metadata, ...current } const locator = JSON.parse(row.locator) as Record @@ -387,17 +387,76 @@ export class SupervisionReviewStore { const reference = kind === 'knowledge' ? Number(key.slice(last + 1)) : 0 // The view predicate pushes record_id into each branch's message/task primary-key lookup. return this.db.prepare(`SELECT type, owner, substr(title, 1, 240) AS title, occurred, alternate_time, - locator, review_revision(context) AS current_revision, substr(body, ? + 1, ?) AS content + locator, review_revision(context) AS current_revision, substr(body, ? + 1, ?) AS content, length(body) AS body_length FROM supervision_review_current WHERE record_id = ? AND reference_index = ? AND type = ?`).get( offset, characters, id, reference, kind === 'message' ? 'conversation' : kind) as { type: SupervisionEvidence['sourceType']; owner: string; title: string; occurred: string; - alternate_time: string; locator: string; current_revision: string; content: string + alternate_time: string; locator: string; current_revision: string; content: string; body_length: number } | undefined } - private sourceChanged(runId: string, source: string): never { - this.db.prepare("UPDATE supervision_review_runs SET state_json = json_set(state_json, '$.restartRequired', json('true')) WHERE run_id = ?").run(runId) - throw new Error(`Review source changed or was removed (${source}); start a new review`) + /** + * Brings one frozen source in line with the business database without failing the run. + * A changed source is re-frozen at its current revision and reviewed again from the start + * (leaves that used it are discarded with their navigation); a removed or no longer visible + * source is omitted. Returns false when the source already matches. + */ + reconcileSource(runId: string, source: string): boolean { + const frozen = this.db.prepare('SELECT revision FROM supervision_review_sources WHERE run_id = ? AND source = ?').get(runId, source) as { revision: string } | undefined + if (!frozen) return false + const live = this.currentSource(source, 0, 0) + if (live?.current_revision === frozen.revision) return false + if (this.omitDeletedConversations(runId)) return true + const current = live && { revision: live.current_revision, length: Number(live.body_length) } + const state = this.load(runId) + const revise = current !== undefined && current.length > 0 && (state.revisedSources ?? 0) < MAX_REVISED_REVIEW_SOURCES + this.db.exec('SAVEPOINT reconcile_review_source') + try { + this.discardLeaves(runId, new Set([source])) + if (revise && current) { + this.db.prepare(`UPDATE supervision_review_sources SET revision = ?, length = ?, initial_offset = 0, processed_offset = 0 + WHERE run_id = ? AND source = ?`).run(current.revision, current.length, runId, source) + } else this.db.prepare('DELETE FROM supervision_review_sources WHERE run_id = ? AND source = ?').run(runId, source) + this.db.prepare(`UPDATE supervision_review_runs SET state_json = json_set(json_remove(state_json, '$.restartRequired'), + '${revise ? '$.revisedSources' : '$.omittedSources'}', COALESCE(json_extract(state_json, '${revise ? '$.revisedSources' : '$.omittedSources'}'), 0) + 1, + '$.phase', 'extracting') WHERE run_id = ?`).run(runId) + this.db.exec('RELEASE reconcile_review_source') + return true + } catch (error) { this.db.exec('ROLLBACK TO reconcile_review_source; RELEASE reconcile_review_source'); throw error } + } + + /** + * Discards every leaf containing an affected source, transitively over mixed leaves, and + * their dependent navigation. Surviving sources of those leaves restart at their initial offset. + */ + private discardLeaves(runId: string, affected: Set): void { + const invalid = new Set() + let changed = true + while (changed) { + changed = false + for (let offset = 0;; offset += 10) { + const batches = this.db.prepare(`SELECT b.id, json_group_array(json_extract(item.value, '$.locator.source')) AS sources + FROM (SELECT id, evidence_json FROM supervision_review_batches WHERE run_id = ? ORDER BY rowid LIMIT 10 OFFSET ?) b, + json_each(b.evidence_json) item GROUP BY b.id`).all(runId, offset) + if (!batches.length) break + for (const batch of batches) { + const sources = JSON.parse(String(batch.sources)) as string[] + if (invalid.has(String(batch.id)) || !sources.some(source => affected.has(source))) continue + invalid.add(String(batch.id)) + for (const source of sources) affected.add(source) + changed = true + } + } + } + const invalidIds = JSON.stringify([...invalid]) + this.db.prepare(`WITH RECURSIVE invalid(id) AS ( + SELECT value FROM json_each(?) UNION + SELECT n.id FROM supervision_review_navigation n, json_each(n.children_json) child + JOIN invalid i ON i.id = child.value WHERE n.run_id = ? + ) DELETE FROM supervision_review_navigation WHERE run_id = ? AND id IN (SELECT id FROM invalid)`) + .run(invalidIds, runId, runId) + this.db.prepare('DELETE FROM supervision_review_batches WHERE run_id = ? AND id IN (SELECT value FROM json_each(?))').run(runId, invalidIds) + for (const source of affected) this.db.prepare('UPDATE supervision_review_sources SET processed_offset = initial_offset WHERE run_id = ? AND source = ?').run(runId, source) } save(runId: string, projectId: string, conversationId: string, evidence: SupervisionEvidence[], output: SupervisionSummaryOutput): void { @@ -405,11 +464,13 @@ export class SupervisionReviewStore { try { this.omitDeletedConversations(runId) if (!this.hasSources(runId, evidence)) { this.db.exec('COMMIT'); return } + // A source changed while the model was working: drop this answer, keep the run going. + const stale = evidence.find(item => this.currentSource(String(item.locator!.source), 0, 0)?.current_revision !== item.locator!.revision) + if (stale) { this.reconcileSource(runId, String(stale.locator!.source)); this.db.exec('COMMIT'); return } const advance = this.db.prepare(`UPDATE supervision_review_sources SET processed_offset = ? WHERE run_id = ? AND source = ? AND revision = ? AND processed_offset = ?`) for (const item of evidence) { const locator = item.locator! - if (this.currentSource(String(locator.source), 0, 0)?.current_revision !== locator.revision) this.sourceChanged(runId, String(locator.source)) if (advance.run(Number(locator.end), runId, String(locator.source), String(locator.revision), Number(locator.start)).changes !== 1) { throw new Error('Review coverage has a gap or duplicate batch') } @@ -438,13 +499,9 @@ export class SupervisionReviewStore { const batches = this.db.prepare(`SELECT COUNT(*) AS batches, COALESCE(SUM(characters), 0) AS characters FROM supervision_review_batches WHERE run_id = ?`).get(runId)! const state = this.load(runId) - const error = String(this.db.prepare('SELECT error FROM supervision_runs WHERE id = ?').get(runId)?.error ?? '') - const failedSource = /Review source changed or was removed \((.+)\);/.exec(error)?.[1] - const deletedFailure = failedSource && this.db.prepare(`SELECT 1 FROM supervision_review_sources s - WHERE s.run_id = ? AND s.source = ? AND s.conversation_id != '' - AND NOT EXISTS (SELECT 1 FROM conversations c WHERE c.id = s.conversation_id)`).get(runId, failedSource) const navigation = this.db.prepare('SELECT COUNT(*) AS count FROM supervision_review_navigation WHERE run_id = ?').get(runId)! - return { runId, omittedSources: state.omittedSources, phase: state.phase, stories: state.stories, navigationNodes: Number(navigation.count), settings: state.config, restartRequired: deletedFailure ? false : state.restartRequired, batches: Number(batches.batches), characters: Number(batches.characters), sources: Number(sources.sources), + // Changed sources are reconciled on resume, so no saved run requires a restart. + return { runId, omittedSources: state.omittedSources, revisedSources: state.revisedSources, phase: state.phase, stories: state.stories, navigationNodes: Number(navigation.count), settings: state.config, batches: Number(batches.batches), characters: Number(batches.characters), sources: Number(sources.sources), remainingSources: Number(sources.remaining), complete: this.db.prepare("SELECT id FROM supervision_runs WHERE id = ? AND status IN ('completed', 'no_change')").get(runId) !== undefined } } @@ -467,41 +524,11 @@ export class SupervisionReviewStore { this.db.exec('SAVEPOINT omit_review_sources') try { const removed = new Set(deleted.map(row => String(row.source))) - const affected = new Set(removed) - const invalid = new Set() // A mixed leaf cannot be filtered safely: its prose may combine every input. // Reset its surviving sources and discard their other leaves to avoid duplicate coverage. - let changed = true - while (changed) { - changed = false - for (let offset = 0;; offset += 10) { - const batches = this.db.prepare(`SELECT b.id, json_group_array(json_extract(item.value, '$.locator.source')) AS sources - FROM (SELECT id, evidence_json FROM supervision_review_batches WHERE run_id = ? ORDER BY rowid LIMIT 10 OFFSET ?) b, - json_each(b.evidence_json) item GROUP BY b.id`).all(runId, offset) - if (!batches.length) break - for (const batch of batches) { - const sources = JSON.parse(String(batch.sources)) as string[] - if (invalid.has(String(batch.id)) || !sources.some(source => affected.has(source))) continue - invalid.add(String(batch.id)) - for (const source of sources) affected.add(source) - changed = true - } - } - } - const invalidIds = JSON.stringify([...invalid]) - this.db.prepare(`WITH RECURSIVE invalid(id) AS ( - SELECT value FROM json_each(?) UNION - SELECT n.id FROM supervision_review_navigation n, json_each(n.children_json) child - JOIN invalid i ON i.id = child.value WHERE n.run_id = ? - ) DELETE FROM supervision_review_navigation WHERE run_id = ? AND id IN (SELECT id FROM invalid)`) - .run(invalidIds, runId, runId) - this.db.prepare('DELETE FROM supervision_review_batches WHERE run_id = ? AND id IN (SELECT value FROM json_each(?))').run(runId, invalidIds) - for (const source of affected) this.db.prepare('UPDATE supervision_review_sources SET processed_offset = initial_offset WHERE run_id = ? AND source = ?').run(runId, source) + this.discardLeaves(runId, new Set(removed)) for (const source of removed) this.db.prepare('DELETE FROM supervision_review_sources WHERE run_id = ? AND source = ?').run(runId, source) - const error = String(this.db.prepare('SELECT error FROM supervision_runs WHERE id = ?').get(runId)?.error ?? '') - const failedSource = /Review source changed or was removed \((.+)\);/.exec(error)?.[1] - if (failedSource && removed.has(failedSource)) this.db.prepare("UPDATE supervision_review_runs SET state_json = json_remove(state_json, '$.restartRequired') WHERE run_id = ?").run(runId) - this.db.prepare(`UPDATE supervision_review_runs SET state_json = json_set(state_json, + this.db.prepare(`UPDATE supervision_review_runs SET state_json = json_set(json_remove(state_json, '$.restartRequired'), '$.omittedSources', COALESCE(json_extract(state_json, '$.omittedSources'), 0) + ?, '$.phase', 'extracting') WHERE run_id = ?`).run(removed.size, runId) this.db.exec('RELEASE omit_review_sources') return true diff --git a/src/main/assistant/supervision-review.test.ts b/src/main/assistant/supervision-review.test.ts index 72a91a7c..a5a6d239 100644 --- a/src/main/assistant/supervision-review.test.ts +++ b/src/main/assistant/supervision-review.test.ts @@ -322,8 +322,60 @@ it('rejects changed source versions and rolls back offsets when batch persistenc expect(f.sql.prepare('SELECT processed_offset FROM supervision_review_sources').get()!.processed_offset).toBe(0) f.sql.exec('DROP TRIGGER fail_batch') f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('Changed', f.conversations[0]!.messages[0]!.id) - await expect(f.service().resume(runId)).rejects.toThrow('source changed') - expect(f.db.listSupervisionActivity()[0]!.status).toBe('failed') + const resumed = await f.service().resume(runId) + expect(resumed).toMatchObject({ status: 'completed', coverage: { revisedSources: 1, characters: 'Changed'.length, complete: true } }) + expect(f.db.supervisionReviewStore().batches(runId).flatMap(batch => batch.evidence).map(item => item.content)).toEqual(['Changed']) +}) + +it('re-reviews a source rewritten while its batch is with the model and discards the stale answer', async () => { + const f = await fixture([['Original reply'], ['Other conversation']], { concurrency: 1 }) + const target = f.conversations[0]!.messages[0]!.id + const base = f.summarize.getMockImplementation()! + let rewritten = false + f.summarize.mockImplementation(async input => { + if (!rewritten && input.evidence.some(item => item.locator?.messageId === target)) { + rewritten = true + f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('Original reply, continued after reconnect', target) + } + return { ...await base(input), summary: input.evidence.map(item => item.content).join(', ') } + }) + const result = await f.service().run(request) + expect(result).toMatchObject({ status: 'completed', coverage: { revisedSources: 1, sources: 2, remainingSources: 0, complete: true } }) + const contents = f.db.supervisionReviewStore().batches(result.runId!).flatMap(batch => batch.evidence).map(item => item.content) + expect(contents).toContain('Original reply, continued after reconnect') + expect(contents).not.toContain('Original reply') +}) + +it('discards a mixed leaf when one of its sources changes and reviews only current versions', async () => { + const f = await fixture([['A', 'B']], { concurrency: 1 }) + const store = f.db.supervisionReviewStore() + const runId = f.db.startSupervisionRun(request) + store.initialize(runId, { request, config: f.config }) + const conversation = f.conversations[0]! + const evidence = store.chunk(runId, conversation.projectId, conversation.id, f.config) + expect(evidence.map(item => item.content)).toEqual(['A', 'B']) + store.save(runId, conversation.projectId, conversation.id, evidence, empty) + store.saveNavigation(runId, 'dependent', [store.batches(runId)[0]!.id], empty) + f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('B2', conversation.messages[1]!.id) + f.db.failSupervisionRun(runId, 'Provider failed') + await f.service().resume(runId) + expect(store.navigation(runId, 'dependent')).toBeUndefined() + expect(store.batches(runId).flatMap(batch => batch.evidence).map(item => item.content)).toEqual(['A', 'B2']) + expect(store.progress(runId)).toMatchObject({ revisedSources: 1, complete: true }) +}) + +it('omits a source that is no longer visible and resumes old runs failed by the earlier restart rule', async () => { + const f = await fixture([['Hidden'], ['Kept']], { concurrency: 1 }) + const store = f.db.supervisionReviewStore() + const runId = f.db.startSupervisionRun(request) + store.initialize(runId, { request, config: f.config }) + f.sql.prepare("UPDATE supervision_review_runs SET state_json = json_set(state_json, '$.restartRequired', json('true')) WHERE run_id = ?").run(runId) + f.db.failSupervisionRun(runId, `Review source changed or was removed (message:${f.conversations[0]!.messages[0]!.id}); start a new review`) + f.sql.prepare("UPDATE conversations SET status = 'archived' WHERE id = ?").run(f.conversations[0]!.id) + expect(store.progress(runId).restartRequired).toBeUndefined() + const result = await f.service().resume(runId) + expect(result).toMatchObject({ status: 'completed', coverage: { omittedSources: 1, sources: 1, complete: true } }) + expect(result.output.summary).not.toContain('Hidden') }) it('round-robins projects and conversations and keeps semantic chunks independent of database page size', async () => { @@ -381,20 +433,19 @@ it('continues an existing automatic timeline checkpoint without rereading its su expect(f.sql.prepare('SELECT processed_offset FROM review_checkpoints').get()!.processed_offset).toBe(3300) }) -it('keeps old-version facts but does not permanently block automatic review after a source changes', async () => { +it('lets the next automatic review continue a failed run whose source changed in the meantime', async () => { const f = await fixture([['x'.repeat(2500)]], { concurrency: 1 }) f.summarize.mockImplementationOnce(async () => empty).mockRejectedValueOnce(new Error('Provider failed')) const automatic = { ...request, trigger: 'heartbeat' as const } await expect(f.service().run(automatic)).rejects.toThrow('Provider failed') const runId = f.db.listSupervisionActivity()[0]!.id - const saved = f.db.supervisionReviewStore().batches(runId) + expect(f.db.supervisionReviewStore().batches(runId)).toHaveLength(1) f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('New version', f.conversations[0]!.messages[0]!.id) - await expect(f.service().resume(runId)).rejects.toThrow('source changed') - expect(f.db.supervisionReviewStore().progress(runId).restartRequired).toBe(true) const result = await f.service().run(automatic) - expect(result.runId).not.toBe(runId) - expect(result.status).toBe('completed') - expect(f.db.supervisionReviewStore().batches(runId)).toEqual(saved) + expect(result.runId).toBe(runId) + expect(result).toMatchObject({ status: 'completed', coverage: { revisedSources: 1, characters: 'New version'.length } }) + // The leaf of the old version is discarded; only the current version is published. + expect(f.db.supervisionReviewStore().batches(runId).flatMap(batch => batch.evidence).map(item => item.content)).toEqual(['New version']) }) it('makes a manual review incremental by default and keeps explicit re-analysis out of shared progress', async () => { @@ -455,7 +506,8 @@ it('omits a deleted conversation, its knowledge and tasks before resume without const store = f.db.supervisionReviewStore() const kept = store.batches(runId).filter(batch => batch.conversationId !== deleted.id) expect(f.db.deleteLocalConversation(deleted.id)).toBe(true) - expect(store.progress(runId)).toMatchObject({ omittedSources: 3, sources: 1, batches: kept.length, restartRequired: undefined }) + expect(store.progress(runId)).toMatchObject({ omittedSources: 3, sources: 1, batches: kept.length }) + expect(store.progress(runId).restartRequired).toBeUndefined() expect(store.batches(runId)).toEqual(kept) f.sql.exec('DROP TRIGGER fail_publication') const calls = f.summarize.mock.calls.length @@ -542,7 +594,7 @@ it('resets mixed-leaf survivors only and invalidates their dependent navigation expect(store.batches(runId)).toContainEqual(kept) }) -it('recovers the earlier deletion failure but keeps archived and modified existing conversations as explicit failures', async () => { +it('recovers earlier deletion failures and omits archived conversations instead of failing', async () => { const f = await fixture([['Source']]) const store = f.db.supervisionReviewStore() const id = f.db.startSupervisionRun(request) @@ -550,11 +602,11 @@ it('recovers the earlier deletion failure but keeps archived and modified existi f.sql.prepare("UPDATE supervision_review_runs SET state_json = json_set(state_json, '$.restartRequired', json('true')) WHERE run_id = ?").run(id) f.db.failSupervisionRun(id, `Review source changed or was removed (message:${f.conversations[0]!.messages[0]!.id}); start a new review`) f.sql.prepare('DELETE FROM conversations WHERE id = ?').run(f.conversations[0]!.id) - expect(store.progress(id).restartRequired).toBe(false) + expect(store.progress(id).restartRequired).toBeUndefined() await expect(f.service().resume(id)).resolves.toMatchObject({ status: 'no_change', coverage: { omittedSources: 1 } }) const g = await fixture([['Archived source']]) g.summarize.mockRejectedValueOnce(new Error('Pause')) await expect(g.service().run(request)).rejects.toThrow('Pause') g.sql.prepare("UPDATE conversations SET status = 'archived' WHERE id = ?").run(g.conversations[0]!.id) - await expect(g.service().resume(g.db.listSupervisionActivity()[0]!.id)).rejects.toThrow('source changed') + await expect(g.service().resume(g.db.listSupervisionActivity()[0]!.id)).resolves.toMatchObject({ status: 'no_change', coverage: { omittedSources: 1 } }) }) diff --git a/src/main/assistant/supervision-worker.test.ts b/src/main/assistant/supervision-worker.test.ts index 41d98def..e92e841f 100644 --- a/src/main/assistant/supervision-worker.test.ts +++ b/src/main/assistant/supervision-worker.test.ts @@ -96,10 +96,15 @@ it('rebuilds a partially written manifest after restart and validates completed await f.storage.resumeSupervisionReview(id, new AbortController().signal) expect(f.db.supervisionReviewStore().progress(id).sources).toBe(422) expect(f.db.supervisionReviewStore().load(id).initializing).toBeUndefined() - // Existing saved runs have no initialization flag and retain their frozen versions. - f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('Changed', f.conversations[0]!.messages[0]!.id) - await expect(f.storage.resumeSupervisionReview(id, new AbortController().signal)).rejects.toThrow('source changed') - expect(f.db.supervisionReviewStore().load(id).restartRequired).toBe(true) + // A source changed after freezing is re-frozen at its new version instead of failing the run. + const changedId = f.conversations[0]!.messages[0]!.id + f.sql.prepare('UPDATE messages SET content = ? WHERE id = ?').run('Changed', changedId) + await f.storage.resumeSupervisionReview(id, new AbortController().signal) + const store = f.db.supervisionReviewStore() + expect(store.progress(id)).toMatchObject({ sources: 422, revisedSources: 1 }) + expect(store.load(id).restartRequired).toBeUndefined() + expect(f.sql.prepare('SELECT length, processed_offset FROM supervision_review_sources WHERE run_id = ? AND source = ?') + .get(id, `message:${changedId}`)).toMatchObject({ length: 'Changed'.length, processed_offset: 0 }) }) it.each(['paused', 'cancelled'] as const)('accepts %s during production initialization, holds the slot, and never calls the model', async status => { diff --git a/src/renderer/src/SupervisionReviewStages.tsx b/src/renderer/src/SupervisionReviewStages.tsx index 4eb3333d..d11e1f33 100644 --- a/src/renderer/src/SupervisionReviewStages.tsx +++ b/src/renderer/src/SupervisionReviewStages.tsx @@ -24,6 +24,7 @@ export function SupervisionReviewStages({ row, stopping }: { row: SupervisionAct

{t('activity.coverage', { batches: progress.batches, remaining: progress.remainingSources })}

{!!progress.omittedSources &&

{t('activity.omittedSources', { count: progress.omittedSources })}

} + {!!progress.revisedSources &&

{t('activity.revisedSources', { count: progress.revisedSources })}

} {!progress.complete && !progress.phase &&

{t('activity.stageUnknown')}

} {row.supervisionStatus === 'running' && !stopping &&

{t('reviewSettings.inFlight', { count: progress.inFlight ?? 0 })}

} diff --git a/src/renderer/src/SupervisorActivity.tsx b/src/renderer/src/SupervisorActivity.tsx index 8eb67e66..501c9d3f 100644 --- a/src/renderer/src/SupervisorActivity.tsx +++ b/src/renderer/src/SupervisorActivity.tsx @@ -229,6 +229,7 @@ export function SupervisorActivity({ active, projects, onOpenResult, configId, c {row.reviewProgress &&

{t('reviewSettings.progress', { batches: row.reviewProgress.batches, characters: row.reviewProgress.characters, remaining: row.reviewProgress.remainingSources })}

} {!!row.reviewProgress?.omittedSources &&

{t('activity.omittedSources', { count: row.reviewProgress.omittedSources })}

} + {!!row.reviewProgress?.revisedSources &&

{t('activity.revisedSources', { count: row.reviewProgress.revisedSources })}

} {row.reviewProgress?.navigationNodes !== undefined &&

{t('activity.navigationSaved', { count: row.reviewProgress.navigationNodes })}

} {row.summary &&

{row.summary}

} {row.reviewProgress?.settings &&
{t('reviewSettings.configuration')} diff --git a/src/renderer/src/i18n/locales/en-US/heartbeat.ts b/src/renderer/src/i18n/locales/en-US/heartbeat.ts index 39603b81..20d9c7f7 100644 --- a/src/renderer/src/i18n/locales/en-US/heartbeat.ts +++ b/src/renderer/src/i18n/locales/en-US/heartbeat.ts @@ -64,6 +64,7 @@ export const heartbeat = { stageUnknown: 'This historical run has no saved stage. Zero remaining sources does not mean the review was published.', coverage: '{{batches}} saved batches · {{remaining}} sources remaining', omittedSources: '{{count}} sources skipped because their conversations were deleted. Coverage includes only retained sources. If none remain, the review finishes without an empty report.', + revisedSources: '{{count}} sources changed during the review and were reviewed again at their latest version; batches of older versions were discarded.', navigationSaved: '{{count}} navigation merges saved', resumeMerge: 'Continue merging', resumeSave: 'Retry publication', runDetails: 'Run details and configuration', treePageHint: 'Expand projects and conversations to browse this page of saved batches. Tasks without a saved result are not listed here.', diff --git a/src/renderer/src/i18n/locales/zh-CN/heartbeat.ts b/src/renderer/src/i18n/locales/zh-CN/heartbeat.ts index da65819d..74642d45 100644 --- a/src/renderer/src/i18n/locales/zh-CN/heartbeat.ts +++ b/src/renderer/src/i18n/locales/zh-CN/heartbeat.ts @@ -61,6 +61,7 @@ export const heartbeat = { stageUnknown: '此历史运行未保存阶段信息。剩余来源为零不代表回顾已发布。', coverage: '已保存 {{batches}} 批 · 剩余 {{remaining}} 个来源', omittedSources: '会话已删除，已跳过 {{count}} 个来源；覆盖统计仅包含保留的来源。没有剩余来源时结束回顾，不生成空报告。', + revisedSources: '回顾期间有 {{count}} 个来源被修改，已按最新内容重新回顾；旧版本的批次已丢弃。', navigationSaved: '已保存 {{count}} 次导航合并', resumeMerge: '继续合并摘要', resumeSave: '重试发布', runDetails: '运行详情与配置', treePageHint: '按项目、会话展开本页已保存批次。这里不包含尚未保存结果的任务。', diff --git a/src/shared/supervision-review-contracts.ts b/src/shared/supervision-review-contracts.ts index e5257de2..433c38da 100644 --- a/src/shared/supervision-review-contracts.ts +++ b/src/shared/supervision-review-contracts.ts @@ -39,6 +39,8 @@ export type SupervisionReviewProgress = { remainingSources: number /** Sources excluded because their owning conversation was deleted; not reviewed coverage. */ omittedSources?: number + /** Sources that changed during the run and were reviewed again at their newer version. */ + revisedSources?: number complete: boolean restartRequired?: boolean inFlight?: number