fix: retry unusable model output in smaller review batches
Leaf batches retry at half and a quarter of the text budget, navigation merges retry once; provider errors are not retried. Adds the long-session acceptance script and records the results.
This commit is contained in:
@@ -23,7 +23,8 @@ it('production six-leaf merge failure resumes navigation-only JSON and publishes
|
||||
let merges = 0
|
||||
f.respond.mockImplementation(prompt => {
|
||||
if (!prompt.includes('These inputs are navigation summaries')) return leaf(prompt)
|
||||
if (++merges === 2) return { summary: 'Invalid navigation', changeDigest: '', openItems: [], events: [{ sourceReferenceIds: ['invented'] }] }
|
||||
// The second merge is unusable on both attempts (one retry), so the run fails there.
|
||||
if (++merges === 2 || merges === 3) return { summary: 'Invalid navigation', changeDigest: '', openItems: [], events: [{ sourceReferenceIds: ['invented'] }] }
|
||||
return { summary: 'Combined navigation', changeDigest: 'Retained leaf decisions', openItems: ['Check sources'] }
|
||||
})
|
||||
await expect(f.service().run(f.request)).rejects.toThrow()
|
||||
@@ -51,7 +52,7 @@ it.each(['missing arrays', 'unknown source', 'unknown entity'])('production leaf
|
||||
await expect(f.service().run(f.request)).rejects.toThrow()
|
||||
expect(f.db.listSupervisionActivity()[0]!.reviewProgress).toMatchObject({ phase: 'extracting', batches: 0, remainingSources: 1, complete: false })
|
||||
expect(f.db.listSupervisionResults()).toHaveLength(0)
|
||||
expect(f.respond).toHaveBeenCalledTimes(1)
|
||||
expect(f.respond).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('persists publication failure and resumes with saved navigation without model calls', async () => {
|
||||
@@ -214,7 +215,7 @@ it.each(['Atlas', 'optional UUID from KNOWN ENTITIES only', ' ', '', null])('pre
|
||||
expect(f.db.listSupervisionCandidates(f.request)).toEqual(expect.arrayContaining(before))
|
||||
})
|
||||
|
||||
it.each([randomUUID()])('production factory rejects unknown UUID %s without publishing or retrying', async persistedId => {
|
||||
it.each([randomUUID()])('production factory rejects unknown UUID %s without publishing, after the bounded output retries', async persistedId => {
|
||||
const f = fixture()
|
||||
const original = f.respond.getMockImplementation()!
|
||||
f.respond.mockImplementation(prompt => {
|
||||
@@ -224,7 +225,7 @@ it.each([randomUUID()])('production factory rejects unknown UUID %s without publ
|
||||
await expect(f.service().run(f.request)).rejects.toThrow(/leaf batch: entities\[0\].persistedId/)
|
||||
expect(f.db.listSupervisionResults()).toHaveLength(0)
|
||||
expect(f.db.listSupervisionActivity()[0]!.status).toBe('failed')
|
||||
expect(f.respond).toHaveBeenCalledTimes(1)
|
||||
expect(f.respond).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it.each(['known_999', 'Atlas', ' ', 'optional candidate reference'])('rejects unknown explicit candidate reference %s', async candidateRef => {
|
||||
@@ -236,7 +237,7 @@ it.each(['known_999', 'Atlas', ' ', 'optional candidate reference'])('rejects un
|
||||
})
|
||||
await expect(f.service().run(f.request)).rejects.toThrow('candidateRef is outside KNOWN ENTITIES')
|
||||
expect(f.db.listSupervisionResults()).toHaveLength(0)
|
||||
expect(f.respond).toHaveBeenCalledTimes(1)
|
||||
expect(f.respond).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('production factory rejects a real UUID from another scope and preserves the existing graph', async () => {
|
||||
@@ -251,7 +252,7 @@ it('production factory rejects a real UUID from another scope and preserves the
|
||||
await expect(f.service().run({ ...f.request, scope: { kind: 'projects', projectIds: [f.db.listProjects()[0]!.id] } })).rejects.toThrow('outside KNOWN ENTITIES')
|
||||
expect(f.db.listSupervisionResults()).toHaveLength(1)
|
||||
expect(f.db.listSupervisionCandidates(f.request)).toEqual(candidates)
|
||||
expect(f.respond).toHaveBeenCalledTimes(2)
|
||||
expect(f.respond).toHaveBeenCalledTimes(4)
|
||||
})
|
||||
|
||||
it('reuses actual legacy text-key shapes through strict UUID aliases and preserves confirmed fields and graph links', async () => {
|
||||
|
||||
@@ -249,6 +249,43 @@ it('reopens saved batches after failure, resumes exact Unicode offsets and only
|
||||
expect(f.summarize).toHaveBeenCalledTimes(calls)
|
||||
})
|
||||
|
||||
it('retries an unusable leaf answer on a smaller part of the same text without skipping any of it', async () => {
|
||||
const f = await fixture([['y'.repeat(2000)]], { concurrency: 1 })
|
||||
// Invalid JSON, then an answer citing a source that is not in the batch, then usable answers.
|
||||
f.summarize.mockResolvedValueOnce('not json' as never).mockImplementationOnce(async () => ({ ...empty,
|
||||
events: [{ title: 'Bad', description: 'x', occurredAt: '2026-09-21T00:00:00.000Z', eventType: 'discussion' as const, entityIds: [], sourceReferenceIds: ['missing'] }] }))
|
||||
const result = await f.service().run(request)
|
||||
expect(result.status).toBe('completed')
|
||||
const evidence = f.db.supervisionReviewStore().batches(result.runId!, 100).flatMap(row => row.evidence)
|
||||
// Third attempt (a quarter of 1000, at least 500) succeeded; the rest of the text continued at full size.
|
||||
expect(evidence.map(item => item.content.length)).toEqual([500, 1000, 500])
|
||||
expect(evidence.map(item => item.content).join('')).toBe('y'.repeat(2000))
|
||||
})
|
||||
|
||||
it('asks a navigation merge again once when its answer is unusable', async () => {
|
||||
const f = await fixture([['n'.repeat(1500)]], { concurrency: 1 })
|
||||
// Two leaves succeed; the merge first returns invalid JSON, then a usable answer.
|
||||
f.summarize.mockImplementation(async (input: SupervisorSummarizerRequest) => input.evidence[0]?.sourceType === 'note' ? empty
|
||||
: { ...empty, events: input.evidence.map(source => ({ title: 'Fact', description: 'x', occurredAt: source.occurredAt, eventType: 'discussion' as const, entityIds: [], sourceReferenceIds: [source.id] })) })
|
||||
let navigation = 0
|
||||
const base = f.summarize.getMockImplementation()!
|
||||
f.summarize.mockImplementation(async input => input.evidence[0]?.sourceType === 'note' && navigation++ === 0 ? 'not json' as never : base(input))
|
||||
const result = await f.service().run(request)
|
||||
expect(result.status).toBe('completed')
|
||||
expect(navigation).toBe(2)
|
||||
})
|
||||
|
||||
it('fails after the retries and does not retry provider errors', async () => {
|
||||
const f = await fixture([['z'.repeat(1500)]], { concurrency: 1 })
|
||||
f.summarize.mockResolvedValue('not json' as never)
|
||||
await expect(f.service().run(request)).rejects.toThrow('无效 JSON')
|
||||
expect(f.summarize).toHaveBeenCalledTimes(3)
|
||||
const g = await fixture([['z'.repeat(1500)]], { concurrency: 1 })
|
||||
g.summarize.mockRejectedValueOnce(new Error('Provider down'))
|
||||
await expect(g.service().run(request)).rejects.toThrow('Provider down')
|
||||
expect(g.summarize).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('continues extraction and navigation beyond the saved execution budget', async () => {
|
||||
const f = await fixture([['x'.repeat(2500)]], { concurrency: 1, executionSeconds: 30 })
|
||||
const clock = vi.spyOn(Date, 'now').mockReturnValue(time)
|
||||
|
||||
@@ -170,6 +170,11 @@ function validateReferences(
|
||||
}
|
||||
}
|
||||
|
||||
/** The model answered, but the answer was unusable (bad JSON, schema, or references). Retrying smaller may help. */
|
||||
export class SupervisionOutputError extends Error {}
|
||||
/** Leaf batches with unusable output are retried at half, then a quarter of the text budget, then the run fails. */
|
||||
const outputRetries = 2
|
||||
|
||||
export class SupervisorService {
|
||||
private active?: { settled: Promise<unknown>; runId?: string; stopping?: 'paused' | 'cancelled' }
|
||||
private readonly controllers = new Map<string, AbortController>()
|
||||
@@ -358,9 +363,13 @@ export class SupervisorService {
|
||||
authorizeTool: async name => { throw new Error(`Supervisor tools are disabled: ${name}`) } })
|
||||
} finally { activity.inFlight-- }
|
||||
controller.signal.throwIfAborted()
|
||||
const output = parseModelOutput(raw, navigation ? [] : candidates, `Supervision ${runId} ${navigation ? 'navigation merge' : 'leaf batch'}`, navigation)
|
||||
validateReferences(output, evidence)
|
||||
return output
|
||||
try {
|
||||
const output = parseModelOutput(raw, navigation ? [] : candidates, `Supervision ${runId} ${navigation ? 'navigation merge' : 'leaf batch'}`, navigation)
|
||||
validateReferences(output, evidence)
|
||||
return output
|
||||
} catch (error) {
|
||||
throw new SupervisionOutputError(error instanceof Error ? error.message : String(error), { cause: error })
|
||||
}
|
||||
}
|
||||
for (;;) {
|
||||
if (controller.signal.aborted) return stopped()
|
||||
@@ -368,11 +377,24 @@ export class SupervisorService {
|
||||
if (!groups.length) break
|
||||
const results = await Promise.allSettled(groups.map(async group => {
|
||||
controller.signal.throwIfAborted()
|
||||
const evidence = db.chunk(runId, group.projectId, group.conversationId, config)
|
||||
if (!evidence.length) throw new Error('Pending source produced no reviewable content')
|
||||
const output = await summarize(evidence, false, await candidatesFor(group.projectId))
|
||||
controller.signal.throwIfAborted()
|
||||
db.save(runId, group.projectId, group.conversationId, evidence, output)
|
||||
const candidates = await candidatesFor(group.projectId)
|
||||
// An unusable answer is retried on a smaller prefix of the same pending text; nothing is skipped,
|
||||
// the rest of the text stays pending for the next batch.
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
const scale = 2 ** attempt
|
||||
const sized = attempt ? { ...config, batchCharacters: Math.max(500, Math.floor(config.batchCharacters / scale)),
|
||||
batchMessages: Math.max(1, Math.ceil(config.batchMessages / scale)) } : config
|
||||
const evidence = db.chunk(runId, group.projectId, group.conversationId, sized)
|
||||
if (!evidence.length) throw new Error('Pending source produced no reviewable content')
|
||||
try {
|
||||
const output = await summarize(evidence, false, candidates)
|
||||
controller.signal.throwIfAborted()
|
||||
db.save(runId, group.projectId, group.conversationId, evidence, output)
|
||||
return
|
||||
} catch (error) {
|
||||
if (!(error instanceof SupervisionOutputError) || attempt >= outputRetries || controller.signal.aborted) throw error
|
||||
}
|
||||
}
|
||||
}))
|
||||
if (controller.signal.aborted) return stopped()
|
||||
const failure = results.find(result => result.status === 'rejected')
|
||||
@@ -393,7 +415,12 @@ export class SupervisorService {
|
||||
if (cached) return { id, output: cached }
|
||||
const evidence = [left, right].map(card => ({ id: card.id, sourceType: 'note' as const, sourceId: card.id,
|
||||
title: 'Retained review navigation', content: card.output.summary, occurredAt: request.timeRange.to }))
|
||||
const output = await summarize(evidence, true)
|
||||
// A navigation merge has fixed inputs; an unusable answer is asked again once before the run fails.
|
||||
let output: SupervisionSummaryOutput
|
||||
try { output = await summarize(evidence, true) } catch (error) {
|
||||
if (!(error instanceof SupervisionOutputError) || controller.signal.aborted) throw error
|
||||
output = await summarize(evidence, true)
|
||||
}
|
||||
controller.signal.throwIfAborted()
|
||||
db.saveNavigation(runId, id, [left.id, right.id], output)
|
||||
return { id, output }
|
||||
|
||||
Reference in New Issue
Block a user