/**
 * @arbiter/value-graph tests — compile → run separation for BOTH queries and
 * invalidations/eager updates, pull-on-duplex-push queries, TTL + version
 * staleness, dependent-DAG cascade, all DSL value types, swappable stores.
 *
 * Resolver contract: CALLBACK ORIENTED (no async/await/promises). Resolvers may
 * call `cb(err, result)` asynchronously (returning `undefined`) or return the
 * result synchronously. A returned Promise is rejected.
 */
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, readdirSync, existsSync, rmSync, readFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
  ValueGraph, createMapStore, createFileStore,
  encodeSnapshot, decodeSnapshot, encodeGraphSnapshot, decodeGraphSnapshot,
  encodeParams, decodeParams
} from '../src/index.js';
import { createSink } from '@push-stream-std/push-stream-base';

// Promise-free library, promise-ful test harness: adapt the callback API.
function get(vg, subject, rel, params = {}) {
  return new Promise((resolve, reject) => {
    vg.get(subject, rel, params, (err, result) => err ? reject(err) : resolve(result));
  });
}
function run(vg, plan) {
  return new Promise((resolve, reject) => {
    vg.run(plan, (err, result) => err ? reject(err) : resolve(result));
  });
}

function collect(stream, { until, timeout = 4000 } = {}) {
  return new Promise((resolve, reject) => {
    const items = [];
    const timer = setTimeout(() => reject(new Error(`collect timeout: ${JSON.stringify(items)}`)), timeout);
    stream.pipe(createSink(
      (data) => {
        items.push(data);
        if (until && until(items)) { clearTimeout(timer); resolve(items); }
      },
      (err) => { clearTimeout(timer); err ? reject(err) : resolve(items); }
    ));
  });
}

describe('ValueGraph compile → run (queries)', () => {
  it('prunes cached subtrees into a single trivial leaf, then runs it', async () => {
    let calls = 0;
    const vg = new ValueGraph();
    vg.compute('double_balance', (s, p, { deps }) => { calls++; return { value: deps.base_balance * 2, unit: 'usd_cents' }; }, { dependsOn: ['base_balance'] });
    vg.compute('base_balance', (s, p, ctx, cb) => cb(null, 21));

    const p1 = vg.plan('t:1', 'double_balance', {});
    assert.equal(p1.size, 2);
    assert.deepEqual(p1.nodes.get('double_balance').parents, ['base_balance']);
    assert.equal((await run(vg, p1)).value, 42);

    const p2 = vg.plan('t:1', 'double_balance', {});
    assert.equal(p2.nodes.get('double_balance').trivial, true);
    assert.deepEqual(p2.nodes.get('double_balance').parents, []);
    assert.equal(p2.size, 1, 'cached subtree pruned');
    assert.equal(calls, 1, 'no recompute');
    assert.equal((await run(vg, p2)).value, 42);
  });

  it('accepts synchronous resolver returns (pure compute)', async () => {
    const vg = new ValueGraph();
    vg.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
    vg.compute('base_balance', () => 21);
    assert.equal((await get(vg, 't:1', 'double_balance', {})).value, 42);
  });

  it('rejects a Promise-returning resolver (promise-free contract)', async () => {
    const vg = new ValueGraph();
    vg.compute('bad', async () => 42);
    await assert.rejects(() => get(vg, 'u:1', 'bad', {}), /returned a Promise/);
  });

  it('recomputes after TTL expiry (plan marks the node non-trivial)', async () => {
    let t = 0;
    let value = 10;
    const vg = new ValueGraph({ clock: () => t, defaultTTL: 1000 });
    vg.compute('quota', (s, p, ctx, cb) => cb(null, value));

    assert.equal((await get(vg, 't:1', 'quota', {})).value, 10);
    assert.equal(vg.plan('t:1', 'quota', {}).nodes.get('quota').trivial, true);
    t = 1500;
    value = 20;
    assert.equal(vg.plan('t:1', 'quota', {}).nodes.get('quota').trivial, false, 'expired → non-trivial');
    assert.equal((await get(vg, 't:1', 'quota', {})).value, 20);
  });

  it('resolves a query duplex with repeated get pushes (pull on duplex push)', async () => {
    let calls = 0;
    const vg = new ValueGraph();
    vg.compute('balance', (s, p, ctx, cb) => { calls++; cb(null, { value: 42, unit: 'usd_cents' }); });

    const q = vg.query('user:alice', 'balance', {});
    const events = collect(q.source, { until: (items) => items.filter(i => i.value !== undefined).length >= 2 });
    q.sink.write({ get: true });
    q.sink.write({ get: true });
    const items = await events;

    const values = items.filter(i => i.value !== undefined).map(i => i.value);
    assert.deepEqual(values, [42, 42]);
    assert.equal(calls, 1, 'second get served from the cached trivial plan');
  });

  it('rejects a dependency cycle at compile time', async () => {
    const vg = new ValueGraph();
    vg.compute('a', () => 1, { dependsOn: ['b'] });
    vg.compute('b', () => 2, { dependsOn: ['a'] });
    await assert.rejects(() => get(vg, 'x', 'a', {}), /cycle/);
  });
});

