TREK/server/tests/integration/plugins/supervisor.test.ts

/**
 * End-to-end proof of the isolated plugin runtime (#plugins, M1). Forks a REAL
 * child process, loads a fixture plugin, and verifies:
 *   - onLoad runs isolated and its ctx.db round-trips through RPC to the host,
 *   - an ungranted capability is refused across the fork boundary,
 *   - a plugin that throws in onLoad fails activation without touching the host,
 *   - disable() tears the child down.
 * The child runs its own process — its crash/throw can never reach this test.
 */
import { PluginDataDb } from '../../../src/nest/plugins/host/plugin-data.service';
import { PluginRpcHost, type HostDeps } from '../../../src/nest/plugins/host/rpc-host';
import { createTestPluginRegistry } from '../../../src/nest/plugins/host/rpc-kit/testing';
import { DbRpc } from '../../../src/nest/plugins/host/rpc/db.rpc';
import type { PluginUserSettingsService } from '../../../src/nest/plugins/plugin-user-settings.service';
import {
  PluginSupervisor,
  type SupervisorHooks,
  type SupervisorTuning,
} from '../../../src/nest/plugins/supervisor/plugin-supervisor';
import { createTestOrm, type TestOrm } from '../../helpers/test-orm';

import Database from 'better-sqlite3';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';

/**
 * DbRpc also carries `settings.get`, which needs the per-user settings store. No
 * fixture below calls it, and none of these plugins has a stored setting anyway, so
 * the stub is the one read DbRpc makes, typed off the real signature and answering
 * "nothing stored" exactly as the real service would.
 */
const stubUserSettings: Pick<PluginUserSettingsService, 'readOne'> = { readOne: () => undefined };

let codeRoot: string;
let dataRoot: string;
let sup: PluginSupervisor;

const broadcasts: unknown[] = [];

function writePlugin(id: string, source: string): void {
  const dir = path.join(codeRoot, id, 'server');
  fs.mkdirSync(dir, { recursive: true });
  fs.writeFileSync(path.join(dir, 'index.js'), source);
}

function makeSupervisor(
  events: Array<{ topic: string; data: unknown }>,
  tuning: SupervisorTuning = {},
): PluginSupervisor {
  const createRpcHost = (id: string, granted: ReadonlySet<string>): PluginRpcHost => {
    const deps: HostDeps = {
      data: new PluginDataDb(id),
      callPlugin: async () => undefined,
      emitPluginEvent: async (event, payload) => {
        broadcasts.push({ id, event, payload });
      },
    };
    // Only db.* is exercised from a child here; the rest of the surface has its own
    // unit suites and would drag every domain service into this integration test.
    return new PluginRpcHost(
      id,
      granted,
      deps,
      createTestPluginRegistry([new DbRpc(stubUserSettings as PluginUserSettingsService)]),
    );
  };
  const hooks: SupervisorHooks = {
    onEvent: (_id, topic, data) => events.push({ topic, data }),
    onLog: (_id, level, msg, meta) => events.push({ topic: '__log', data: { level, msg, meta } }),
  };
  // task-6-fix-brief.md item 1: every real `ctx.*` round-trip below dispatches
  // through PluginSupervisor.onMessage's 'req' branch over real child-process
  // IPC, which now THROWS without a resolveOrm thunk rather than dispatching
  // unwrapped — so this double needs a real (if otherwise unused) ORM, even
  // though PluginDataDb/DbRpc here are raw better-sqlite3, not MikroORM.
  return new PluginSupervisor(createRpcHost, hooks, tuning, () => ormHandle.orm);
}

/**
 * Pull the `meta` of a named `ctx.log.info(msg, meta)` call. Fixtures report
 * structured diagnostics this way now that the child's raw `process.send` is
 * sealed off (installIpcGuard) — the SDK log channel is the only route out.
 */
function logMeta<T = Record<string, unknown>>(events: Array<{ topic: string; data: unknown }>, msg: string): T {
  const hit = events.find((e) => e.topic === '__log' && (e.data as { msg?: string }).msg === msg);
  return (hit && (hit.data as { meta?: unknown }).meta) as T;
}

const ormDb = new Database(':memory:');
let ormHandle: TestOrm;

beforeAll(async () => {
  codeRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'trekplug-code-'));
  dataRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'trekplug-pdata-'));
  process.env.TREK_PLUGINS_DIR = codeRoot;
  process.env.TREK_PLUGINS_DATA_DIR = dataRoot;
  ormHandle = await createTestOrm(ormDb);
});
afterAll(async () => {
  delete process.env.TREK_PLUGINS_DIR;
  delete process.env.TREK_PLUGINS_DATA_DIR;
  fs.rmSync(codeRoot, { recursive: true, force: true });
  fs.rmSync(dataRoot, { recursive: true, force: true });
  await ormHandle.close();
  ormDb.close();
});
afterEach(async () => {
  await sup?.shutdownAll();
});

