// FED R4 (#1801, federation design 3.7 and the R4 row of section 7): revocation. A loss of a user's
// grant ends the claim of every run on that user's threads, from one of three triggers: the timer's
// accessSweep, the human's access.reload, and the next request of any user once the file changed.
// The access document and its stamp come in as functions, as the runtime passes loadAccess and an
// lstat key; `edit` changes the document and the stamp together, as a hand edit does. BUILD.md
// (agents/morgana/work/queue-295) names the mutant each test kills.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
import { ACCESS_SWEEP_MS, startBroker } from '../src/runtime.mjs';

const shipped = (name) =>
  JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
const grant = (over = {}) => ({
  id: `g-${over.user ?? 'alice'}`, user: 'alice', business: 'demo', projects: ['*'], instances: ['assistant'], level: 'collaborator', ...over,
});
const BUSINESSES = {
  demo: {
    id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' }, vars: { 'bus.tester': true }, plugins: { harness: ['pi'] },
    roles: {
      chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
      assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
    },
    definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging } },
  },
};

// Alice and Bob each hold a collaborator grant and own a thread; the human owns thread H. With
// `runs`, each thread has a claimed run: a-run on A, b-run on B, h-run on H.
function setup(t, { runs = true, grants = [grant(), grant({ user: 'bob' })] } = {}) {
  const root = mkdtempSync(join(tmpdir(), 'bus-access-revoke-'));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  // `reads` counts loader calls; `rev` is the stamp, which a hand edit changes.
  const access = { doc: { accessVersion: 1, revision: 1, users: [], grants }, reads: 0, rev: 1 };
  const broker = () => new Broker({
    store,
    businesses: BUSINESSES,
    access: () => {
      access.reads++;
      return access.doc;
    },
    accessStamp: () => String(access.rev),
  });
  const b = broker();
  const caps = {
    human: b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }),
    tester: b.bindTester({ business: 'demo' }),
    reader: b.bindReader({ business: 'demo' }),
  };
  const call = (who, verb, args = {}) => b.request(caps[who], { verb, args });
  const code = (who, verb, args = {}) => {
    try {
      call(who, verb, args);
    } catch (e) {
      return e.code;
    }
    return null;
  };
  const edit = (fn) => {
    access.doc = { ...access.doc, revision: access.doc.revision + 1, grants: fn(access.doc.grants) };
    access.rev++;
  };
  const record = (name, thread) => ({ business: 'demo', role: 'assistant', run: `${name}-run`, harness: 'pi', thread });
  const launch = (name, thread) => {
    caps[name] = b.bindLaunch(record(name, thread));
    return call(name, 'role.claim');
  };
  // The latest claim row of a run: its op, reason and by, spread off the store's null prototype.
  const claim = (run) => ({ ...store.get('SELECT op, reason, by FROM thread_claims WHERE holder_run=? ORDER BY seq DESC LIMIT 1', run) });
  const P = call('human', 'project.create', { name: 'shared' }).id;
  caps.alice = b.bindUser({ business: 'demo', user: 'alice' });
  caps.bob = b.bindUser({ business: 'demo', user: 'bob' });
  const A = call('alice', 'thread.create', { project: P, name: 'alice work' }).id;
  const B = call('bob', 'thread.create', { project: P, name: 'bob work' }).id;
  const H = call('human', 'thread.create', { project: P, name: 'owner work' }).id;
  if (runs) {
    launch('a', A);
    launch('b', B);
    launch('h', H);
  }
  return { b, store, access, caps, call, code, edit, record, launch, claim, broker, P, A, B, H };
}

const withoutAlice = (gs) => gs.filter((g) => g.user !== 'alice');

test('R4: the timer sweep after a hand edit ends only the lost user\'s run; its next call refuses thread-revoked', (t) => {
  const { b, code, edit, claim, A } = setup(t);
  edit(withoutAlice);
  const ended = b.accessSweep();
  assert.equal(ended.length, 1);
  assert.deepEqual({ ...ended[0], session: undefined }, { business: 'demo', thread: A, session: undefined, run: 'a-run' });
  assert.equal(typeof ended[0].session, 'string');
  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-timer' });
  assert.equal(code('a', 'message.receive'), 'thread-revoked', 'the ended run');
  assert.equal(code('b', 'message.receive'), null, 'another user\'s run stays');
  assert.equal(code('h', 'message.receive'), null, 'the human\'s run stays');
  assert.deepEqual([claim('b-run').op, claim('h-run').op], ['claim', 'claim']);
  // Ended once: a second sweep finds nothing to end.
  assert.deepEqual(b.accessSweep(), []);
});

