nexu-io--open-design/apps/daemon/tests/plugins-atom-registry.test.ts

// Stage D of plugin-driven-flow-plan — atom worker registry tests.
//
// Covers:
//   - Registration / lookup / clearing surface.
//   - `runStageWithRegistry` aggregates signals across multiple atoms
//     using pessimistic merge (lowest number / false-wins boolean).
//   - Permissive defaults preserve happy-path convergence when no
//     atom registers a real worker.
//   - Worker errors are captured as notes instead of crashing the
//     stage so a bad atom never blocks the run.
//   - The built-in `critique-theater` worker reads
//     `run_devloop_iterations` and surfaces the lowest numeric score
//     it can find — and falls through silently when no score is
//     present yet (e.g. the first iteration before the agent has
//     written a critique).
//   - Built-in registration is idempotent and the test-only reset
//     hook restores the install flag for the next case.
//
// The registry is module-scoped so each `beforeEach` calls
// `clearAtomWorkers()` + `resetBuiltInAtomWorkersForTests()` to
// guarantee independence between cases.

import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { mkdtemp, rm } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import Database from 'better-sqlite3';
import type { AppliedPluginSnapshot, PipelineStage } from '@open-design/contracts';
import { migratePlugins } from '../src/plugins/persistence.js';
import {
  PERMISSIVE_DEFAULT_SIGNALS,
  clearAtomWorkers,
  getAtomWorker,
  listRegisteredAtomIds,
  registerAtomWorker,
  runStageWithRegistry,
  type AtomWorkerContext,
} from '../src/plugins/atoms/registry.js';
import {
  registerBuiltInAtomWorkers,
  resetBuiltInAtomWorkersForTests,
} from '../src/plugins/atoms/built-ins.js';

let db: Database.Database;
let tmpDir: string;

beforeEach(async () => {
  tmpDir = await mkdtemp(path.join(os.tmpdir(), 'od-atom-reg-'));
  db = new Database(path.join(tmpDir, 'test.sqlite'));
  db.exec(`
    CREATE TABLE projects (id TEXT PRIMARY KEY, name TEXT);
    CREATE TABLE conversations (id TEXT PRIMARY KEY, project_id TEXT, title TEXT);
  `);
  migratePlugins(db);
  clearAtomWorkers();
  resetBuiltInAtomWorkersForTests();
});

afterEach(async () => {
  db.close();
  await rm(tmpDir, { recursive: true, force: true });
});

function fakeSnapshot(): AppliedPluginSnapshot {
  return {
    snapshotId:          'snap-1',
    projectId:           'project-1',
    conversationId:      'conv-A',
    runId:               'run-1',
    pluginId:            'sample-plugin',
    pluginVersion:       '1.0.0',
    pluginTitle:         'Sample',
    pluginDescription:   '',
    manifestSourceDigest: 'digest-1',
    taskKind:            'new-generation',
    inputs:              {},
    resolvedContext:     { items: [] },
    capabilitiesGranted:  ['prompt:inject'],
    capabilitiesRequired: ['prompt:inject'],
    assetsStaged:         [],
    connectorsRequired:   [],
    connectorsResolved:   [],
    mcpServers:           [],
    query:                '',
    createdAt:            Date.now(),
  } as unknown as AppliedPluginSnapshot;
}

function ctxFor(stage: PipelineStage, iteration = 0): AtomWorkerContext {
  return {
    db,
    runId:          'run-1',
    projectId:      'project-1',
    conversationId: 'conv-A',
    stage,
    iteration,
    snapshot:       fakeSnapshot(),
  };
}

describe('atom registry: registration + lookup', () => {
  it('returns undefined for unregistered ids and lists registered ones alphabetically', () => {
    expect(getAtomWorker('nope')).toBeUndefined();
    registerAtomWorker({ id: 'zebra', run: () => ({}) });
    registerAtomWorker({ id: 'alpha', run: () => ({}) });
    expect(listRegisteredAtomIds()).toEqual(['alpha', 'zebra']);
  });

  it('clears the registry on demand', () => {
    registerAtomWorker({ id: 'temp', run: () => ({}) });
    expect(listRegisteredAtomIds()).toContain('temp');
    clearAtomWorkers();
    expect(listRegisteredAtomIds()).toEqual([]);
  });
});

