A failed reply queued with its landing unknown can reach the primary twice, or not at all, when its agent is released #623

// Probe: does the primary hear a released
// run's failed reply twice, or not at all, when a queued entry whose landing
// is unknown (#607) did land in the run's window? Each case prints how many
// by-name notices of worker's failed reply the primary's window holds.
import { describe, it, afterEach } from 'node:test';
import { mkdtempSync, rmSync, readFileSync, existsSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { AgentFramework } from '../src/index.js';
import type { Module, ModuleContext, ToolDefinition, ToolCall, ToolResult, EventResponse } from '../src/index.js';
import type { ContentBlock } from '@animalabs/membrane';
import { MockMembrane } from './helpers/mock-membrane.js';

class NoopModule implements Module {
  readonly name = 'sender';
  async start(_ctx: ModuleContext): Promise<void> {}
  async stop(): Promise<void> {}
  getTools(): ToolDefinition[] { return []; }
  async handleToolCall(_call: ToolCall): Promise<ToolResult> { return { success: true, data: {} }; }
  async onProcess(): Promise<EventResponse> { return {}; }
}

type Internals = {
  initializeMcpl(servers: []): Promise<void>;
  deferredMessages: Array<Record<string, unknown>>;
  quiesced: boolean;
  deferredWritesPersisted: boolean;
  releasesAfterResume: Map<string, unknown>;
  persistDeferredWrites(): void;
  resume?: () => Promise<unknown>;
};
const internals = (f: AgentFramework) => f as unknown as Internals;

const FAILED = { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 };
const NOTICE: ContentBlock[] = [{ type: 'text', text: '[discord-send-failed] Your previous reply (5 chars) could not be delivered to missing-channel (no registered channel).' }];
const byNameNotices = (f: AgentFramework): number =>
  f.getAgent('primary')!.getContextManager().getAllMessages()
    .filter((m) => JSON.stringify(m.content).includes('A reply by worker')).length;
const queuedForPrimary = (f: AgentFramework): number =>
  internals(f).deferredMessages.filter((m) => m.forAgent === undefined && JSON.stringify(m.content).includes('A reply by worker')).length;

const dirs: string[] = [];
afterEach(() => { for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }); });

async function open(storePath: string): Promise<AgentFramework> {
  const framework = await AgentFramework.create({
    storePath,
    membrane: new MockMembrane().asMembrane(),
    agents: [{ name: 'primary', model: 'test-model', systemPrompt: 'Primary.', proseRouting: 'locus' }],
    modules: [new NoopModule()],
  });
  await internals(framework).initializeMcpl([]);
  return framework;
}

/** A run 'worker' whose failed-reply notice E is queued with its landing
 *  unknown; `landed` stores E's copy in worker's window first, as a write
 *  that landed before it threw. */
async function setUp(framework: AgentFramework, landed: boolean) {
  const created = await framework.createEphemeralAgent({
    name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'explicit',
  });
  created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]);
  if (landed) created.contextManager.addMessage('user', NOTICE, { ...FAILED, deferredWriteId: 'w-e' } as never);
  internals(framework).deferredMessages.push({
    id: 'w-e', seq: 1_000_000, participant: 'user', forAgent: 'worker', content: NOTICE, metadata: FAILED,
    heldAt: Date.now(), landing: 'unknown',
  });
  return created;
}

async function endRun(framework: AgentFramework, created: Awaited<ReturnType<typeof setUp>>) {
  await framework.runEphemeralToCompletion(created.agent, created.contextManager, { startupTimeoutMs: 150 }).catch(() => {});
}

const quietly = async <T>(fn: () => Promise<T>): Promise<T> => {
  const error = console.error; const warn = console.warn;
  console.error = () => {}; console.warn = () => {};
  try { return await fn(); } finally { console.error = error; console.warn = warn; }
};

