228 lines
17 KiB
JavaScript
228 lines
17 KiB
JavaScript
const { test } = require('node:test')
|
|||
|
|
const assert = require('node:assert/strict')
|
||
|
|
const crypto = require('node:crypto')
|
||
|
|
const fs = require('node:fs')
|
||
|
|
const path = require('node:path')
|
||
|
|
const vm = require('node:vm')
|
||
|
|
const policy = require('./wiki_recovery_policy.cjs')
|
||
|
|
const seventh = require('./wiki_seventh_recovery_review.cjs')
|
||
|
|
const scheduler = require('./wiki_recovery_scheduler_policy.cjs')
|
||
|
|
const sha = bytes => crypto.createHash('sha256').update(bytes).digest('hex')
|
||
|
|
const reason = 'Reviewed fresh source identity correction and published exact frozen candidate build'
|
||
|
|
|
||
|
|
function fixture() {
|
||
|
|
const now = Date.now(), scope = seventh.seventhRecoveryScope, pins = seventh.seventhPins
|
||
|
|
const iso = offset => new Date(now + offset).toISOString()
|
||
|
|
const source = { ...scope, id: 10, eligible: true, pending: 0, chunks: 624 }
|
||
|
|
const events = seventh.seventhRequiredTests.flatMap(Test => ['run', 'pass'].map(Action => ({
|
||
|
|
Time: iso(Action === 'run' ? -9000 : -8000), Package: pins.test_package, Action, Test })))
|
||
|
|
events.push({ Time: iso(-7500), Package: pins.test_package, Action: 'pass' },
|
||
|
|
{ Time: iso(-7500), Package: 'github.com/Tencent/WeKnora/internal/agent', Action: 'pass' })
|
||
|
|
const testBytes = events.map(event => JSON.stringify(event)).join('\n') + '\n'
|
||
|
|
const evidence = { kind: pins.kind, run_id: '00000000-0000-4000-8000-000000000007', status: 'passed',
|
||
|
|
source_tree_hash: 'a'.repeat(64), binary_sha256: 'b'.repeat(64), fix_release: 'public-20260921-180000',
|
||
|
|
api_image_id: 'sha256:' + 'c'.repeat(64), model_calls: 0, production_writes: 0, prior_evidence_preserved: true,
|
||
|
|
started_at: iso(-10000), completed_at: iso(-7000), regression_tests_sha256: sha(testBytes) }
|
||
|
|
const semantic = { kind: pins.kind, run_id: evidence.run_id, source_tree_hash: evidence.source_tree_hash,
|
||
|
|
binary_sha256: evidence.binary_sha256, regression_tests_sha256: sha(testBytes), reviewed_by: 'independent-reviewer',
|
||
|
|
reviewed_at: iso(-6000), approved: true, candidate_count_preserved: true, aliases_not_removed: true,
|
||
|
|
no_extra_retry: true, prior_evidence_preserved: true, notes: 'Reviewed exact source quote guard, aliases, metadata retention and bounded retry.' }
|
||
|
|
const semanticBytes = JSON.stringify(semantic)
|
||
|
|
const build = { source_tree_hash: evidence.source_tree_hash, binary_sha256: evidence.binary_sha256,
|
||
|
|
fix_release: evidence.fix_release, api_image_id: evidence.api_image_id, regression_tests_sha256: sha(testBytes),
|
||
|
|
source_kind: 'frozen_published_candidate', baseline_graph_difference: seventh.seventhGraphDifference }
|
||
|
|
const buildBytes = JSON.stringify(build)
|
||
|
|
evidence.semantic_review_sha256 = sha(semanticBytes)
|
||
|
|
evidence.build_manifest_sha256 = sha(buildBytes)
|
||
|
|
const bytes = JSON.stringify(evidence)
|
||
|
|
const review = { ...scope, max_attempt: 7, previous_attempt: 6, previous_pending_op_id: 4145,
|
||
|
|
previous_error_code: 'EXTRACT_FAILED', budget_version: 1, public_gates_passed: true, max_provider_requests: 1000,
|
||
|
|
budget_started_at: iso(-1000), deadline_at: iso(7199000), fix_release: evidence.fix_release,
|
||
|
|
api_image_id: evidence.api_image_id, published_at: iso(-5000), candidate_diagnostic: {
|
||
|
|
kind: pins.kind, run_id: evidence.run_id, evidence_sha256: sha(bytes), source_tree_hash: evidence.source_tree_hash,
|
||
|
|
binary_sha256: evidence.binary_sha256, build_manifest_sha256: evidence.build_manifest_sha256, semantic_reviewed: true } }
|
||
|
|
const previous = { ...scope, attempt_number: 6, pending_op_id: 4145, recovery_status: 'stopped_budget',
|
||
|
|
new_failures: 0, error_code: 'EXTRACT_FAILED', span_status: 'failed', pending: false, pending_finalize: 0,
|
||
|
|
queued_at: pins.previous_queued_at, read_progress: { candidates: { ...seventh.seventhReadProgress } },
|
||
|
|
recovery_budget: { ...scope, attempt_number: 6, pending_op_id: 4145, status: 'stopped_budget', stop_reason: 'phase_failed',
|
||
|
|
last_phase: 'wiki_candidate_slug', reserved_provider_requests: 3, reviewed_release: pins.previous_release,
|
||
|
|
reviewed_api_image: pins.previous_image, paused_payload: JSON.stringify([seventh.seventhPausedOperation]) } }
|
||
|
|
const pilot = { ...scope, dead_letter_id: 31, pending: false, recovery_status: 'wiki_recovered', span_status: 'done',
|
||
|
|
pages_dropped: 0, failed_slug_writes: 0, missing_expected_pages: 0, failed_page_spans: 0, current_pages: 1,
|
||
|
|
published_summaries: 1, expected_pages: 1, pages_written: 1, pages_total: 1 }
|
||
|
|
const receipt = { ...scope, attempt_number: 7, pending_op_id: 7000, pending: true, pending_finalize: 0,
|
||
|
|
deployed_fix_review: review, retry_reason: reason, new_failures: 0, recovery_budget: { ...scope,
|
||
|
|
attempt_number: 7, pending_op_id: 7000, status: 'active', max_provider_requests: 1000, reserved_provider_requests: 0,
|
||
|
|
deadline_at: review.deadline_at, reviewed_release: review.fix_release, reviewed_api_image: review.api_image_id } }
|
||
|
|
return { source, review, previous, pilot, receipt, evidence, bytes, events, testBytes, semantic, semanticBytes, build, buildBytes }
|
||
|
|
}
|
||
|
|
function validateEvidence(data) {
|
||
|
|
return seventh.validateSeventhCorrectionEvidence(data.review, data.bytes, data.evidence.source_tree_hash,
|
||
|
|
data.evidence.binary_sha256, data.testBytes, data.semanticBytes, data.buildBytes, data.evidence.binary_sha256)
|
||
|
|
}
|
||
|
|
function reseal(data) {
|
||
|
|
data.evidence.regression_tests_sha256 = sha(data.testBytes)
|
||
|
|
data.evidence.semantic_review_sha256 = sha(data.semanticBytes)
|
||
|
|
data.evidence.build_manifest_sha256 = sha(data.buildBytes)
|
||
|
|
data.bytes = JSON.stringify(data.evidence)
|
||
|
|
data.review.candidate_diagnostic.evidence_sha256 = sha(data.bytes)
|
||
|
|
data.review.candidate_diagnostic.build_manifest_sha256 = data.evidence.build_manifest_sha256
|
||
|
|
}
|
||
|
|
|
||
|
|
test('seventh exception preserves predecessor, is idempotent and never permits attempt eight', () => {
|
||
|
|
const d = fixture(), saved = JSON.stringify(d.previous)
|
||
|
|
assert.equal(policy.maxReviewedAttempt, 4)
|
||
|
|
assert.deepEqual(policy.planRecovery([d.source], [d.pilot, d.previous], 7, reason, d.review), [d.source])
|
||
|
|
assert.deepEqual(policy.planRecovery([d.source], [d.pilot, d.previous, d.receipt], 7, reason, d.review), [])
|
||
|
|
assert.doesNotThrow(() => policy.validateFinalRecoveryTrigger(d.receipt, { ...d.source, pending: 1 }, [d.pilot, d.previous, d.receipt], d.review))
|
||
|
|
assert.equal(JSON.stringify(d.previous), saved)
|
||
|
|
assert.equal(policy.supportedReviewedAttempt(8), false)
|
||
|
|
assert.throws(() => policy.planRecovery([d.source], [d.pilot, d.previous], 8, reason, { ...d.review, max_attempt: 8 }))
|
||
|
|
const terminal = { ...d.receipt, pending: false, recovery_status: 'stopped_budget', recovery_budget: { ...d.receipt.recovery_budget, status: 'stopped_budget' } }
|
||
|
|
assert.equal(scheduler.recoverySchedulingGate({ recoveries: [terminal], remaining_not_attempted: 0 }).state, 'stopped_for_review')
|
||
|
|
assert.deepEqual(scheduler.remainingRecoveryCandidates({ candidates: [d.source], recoveries: [terminal] }), [])
|
||
|
|
})
|
||
|
|
|
||
|
|
test('seventh recovery refuses predecessor, scope, progress or budget drift', () => {
|
||
|
|
const d = fixture()
|
||
|
|
for (const key of Object.keys(seventh.seventhRecoveryScope)) {
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery({ ...d.source, [key === 'dead_letter_id' ? 'id' : key]: 'changed' }, d.previous, d.review))
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, d.previous, { ...d.review, [key]: 'changed' }))
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, { ...d.previous, recovery_budget: { ...d.previous.recovery_budget, [key]: 'changed' } }, d.review))
|
||
|
|
}
|
||
|
|
for (const change of [{ pending: true }, { pending_finalize: 1 }, { attempt_number: 5 }, { pending_op_id: 4144 },
|
||
|
|
{ recovery_status: 'failed' }, { span_status: 'done' }, { error_code: 'OTHER' }, { new_failures: 1 },
|
||
|
|
{ queued_at: new Date().toISOString() }, { read_progress: {} }]) {
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, { ...d.previous, ...change }, d.review))
|
||
|
|
}
|
||
|
|
for (const key of Object.keys(seventh.seventhReadProgress)) {
|
||
|
|
const progress = { ...seventh.seventhReadProgress, [key]: 'changed' }
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, { ...d.previous, read_progress: { candidates: progress } }, d.review))
|
||
|
|
}
|
||
|
|
for (const change of [{ status: 'active' }, { stop_reason: 'guard_rejected' }, { last_phase: 'wiki_summary' },
|
||
|
|
{ reserved_provider_requests: 4 }, { reviewed_release: 'changed' }, { reviewed_api_image: 'changed' }, { paused_payload: '[]' }]) {
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, { ...d.previous, recovery_budget: { ...d.previous.recovery_budget, ...change } }, d.review))
|
||
|
|
}
|
||
|
|
for (const field of Object.keys(seventh.seventhPausedOperation.payload)) {
|
||
|
|
const paused = structuredClone(seventh.seventhPausedOperation); paused.payload[field] = 'changed'
|
||
|
|
assert.throws(() => seventh.validateSeventhRecovery(d.source, { ...d.previous,
|
||
|
|
recovery_budget: { ...d.previous.recovery_budget, paused_payload: JSON.stringify([paused]) } }, d.review))
|
||
|
|
}
|
||
|
|
for (const change of [{ max_provider_requests: 1001 }, { max_provider_requests: 0 }, { budget_version: 0 },
|
||
|
|
{ previous_pending_op_id: 4144 }, { previous_attempt: 5 }, { public_gates_passed: false },
|
||
|
|
{ fix_release: seventh.seventhPins.baseline_release }, { api_image_id: seventh.seventhPins.baseline_image },
|
||
|
|
{ deadline_at: new Date(Date.parse(d.review.budget_started_at) + 7200001).toISOString() }]) {
|
||
|
|
assert.throws(() => seventh.validateSeventhRecoveryReview({ ...d.review, ...change }))
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
test('fresh immutable evidence binds tests, separate semantic review, source, binary and build', () => {
|
||
|
|
const d = fixture()
|
||
|
|
assert.doesNotThrow(() => validateEvidence(d))
|
||
|
|
for (const key of ['bytes', 'testBytes', 'semanticBytes', 'buildBytes']) {
|
||
|
|
assert.throws(() => validateEvidence({ ...d, [key]: d[key] + '\n' }), key)
|
||
|
|
}
|
||
|
|
for (const change of [{ model_calls: 1 }, { production_writes: 1 }, { prior_evidence_preserved: false },
|
||
|
|
{ status: 'failed' }, { started_at: seventh.seventhPins.baseline_published_at },
|
||
|
|
{ completed_at: d.review.published_at }, { run_id: 'changed' }, { fix_release: 'changed' }, { api_image_id: 'changed' }]) {
|
||
|
|
const altered = fixture(); Object.assign(altered.evidence, change); reseal(altered)
|
||
|
|
assert.throws(() => validateEvidence(altered), JSON.stringify(change))
|
||
|
|
}
|
||
|
|
assert.throws(() => seventh.validateSeventhCorrectionEvidence(d.review, d.bytes, 'd'.repeat(64), d.evidence.binary_sha256,
|
||
|
|
d.testBytes, d.semanticBytes, d.buildBytes, d.evidence.binary_sha256))
|
||
|
|
assert.throws(() => seventh.validateSeventhCorrectionEvidence(d.review, d.bytes, d.evidence.source_tree_hash, 'd'.repeat(64),
|
||
|
|
d.testBytes, d.semanticBytes, d.buildBytes, d.evidence.binary_sha256))
|
||
|
|
assert.throws(() => seventh.validateSeventhCorrectionEvidence(d.review, d.bytes, d.evidence.source_tree_hash, d.evidence.binary_sha256,
|
||
|
|
d.testBytes, d.semanticBytes, d.buildBytes, 'd'.repeat(64)))
|
||
|
|
})
|
||
|
|
|
||
|
|
test('cached, skipped or failed correction regressions and unapproved semantic reviews cannot pass', () => {
|
||
|
|
for (const mutate of [d => { d.events[0].Action = 'output' }, d => { d.events[1].Action = 'skip' },
|
||
|
|
d => { d.events[1].Action = 'fail' }, d => { d.events[1].Package = 'github.com/Tencent/WeKnora/internal/agent' },
|
||
|
|
d => { d.events[0].Time = seventh.seventhPins.baseline_published_at },
|
||
|
|
d => { d.events = d.events.filter(e => e.Test) }]) {
|
||
|
|
const d = fixture(); mutate(d); d.testBytes = d.events.map(e => JSON.stringify(e)).join('\n')
|
||
|
|
d.semantic.regression_tests_sha256 = sha(d.testBytes); d.build.regression_tests_sha256 = sha(d.testBytes)
|
||
|
|
d.semanticBytes = JSON.stringify(d.semantic); d.buildBytes = JSON.stringify(d.build); reseal(d)
|
||
|
|
assert.throws(() => validateEvidence(d))
|
||
|
|
}
|
||
|
|
for (const change of [{ approved: false }, { candidate_count_preserved: false }, { aliases_not_removed: false },
|
||
|
|
{ no_extra_retry: false }, { prior_evidence_preserved: false }, { reviewed_by: '' }, { notes: '' },
|
||
|
|
{ source_tree_hash: 'e'.repeat(64) }, { reviewed_at: 'invalid' }]) {
|
||
|
|
const d = fixture(); Object.assign(d.semantic, change); d.semanticBytes = JSON.stringify(d.semantic); reseal(d)
|
||
|
|
assert.throws(() => validateEvidence(d))
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
function runCLI(d, trigger = false, drift) {
|
||
|
|
let transaction, triggered = 0
|
||
|
|
const errors = []
|
||
|
|
const context = { __dirname, process: { argv: ['node', 'recover_wiki_deadletters.cjs', ...(trigger ? ['--trigger', '--ids=10'] :
|
||
|
|
['--apply', '--ids=10', '--attempt=7', `--reason=${reason}`, '--review-file=/fixture/review.json'])] },
|
||
|
|
console: { log() {}, error(value) { errors.push(value) } }, require(name) {
|
||
|
|
if (name === 'node:fs') return { mkdirSync() {}, writeFileSync() {}, readFileSync(file) {
|
||
|
|
if (file === '/fixture/review.json') return JSON.stringify(d.review)
|
||
|
|
if (path.basename(file) === 'public-release.json') return JSON.stringify({ stage: 'published', release: d.review.fix_release,
|
||
|
|
api_image_id: d.review.api_image_id, published_at: d.review.published_at })
|
||
|
|
const files = { 'evidence.json': d.bytes, 'regression-tests.jsonl': d.testBytes, 'semantic-review.json': d.semanticBytes }
|
||
|
|
assert.ok(Object.hasOwn(files, path.basename(file)))
|
||
|
|
return files[path.basename(file)] + (drift === path.basename(file) ? '\n' : '')
|
||
|
|
} }
|
||
|
|
if (name === './wiki_seventh_recovery_review.cjs') return { ...seventh, verifySeventhBuildFiles() {
|
||
|
|
assert.notEqual(drift, 'source-manifest')
|
||
|
|
return { sourceTreeHash: d.evidence.source_tree_hash, frozenBinarySHA: d.evidence.binary_sha256, buildManifestBytes: d.buildBytes }
|
||
|
|
} }
|
||
|
|
if (name === './wiki_recovery_budget_sql.cjs') return { recoveryFingerprintsSQL: () => ({
|
||
|
|
source: `'${d.source.source_fingerprint}'::text`, model: `'${d.source.model_fingerprint}'::text` }) }
|
||
|
|
if (name === 'node:child_process') return { execFileSync(command, args, options = {}) {
|
||
|
|
assert.equal(command, 'docker')
|
||
|
|
if (args.includes('inspect')) return JSON.stringify({ running: true, release: d.review.fix_release, image_id: d.review.api_image_id })
|
||
|
|
if (args.includes('sha256sum')) return (drift === 'binary' ? 'e'.repeat(64) : d.evidence.binary_sha256) + ' /app/WeKnora\n'
|
||
|
|
if (args.includes('cp')) return ''
|
||
|
|
if (args.includes('/tmp/wiki-backfill-trigger')) { triggered++; return '{}' }
|
||
|
|
assert.ok(args.includes('psql'))
|
||
|
|
if (options.input.startsWith('BEGIN;')) { transaction = options.input; throw new Error('Captured transaction') }
|
||
|
|
if (options.input.includes('to_regclass')) return 't'
|
||
|
|
if (options.input.includes('FROM (SELECT ledger.*')) return JSON.stringify([d.pilot, d.previous, ...(trigger ? [d.receipt] : [])])
|
||
|
|
if (options.input.includes('SELECT coalesce(json_agg(row_to_json(d))')) return '[]'
|
||
|
|
return JSON.stringify([{ ...d.source, pending: trigger ? 1 : 0 }])
|
||
|
|
} }
|
||
|
|
return require(name)
|
||
|
|
} }
|
||
|
|
vm.runInNewContext(fs.readFileSync(path.join(__dirname, 'recover_wiki_deadletters.cjs'), 'utf8'), context)
|
||
|
|
return { transaction, triggered, errors }
|
||
|
|
}
|
||
|
|
|
||
|
|
test('actual CLI verifies immutable correction artifacts before enqueue and every explicit trigger', () => {
|
||
|
|
const d = fixture()
|
||
|
|
const first = runCLI(d)
|
||
|
|
assert.ok(first.transaction, first.errors.join('\n'))
|
||
|
|
const triggerRun = runCLI(d, true)
|
||
|
|
assert.equal(triggerRun.triggered, 1, triggerRun.errors.join('\n'))
|
||
|
|
for (const drift of ['evidence.json', 'regression-tests.jsonl', 'semantic-review.json', 'source-manifest', 'binary']) {
|
||
|
|
assert.equal(runCLI(d, false, drift).transaction, undefined, drift)
|
||
|
|
assert.equal(runCLI(d, true, drift).triggered, 0, drift)
|
||
|
|
}
|
||
|
|
d.previous.recovery_budget.reserved_provider_requests = 4
|
||
|
|
assert.equal(runCLI(d).transaction, undefined)
|
||
|
|
assert.equal(runCLI(d, true).triggered, 0)
|
||
|
|
})
|
||
|
|
|
||
|
|
test('actual atomic enqueue SQL rechecks exact stopped predecessor and checkpoint under locks', () => {
|
||
|
|
const sql = runCLI(fixture()).transaction
|
||
|
|
for (const required of ['p.attempt_number=6 AND p.pending_op_id=4145', "b.status='stopped_budget' AND b.stop_reason='phase_failed'",
|
||
|
|
"b.last_phase='wiki_candidate_slug' AND b.reserved_provider_requests=3", 'b.source_fingerprint=', 'b.model_fingerprint=',
|
||
|
|
'b.reviewed_release=', 'b.reviewed_api_image=', 'jsonb_array_length(b.paused_payload::jsonb)=1', 'b.paused_payload::jsonb @>',
|
||
|
|
"span.status='failed' AND span.error_code='EXTRACT_FAILED'", 'ORDER BY span.started_at DESC,span.id DESC LIMIT 1',
|
||
|
|
'checkpoint.payload::jsonb=', seventh.seventhPins.checkpoint_fingerprint, 'later.attempt_number>6', 'failure.id>p.dead_letter_highwater',
|
||
|
|
"interval '120 minutes'", '1000 BETWEEN 1 AND 1000', 'AND d.id=10 AND d.tenant_id=10059',
|
||
|
|
'LOCK TABLE wiki_recovery_budgets, task_dead_letters, knowledge_processing_spans, wiki_ingest_checkpoints IN SHARE MODE;',
|
||
|
|
'WHERE EXISTS (SELECT 1 FROM budget b WHERE b.pending_op_id=source.recovery_op_id)', 'FROM source d JOIN inserted i ON i.id=d.recovery_op_id;']) assert.ok(sql.includes(required), required)
|
||
|
|
assert.ok(sql.indexOf('pg_advisory_xact_lock') < sql.indexOf('LOCK TABLE wiki_recovery_budgets'))
|
||
|
|
assert.ok(sql.indexOf('LOCK TABLE wiki_recovery_budgets') < sql.indexOf('WITH source AS'))
|
||
|
|
assert.ok(sql.indexOf('INSERT INTO wiki_recovery_budgets') < sql.indexOf('INSERT INTO task_pending_ops'))
|
||
|
|
assert.match(sql, /^BEGIN;/)
|
||
|
|
assert.match(sql, /COMMIT;$/)
|
||
|
|
assert.doesNotMatch(sql, /DELETE FROM|UPDATE wiki_dead_letter_recoveries|UPDATE wiki_recovery_budgets/)
|
||
|
|
})
|