test('R4: an expiry with no write and no request ends the run at the next timer sweep', (t) => {
  t.mock.timers.enable({ apis: ['Date'], now: Date.parse('2026-10-23T00:00:00Z') });
  const { b, code, claim } = setup(t, { grants: [grant({ expires: '2026-10-23T00:01:00Z' }), grant({ user: 'bob' })] });
  assert.deepEqual(b.accessSweep(), [], 'live');
  t.mock.timers.tick(60_001);
  // Bob's request sees the same stamp, so it doesn't sweep: the timer is what catches an expiry.
  assert.equal(code('bob', 'thread.list'), null);
  assert.equal(claim('a-run').op, 'claim');
  assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
  assert.equal(code('a', 'message.receive'), 'thread-revoked');
  assert.equal(code('b', 'message.receive'), null);
});

test('R4: a downgrade to viewer is a loss', (t) => {
  const { b, code, edit, claim } = setup(t);
  edit((gs) => gs.map((g) => (g.user === 'alice' ? { ...g, level: 'viewer' } : g)));
  assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-timer' });
  assert.equal(code('b', 'message.receive'), null);
});

// sage-3139 Q1: no timer runs here, so only the request trigger can end alice's run.
test('R4: another user\'s request after a hand edit ends the lost user\'s run, and its own run stays', (t) => {
  const { code, edit, claim } = setup(t);
  edit(withoutAlice);
  assert.equal(claim('a-run').op, 'claim', 'nothing ends before a request');
  assert.equal(code('a', 'message.receive'), null, 'an agent request is not a trigger');
  assert.equal(code('bob', 'thread.list'), null);
  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-change' });
  assert.equal(code('a', 'message.receive'), 'thread-revoked');
  assert.equal(code('b', 'message.receive'), null);
  assert.equal(code('h', 'message.receive'), null);
});

// sage-3139 Q1, the cost rule: a request only stats the file, and reads it once per change.
test('R4: a request reads the access file only when its stamp changed, once per change', (t) => {
  const { access, code, edit } = setup(t);
  const reads = access.reads;
  assert.equal(code('bob', 'thread.list'), null);
  assert.equal(code('alice', 'project.list'), null);
  assert.equal(access.reads, reads, 'unchanged: no read');
  edit((gs) => gs);
  assert.equal(code('bob', 'thread.list'), null);
  assert.equal(code('alice', 'thread.list'), null);
  assert.equal(access.reads, reads + 1, 'changed: one read for both requests');
});

test('R4: access.reload ends the lost runs at once for the human; the tester, a reader, an agent and a user are refused', (t) => {
  const { b, code, call, edit, claim, A } = setup(t);
  assert.equal(code('tester', 'access.reload'), 'human-required', 'tester');
  assert.equal(code('reader', 'access.reload'), 'read-only', 'reader');
  assert.equal(code('b', 'access.reload'), 'human-required', 'agent');
  assert.equal(code('bob', 'access.reload'), 'access-refused', 'user');
  assert.deepEqual(call('human', 'access.reload'), { ended: [] }, 'nothing lost');
  assert.equal(code('human', 'access.reload', { force: true }), 'invalid-request');
  edit(withoutAlice);
  const { ended } = call('human', 'access.reload');
  assert.deepEqual(ended.map((h) => [h.thread, h.run]), [[A, 'a-run']]);
  assert.equal(typeof ended[0].session, 'string');
  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'jason' });
  assert.equal(code('a', 'message.receive'), 'thread-revoked');
  assert.equal(code('b', 'message.receive'), null);
  assert.deepEqual(b.accessSweep(), []);
});

test('R4: threadsState marks revoked: true on the lost user\'s thread only, a pending message included', (t) => {
  const { b, call, edit, A, B, H } = setup(t, { runs: false });
  call('alice', 'thread.send', { thread: A, body: 'for A' });
  call('bob', 'thread.send', { thread: B, body: 'for B' });
  const row = (id) => b.threadsState('demo').find((x) => x.thread === id);
  assert.equal(row(A).revoked, undefined, 'covered');
  edit(withoutAlice);
  assert.equal(row(A).revoked, true);
  assert.equal(row(A).pending, 1, 'its message is still pending');
  assert.equal(Object.hasOwn(row(B), 'revoked'), false);
  assert.equal(Object.hasOwn(row(H), 'revoked'), false);
  assert.equal(Object.hasOwn(row(A), 'owner'), false, 'the row keeps its old fields');
});