describe('probe', () => {
  for (const landed of [true, false]) {
    it(`(a) a crash between the release's sync and its queue rewrite, then boot (E ${landed ? 'landed' : 'not landed'})`, async () => {
      const dir = mkdtempSync(join(tmpdir(), 'f374-a-')); dirs.push(dir);
      const storePath = join(dir, 'test.chronicle');
      const framework = await open(storePath);
      let atRelease = -1;
      await quietly(async () => {
        const state = internals(framework);
        state.deferredWritesPersisted = true;
        const created = await setUp(framework, landed);
        state.persistDeferredWrites();
        // The crash point: the release's ack syncs the store, and the
        // queue's rewrite never happens.
        state.persistDeferredWrites = () => {};
        await endRun(framework, created);
        atRelease = byNameNotices(framework) + queuedForPrimary(framework);
        await framework.stop();
      });
      const queueFile = join(storePath, 'recovery', 'deferred-writes.json');
      const fileIds = existsSync(queueFile) ? (JSON.parse(readFileSync(queueFile, 'utf8')).pending as Array<{ id: string }>).map((m) => m.id) : [];
      const reopened = await quietly(() => open(storePath));
      const afterBoot = byNameNotices(reopened) + queuedForPrimary(reopened);
      console.log(`PROBE (a) E ${landed ? 'landed' : 'not landed'}: notices at release ${atRelease}; queue file at crash ${JSON.stringify(fileIds)}; primary's notices after boot ${afterBoot}`);
      await quietly(() => reopened.stop());
    });

    it(`(b) a release postponed by a quiesce, then stop() and boot (E ${landed ? 'landed' : 'not landed'})`, async () => {
      const dir = mkdtempSync(join(tmpdir(), 'f374-b-')); dirs.push(dir);
      const storePath = join(dir, 'test.chronicle');
      const framework = await open(storePath);
      let atStop = '';
      await quietly(async () => {
        const state = internals(framework);
        const created = await setUp(framework, landed);
        // As route-failure-targeting's stop-before-release test: the run is
        // under way when the host quiesces, and its startup watchdog ends it.
        const done = framework.runEphemeralToCompletion(created.agent, created.contextManager, { startupTimeoutMs: 150 });
        state.quiesced = true;
        // quiesce() writes the queue to its record as it sets the flag.
        state.persistDeferredWrites();
        await done.catch(() => {});
        if (!state.releasesAfterResume.has('worker')) throw new Error('the release was not postponed');
        await framework.stop();
        atStop = `queued for the primary ${queuedForPrimary(framework)}, E still queued ${state.deferredMessages.some((m) => m.id === 'w-e')}`;
      });
      const reopened = await quietly(() => open(storePath));
      const state2 = internals(reopened);
      if (state2.quiesced) await quietly(async () => { await (reopened as unknown as { resume(): Promise<unknown> }).resume(); });
      const afterBoot = byNameNotices(reopened) + queuedForPrimary(reopened);
      console.log(`PROBE (b) E ${landed ? 'landed' : 'not landed'}: at stop ${atStop}; primary's notices after boot${state2.quiesced ? ' and resume' : ''} ${afterBoot}`);
      await quietly(() => reopened.stop());
    });
  }

  for (const scanThrows of [true, false]) {
    it(`(c) a release whose unread scan ${scanThrows ? 'throws' : 'reads'} while the id lookup works (E landed)`, async () => {
      const dir = mkdtempSync(join(tmpdir(), 'f374-c-')); dirs.push(dir);
      const framework = await open(join(dir, 'test.chronicle'));
      await quietly(async () => {
        const created = await setUp(framework, true);
        if (scanThrows) {
          const cm = created.contextManager as unknown as { getAllMessages: () => unknown };
          cm.getAllMessages = () => { throw new Error('window unreadable'); };
        }
        await endRun(framework, created);
      });
      console.log(`PROBE (c) scan ${scanThrows ? 'throws' : 'reads'}: primary's notices ${byNameNotices(framework) + queuedForPrimary(framework)}`);
      await quietly(() => framework.stop());
    });
  }

  it('(e) a release after a settle found E landed (un-acked), then a crash before any ack, then boot', async () => {
    const dir = mkdtempSync(join(tmpdir(), 'f374-e-')); dirs.push(dir);
    const storePath = join(dir, 'test.chronicle');
    const framework = await open(storePath);
    let state = '';
    await quietly(async () => {
      const s = internals(framework) as Internals & { settleUnknownLandings(which: (m: { id: string }) => boolean): void; unackedDeferredWrites: Array<{ id: string }> };
      s.deferredWritesPersisted = true;
      const created = await setUp(framework, true);
      // What settleFound does for a marked write found landed: out of the
      // pending queue, into the un-acked set until the next ack. (Here by
      // hand: the run isn't registered until it runs, so the settle itself
      // skips it.)
      const e = s.deferredMessages.find((m) => m.id === 'w-e')!;
      s.deferredMessages.splice(s.deferredMessages.indexOf(e), 1);
      s.unackedDeferredWrites.push(e as unknown as { id: string });
      s.persistDeferredWrites();
      state = `after the settle: pending ${JSON.stringify(s.deferredMessages.map((m) => m.id))}, un-acked ${JSON.stringify(s.unackedDeferredWrites.map((m) => m.id))}`;
      // The crash point: nothing rewrites the queue after this.
      s.persistDeferredWrites = () => {};
      await endRun(framework, created);
      state += `; notices at release ${byNameNotices(framework) + queuedForPrimary(framework)}`;
      await framework.stop();
    });
    const reopened = await quietly(() => open(storePath));
    console.log(`PROBE (e) ${state}; primary's notices after boot ${byNameNotices(reopened) + queuedForPrimary(reopened)}`);
    await quietly(() => reopened.stop());
  });

  for (const read of [false, true]) {
    it(`(d) a release whose id lookup throws while its unread scan reads (E landed, ${read ? 'read' : 'unread'})`, async () => {
      const dir = mkdtempSync(join(tmpdir(), 'f374-d-')); dirs.push(dir);
      const framework = await open(join(dir, 'test.chronicle'));
      let copies = -1;
      await quietly(async () => {
        const created = await setUp(framework, true);
        if (read) created.agent.markContextConsumed();
        const cm = created.contextManager as unknown as { queryMessages: (...args: unknown[]) => { messages: unknown[] } };
        const query = cm.queryMessages.bind(cm);
        cm.queryMessages = () => { throw new Error('search failed'); };
        await endRun(framework, created);
        cm.queryMessages = query;
        copies = (query({ metadata: { deferredWriteId: 'w-e' } }).messages).length;
      });
      console.log(`PROBE (d) lookup throws, E ${read ? 'read' : 'unread'}: primary's notices ${byNameNotices(framework) + queuedForPrimary(framework)}; copies of E in worker's window ${copies}`);
      await quietly(() => framework.stop());
    });
  }
});