xinmi/scripts/wiki_seventh_recovery_review.test.cjs

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