describe('ValueGraph compile → run (invalidation / eager updates)', () => {
  it('invalidate compiles the dependent DAG and recomputes downstream', async () => {
    let base = 10;
    const vg = new ValueGraph();
    vg.compute('base_balance', () => base);
    vg.compute('double_balance', (s, p, { deps }) => ({ value: deps.base_balance * 2, unit: 'usd_cents' }), { dependsOn: ['base_balance'] });
    vg.compute('quad_balance', (s, p, { deps }) => deps.double_balance * 2, { dependsOn: ['double_balance'] });

    assert.equal((await get(vg, 't:1', 'quad_balance', {})).value, 40);
    assert.equal(vg.plan('t:1', 'quad_balance', {}).size, 1, 'cached: single leaf');

    base = 30;
    vg.invalidate('t:1', 'base_balance', {}); // affects base + double + quad transitively

    const p = vg.plan('t:1', 'quad_balance', {});
    assert.equal(p.size, 3, 'invalidation expanded the plan to the full dependent DAG');
    assert.equal((await run(vg, p)).value, 120, 'recomputed down the chain');
  });

  it('pushes a stale trigger down on invalidation and recomputes', async () => {
    let value = 100;
    const vg = new ValueGraph();
    vg.compute('available', () => value);

    const q = vg.query('tenant:acme', 'available', { feature: 'tokens_in:gpt-4' });
    const events = collect(q.source, {
      until: (items) => items.some(i => i.stale) && items.filter(i => i.value !== undefined).length >= 2
    });
    q.sink.write({ get: true });
    value = 50;
    vg.invalidate('tenant:acme', 'available', { feature: 'tokens_in:gpt-4' });
    q.sink.write({ get: true });
    const items = await events;

    assert.ok(items.some(i => i.stale === true), 'stale trigger pushed down');
    assert.ok(items.filter(i => i.value !== undefined).some(i => i.value === 50), 'recomputed to 50');
  });

  it('set compiles the dependent DAG: writes the source, invalidates dependents', async () => {
    let calls = 0;
    const vg = new ValueGraph();
    vg.compute('base_balance', () => 10);
    vg.compute('double_balance', (s, p, { deps }) => { calls++; return deps.base_balance * 2; }, { dependsOn: ['base_balance'] });

    assert.equal((await get(vg, 't:1', 'double_balance', {})).value, 20);
    // Eager update from outside (e.g. a ledger posting).
    vg.set('t:1', 'base_balance', {}, { value: 50, unit: 'usd_cents' });

    // double_balance is no longer cached — the set invalidated it.
    assert.equal(vg.plan('t:1', 'double_balance', {}).nodes.get('double_balance').trivial, false);
    const got = await get(vg, 't:1', 'double_balance', {});
    assert.equal(got.value, 100, 'dependent recomputed after eager set');
  });

  it('cascades a stale trigger to dependent queries on set', async () => {
    const vg = new ValueGraph();
    vg.compute('base_balance', () => 10);
    vg.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });

    const q = vg.query('t:1', 'double_balance', {});
    const events = collect(q.source, { until: (items) => items.some(i => i.stale) });
    q.sink.write({ get: true });
    vg.set('t:1', 'base_balance', {}, { value: 99, unit: 'usd_cents' });
    const items = await events;
    assert.ok(items.some(i => i.stale === true && i.relation === 'double_balance' && i.via === 'base_balance'));
  });
});

describe('ValueGraph value types', () => {
  it('supports number, string, boolean, timestamp, and arrays', async () => {
    const vg = new ValueGraph();
    vg.compute('reputation', () => 7.5);
    vg.compute('role', () => 'admin');
    vg.compute('active', () => true);
    vg.compute('last_login', () => 1700000000000);
    vg.compute('permissions', () => ['read', 'write']);

    assert.equal((await get(vg, 'u:1', 'reputation', {})).value, 7.5);
    assert.equal((await get(vg, 'u:1', 'role', {})).value, 'admin');
    assert.equal((await get(vg, 'u:1', 'active', {})).value, true);
    assert.equal((await get(vg, 'u:1', 'last_login', {})).value, 1700000000000);
    assert.deepEqual((await get(vg, 'u:1', 'permissions', {})).value, ['read', 'write']);
  });

  it('supports interval values ({ lower, upper }) for possibilistic estimates', async () => {
    const vg = new ValueGraph();
    vg.compute('estimated_tokens', () => ({ lower: 800, upper: 1400 }));
    const entry = await get(vg, 't:1', 'estimated_tokens', {});
    assert.deepEqual(entry.value, { lower: 800, upper: 1400 });
  });

  it('rejects an unregistered value type', async () => {
    const vg = new ValueGraph();
    vg.compute('bad', () => ({ value: { nested: true } }));
    await assert.rejects(() => get(vg, 'u:1', 'bad', {}), /must return a value/);
  });
});

