xinmi/scripts/verify_graph_extraction_probe.test.cjs

const assert = require('node:assert/strict')
const fs = require('node:fs')
const path = require('node:path')
const vm = require('node:vm')
const os = require('node:os')
const crypto = require('node:crypto')
const test = require('node:test')

const code = fs.readFileSync(path.join(__dirname, 'verify_graph_extraction_probe.cjs'), 'utf8')
const chunk = 'e40b74eb-0ac7-4ff6-a107-00ce5adfe610'
const source = 'Reviewed source.'
const hash = crypto.createHash('sha256').update(source).digest('hex')
const modelID = '404e005b-83ae-5e2e-a3b3-999f9d5cd47e'
const env = { GRAPH_PROBE_CHUNK: chunk, GRAPH_PROBE_SOURCE_SHA256: hash, GRAPH_PROBE_MODEL_ID: modelID, GRAPH_PROBE_REPORT_TAG: 'bounded-test' }

function runProbe(directory, overrides = {}, changeFS, report, observed = {}) {
  let reads = 0, spawned = 0, stage
  const fakeFS = { ...fs, ...changeFS }
  const child = {
    execFileSync: (_command, args, opts) => {
      reads++
      if (args.includes('inspect')) return JSON.stringify([])
      assert.match(opts.input, /BEGIN READ ONLY/)
      assert.match(opts.input, /k\.parse_status='completed'/)
      assert.match(opts.input, /c\.is_enabled=true/)
      return JSON.stringify({model: {id:modelID,tenant_id:10059,type:'KnowledgeQA',parameters:{api_key:''}},chunk:{id:chunk,tenant_id:10059,content:source},extract_config:{}})
    },
    spawn: (command, args) => {
      spawned++
      stage = 'spawn'
      observed.command = command
      observed.args = args
      if (!report) throw new Error('TEST_STOP')
      return {
        stdin: { end: input => { observed.input = JSON.parse(input) } }, stderr: { resume: () => {} },
        stdout: { on: (_event, receive) => receive(Buffer.from('GRAPH_EXTRACTION_PROBE_RESULT=' + JSON.stringify(report) + '\n')) },
        on: (event, callback) => { if (event === 'close') callback(0) },
      }
    },
  }
  const context = { __dirname: path.join(directory, 'src/scripts'), Buffer,
    console: { error: text => { stage = JSON.parse(text).stage }, log: () => { stage = 'complete'; throw new Error('TEST_STOP') } },
    process: { env: { ...env, ...overrides }, exit: () => { throw new Error('TEST_STOP') } },
    require: name => name === 'node:child_process' ? child : name === 'node:fs' ? fakeFS : require(name) }
  assert.throws(() => vm.runInNewContext(code, context), /TEST_STOP/)
  return { reads, spawned, stage }
}

function fixture(t) {
  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-probe-'))
  fs.mkdirSync(path.join(directory, 'acceptance'), { mode: 0o700 })
  t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
  return directory
}

for (const tc of [
  {name:'exact scope reaches bounded probe',env:{},stage:'spawn',reads:2},
  {name:'changed source stops before credential access',env:{GRAPH_PROBE_SOURCE_SHA256:'0'.repeat(64)},stage:'source_validation',reads:1},
  {name:'invalid hash',env:{GRAPH_PROBE_SOURCE_SHA256:'oops'},stage:'input',reads:0},
  {name:'uppercase hash',env:{GRAPH_PROBE_SOURCE_SHA256:hash.toUpperCase()},stage:'input',reads:0},
  {name:'tag cannot traverse paths',env:{GRAPH_PROBE_REPORT_TAG:'../other'},stage:'input',reads:0},
  {name:'missing tag',env:{GRAPH_PROBE_REPORT_TAG:''},stage:'input',reads:0},
  {name:'missing model',env:{GRAPH_PROBE_MODEL_ID:''},stage:'input',reads:0},
  {name:'missing hash',env:{GRAPH_PROBE_SOURCE_SHA256:''},stage:'input',reads:0},
]) {
  test(tc.name, t => {
    const result = runProbe(fixture(t), tc.env)
    assert.equal(result.stage, tc.stage)
    assert.equal(result.reads, tc.reads)
    assert.equal(result.spawned, tc.stage === 'spawn' ? 1 : 0)
  })
}

test('a failed launch spends the allocation durably before any report exists', t => {
  const directory = fixture(t)
  assert.equal(runProbe(directory).stage, 'spawn')
  const files = fs.readdirSync(path.join(directory, 'acceptance'))
  assert.equal(files.length, 1)
  assert.match(files[0], /\.spent\.json$/)
  const receipt = path.join(directory, 'acceptance', files[0])
  const before = fs.readFileSync(receipt)
  assert.equal(fs.statSync(receipt).mode & 0o777, 0o600)
  assert.equal(JSON.parse(before).allocated_http_sends, 3)
  assert.deepEqual(runProbe(directory), {stage:'execution_receipt', reads:0, spawned:0})
  assert.deepEqual(fs.readFileSync(receipt), before)
})

