stack/packages/bus/tests/spec-socket.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { connect } from 'node:net';
import { join } from 'node:path';
import { fixture } from './spec-fixtures.mjs';
import { serve } from '../src/server.mjs';
import { Client } from '../src/client.mjs';
async function socketFixture(t, options = {}) {
  const f = await fixture(t), path = join(f.root, 'bus.sock');
  const server = await serve({ broker: f.broker, path, ...options });
  t.after(() => server.close());
  return { ...f, path, client: new Client({ path, cap: f.coder }) };
}

test('real socket awaits spec reads, record writes and cited raises; other requests still return normally', async t => {
  const f = await socketFixture(t);
  const spec = await f.client.call('spec.read', { project: 'stack' });
  assert.equal(spec.commit, f.commit); assert.match(spec.content, /North.star/i);
  const result = await f.client.call('record.write', { project: 'stack', path: 'docs/socket.md', record: '# Socket\n' });
  assert.match(result.commit, /^[0-9a-f]{40}$/);
  const d = await f.client.call('decision.raise', { action: 'deploy', question: 'Ship?',
    options: [{ key: 'yes', text: 'Yes' }, { key: 'no', text: 'No' }], recommendation: 'yes', blocking: false,
    project: 'stack', spec_path: spec.path, spec_commit: spec.commit });
  assert.equal(f.store.get('SELECT spec_commit FROM decisions WHERE id=?', d.id).spec_commit, spec.commit);
  assert.ok(Array.isArray(await f.client.call('agents')));
});

test('real socket preserves asynchronous refusal codes and disconnects its transient proof', async t => {
  let transient;
  const f = await fixture(t), path = join(f.root, 'bus.sock');
  const server = await serve({ broker: f.broker, path, authenticateHuman: () => {
    transient = f.broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
    return transient;
  } });
  t.after(() => server.close());
  const client = new Client({ path, cap: f.coder });
  await assert.rejects(client.call('spec.read', { project: 'missing' }), { code: 'project-not-declared' });
  await assert.rejects(new Client({ path, human: { fixture: true } }).call('spec.read', { project: 'stack' }), { code: 'agent-required' });
  assert.throws(() => f.broker.identity(transient), /unauthenticated/);
  assert.ok(Array.isArray(await client.call('agents')));
});

test('two real socket writes to one repository yield one commit and one project-busy refusal', async t => {
  const f = await socketFixture(t);
  const results = await Promise.allSettled(['one', 'two'].map(name =>
    new Client({ path: f.path, cap: f.coder }).call('record.write', {
      project: 'stack', path: `docs/${name}.md`, record: `# ${name}\n`,
    })));
  assert.equal(results.filter(r => r.status === 'fulfilled').length, 1);
  const refused = results.find(r => r.status === 'rejected');
  assert.equal(refused.reason.code, 'project-busy');
  assert.equal(f.store.all("SELECT body FROM events WHERE json_extract(body,'$.action')='record.write'").length, 1);
});

for (const rejects of [false, true]) test(`client disconnect during await tolerates late ${rejects ? 'refusal' : 'reply'}`, async t => {
  const f = await fixture(t), path = join(f.root, 'bus.sock');
  let release, admitted, finished;
  const holding = new Promise(resolve => { release = resolve; });
  const started = new Promise(resolve => { admitted = resolve; });
  const completed = new Promise(resolve => { finished = resolve; });
  const broker = { request(cap, request) {
    if (request.verb !== 'spec.read') return f.broker.request(cap, request);
    admitted();
    return holding.then(() => f.broker.request(cap, { verb: 'spec.read',
      args: { project: rejects ? 'missing' : 'stack' } })).finally(finished);
  } };
  const server = await serve({ broker, path });
  t.after(() => server.close());
  const socket = connect(path);
  socket.on('error', () => {});
  socket.write(JSON.stringify({ cap: f.coder, verb: 'spec.read', args: { project: 'stack' } }) + '\n');
  await started;
  const closed = new Promise(resolve => socket.once('close', resolve));
  socket.destroy(); await closed;
  release(); await completed;
  // A further socket round-trip proves the server remains usable after the late completion.
  assert.ok(Array.isArray(await new Client({ path, cap: f.coder }).call('agents')));
});


test('N-a: slow admitted record.write returns its committed result after the idle deadline', async t => {
  const f = await fixture(t), path = join(f.root, 'bus.sock');
  const broker = { async request(cap, request) {
    if (request.verb === 'record.write') await new Promise(resolve => setTimeout(resolve, 150));
    return f.broker.request(cap, request);
  } };
  const server = await serve({ broker, path, timeout: 40 });
  t.after(() => server.close());
  const written = await new Client({ path, cap: f.coder }).call('record.write', {
    project: 'stack', path: 'docs/slow.md', record: '# Slow write\n',
  });
  assert.match(written.commit, /^[0-9a-f]{40}$/);
  const receipts = f.store.all("SELECT body FROM events WHERE kind='action.allowed' AND json_extract(body,'$.action')='record.write'");
  assert.equal(receipts.length, 1);
  assert.equal(JSON.parse(receipts[0].body).commit, written.commit);
});