describe('ValueGraph typed values (DSL-declared returnType)', () => {
  it('exposes the declared spec via relationSpec', () => {
    const vg = new ValueGraph();
    vg.define('balance', { operator: 'source', returnType: 'number', params: [{ name: 'tenant', type: 'Employee', isArray: false }], fn: () => 0 });
    const spec = vg.relationSpec('balance');
    assert.equal(spec.returnType, 'number');
    assert.equal(spec.params[0].name, 'tenant');
    assert.equal(spec.operator, 'source');
    assert.equal(vg.relationSpec('nope'), null);
  });

  it('set enforces the declared return type', () => {
    const vg = new ValueGraph();
    vg.define('role', { operator: 'source', returnType: 'string', fn: () => null });
    vg.set('t:1', 'role', {}, { value: 'admin' });
    assert.throws(() => vg.set('t:1', 'role', {}, { value: 42 }), /must match declared type 'string'/);
  });

  it('resolver results are validated against the declared return type', async () => {
    const vg = new ValueGraph();
    vg.define('role', { operator: 'source', returnType: 'string', fn: () => 42 });
    await assert.rejects(() => get(vg, 't:1', 'role', {}), /must match declared type 'string'/);
  });

  it('accepts interval values via set (consistent with the resolver path)', async () => {
    const vg = new ValueGraph();
    vg.define('estimate', { operator: 'source', returnType: 'interval', fn: () => null });
    vg.set('t:1', 'estimate', {}, { value: { lower: 100, upper: 200 } });
    const e = await get(vg, 't:1', 'estimate', {});
    assert.deepEqual(e.value, { lower: 100, upper: 200 });
  });

  it('entity-typed values accept keys (non-primitive type names)', () => {
    const vg = new ValueGraph();
    vg.define('owner', { operator: 'source', returnType: 'Employee', fn: () => null });
    vg.set('t:1', 'owner', {}, { value: 'user:alice' }); // key accepted
    assert.throws(() => vg.set('t:1', 'owner', {}, { value: { id: 1 } }), /requires a value/); // object not a valid key
  });
});

describe('ValueGraph invalidation guards', () => {
  it('invalidate(subject) without a relation throws instead of clearing the whole store', () => {
    const vg = new ValueGraph();
    vg.compute('a', () => 1);
    assert.throws(() => vg.invalidate('t:1'), /requires a valueRelation/);
  });

  it('invalidate() with no arguments still clears the whole store', async () => {
    const vg = new ValueGraph();
    vg.compute('a', () => 1);
    await get(vg, 't:1', 'a', {});
    assert.equal(vg.plan('t:1', 'a', {}).nodes.get('a').trivial, true);
    vg.invalidate();
    assert.equal(vg.plan('t:1', 'a', {}).nodes.get('a').trivial, false, 'cleared');
  });
});

describe('ValueGraph backing store', () => {
  it('uses a swappable backing store', async () => {
    const store = createMapStore();
    let put = 0;
    const tracked = { ...store, set(key, value) { put++; store.set(key, value); } };
    const vg = new ValueGraph({ store: tracked });
    vg.compute('score', () => 7);
    assert.equal((await get(vg, 'u:1', 'score', {})).value, 7);
    assert.equal(put, 1);
  });

  it('returns null when no compute is registered and nothing is cached', async () => {
    const vg = new ValueGraph();
    assert.equal(await get(vg, 'u:1', 'nope', {}), null);
  });
});