test('receipt create and fsync errors stop before launching a model process', t => {
  for (const operation of ['openSync', 'fsyncSync']) {
    const directory = fixture(t)
    const result = runProbe(directory, {}, { [operation]: () => { throw new Error('offline storage fault') } })
    assert.equal(result.stage, 'execution_receipt')
    assert.equal(result.spawned, 0)
  }
})

test('exclusive receipt creation closes the check-create race', t => {
  const directory = fixture(t)
  assert.equal(runProbe(directory).stage, 'spawn')
  const result = runProbe(directory, {}, {existsSync: () => false})
  assert.equal(result.stage, 'execution_receipt')
  assert.equal(result.spawned, 0)
})

test('only valid current execution accounting can complete a probe report', t => {
  const valid = { diagnostics_version: 2, execution_id: `graph-extraction-probe-${chunk}-${modelID}-bounded-test.json`,
    http_send_limit: 3, http_send_reservations: 2, model_calls: 2, passed: true }
  for (const change of [{}, {diagnostics_version:1}, {execution_id:'other'}, {http_send_limit:4},
    {http_send_reservations:4}, {http_send_reservations:0}, {model_calls:3}]) {
    const directory = fixture(t)
    const result = runProbe(directory, {}, undefined, {...valid, ...change})
    assert.equal(result.stage, Object.keys(change).length === 0 ? 'complete' : 'model_probe_budget')
    assert.deepEqual(runProbe(directory), {stage:Object.keys(change).length === 0 ? 'report_write' : 'execution_receipt',reads:0,spawned:0})
  }
})

test('invalid send caps stop before setup, receipt creation or process launch', t => {
  for (const value of ['', '0', '4', '-1', '2.0', '02', ' 2', '2 ', '2\n', '2\r\n', 'NaN', 'Infinity']) {
    const directory = fixture(t)
    assert.deepEqual(runProbe(directory, {GRAPH_PROBE_MAX_SENDS:value}), {stage:'input', reads:0, spawned:0}, value)
    assert.deepEqual(fs.readdirSync(path.join(directory, 'acceptance')), [], value)
  }
})

test('selected and default send caps agree in receipt, input, child environment and report', t => {
  for (const cap of [undefined, '1', '2', '3']) {
    const directory = fixture(t)
    const observed = {}
    const limit = Number(cap ?? '3')
    const report = { diagnostics_version: 2, execution_id: `graph-extraction-probe-${chunk}-${modelID}-bounded-test.json`,
      http_send_limit: limit, http_send_reservations: limit, model_calls: 1, passed: true }
    const result = runProbe(directory, {GRAPH_PROBE_MAX_SENDS:cap}, undefined, report, observed)
    assert.equal(result.stage, 'complete')
    assert.equal(observed.command, 'docker')
    assert.equal(observed.input.max_http_sends, limit)
    assert.equal(observed.args.filter(value => value.startsWith('GRAPH_PROBE_MAX_SENDS=')).length, 1)
    assert.ok(observed.args.includes(`GRAPH_PROBE_MAX_SENDS=${limit}`))
    const receiptName = fs.readdirSync(path.join(directory, 'acceptance')).find(name => name.endsWith('.spent.json'))
    const receipt = JSON.parse(fs.readFileSync(path.join(directory, 'acceptance', receiptName)))
    assert.equal(receipt.allocated_http_sends, limit)
    assert.equal(receipt.execution_id, observed.input.execution_id)
    assert.equal(receipt.model_deadline_seconds, 180)
  }
})

test('a tighter cap rejects mismatched or excessive return accounting and stays spent', t => {
  const valid = { diagnostics_version: 2, execution_id: `graph-extraction-probe-${chunk}-${modelID}-bounded-test.json`,
    http_send_limit: 2, http_send_reservations: 2, model_calls: 2, passed: true }
  for (const change of [{http_send_limit:3}, {http_send_limit:1}, {http_send_limit:'2'},
    {http_send_reservations:3}, {http_send_reservations:-1}, {http_send_reservations:1.5}]) {
    const directory = fixture(t)
    assert.equal(runProbe(directory, {GRAPH_PROBE_MAX_SENDS:'2'}, undefined, {...valid, ...change}).stage, 'model_probe_budget')
    const files = fs.readdirSync(path.join(directory, 'acceptance'))
    assert.equal(files.length, 1)
    const receiptPath = path.join(directory, 'acceptance', files[0])
    const before = fs.readFileSync(receiptPath)
    assert.equal(JSON.parse(before).allocated_http_sends, 2)
    assert.deepEqual(runProbe(directory, {GRAPH_PROBE_MAX_SENDS:'3'}), {stage:'execution_receipt', reads:0, spawned:0})
    assert.deepEqual(fs.readFileSync(receiptPath), before)
  }
})