describe('runStageWithRegistry: signal aggregation', () => {
  it('falls through to permissive defaults when no atom has a registered worker', async () => {
    const stage: PipelineStage = { id: 's', atoms: ['unknown-1', 'unknown-2'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals).toEqual(PERMISSIVE_DEFAULT_SIGNALS);
    expect(out.observedAtoms).toEqual([]);
    expect(out.notes).toEqual([]);
    expect(out.critiqueSummary).toBeNull();
  });

  it('pessimistically merges numeric signals (lowest wins) across multiple atoms', async () => {
    registerAtomWorker({
      id:  'judge-low',
      run: () => ({ signals: { 'critique.score': 2 } }),
    });
    registerAtomWorker({
      id:  'judge-high',
      run: () => ({ signals: { 'critique.score': 5 } }),
    });
    const stage: PipelineStage = { id: 's', atoms: ['judge-low', 'judge-high'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(2);
    expect(out.observedAtoms).toEqual(['judge-low', 'judge-high']);
  });

  it('pessimistically merges boolean signals (false wins) across atoms', async () => {
    registerAtomWorker({
      id:  'gate-pass',
      run: () => ({ signals: { 'preview.ok': true } }),
    });
    registerAtomWorker({
      id:  'gate-fail',
      run: () => ({ signals: { 'preview.ok': false } }),
    });
    const stage: PipelineStage = { id: 's', atoms: ['gate-pass', 'gate-fail'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['preview.ok']).toBe(false);
  });

  it('captures worker errors as notes without crashing the stage', async () => {
    registerAtomWorker({
      id:  'boom',
      run: () => {
        throw new Error('boom');
      },
    });
    registerAtomWorker({
      id:  'ok',
      run: () => ({ signals: { 'critique.score': 4 }, note: 'looks fine' }),
    });
    const stage: PipelineStage = { id: 's', atoms: ['boom', 'ok'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(4);
    expect(out.notes.some((n) => n.includes('worker error: boom'))).toBe(true);
    expect(out.notes.some((n) => n.includes('looks fine'))).toBe(true);
    expect(out.critiqueSummary).not.toBeNull();
  });

  it('awaits async workers and applies their signals to the merge', async () => {
    registerAtomWorker({
      id:  'slow-judge',
      run: async () => {
        await new Promise((resolve) => setTimeout(resolve, 5));
        return { signals: { 'critique.score': 1 } };
      },
    });
    const stage: PipelineStage = { id: 's', atoms: ['slow-judge'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(1);
  });
});

describe('built-in critique-theater worker', () => {
  beforeEach(() => {
    registerBuiltInAtomWorkers();
  });

  it('returns permissive (no score) when no devloop row exists yet', async () => {
    const stage: PipelineStage = { id: 'critique', atoms: ['critique-theater'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals).toEqual(PERMISSIVE_DEFAULT_SIGNALS);
    expect(out.observedAtoms).toEqual(['critique-theater']);
    expect(out.critiqueSummary).toBeNull();
  });

  it('parses a numeric score=N from critique_summary and surfaces the latest iteration that has one', async () => {
    insertIteration('critique', 1, 'critique panel: score=2');
    insertIteration('critique', 2, 'no parseable score here');
    insertIteration('critique', 3, 'score=4');
    const stage: PipelineStage = { id: 'critique', atoms: ['critique-theater'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(4);
    expect(out.critiqueSummary).toContain('latest critique score=4 from iteration 3');
  });

  it('falls back to the most recent parseable iteration when the newest row has no score', async () => {
    insertIteration('critique', 1, 'score=3');
    insertIteration('critique', 2, 'unrelated notes only');
    const stage: PipelineStage = { id: 'critique', atoms: ['critique-theater'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(3);
  });

  it('only looks at iterations for the active stage, ignoring siblings', async () => {
    insertIteration('discovery', 1, 'score=1');
    insertIteration('critique', 1, 'score=4');
    const stage: PipelineStage = { id: 'critique', atoms: ['critique-theater'] };
    const out = await runStageWithRegistry(ctxFor(stage));
    expect(out.signals['critique.score']).toBe(4);
  });
});

describe('registerBuiltInAtomWorkers: idempotency', () => {
  it('registers every FIRST_PARTY_ATOM exactly once even on repeat calls', () => {
    registerBuiltInAtomWorkers();
    const first = listRegisteredAtomIds();
    registerBuiltInAtomWorkers();
    const second = listRegisteredAtomIds();
    expect(second).toEqual(first);
    expect(first).toContain('critique-theater');
    expect(first).toContain('file-write');
    expect(first).toContain('media-image');
  });
});

function insertIteration(stageId: string, iteration: number, summary: string): void {
  db.prepare(
    `INSERT INTO run_devloop_iterations
       (id, run_id, stage_id, iteration, artifact_diff_summary, critique_summary, tokens_used, ended_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
  ).run(`iter-${stageId}-${iteration}`, 'run-1', stageId, iteration, null, summary, null, Date.now());
}