describe('ValueGraph larger-than-memory (disk-backed store)', () => {
  function tmpDir() {
    return mkdtempSync(join(tmpdir(), 'value-graph-'));
  }

  it('persists entries to disk and reads them back into a fresh graph instance', async () => {
    const dir = tmpDir();
    try {
      // First process: compute and cache against the disk store.
      let calls = 0;
      const vg1 = new ValueGraph({ store: createFileStore(dir) });
      vg1.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
      vg1.compute('base_balance', (s, p, ctx, cb) => { calls++; cb(null, 21); });
      assert.equal((await get(vg1, 't:1', 'double_balance', {})).value, 42);
      assert.equal(calls, 1);
      assert.ok(readdirSync(dir).length > 0, 'entries written to disk');

      // Second process: a brand-new graph over the SAME directory.
      const vg2 = new ValueGraph({ store: createFileStore(dir) });
      vg2.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
      vg2.compute('base_balance', () => { throw new Error('should not recompute from disk'); });
      const plan = vg2.plan('t:1', 'double_balance', {});
      assert.equal(plan.nodes.get('double_balance').trivial, true, 'cached leaf restored from disk');
      assert.equal((await run(vg2, plan)).value, 42, 'value read from disk, no recompute');
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });

  it('keeps only a bounded number of entries resident (hot-page cache)', async () => {
    const dir = tmpDir();
    try {
      const vg = new ValueGraph({ store: createFileStore(dir, { maxResident: 2 }) });
      for (let i = 0; i < 5; i++) {
        vg.compute(`v${i}`, () => i * 10);
      }
      for (let i = 0; i < 5; i++) {
        assert.equal((await get(vg, 't', `v${i}`, {})).value, i * 10);
      }
      // All five entries persist on disk even though only 2 are resident.
      assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 5);
      const again = await get(vg, 't', 'v0', {}); // evicted from cache → reloaded from disk
      assert.equal(again.value, 0);
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });

  it('invalidate/set delete the persisted files', async () => {
    const dir = tmpDir();
    try {
      const vg = new ValueGraph({ store: createFileStore(dir) });
      vg.compute('balance', () => 100);
      assert.equal((await get(vg, 't:1', 'balance', {})).value, 100);
      assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 1);

      vg.invalidate('t:1', 'balance', {});
      assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 0, 'invalidate removed the file');

      vg.set('t:1', 'balance', {}, { value: 42, unit: 'usd_cents' });
      assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 1, 'set persisted the file');
      assert.equal((await get(vg, 't:1', 'balance', {})).value, 42);
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });

  it('honors the store contract for a query duplex on a disk-backed graph', async () => {
    const dir = tmpDir();
    try {
      const vg = new ValueGraph({ store: createFileStore(dir) });
      vg.compute('available', () => 88);
      const q = vg.query('tenant:acme', 'available', { feature: 'gpt-4' });
      const events = collect(q.source, { until: (items) => items.some((i) => i.value === 88) });
      q.sink.write({ get: true });
      await events;
      assert.ok(readdirSync(dir).some((f) => f.endsWith('.bin')), 'query result persisted to disk');
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });
});