describe('PluginSupervisor — isolated runtime', () => {
  it('loads a plugin in a child process: ctx.db round-trips and an ungranted capability is denied', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);

    writePlugin(
      'hello',
      `module.exports = {
        async onLoad(ctx) {
          await ctx.db.migrate('001', 'CREATE TABLE kv (k TEXT PRIMARY KEY, v TEXT)');
          await ctx.db.exec('INSERT INTO kv (k, v) VALUES (?, ?)', ['greeting', ctx.config.greeting || 'none']);
          const rows = await ctx.db.query('SELECT v FROM kv WHERE k = ?', 'greeting');
          let tripsDenied = false;
          try { await ctx.trips.getById(1, 42); } catch (e) { tripsDenied = /PERMISSION_DENIED/.test(e.message); }
          ctx.log.info('selftest complete');
          ctx.log.info('diag', { value: rows[0] && rows[0].v, tripsDenied });
        }
      };`,
    );

    await sup.activate('hello', new Set(['db:own']), { greeting: 'hi there' });
    expect(sup.isActive('hello')).toBe(true);

    const diag = logMeta<{ value: string; tripsDenied: boolean }>(events, 'diag');
    expect(diag).toBeTruthy();
    expect(diag.value).toBe('hi there'); // instance config reached the child, db round-trip worked
    expect(diag.tripsDenied).toBe(true); // ungranted db:read:trips was refused across the fork boundary
    // the plugin's ctx.log surfaced through the supervisor's log hook
    expect(events.some((e) => e.topic === '__log')).toBe(true);
  });

  it('dispatches invoke.hook + invoke.event across the fork (provider hooks + event subscriptions)', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);
    writePlugin(
      'provider',
      `module.exports = {
        hooks: {
          placeDetailProvider: { async getDetails(placeId, ctx) { return [{ label: 'placeId', value: String(placeId) }]; } },
        },
        events: [
          { on: 'place:created', async handler(e, ctx) { ctx.log.info('got-event', e); } },
          { on: 'other:thing', handler() { throw new Error('non-matching subscription must not run'); } },
        ],
      };`,
    );
    await sup.activate('provider', new Set(['hook:place-detail-provider', 'events:subscribe']), {});

    // host->plugin hook: the child runs getDetails(7, ctx) and returns its result
    const hookRes = await sup.invoke(
      'provider',
      'invoke.hook',
      { hook: 'placeDetailProvider', fn: 'getDetails', args: [7] },
      { actingUserId: 5 },
    );
    expect(hookRes).toEqual([{ label: 'placeId', value: '7' }]);

    // host->plugin event: ONLY the matching subscription runs (the 'other:thing' one throws if hit)
    await sup.invoke('provider', 'invoke.event', { event: 'place:created', tripId: 3 }, { actingUserId: undefined });
    const got = logMeta<{ event: string; tripId: number }>(events, 'got-event');
    expect(got).toEqual({ event: 'place:created', tripId: 3 });
  });

  it('reports its MCP tools at load and dispatches a tool call as the requesting user', async () => {
    // plugin-host-entry.ts is excluded from coverage (it runs in a forked
    // subprocess), so this is the only thing proving the loaded payload and the
    // invoke path for mcpToolProvider.
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);
    writePlugin(
      'toolbox',
      `module.exports = {
        hooks: {
          mcpToolProvider: {
            tools: ['echo', 'whoami'],
            async callTool(call, ctx) { return { name: call.name, args: call.args }; },
          },
        },
      };`,
    );
    await sup.activate('toolbox', new Set(['mcp:tools']), {});

    expect(sup.mcpToolsOf('toolbox')).toEqual(['echo', 'whoami']);
    expect(sup.providersOf('mcpToolProvider')).toContain('toolbox');
    expect([...sup.grantsOf('toolbox')]).toContain('mcp:tools');

    const res = await sup.invoke(
      'toolbox',
      'invoke.hook',
      { hook: 'mcpToolProvider', fn: 'callTool', args: [{ name: 'echo', args: { v: 1 } }] },
      { actingUserId: 5 },
    );
    expect(res).toEqual({ name: 'echo', args: { v: 1 } });
  });

  it('reports the functions each hook carries at load, class methods included (#2221)', async () => {
    // Only the forked child knows whether searchProvider has the optional suggest, and
    // plugin-host-entry.ts is excluded from coverage, so this is the proof that the
    // report reaches the host for a literal and for a class instance alike.
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);
    writePlugin(
      'typeahead',
      `class LocalIndex {
        async search() { return []; }
        async suggest(request) { return [{ id: 'x', name: request.query, lat: 1, lng: 2 }]; }
        // Reading the hook's functions must not run a getter, let alone fail the load on one.
        get broken() { throw new Error('getter ran at load'); }
      }
      module.exports = { hooks: { searchProvider: new LocalIndex() } };`,
    );
    writePlugin('search-only', `module.exports = { hooks: { searchProvider: { async search() { return []; } } } };`);
    await sup.activate('typeahead', new Set(['hook:search-provider']), {});
    await sup.activate('search-only', new Set(['hook:search-provider']), {});

    expect(sup.providersOf('searchProvider').sort()).toEqual(['search-only', 'typeahead']);
    expect(sup.providersOf('searchProvider', 'suggest')).toEqual(['typeahead']);
    const res = await sup.invoke(
      'typeahead',
      'invoke.hook',
      { hook: 'searchProvider', fn: 'suggest', args: [{ query: 'ic', limit: 3 }] },
      { actingUserId: 5 },
    );
    expect(res).toEqual([{ id: 'x', name: 'ic', lat: 1, lng: 2 }]);
  });

  it('stops reporting MCP tools once the plugin is no longer active', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);
    writePlugin(
      'toolbox2',
      `module.exports = { hooks: { mcpToolProvider: { tools: ['echo'], async callTool() { return 1; } } } };`,
    );
    await sup.activate('toolbox2', new Set(['mcp:tools']), {});
    expect(sup.mcpToolsOf('toolbox2')).toEqual(['echo']);

    await sup.disable('toolbox2');
    expect(sup.mcpToolsOf('toolbox2')).toEqual([]);
    expect(sup.providersOf('mcpToolProvider')).not.toContain('toolbox2');
  });

  it('seals the raw IPC surface: a plugin cannot forge/sniff over process.send/on(message), and a forged init cannot reopen egress', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);

    // The plugin runs each raw-IPC attack in try/catch and reports the outcome
    // over the ONLY channel it still has — the SDK log (ctx.log → realSend).
    // No egress is granted, so a fetch to an undeclared host must fail with an
    // `egress:` error; if a forged `init` reopened egress, the second fetch would
    // instead fail with a network error (or succeed) — that is the discriminator.
    writePlugin(
      'attacker',
      `module.exports = {
        async onLoad(ctx) {
          const results = {};
          const probe = (name, fn) => { try { fn(); results[name] = 'NOT_BLOCKED'; } catch { results[name] = 'blocked'; } };
          probe('send_evt', () => process.send({ k: 'evt', topic: 'loaded', data: { routes: [{ i: 0, method: 'GET', path: '/pwned', auth: false }] } }));
          probe('on_message', () => process.on('message', () => {}));
          probe('once_message', () => process.once('message', () => {}));
          probe('prepend_message', () => process.prependListener('message', () => {}));
          probe('removeAll_message', () => process.removeAllListeners('message'));
          probe('removeAll_bare', () => process.removeAllListeners());
          probe('forged_req', () => process.send({ k: 'req', method: 'trips.getById', params: { tripId: 1, _inv: 'forged' } }));
          probe('disconnect', () => process.disconnect());

          const fetchOutcome = async () => {
            try { await fetch('https://blocked.example/x'); return 'NO_ERROR'; }
            catch (e) { return /egress/.test(e.message) ? 'egress' : 'other'; }
          };
          const egressBefore = await fetchOutcome();
          // process.emit is intentionally NOT sealed (Node delivers inbound IPC
          // through it); the one-shot init guard must make this forged init a no-op.
          let emitThrew = false;
          try { process.emit('message', { k: 'evt', topic: 'init', data: { egress: ['*'] } }); }
          catch { emitThrew = true; }
          const egressAfter = await fetchOutcome();

          // the legit SDK channel must still round-trip through the captured realSend
          const legit = await ctx.db.query('SELECT 1 AS ok');

          ctx.log.info('attacks', { results, egressBefore, egressAfter, emitThrew, legit: legit[0] && legit[0].ok });
        },
      };`,
    );

    await sup.activate('attacker', new Set(['db:own']), {});
    // Sealing the IPC surface must NOT prevent a benign plugin from loading.
    expect(sup.isActive('attacker')).toBe(true);

    const a = logMeta<{
      results: Record<string, string>;
      egressBefore: string;
      egressAfter: string;
      emitThrew: boolean;
      legit: number;
    }>(events, 'attacks');
    expect(a).toBeTruthy();

    // Every raw-IPC operation is blocked.
    expect(a.results.send_evt).toBe('blocked'); // can't forge lifecycle/evt envelopes
    expect(a.results.on_message).toBe('blocked'); // can't add a 'message' listener to eavesdrop
    expect(a.results.once_message).toBe('blocked');
    expect(a.results.prepend_message).toBe('blocked');
    expect(a.results.removeAll_message).toBe('blocked'); // can't unhook the trusted handler
    expect(a.results.removeAll_bare).toBe('blocked');
    expect(a.results.forged_req).toBe('blocked'); // can't forge a req with someone else's _inv
    expect(a.results.disconnect).toBe('blocked');

    // The forged `init` (via the un-sealed process.emit) is a no-op: egress stays sealed.
    expect(a.egressBefore).toBe('egress');
    expect(a.egressAfter).toBe('egress');
    expect(a.emitThrew).toBe(false);

    // The legitimate transport still works — the seal (incl. the channel lock)
    // did not break realSend.
    expect(a.legit).toBe(1);

    // The host never registered the forged route table (the real plugin declares none).
    expect(sup.routesOf('attacker').some((r) => r.path === '/pwned')).toBe(false);
  });

  it('injects require(trek-plugin-sdk) — a scaffold-style plugin loads with no node_modules', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);

    writePlugin(
      'scaffolded',
      `const { definePlugin, PLUGIN_API_VERSION } = require('trek-plugin-sdk');
      let testingError = '';
      try { require('trek-plugin-sdk/testing'); } catch (e) { testingError = e.message; }
      module.exports = definePlugin({
        async onLoad(ctx) {
          ctx.log.info('diag', { api: PLUGIN_API_VERSION, fn: typeof definePlugin, testingError });
        },
        routes: [{ method: 'GET', path: '/hello', auth: true, async handler() { return { status: 200 }; } }],
      });`,
    );

    await sup.activate('scaffolded', new Set(['db:own']), {});
    expect(sup.isActive('scaffolded')).toBe(true);

    const diag = logMeta<{ api: number; fn: string; testingError: string }>(events, 'diag');
    expect(diag).toBeTruthy();
    expect(diag.api).toBe(1); // the injected shim, not a vendored copy
    expect(diag.fn).toBe('function');
    expect(diag.testingError).toMatch(/build\/test-time/); // subpaths fail with a pointed message
  });

  it('a plugin that throws in onLoad fails activation and is marked error — the host survives', async () => {
    const events: Array<{ topic: string; data: unknown }> = [];
    sup = makeSupervisor(events);
    writePlugin('boom', `module.exports = { async onLoad() { throw new Error('kaboom'); } };`);

    await expect(sup.activate('boom', new Set(['db:own']), {})).rejects.toThrow(/kaboom/);
    expect(sup.isActive('boom')).toBe(false);
    expect(sup.statusOf('boom')).toBe('error');
  });

  it('disable() stops a running plugin', async () => {
    sup = makeSupervisor([]);
    writePlugin('stopme', `module.exports = { async onLoad() {} };`);
    await sup.activate('stopme', new Set(), {});
    expect(sup.isActive('stopme')).toBe(true);
    await sup.disable('stopme');
    expect(sup.isActive('stopme')).toBe(false);
  });

  it('auto-disables a plugin that keeps crashing (backoff + crash limit)', async () => {
    sup = makeSupervisor([], { crashLimit: 3, backoffCapMs: 5, crashWindowMs: 60_000 });
    // Exits during onLoad every time -> never reaches "loaded" -> crash loop.
    writePlugin('crasher', `module.exports = { async onLoad() { process.exit(1); } };`);
    await expect(sup.activate('crasher', new Set(), {})).rejects.toThrow(/crashed repeatedly/);
    expect(sup.statusOf('crasher')).toBe('error');
  });

  it('reaps a plugin that stops sending heartbeats', async () => {
    // heartbeatTimeoutMs -1 => any active plugin is "stale"; crashLimit 1 => reap = terminal error.
    sup = makeSupervisor([], { heartbeatTimeoutMs: -1, crashLimit: 1, crashWindowMs: 60_000 });
    writePlugin('idle', `module.exports = { async onLoad() {} };`);
    await sup.activate('idle', new Set(), {});
    expect(sup.isActive('idle')).toBe(true);

    sup.reapStale(); // manual tick — kills the stale child
    await new Promise((r) => setTimeout(r, 400)); // let SIGKILL -> exit -> onCrash settle
    expect(sup.statusOf('idle')).toBe('error');
  });

  it('force-kills a plugin that ignores shutdown (kill grace)', async () => {
    sup = makeSupervisor([], { killGraceMs: 60 });
    // onUnload blocks forever -> shutdown never completes -> SIGKILL after the grace.
    writePlugin('sticky', `module.exports = { async onLoad() {}, async onUnload() { while (true) {} } };`);
    await sup.activate('sticky', new Set(), {});
    await sup.disable('sticky'); // resolves via the kill-grace SIGKILL path
    expect(sup.isActive('sticky')).toBe(false);
  });
});