stack/packages/discord/tests/processing.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { spawnSync } from 'node:child_process';
import { mkdirSync, unlinkSync, readFileSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { createConnector } from '../src/connector.mjs';
import { initializeControl, assertProcessing, transition } from '../../processing-control/src/state.mjs';
import { makeRoot, makeRepo, makeDeployment, binding, message, fakeRest, fakeGateway, fakeEngine } from './helpers.mjs';
import { appendOutbox, ensureJournal, readOutbox, listTurns } from '../src/journal.mjs';
const actor = { business: 'demo', human: 'jason', halted: true };
function fixture() {
  const root = makeRoot(); initializeControl(root);
  const journalDir = join(root, 'journal'), rest = fakeRest(), gateway = fakeGateway(), engine = fakeEngine({ replies: [{ text: 'reply' }] });
  const processing = () => assertProcessing(root);
  const options = { binding: binding(), journalDir, rest, gateway, engine, processing };
  return { root, journalDir, rest, gateway, engine, options };
}
test('halted connector refuses before outbox reconciliation, engine or gateway start', async () => {
  const f = fixture(); ensureJournal(f.journalDir);
  appendOutbox(f.journalDir, { nonce: 'pending', channelId: '100000000000000200', status: 'intent', content: 'do not replay', at: new Date().toISOString() });
  transition(f.root, actor);
  await assert.rejects(createConnector(f.options).start(), e => e.code === 'processing-halted');
  assert.equal(f.rest.calls.length, 0); assert.equal(f.engine.started, false);
  const corrupt = fixture(); ensureJournal(corrupt.journalDir);
  const connector = createConnector(corrupt.options);
  transition(corrupt.root, actor);
  writeFileSync(join(corrupt.journalDir, 'outbox.jsonl'), '{broken\n', { mode: 0o600 });
  await assert.rejects(connector.start(), e => e.code === 'processing-halted');
  assert.equal(corrupt.rest.calls.length, 0); assert.equal(corrupt.engine.started, false);
});
test('halt arriving during startup reconciliation prevents engine and gateway start', async () => {
  const f = fixture(); ensureJournal(f.journalDir);
  appendOutbox(f.journalDir, { nonce: 'pending', channelId: '100000000000000200', status: 'intent', content: 'first', at: new Date().toISOString() });
  const create = f.rest.createMessage;
  f.rest.createMessage = async (...args) => { const result = await create(...args); transition(f.root, actor); return result; };
  await assert.rejects(createConnector(f.options).start(), e => e.code === 'processing-halted');
  assert.equal(f.engine.started, false);
});
test('halted connector refuses a new inbound turn before claiming inbox or invoking the engine', async () => {
  const f = fixture(), connector = createConnector(f.options); await connector.start();
  transition(f.root, actor);
  const answer = await connector.handleMessage(message());
  assert.equal(answer.accepted, false); assert.equal(answer.reason, 'processing-halted');
  assert.equal(connector.inboxSize, 0); assert.equal(f.engine.prompts.length, 0);
  await connector.stop();
});
test('halt during a model turn never sends its answer or a fallback failure message', async () => {
  const f = fixture();
  f.engine.prompt = async () => { transition(f.root, actor); return { text: 'must not send', usage: { input: 1, output: 1 } }; };
  const connector = createConnector(f.options); await connector.start();
  const turn = await connector.handleMessage(message()); await turn.turn;
  assert.equal(f.rest.calls.length, 0);
  assert.equal(listTurns(f.journalDir)[0].error.code, 'processing-halted');
  await connector.stop();
});
test('Discord run and recover refuse a halted root before reading a token or reaching network', () => {
  const root = makeRoot(), repo = makeRepo(root), d = makeDeployment(root);
  mkdirSync(d.dataRoot, { recursive: true, mode: 0o700 }); initializeControl(d.dataRoot);
  transition(d.dataRoot, actor); unlinkSync(d.tokenFile);
  for (const command of ['run', 'recover']) {
    const r = spawnSync(process.execPath, [new URL('../src/cli.mjs', import.meta.url).pathname, command, d.raw.name, '--config', d.config, '--repo', repo], { encoding: 'utf8', timeout: 5000 });
    assert.equal(r.status, 3, r.stderr); assert.match(r.stderr, /processing-halted/);
  }
});

test('Discord engine publishes a verified group and a root halt kills it independently of the connector', async t => {
  const { createEngine } = await import('../src/engine-pi.mjs');
  const { listRuntimes, sweepRuntimes, processIdentity } = await import('../../processing-control/src/runtimes.mjs');
  const f = fixture();
  const engine = createEngine({ command: process.execPath,
    args: [new URL('./fake-pi.mjs', import.meta.url).pathname], cwd: f.root,
    env: { ...process.env }, processingRoot: f.root, binding: 'test-seat' });
  t.after(() => engine.stop({ graceMs: 50 }));
  const child = engine.start();
  const tracked = listRuntimes(f.root);
  assert.equal(tracked.length, 1); assert.equal(tracked[0].pid, child.pid);
  assert.equal(tracked[0].kind, 'discord'); assert.equal(tracked[0].group, true);
  const halted = transition(f.root, actor);
  await assert.rejects(engine.prompt('not admitted'), e => e.code === 'processing-halted');
  const result = await sweepRuntimes(f.root, halted.generation, { graceMs: 80, killWaitMs: 1000 });
  assert.equal(result.quiescent, true); assert.equal(processIdentity(child.pid), null);
});