describe('ValueGraph operator model (ADR-000 OWA)', () => {
  function make(operators, parents) {
    const vg = new ValueGraph();
    for (const [rel, value] of Object.entries(parents)) {
      vg.define(rel, { operator: 'source', fn: () => value });
    }
    for (const [rel, op] of Object.entries(operators)) {
      vg.define(rel, { operator: `fusion:${op}`, parents: Object.keys(parents) });
    }
    return vg;
  }

  it('applies the OWA default weight distributions', async () => {
    // values 10, 20, 30
    const vg = make({}, { a: 10, b: 20, c: 30 });
    vg.define('mx', { operator: 'fusion:max', parents: ['a', 'b', 'c'] });
    vg.define('mn', { operator: 'fusion:min', parents: ['a', 'b', 'c'] });
    vg.define('av', { operator: 'fusion:average', parents: ['a', 'b', 'c'] });
    vg.define('sm', { operator: 'fusion:sum_unbounded', parents: ['a', 'b', 'c'] });
    vg.define('md', { operator: 'fusion:median', parents: ['a', 'b', 'c'] });

    assert.equal((await get(vg, 't', 'mx', {})).value, 30);
    assert.equal((await get(vg, 't', 'mn', {})).value, 10);
    assert.ok(Math.abs((await get(vg, 't', 'av', {})).value - 20) < 1e-9, 'average ≈ 20');
    assert.equal((await get(vg, 't', 'sm', {})).value, 60);
    assert.equal((await get(vg, 't', 'md', {})).value, 20);
  });

  it('applies custom OWA weights', async () => {
    const vg = make({}, { a: 100, b: 0 });
    vg.define('custom', { operator: 'fusion:custom', parents: ['a', 'b'], weights: [0.75, 0.25] });
    // sorted [100, 0]; 0.75*100 + 0.25*0 = 75
    assert.equal((await get(vg, 't', 'custom', {})).value, 75);
  });

  it('supports optimistic / pessimistic / top2', async () => {
    const vg = make({}, { a: 10, b: 20, c: 30 });
    vg.define('opt', { operator: 'fusion:optimistic', parents: ['a', 'b', 'c'] });
    vg.define('pes', { operator: 'fusion:pessimistic', parents: ['a', 'b', 'c'] });
    vg.define('t2', { operator: 'fusion:top2', parents: ['a', 'b', 'c'] });
    // optimistic [0.5,0.25,0.125]→ normalized; pes reversed; top2 = 0.5*30+0.5*20 = 25
    assert.ok((await get(vg, 't', 'opt', {})).value > 20, 'optimistic favors high values');
    assert.ok((await get(vg, 't', 'pes', {})).value < 20, 'pessimistic favors low values');
    assert.equal((await get(vg, 't', 't2', {})).value, 25);
  });

  it('supports non-numeric max/min/majority (strings)', async () => {
    const vg = make({}, { a: 'low', b: 'high', c: 'high' });
    vg.define('mx', { operator: 'fusion:max', parents: ['a', 'b', 'c'] });
    vg.define('maj', { operator: 'fusion:majority', parents: ['a', 'b', 'c'] });
    assert.equal((await get(vg, 't', 'mx', {})).value, 'low' > 'high' ? 'low' : 'high');
    assert.equal((await get(vg, 't', 'maj', {})).value, 'high');
  });

  it('computes via an operator node with parents (deps passed via ctx)', async () => {
    const vg = new ValueGraph();
    vg.define('base', { operator: 'source', fn: (s, p, ctx, cb) => cb(null, 7) });
    vg.define('twice', { operator: 'compute', parents: ['base'], fn: (subject, params, { deps }) => deps.base * 2 });
    assert.equal((await get(vg, 't', 'twice', {})).value, 14);
  });

  it('reads an attribute via the resolveAttribute callback hook', async () => {
    const graph = new Map([['user:1', { balance: 500 }]]);
    const vg = new ValueGraph({
      resolveAttribute: (nodeKey, path, params, ctx, cb) => {
        const node = graph.get(nodeKey);
        cb(null, node ? node[path] : undefined);
      }
    });
    vg.define('balance', { operator: 'attribute', attribute: 'balance' });
    assert.equal((await get(vg, 'user:1', 'balance', {})).value, 500);
  });

  it('resolves a pattern measure via the resolvePattern callback hook, lazily and cached', async () => {
    let calls = 0;
    const edges = [['u:1', 'r', 'x'], ['u:1', 'r', 'y'], ['u:1', 'r', 'z']];
    const vg = new ValueGraph({
      resolvePattern: (pattern, subject, params, ctx, cb) => {
        calls++;
        cb(null, edges.filter(e => e[0] === subject && e[1] === pattern.relation).length);
      }
    });
    vg.define('degree', { operator: 'pattern', pattern: { relation: 'r' } });
    assert.equal((await get(vg, 'u:1', 'degree', {})).value, 3);
    assert.equal((await get(vg, 'u:1', 'degree', {})).value, 3);
    assert.equal(calls, 1, 'pattern cached');
  });

  it('invalidating a parent recomputes a fusion node', async () => {
    let b = 10;
    const vg = new ValueGraph();
    vg.define('a', { operator: 'source', fn: () => 5 });
    vg.define('base', { operator: 'source', fn: () => b });
    vg.define('sum', { operator: 'fusion:sum_unbounded', parents: ['a', 'base'] });
    assert.equal((await get(vg, 't', 'sum', {})).value, 15);
    b = 100;
    vg.invalidate('t', 'base', {});
    assert.equal((await get(vg, 't', 'sum', {})).value, 105, 'fusion recomputed after parent invalidation');
  });
});