// Dewey's B1 (dewey-0871): the launcher ticks faster than the timer, so a grant that expires with no
// write must not reach it as revoked: true with the claim still held.
test('R4: threadsState after an expiry with no write and no sweep ends the claim before it marks the row revoked', (t) => {
  t.mock.timers.enable({ apis: ['Date'], now: Date.parse('2026-10-23T00:00:00Z') });
  const { b, code, claim, store, A, B } = setup(t, { grants: [grant({ expires: '2026-10-23T00:01:00Z' }), grant({ user: 'bob' })] });
  const row = (id) => b.threadsState('demo').find((x) => x.thread === id);
  assert.equal(row(A).revoked, undefined, 'live');
  t.mock.timers.tick(60_001);
  const a = row(A);
  assert.deepEqual([a.revoked, a.holder], [true, null]);
  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-expiry' });
  assert.equal(code('a', 'message.receive'), 'thread-revoked');
  assert.deepEqual([row(B).holder, claim('b-run').op], ['b-run', 'claim']);
  // Ended once: the next call writes no second revoke.
  row(A);
  assert.equal(store.get("SELECT count(*) AS n FROM thread_claims WHERE holder_run='a-run' AND op='revoke'").n, 1);
});

test('R4: a new run of a revoked thread refuses thread-revoked at bind and at claim; a rebind is allowed', (t) => {
  const { b, call, edit, record, broker, A, B } = setup(t, { runs: false });
  const late = b.bindLaunch(record('late', A));
  edit(withoutAlice);
  assert.throws(() => b.request(late, { verb: 'role.claim' }), { code: 'thread-revoked' }, 'bound before, claimed after');
  assert.throws(() => b.bindLaunch(record('new', A)), { code: 'thread-revoked' }, 'bound after');
  assert.ok(b.bindLaunch(record('other', B)), 'bob\'s thread binds');
  call('bob', 'thread.list');
  // D1: a later broker over the same store rebinds the run an earlier one bound (recover()), so the
  // host can end it.
  const again = broker();
  assert.ok(again.bindLaunch(record('late', A)));
  assert.equal(again.endLaunch({ business: 'demo', run: 'late-run', reason: 'recovered' }).released, false);
});

// The runtime's timer (design 3.7 trigger 2) on a real 0600 access file read by loadAccess. The mocked
// setInterval stands in for 30 s; the run is watched through its own calls, since threadsState and a
// user request would sweep by themselves.
test('R4 runtime: a hand edit ends the run at the first timer period, intervalMs is checked, and close stops the timer', async (t) => {
  t.mock.timers.enable({ apis: ['setInterval'] });
  const root = mkdtempSync(join(tmpdir(), 'bus-access-timer-'));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  const dataRoot = join(root, 'data');
  mkdirSync(dataRoot, { mode: 0o700 });
  const file = join(root, 'access.json');
  const write = (grants) => writeFileSync(file, JSON.stringify({
    accessVersion: 1, revision: 1, users: [{ id: 'alice', home: 'local', subjects: [{ provider: 'authentik', sub: 'alice-sub' }] }], grants,
  }), { mode: 0o600 });
  write([grant()]);
  const context = { providers: ['authentik'], peers: [], businesses: { demo: { human: 'jason', instances: ['chief', 'assistant'] } }, ssoSubjects: [] };
  for (const intervalMs of [0, -1, 1.5, '30000'])
    await assert.rejects(startBroker({ dataRoot, businesses: BUSINESSES, access: { file, context, intervalMs } }), { code: 'invalid-host-request' }, String(intervalMs));
  const runtime = await startBroker({ dataRoot, businesses: BUSINESSES, access: { file, context } });
  let closed = false;
  t.after(() => closed || runtime.close());
  const b = runtime.broker;
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const alice = b.bindUser({ business: 'demo', user: 'alice' });
  const P = b.request(human, { verb: 'project.create', args: { name: 'shared' } }).id;
  const A = b.request(alice, { verb: 'thread.create', args: { project: P, name: 'alice work' } }).id;
  const run = b.bindLaunch({ business: 'demo', role: 'assistant', run: 'a-run', harness: 'pi', thread: A });
  b.request(run, { verb: 'role.claim' });
  const receive = () => {
    try {
      b.request(run, { verb: 'message.receive' });
    } catch (e) {
      return e.code;
    }
    return null;
  };
  write([]);
  t.mock.timers.tick(ACCESS_SWEEP_MS - 1);
  assert.equal(receive(), null, 'before the first period');
  t.mock.timers.tick(1);
  assert.equal(receive(), 'thread-revoked', 'at the first period');
  // After close the timer is gone: the sweep is never called again.
  let sweeps = 0;
  b.accessSweep = () => void sweeps++;
  t.mock.timers.tick(ACCESS_SWEEP_MS);
  assert.equal(sweeps, 1, 'one more period, one sweep');
  closed = true;
  await runtime.close();
  t.mock.timers.tick(ACCESS_SWEEP_MS * 3);
  assert.equal(sweeps, 1, 'closed');
});