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); });