describe('ValueGraph blackbox node (function of its dependencies)', () => {
  it('receives parent values (ctx.deps) and full entries (ctx.entries)', async () => {
    const vg = new ValueGraph();
    vg.define('base', { operator: 'source', fn: (s, p, ctx, cb) => cb(null, { value: 21, unit: 'usd_cents' }) });
    let seen = null;
    vg.define('total', {
      operator: 'blackbox',
      parents: ['base'],
      fn: (subject, params, ctx, cb) => {
        seen = ctx;
        cb(null, ctx.deps.base * 2);
      }
    });

    const entry = await get(vg, 't:1', 'total', {});
    assert.equal(entry.value, 42);
    assert.equal(seen.deps.base, 21, 'ctx.deps exposes parent VALUES');
    assert.equal(seen.entries.base.value, 21, 'ctx.entries exposes the full parent result');
    assert.equal(seen.entries.base.unit, 'usd_cents');
    assert.equal(seen.entries.base.source, 'source');
    assert.equal(seen.entries.base.fresh, true);
  });

  it('queries an external service keyed on parent edges (callback style)', async () => {
    const vg = new ValueGraph();
    const edges = [['u:1', 'r', 'x'], ['u:1', 'r', 'y']];
    vg.define('related', {
      operator: 'pattern',
      pattern: { relation: 'r' },
      fn: (s, p, ctx, cb) => cb(null, edges.filter(e => e[0] === s))
    });
    // Fake Overlay: given the related edges, score the subject asynchronously.
    const overlayQuery = (edgeList, cb) => {
      setTimeout(() => cb(null, { value: edgeList.length * 10, unit: 'score', source: 'overlay:evals' }), 5);
    };
    vg.define('score', {
      operator: 'blackbox',
      parents: ['related'],
      fn: (subject, params, ctx, cb) => overlayQuery(ctx.deps.related, cb)
    });

    const entry = await get(vg, 'u:1', 'score', {});
    assert.equal(entry.value, 20);
    assert.equal(entry.source, 'overlay:evals');
  });

  it('recomputes a blackbox node when a parent edge set changes', async () => {
    const edges = [['u:1', 'r', 'x']];
    const vg = new ValueGraph();
    vg.define('related', {
      operator: 'pattern',
      pattern: { relation: 'r' },
      fn: (s, p, ctx, cb) => cb(null, edges.filter(e => e[0] === s))
    });
    vg.define('count', {
      operator: 'blackbox',
      parents: ['related'],
      fn: (s, p, { deps }) => deps.related.length
    });

    assert.equal((await get(vg, 'u:1', 'count', {})).value, 1);
    edges.push(['u:1', 'r', 'y'], ['u:1', 'r', 'z']);
    vg.invalidate('u:1', 'related', {});
    assert.equal((await get(vg, 'u:1', 'count', {})).value, 3, 'blackbox re-run against the new edges');
  });

  it('requires an fn', async () => {
    const vg = new ValueGraph();
    vg.define('broken', { operator: 'blackbox', parents: [] });
    await assert.rejects(() => get(vg, 't', 'broken', {}), /blackbox node 'broken' needs an fn/);
  });
});

