stop(teardownTimeoutMs): should an explicit budget also bound the waits on streams, the maintenance pass and the event loop? #373

// Does an explicit teardownTimeoutMs bound stop()'s waits on a stream and on the
// event loop? Run from a built agent-framework checkout (dist/).
//  C. An agent's stream never yields again, and its cancel can't reach it.
//  D. The event loop is stuck in a module's onProcess.
// stop(1000) is called in each; its state is printed 3 s later. Both are left
// pending, and the process exits at the end.
import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';

const root = process.cwd();
const { AgentFramework } = await import(join(root, 'dist/src/index.js'));
const { MockMembrane, MockYieldingStream, createMockResponse } = await import(join(root, 'dist/test/helpers/mock-membrane.js'));

class NeverStream extends MockYieldingStream {
  async *[Symbol.asyncIterator]() { await new Promise(() => {}); }
}

class EventsModule {
  name = 'events';
  constructor() { this.hanging = new Promise((r) => { this.markHanging = r; }); }
  async start() {}
  async stop() {}
  getTools() { return []; }
  async handleToolCall() { return { success: true }; }
  async onProcess(event) {
    if (event.type === 'external-message' && String(event.content) === 'hang') {
      this.markHanging();
      await new Promise(() => {});
    }
    if (event.type === 'external-message' && String(event.content) === 'go') {
      return { addMessages: [{ participant: 'Antra', content: [{ type: 'text', text: 'go' }] }], requestInference: true };
    }
    return {};
  }
}

const watch = (promise) => {
  const state = { now: 'pending' };
  promise.then(() => { state.now = 'resolved'; }, (e) => { state.now = `rejected: ${String(e?.message ?? e).slice(0, 160)}`; });
  return state;
};

async function caseC() {
  const dir = mkdtempSync(join(tmpdir(), 'probe-budget-c-'));
  const membrane = new MockMembrane();
  membrane.streamYielding = (request) => {
    membrane.calls.push(request);
    return new NeverStream([createMockResponse([{ type: 'text', text: 'never' }])]);
  };
  const framework = await AgentFramework.create({
    storePath: join(dir, 'store'), membrane: membrane.asMembrane(),
    agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], modules: [new EventsModule()],
  });
  framework.start();
  framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} });
  const streams = framework.activeStreams;
  for (let i = 0; i < 400 && !streams.has('agent'); i++) await new Promise((r) => setTimeout(r, 5));
  console.log(`C. stream driven: ${streams.has('agent')}`);
  const t0 = Date.now();
  const stop = watch(framework.stop(1000));
  await new Promise((r) => setTimeout(r, 3000));
  console.log(`C. a stream that never settles: stop(1000) is ${stop.now} at ${Date.now() - t0} ms`);
}

async function caseD() {
  const dir = mkdtempSync(join(tmpdir(), 'probe-budget-d-'));
  const events = new EventsModule();
  const framework = await AgentFramework.create({
    storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(),
    agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], modules: [events],
  });
  framework.start();
  framework.pushEvent({ type: 'external-message', source: 'test', content: 'hang', metadata: {} });
  await events.hanging;
  const t0 = Date.now();
  const stop = watch(framework.stop(1000));
  await new Promise((r) => setTimeout(r, 3000));
  console.log(`D. an onProcess that never returns: stop(1000) is ${stop.now} at ${Date.now() - t0} ms`);
}

await caseC();
await caseD();
process.exit(0);