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/) })