describe('ValueGraph compact snapshots (persistence & transport)', () => {
  function tmpDir() {
    return mkdtempSync(join(tmpdir(), 'value-graph-snap-'));
  }

  // A graph whose entries exercise every value type: number, string,
  // boolean, buffer (Uint8Array), array (with nested object + buffer), interval.
  // NOTE: no JS bigint — large integers are Buffers (the wire is JSON-free).
  function richGraph({ store, defaultTTL = 0 } = {}) {
    const vg = new ValueGraph({ store, defaultTTL });
    vg.define('num', { operator: 'source', fn: () => 42 });
    vg.define('str', { operator: 'source', fn: () => 'hello' });
    vg.define('bool', { operator: 'source', fn: () => true });
    vg.define('buf', { operator: 'source', fn: () => new Uint8Array([1, 2, 3, 254, 255]) });
    vg.define('arr', { operator: 'source', fn: () => [1, 'two', new Uint8Array([3, 4]), { nested: true }] });
    vg.define('iv', { operator: 'source', fn: () => ({ lower: 0.5, upper: 1.5 }) });
    vg.define('fusion_risk', {
      operator: 'fusion:custom', parents: ['num', 'buf'], weights: [0.25, 0.75],
      capSum: true, returnType: 'number'
    });
    return vg;
  }

  it('buffer (Uint8Array) values round-trip byte-exactly through the wire', async () => {
    const vg = richGraph();
    await get(vg, 't', 'buf', {});
    const buf = vg.snapshot();
    const snap = decodeGraphSnapshot(buf);
    const value = snap.records.find((r) => r.relation === 'buf').entry.value;
    assert.deepEqual([...value], [1, 2, 3, 254, 255]);
  });

  it('value-level round-trip: encodeSnapshot/decodeSnapshot is exact for all value types', async () => {
    const vg = richGraph();
    // num with a nested-params key (params round-trip through _parseKey, JSON-free)
    await get(vg, 't', 'num', { x: [1, { y: 2 }] });
    await get(vg, 't', 'str', {});
    await get(vg, 't', 'bool', {});
    await get(vg, 't', 'buf', {});
    await get(vg, 't', 'arr', {});
    await get(vg, 't', 'iv', {});

    const buf = vg.snapshot();
    assert.ok(buf.length > 0);
    assert.equal(Buffer.from(buf.subarray(0, 4)).toString('ascii'), 'VGGP', 'graph magic');
    const snap = decodeGraphSnapshot(buf);
    assert.equal(snap.records.length, 6);
    assert.equal(snap.schema.length, 7);
    const byRel = new Map(snap.records.map((r) => [r.relation, r.entry]));
    assert.equal(byRel.get('num').value, 42);
    assert.equal(byRel.get('str').value, 'hello');
    assert.equal(byRel.get('bool').value, true);
    assert.deepEqual([...byRel.get('buf').value], [1, 2, 3, 254, 255]);
    assert.deepEqual(byRel.get('arr').value[0], 1);
    assert.equal(byRel.get('arr').value[1], 'two');
    assert.deepEqual([...byRel.get('arr').value[2]], [3, 4]);
    assert.deepEqual(byRel.get('arr').value[3], { nested: true });
    assert.deepEqual(byRel.get('iv').value, { lower: 0.5, upper: 1.5 });
    assert.deepEqual(snap.records.find((r) => r.relation === 'num').params, { x: [1, { y: 2 }] }, 'nested params round-trip');
  });

  it('graph snapshot restores into a fresh graph: values served WITHOUT recompute', async () => {
    let calls = 0;
    const vg = richGraph();
    await get(vg, 't', 'num', {});
    await get(vg, 't', 'buf', {});
    const buf = vg.snapshot();

    const restored = ValueGraph.restore(buf, {
      resolvers: { num: () => { calls++; return -1; } }
    });
    assert.equal(calls, 0);
    const e = await get(restored, 't', 'num', {});
    assert.equal(e.value, 42);
    assert.equal(e.fresh, true, 'restored entry treated as fresh (version re-stamped)');
    assert.deepEqual([...(await get(restored, 't', 'buf', {})).value], [1, 2, 3, 254, 255]);
    assert.equal(restored.relationSpec('fusion_risk').operator, 'fusion:custom');
    assert.deepEqual(restored.relationSpec('fusion_risk').parents, ['num', 'buf']);
    assert.equal(restored.relationSpec('fusion_risk').returnType, 'number');
  });

  it('schema metadata round-trips: operator, parents, weights, capSum, returnType', async () => {
    const vg = new ValueGraph({ defaultTTL: 0 });
    vg.define('s', { operator: 'source', fn: () => 1 });
    vg.define('f', {
      operator: 'fusion:priority', parents: ['s'], priorities: [0.4],
      ttl: 5000, returnType: 'number', params: [{ name: 'p', type: 'number' }]
    });
    await get(vg, 't', 's', {});
    const restored = ValueGraph.restore(vg.snapshot(), { resolvers: { s: () => 1 } });
    const spec = restored.relationSpec('f');
    assert.equal(spec.operator, 'fusion:priority');
    assert.deepEqual(spec.parents, ['s']);
    assert.equal(spec.ttl, 5000);
    assert.equal(spec.returnType, 'number');
    assert.deepEqual(spec.params, [{ name: 'p', type: 'number' }]);
  });

  it('loadFile/snapshotFile: portable single-file transport between "processes"', async () => {
    const dir = tmpDir();
    try {
      // Process 1: disk-backed graph computes values, writes ONE portable file.
      const vg1 = new ValueGraph({ store: createFileStore(dir), defaultTTL: 0 });
      vg1.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
      vg1.compute('base_balance', (s, p, ctx, cb) => cb(null, 21));
      assert.equal((await get(vg1, 't:1', 'double_balance', {})).value, 42);
      const snapPath = join(dir, 'graph.snap');
      vg1.snapshotFile(snapPath);
      assert.ok(existsSync(snapPath));

      // Process 2: brand-new graph, EMPTY store, reads the portable file.
      // Resolvers re-registered; the resolver would throw if recomputed.
      const loaded = ValueGraph.loadFile(snapPath, {
        resolvers: { base_balance: () => { throw new Error('recompute!'); } }
      });
      assert.equal(loaded.relations.size, 2, 'schema restored');
      const e = await get(loaded, 't:1', 'double_balance', {});
      assert.equal(e.value, 42);
      assert.equal(e.fresh, true, 'no recompute — cache restored from disk');
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });

  it('store-level snapshot: opaque key→entry dump round-trips through any store', async () => {
    const vg = new ValueGraph({ defaultTTL: 0 });
    vg.compute('balance', (s, p, { deps }) => deps.rate * deps.units, { dependsOn: ['rate', 'units'] });
    vg.compute('rate', () => 3);
    vg.compute('units', () => 4);
    assert.equal((await get(vg, 'a', 'balance', {})).value, 12);

    const storeBuf = vg.snapshot({ includeSchema: false });
    const snap = decodeSnapshot(storeBuf);
    assert.equal(snap.entries.length, 3);
    assert.equal(Buffer.from(storeBuf.subarray(0, 4)).toString('ascii'), 'VGST', 'store magic');

    // Feed the raw dump into a fresh map store, then serve from it.
    const fresh = new ValueGraph({ store: createMapStore(), defaultTTL: 0 });
    for (const [key, entry] of snap.entries) fresh.store.set(key, entry);
    assert.equal((await get(fresh, 'a', 'balance', {})).value, 12, 'opaque dump replays into a new store');
  });

  it('createFileStore loadFrom: a store constructed already loaded from a snapshot', async () => {
    const dirA = tmpDir();
    const dirB = tmpDir();
    try {
      const vg = new ValueGraph({ store: createFileStore(dirA), defaultTTL: 0 });
      vg.compute('balance', () => 99);
      assert.equal((await get(vg, 'u:1', 'balance', {})).value, 99);
      const snapPath = join(dirA, 'store.snap');
      vg.store.snapshotFile(snapPath);

      // New store in a DIFFERENT directory, constructed with loadFrom.
      const store = createFileStore(dirB, { loadFrom: snapPath });
      assert.ok(store.get(vg._key('u:1', 'balance', {})) !== undefined, 'snapshot preloaded');
      const vg2 = new ValueGraph({ store, defaultTTL: 0 });
      vg2.compute('balance', () => { throw new Error('recompute!'); });
      assert.equal((await get(vg2, 'u:1', 'balance', {})).value, 99, 'served from the loaded snapshot');
      assert.ok(readdirSync(dirB).filter((f) => f.endsWith('.bin')).length >= 1, 'materialized into working dir');
    } finally {
      rmSync(dirA, { recursive: true, force: true });
      rmSync(dirB, { recursive: true, force: true });
    }
  });

  it('corrupt or truncated snapshots are rejected (CRC guard), never silently mis-read', async () => {
    const vg = new ValueGraph({ defaultTTL: 0 });
    vg.compute('balance', () => 7);
    await get(vg, 'u', 'balance', {});
    const buf = vg.snapshot();
    assert.doesNotThrow(() => decodeGraphSnapshot(buf));

    // Flip a byte inside createdAt: parse succeeds, CRC must reject it.
    const flipped = Buffer.from(buf);
    flipped[9] ^= 0xff;
    assert.throws(() => decodeGraphSnapshot(flipped), /CRC mismatch/);

    // Flip a byte inside a record payload: the parser may desync — the result
    // must be a coherent corruption error, never a silent wrong answer.
    const payloadFlip = Buffer.from(buf);
    payloadFlip[payloadFlip.length - 8] ^= 0xff;
    assert.throws(() => decodeGraphSnapshot(payloadFlip), /CRC mismatch|corrupt/);

    const truncated = Buffer.from(buf.subarray(0, buf.length - 4)); // missing CRC
    assert.throws(() => decodeGraphSnapshot(truncated), /truncated/);

    const badMagic = Buffer.from(buf);
    badMagic[0] = 0x00;
    assert.throws(() => decodeGraphSnapshot(badMagic), /not a value-graph snapshot/);

    // Store flavour gets the same guards.
    const storeBuf = vg.snapshot({ includeSchema: false });
    assert.doesNotThrow(() => decodeSnapshot(storeBuf));
    assert.throws(() => decodeSnapshot(Buffer.from(storeBuf.subarray(0, storeBuf.length - 5))), /truncated|CRC|corrupt/);
  });

  it('instance restore merges a snapshot into a live graph (schema declared if absent)', async () => {
    const src = new ValueGraph({ defaultTTL: 0 });
    src.compute('rate', () => 5);
    assert.equal((await get(src, 'u', 'rate', {})).value, 5);

    const dst = new ValueGraph({ defaultTTL: 0 });
    dst.compute('quota', () => 100);
    await get(dst, 'u', 'quota', {});
    dst.restore(src.snapshot());

    assert.equal(dst.relations.has('rate'), true, 'foreign schema declared');
    assert.equal((await get(dst, 'u', 'rate', {})).value, 5, 'foreign entries loaded');
    assert.equal((await get(dst, 'u', 'quota', {})).value, 100, 'local entries untouched');
  });

  it('disk-backed store snapshot is byte-identical to the in-memory encode', async () => {
    const dir = tmpDir();
    try {
      const vg = new ValueGraph({ store: createFileStore(dir), defaultTTL: 0 });
      vg.compute('balance', () => 11);
      vg.compute('tag', () => 'x');
      await get(vg, 'u:1', 'balance', {});
      await get(vg, 'u:1', 'tag', {});

      const snapPath = join(dir, 'g.snap');
      vg.snapshotFile(snapPath, { createdAt: 12345 });
      const fromFile = readFileSync(snapPath);
      const fromBuffer = vg.snapshot({ createdAt: 12345 });
      // Same schema+records, deterministic createdAt + CRC → identical bytes.
      assert.deepEqual([...fromFile], [...fromBuffer], 'file snapshot == buffer snapshot');
    } finally {
      rmSync(dir, { recursive: true, force: true });
    }
  });
});