/**
 * rc.9 PR-D integration test for the SwmAckQuorum wiring inside
 * DKGAgent. The component itself is covered by
 * `swm-ack-quorum.test.ts`; this file pins the AGENT-side wiring:
 *
 *   1. `share()` registers tracking when preconditions hold
 *      (shareOperationId set, gossip leg active, substrateMembers
 *      non-empty).
 *   2. PROTOCOL_SWM_SHARE_ACK arrivals route into
 *      `swmAckQuorum.onAck()` and count toward quorum.
 *   3. Pre-acked substrate peers (PR-C delivered set) feed
 *      `preAckedFromSubstrate` so a hybrid share that already met
 *      quorum via substrate doesn't fight a watchdog race.
 *   4. The gossip-applied path of `SharedMemoryHandler.handle()`
 *      emits a SwmShareAck back to the publisher.
 *   5. `getSwmAckQuorumStats()` returns pristine zeroes before any
 *      share, and reflects the right counters after.
 */

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { MockChainAdapter } from '@origintrail-official/dkg-chain';
import {
  PROTOCOL_SWM_UPDATE,
  PROTOCOL_SWM_SHARE_ACK,
  encodeSwmShareAck,
} from '@origintrail-official/dkg-core';
import { DKGAgent } from '../src/index.js';
import type { ReliableSendResult } from '../src/p2p/messenger.js';

const SELF_PEER = '12D3KooWSelfPubD';

class CapturingGossip {
  publishes: Array<{ topic: string; bytes: number }> = [];
  subscribers: string[] = [];

  subscribe(_topic: string): void {}
  unsubscribe(_topic: string): void {}
  onMessage(_topic: string, _handler: (topic: string, data: Uint8Array, from?: string) => void | Promise<void>): void {}
  async publish(topic: string, data: Uint8Array): Promise<void> {
    this.publishes.push({ topic, bytes: data.byteLength });
  }
  getSubscribers(_topic: string): string[] {
    return [...this.subscribers];
  }
}

interface SendReliableCall {
  /**
   * Which messenger method was invoked. rc.9 PR-D codex
   * follow-up #D8: pre-D8 we only tracked `messageId` and
   * relied on it being `undefined` to infer "this was a
   * `sendToPeer` call, not `sendReliable`" — but the
   * production `maybeEmitSwmShareAck` doesn't pass
   * `opts.messageId` to `sendReliable` either, so a
   * regression that re-routed acks back through `sendReliable`
   * would have passed the test. Tracking the method name
   * explicitly fixes the assertion's signal-to-noise.
   */
  method: 'sendReliable' | 'sendToPeer';
  peerId: string;
  protocolId: string;
  bytes: number;
  messageId?: string;
}

/**
 * Stub `messenger.sendReliable` with two extras over the PR-C
 * integration helper:
 *   - Capture `messageId` (we assert top-up uses the documented
 *     `swm-topup-` prefix).
 *   - Default to a successful delivery if the lookup returns
 *     undefined, so ack-quorum top-up calls don't blow up just
 *     because the test only configured the initial fan-out
 *     responses.
 */
function stubMessengerSendReliable(
  results: Map<string, ReliableSendResult> | ((peerId: string, protocolId: string, messageId?: string) => ReliableSendResult | Error),
): { calls: SendReliableCall[]; install: (agent: DKGAgent) => void } {
  const calls: SendReliableCall[] = [];
  const lookup = typeof results === 'function'
    ? results
    : (peerId: string): ReliableSendResult | Error => {
      const r = results.get(peerId);
      if (!r) {
        return {
          delivered: true,
          response: new Uint8Array(),
          attempts: 1,
          messageId: 'stub-default-ok',
        };
      }
      return r;
    };
  const install = (agent: DKGAgent): void => {
    const stub = {
      sendReliable: async (
        peerId: string,
        protocolId: string,
        payload: Uint8Array,
        opts?: { messageId?: string },
      ): Promise<ReliableSendResult> => {
        calls.push({ method: 'sendReliable', peerId, protocolId, bytes: payload.byteLength, messageId: opts?.messageId });
        const out = lookup(peerId, protocolId, opts?.messageId);
        if (out instanceof Error) throw out;
        return out;
      },
      // rc.9 PR-D codex follow-up #D1: ack emission uses
      // fire-and-forget `sendToPeer` instead of durable
      // `sendReliable`. We capture into the same `calls` array
      // (with `method: 'sendToPeer'`) so assertions can
      // distinguish the two surfaces — the `protocolId` alone
      // is not enough since the production `sendReliable` and
      // `sendToPeer` paths both have undefined `messageId` when
      // the caller doesn't pass `opts.messageId` (codex #D8).
      sendToPeer: async (
        peerId: string,
        protocolId: string,
        payload: Uint8Array,
      ): Promise<Uint8Array> => {
        calls.push({ method: 'sendToPeer', peerId, protocolId, bytes: payload.byteLength });
        return new Uint8Array();
      },
      // PR-D registers PROTOCOL_SWM_SHARE_ACK and PR-C registers
      // a deliveredResponseClassifier — stub these as no-ops so
      // DKGAgent's constructor / wiring code doesn't crash when
      // it reaches into the stubbed messenger.
      register: (_proto: string, _handler: (data: Uint8Array, from: string) => Promise<Uint8Array>): void => {},
      setResponseDeliveredClassifier: (_proto: string, _fn: (r: Uint8Array) => boolean): void => {},
    };
    (agent as unknown as { messenger: typeof stub }).messenger = stub;
  };
  return { calls, install };
}

async function createAgent(name: string): Promise<DKGAgent> {
  const agent = await DKGAgent.create({
    name: `${name}-${Math.random().toString(36).slice(2)}`,
    chainAdapter: new MockChainAdapter(),
  });
  Object.defineProperty((agent as unknown as { node: object }).node, 'peerId', {
    value: SELF_PEER,
    configurable: true,
  });
  // PR-J: CGMemberEnumerator now filters topic-subscribers
  // through libp2p reachability (connected OR peerStore-known).
  // These tests don't drive a real libp2p — without a stub the
  // filter would strip every fake-peer subscriber and `track()`
  // would never fire. Treat ALL test peers as dialable; the
  // PR-J test that actually exercises the filter lives in
  // `enumerate-cg-members.test.ts` where it can drive the
  // predicate directly.
  installAllReachableLibp2pStub(agent);
  return agent;
}

/**
 * Install a `node.libp2p` shim that reports test peers as
 * dialable. Matches the pre-PR-J behaviour the integration tests
 * expect — they care about ack-quorum / fan-out wiring, not
 * reachability filtering. The PR-J filter is exercised in
 * `enumerate-cg-members.test.ts` where the predicate is the unit
 * under test.
 *
 * Why a lazy gossip read: PR-J round 2 (codex RED #4) routes the
 * `isPeerDialable` predicate through `peerIdFromString` before
 * `peerStore.get`. Short test peer IDs like `12D3KooWPeerA` are
 * not valid Ed25519 base58 strings, so `peerIdFromString` would
 * throw → catch → false → substrate target subset empties →
 * the whole fan-out stops. We side-step by making `getPeers()`
 * return whatever gossip.subscribers currently contains: the
 * first branch of `isPeerDialable` (`getPeers().some(...)`)
 * matches and short-circuits before peerStore is touched.
 *
 * For tests whose substrate fan-out targets peer IDs NOT in
 * gossip.subscribers (e.g. watchdog-top-up tests with custom
 * `missingPeers`), they can also push into
 * `_extraDialablePeerIds` after install.
 */
function installAllReachableLibp2pStub(agent: DKGAgent): { extraDialablePeerIds: string[] } {
  const extraDialablePeerIds: string[] = [];
  const allReachableIds = (): string[] => {
    const gossip = (agent as unknown as { gossip?: { subscribers?: string[] } }).gossip;
    const fromGossip = gossip?.subscribers ?? [];
    return [...fromGossip, ...extraDialablePeerIds];
  };
  const stub = {
    getPeers: (): Array<{ toString: () => string }> => {
      return allReachableIds().map((id) => ({ toString: () => id }));
    },
    // PR-K: isPeerDialable now uses getConnections(pid) and checks
    // for non-limited connections. Return a single direct-style
    // connection object (no `limits` property) for any gossip-
    // subscribed peer so the predicate stays "everything reachable"
    // in tests that don't care about the PR-K filter itself (the
    // PR-K unit tests pass the predicate directly in
    // `enumerate-cg-members.test.ts`).
    getConnections: (pid?: unknown): Array<{ remoteAddr: { toString: () => string } }> => {
      const idStr = pid ? (pid as { toString: () => string }).toString() : '';
      return allReachableIds().includes(idStr)
        ? [{ remoteAddr: { toString: () => '/ip4/127.0.0.1/tcp/0' } }]
        : [];
    },
    peerStore: {
      get: async (_peerId: unknown) => ({
        addresses: [{ multiaddr: { toString: () => '/ip4/127.0.0.1/tcp/0' } }],
      }),
    },
  };
  Object.defineProperty((agent as unknown as { node: { libp2p?: unknown } }).node, 'libp2p', {
    value: stub,
    configurable: true,
    writable: true,
  });
  return { extraDialablePeerIds };
}

describe('DKGAgent SwmAckQuorum integration (rc.9 PR-D)', () => {
  let createdAgents: DKGAgent[] = [];

  beforeAll(() => {
    createdAgents = [];
  });

  afterAll(async () => {
    // Stop each created agent so the 5s ack-quorum tick + any
    // other intervals it owns are cleared. The timers use
    // `.unref()` so they wouldn't block process exit, but
    // stopping is the documented contract.
    await Promise.allSettled(createdAgents.map(async (a) => {
      const maybeStop = a as unknown as { stop?: () => Promise<void> };
      if (typeof maybeStop.stop === 'function') {
        try { await maybeStop.stop(); } catch { /* test-shape stub agent */ }
      }
    }));
  });

  function register(agent: DKGAgent): DKGAgent {
    createdAgents.push(agent);
    return agent;
  }

  it('cold-start: getSwmAckQuorumStats returns pristine zeroes', async () => {
    const agent = register(await createAgent('AckQuorumCold'));
    expect(agent.getSwmAckQuorumStats()).toEqual({
      tracked: 0,
      completed: 0,
      watchdogFired: 0,
      deadlineExpired: 0,
      pending: 0,
    });
  });

  it('share() registers tracking after fan-out, substrate-delivered peers pre-ack', async () => {
    const agent = register(await createAgent('AckQuorumTrack'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPeerA', '12D3KooWPeerB', '12D3KooWPeerC'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(new Map([
      ['12D3KooWPeerA', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'mA' }],
      ['12D3KooWPeerB', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'mB' }],
      ['12D3KooWPeerC', {
        delivered: false, queued: true, attempts: 1, messageId: 'mC',
        error: 'transient', nextAttemptAtMs: Date.now() + 1000,
      }],
    ]));
    install(agent);

    await agent.share('cg-ackquorum-track', [{
      subject: 'urn:test:track', predicate: 'http://schema.org/name', object: '"hi"', graph: '',
    }]);

    const stats = agent.getSwmAckQuorumStats();
    expect(stats.tracked).toBe(1);
    expect(stats.pending).toBe(1);
    expect(stats.completed).toBe(0);
  });

  it('substrate-only-delivery reaches quorum at track time (no gossip ack needed)', async () => {
    const agent = register(await createAgent('AckQuorumAllSubstrate'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPeerX', '12D3KooWPeerY'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(new Map([
      ['12D3KooWPeerX', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'mX' }],
      ['12D3KooWPeerY', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'mY' }],
    ]));
    install(agent);

    await agent.share('cg-ackquorum-all-substrate', [{
      subject: 'urn:test:all', predicate: 'http://schema.org/name', object: '"all"', graph: '',
    }]);

    const stats = agent.getSwmAckQuorumStats();
    expect(stats.tracked).toBe(1);
    expect(stats.completed).toBe(1);
    expect(stats.pending).toBe(0);
    expect(stats.watchdogFired).toBe(0);
  });

  /**
   * rc.9 PR-D codex follow-up #D7: drive a real
   * `PROTOCOL_SWM_SHARE_ACK` arrival through the production
   * handler and assert that it lands in the quorum's `acked`
   * set, then reaches the `completed` count once the threshold
   * is met. The pre-D7 version of this test only asserted
   * "one share is pending" — a regression in
   * `handleSwmShareAck()` (e.g. failing to call onAck, or
   * dropping ACKs silently) would not have been caught.
   */
  it('PROTOCOL_SWM_SHARE_ACK arrival increments acked count + completes quorum when threshold met', async () => {
    const agent = register(await createAgent('AckQuorumAckArrival'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPeer1', '12D3KooWPeer2', '12D3KooWPeer3'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable((_peerId, _proto, _msg) => ({
      delivered: false, queued: true, attempts: 1, messageId: 'queued-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    const { shareOperationId } = await agent.share('cg-ackquorum-arrivals', [{
      subject: 'urn:test:arr', predicate: 'http://schema.org/name', object: '"arr"', graph: '',
    }]);
    expect(typeof shareOperationId).toBe('string');

    const stats0 = agent.getSwmAckQuorumStats();
    expect(stats0.tracked).toBe(1);
    expect(stats0.completed).toBe(0);
    expect(stats0.pending).toBe(1);

    // Drive the real handler with three real-shape ACK bytes —
    // exactly the same code path that a libp2p arrival would
    // hit. We get the handler via the same test-only getter
    // production registration uses; this test exercises
    // handleSwmShareAck → SwmAckQuorum.onAck without bypassing
    // anything.
    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();
    const inspect = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(inspect?.acked).toEqual([]);
    expect([...(inspect?.expectedMembers ?? [])].sort()).toEqual([
      '12D3KooWPeer1', '12D3KooWPeer2', '12D3KooWPeer3',
    ]);

    await ackHandler(encodeSwmShareAck({ shareOperationId, ackPeerId: '12D3KooWPeer1' }), '12D3KooWPeer1');
    expect(agent.getSwmAckQuorumStats().completed).toBe(0);
    expect(agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId)?.acked).toEqual(['12D3KooWPeer1']);

    await ackHandler(encodeSwmShareAck({ shareOperationId, ackPeerId: '12D3KooWPeer2' }), '12D3KooWPeer2');
    expect(agent.getSwmAckQuorumStats().completed).toBe(0);

    await ackHandler(encodeSwmShareAck({ shareOperationId, ackPeerId: '12D3KooWPeer3' }), '12D3KooWPeer3');
    // 3/3 acked → quorum complete (default quorumPct is 1.0 in
    // the test config; the record is reaped on completion so
    // inspect() returns undefined).
    expect(agent.getSwmAckQuorumStats().completed).toBe(1);
    expect(agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId)).toBeUndefined();
  });

  it('share() returns shareOperationId — ack arrivals for it complete quorum', async () => {
    const agent = register(await createAgent('AckQuorumWithId'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPeerQ1', '12D3KooWPeerQ2', '12D3KooWPeerQ3', '12D3KooWPeerQ4'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(() => ({
      delivered: false, queued: true, attempts: 1, messageId: 'q-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    const { shareOperationId } = await agent.share('cg-ackquorum-byid', [{
      subject: 'urn:test:byid', predicate: 'http://schema.org/name', object: '"byid"', graph: '',
    }]);
    expect(typeof shareOperationId).toBe('string');

    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => {
        onAck: (op: string, peer: string) => void;
        stats: () => { tracked: number; completed: number; watchdogFired: number; deadlineExpired: number; pending: number };
        inspect: (op: string) => { acked: readonly string[]; expectedMembers: readonly string[]; ackPct: number } | undefined;
      };
    }).getOrCreateSwmAckQuorum();

    expect(quorum.inspect(shareOperationId)?.expectedMembers.sort())
      .toEqual(['12D3KooWPeerQ1', '12D3KooWPeerQ2', '12D3KooWPeerQ3', '12D3KooWPeerQ4']);

    quorum.onAck(shareOperationId, '12D3KooWPeerQ1');
    quorum.onAck(shareOperationId, '12D3KooWPeerQ2');
    quorum.onAck(shareOperationId, '12D3KooWPeerQ3');
    expect(quorum.stats().completed).toBe(0);

    quorum.onAck(shareOperationId, '12D3KooWPeerQ4');
    expect(quorum.stats().completed).toBe(1);
    expect(quorum.inspect(shareOperationId)).toBeUndefined();
  });

  it('legacy callers (no shareOperationId on publishWorkspaceGossip) do not register tracking', async () => {
    const agent = register(await createAgent('AckQuorumLegacy'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWLegacyPeer'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(new Map([
      ['12D3KooWLegacyPeer', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'mL' }],
    ]));
    install(agent);

    const internals = agent as unknown as {
      publishWorkspaceGossip: (cgId: string, msg: Uint8Array, ctx: unknown, signer: unknown) => Promise<void>;
    };
    const dummyMessage = new TextEncoder().encode('not-a-real-wire-message');
    await internals.publishWorkspaceGossip('cg-legacy-caller', dummyMessage, { operationId: 'test' }, null);

    const stats = agent.getSwmAckQuorumStats();
    expect(stats.tracked).toBe(0);
    expect(stats.pending).toBe(0);
  });

  it('SwmShareAck wire shape: handler accepts the same bytes encodeSwmShareAck produces', () => {
    const bytes = encodeSwmShareAck({
      shareOperationId: 'op-roundtrip',
      ackPeerId: '12D3KooWAckRoundtrip',
    });
    expect(bytes).toBeInstanceOf(Uint8Array);
    expect(bytes.byteLength).toBeGreaterThan(0);
    expect(PROTOCOL_SWM_SHARE_ACK).toBe('/dkg/10.0.1/swm-share-ack');
    expect(PROTOCOL_SWM_UPDATE).toBe('/dkg/10.0.1/swm-update');
  });

  /**
   * rc.9 PR-D codex follow-up #D1: ack emission MUST use
   * fire-and-forget `sendToPeer`, NOT durable `sendReliable`.
   * Pre-D1 a publisher peer on a build without
   * PROTOCOL_SWM_SHARE_ACK registered would leave a permanently-
   * queued outbox row per received share, retrying protocol
   * negotiation forever. This test pins the call surface: the
   * ack handler exercised here calls `messenger.sendToPeer`,
   * not `messenger.sendReliable`.
   */
  it('PR-D #D1: ack emission uses sendToPeer (fire-and-forget), not sendReliable (durable)', async () => {
    const agent = register(await createAgent('AckQuorumD1FireForget'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPublisher'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { calls, install } = stubMessengerSendReliable(new Map([
      ['12D3KooWPublisher', { delivered: true, response: new Uint8Array(), attempts: 1, messageId: 'm-pub' }],
    ]));
    install(agent);

    // Invoke the ack emit path directly with a synthetic
    // apply-true outcome — bypasses the gossip mesh dance and
    // pins the messenger call shape.
    const emit = (agent as unknown as {
      maybeEmitSwmShareAck: (o: { applied: true; cgId?: string; shareOperationId?: string; publisherPeerId?: string }) => Promise<void>;
    }).maybeEmitSwmShareAck.bind(agent);

    await emit({
      applied: true,
      cgId: 'cg-d1',
      shareOperationId: 'op-d1-fire-forget',
      publisherPeerId: '12D3KooWPublisher',
    });

    const ackCalls = calls.filter((c) => c.protocolId === PROTOCOL_SWM_SHARE_ACK);
    expect(ackCalls).toHaveLength(1);
    expect(ackCalls[0]?.peerId).toBe('12D3KooWPublisher');
    // rc.9 PR-D codex #D8: assert the METHOD explicitly. The
    // pre-D8 assertion checked `messageId === undefined` which
    // would have passed even if the implementation regressed
    // back to `sendReliable(peer, proto, payload, { timeoutMs })`
    // (no opts.messageId → still undefined). Tracking the
    // method name makes the assertion actually pin the
    // fire-and-forget contract.
    expect(ackCalls[0]?.method).toBe('sendToPeer');

    // Additional defense in depth: zero sendReliable calls
    // hit PROTOCOL_SWM_SHARE_ACK. Catches the regression
    // where a future change might call BOTH for some reason.
    const reliableAckCalls = calls.filter(
      (c) => c.protocolId === PROTOCOL_SWM_SHARE_ACK && c.method === 'sendReliable',
    );
    expect(reliableAckCalls).toEqual([]);
  });

  /**
   * rc.9 PR-D codex follow-up #D2: the ack handler MUST trust
   * the libp2p-authenticated `fromPeerId`, NOT the
   * self-asserted `ack.ackPeerId` in the protobuf body. Pre-D2
   * any peer that learned a `shareOperationId` could spoof
   * acks on behalf of other expected members, suppressing
   * watchdog top-up for those members.
   *
   * This test exercises the handler with a body that claims
   * peerB acked while the transport identity is peerA, and
   * asserts:
   *   1. The spoof attempt is dropped (the quorum's `acked` set
   *      gains no entry).
   *   2. Body == transport → the ack lands as expected.
   */
  it('PR-D #D2: ack handler rejects body/transport peerId mismatch (anti-spoof)', async () => {
    const agent = register(await createAgent('AckQuorumD2AntiSpoof'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWPeerA', '12D3KooWPeerB', '12D3KooWPeerC'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(() => ({
      delivered: false, queued: true, attempts: 1, messageId: 'q-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    const { shareOperationId } = await agent.share('cg-d2-antispoof', [{
      subject: 'urn:test:d2', predicate: 'http://schema.org/name', object: '"d2"', graph: '',
    }]);

    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => {
        onAck: (op: string, peer: string) => void;
        inspect: (op: string) => { acked: readonly string[]; expectedMembers: readonly string[] } | undefined;
      };
    }).getOrCreateSwmAckQuorum();

    expect(quorum.inspect(shareOperationId)?.acked).toEqual([]);

    // Find the PROTOCOL_SWM_SHARE_ACK handler. We can't intercept
    // the stub's register call (PR-D registers BEFORE the stub
    // is installed), so reach into the real Messenger via the
    // production-side getter that DKGAgent exposes. Test goes
    // through the SAME handler in src/dkg-agent.ts that
    // production traffic hits — that's the whole point.
    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();

    // Spoof attempt: body claims peerB, transport says peerA.
    // Expected: ack DROPPED, acked set unchanged.
    const spoofBytes = encodeSwmShareAck({
      shareOperationId,
      ackPeerId: '12D3KooWPeerB',
    });
    await ackHandler(spoofBytes, '12D3KooWPeerA');
    expect(quorum.inspect(shareOperationId)?.acked).toEqual([]);

    // Honest ack: body matches transport. Expected: lands.
    const honestBytes = encodeSwmShareAck({
      shareOperationId,
      ackPeerId: '12D3KooWPeerA',
    });
    await ackHandler(honestBytes, '12D3KooWPeerA');
    expect(quorum.inspect(shareOperationId)?.acked).toEqual(['12D3KooWPeerA']);

    // Empty body ackPeerId is also accepted — back-compat with
    // a future relayed-ack path where the body field might be
    // omitted intentionally.
    const emptyBodyBytes = encodeSwmShareAck({
      shareOperationId,
      ackPeerId: '',
    });
    await ackHandler(emptyBodyBytes, '12D3KooWPeerC');
    const after = quorum.inspect(shareOperationId)?.acked ?? [];
    expect([...after].sort()).toEqual(['12D3KooWPeerA', '12D3KooWPeerC']);
  });

  /**
   * rc.9 PR-D codex follow-up #D5: the quorum tracker MUST
   * register the share BEFORE substrate + gossip publish so a
   * fast receiver's ack lands against a known shareOperationId
   * instead of getting dropped because the record doesn't
   * exist yet. Pre-D5 the order was:
   *   1. Promise.all([substrate, gossip])    ← waits for both
   *   2. quorum.track(...)                   ← registers AFTER
   * A fast receiver could ack between #1 finishing and #2
   * starting; the handler's `quorum.onAck()` would no-op
   * because the shareOperationId wasn't tracked yet, causing
   * undercounted quorum and spurious watchdog top-up.
   *
   * Easiest exercise: intercept the gossip publish so it spies
   * on whether the quorum record exists at publish time, then
   * assert the record was tracked BEFORE the publish call.
   */
  it('PR-D #D5: quorum record is registered BEFORE gossip publish (no pre-track ack drop race)', async () => {
    const agent = register(await createAgent('AckQuorumD5RaceFix'));
    let recordExistedAtPublishTime: boolean | null = null;
    let capturedShareOperationId: string | null = null;
    const spyingGossip = new CapturingGossip();
    spyingGossip.subscribers = ['12D3KooWPeerRace1', '12D3KooWPeerRace2'];
    const originalPublish = spyingGossip.publish.bind(spyingGossip);
    spyingGossip.publish = async (topic: string, data: Uint8Array): Promise<void> => {
      // Inspect the agent's quorum state right when gossip
      // publish is about to fire. By this point the production
      // code MUST have already called quorum.track() — pre-D5
      // it hadn't.
      const stats = agent.getSwmAckQuorumStats();
      recordExistedAtPublishTime = stats.tracked === 1 && stats.pending === 1;
      await originalPublish(topic, data);
    };
    (agent as unknown as { gossip: CapturingGossip }).gossip = spyingGossip;

    const { install } = stubMessengerSendReliable(() => ({
      delivered: false, queued: true, attempts: 1, messageId: 'q-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    const { shareOperationId } = await agent.share('cg-d5-race-fix', [{
      subject: 'urn:test:d5', predicate: 'http://schema.org/name', object: '"d5"', graph: '',
    }]);
    capturedShareOperationId = shareOperationId;

    expect(recordExistedAtPublishTime).toBe(true);
    expect(agent.getSwmAckQuorumRecordSnapshotForTests(capturedShareOperationId)).toBeDefined();
  });

  /**
   * rc.9 PR-D codex follow-up #D6: substrate top-up MUST honour
   * the `ReliableSendResult` it gets back from sendReliable —
   * a successful top-up needs to pipe through onAck so the
   * peer counts toward quorum. Pre-D6 the result was awaited
   * and discarded, so a peer the watchdog successfully topped
   * up via substrate would STAY pending until deadlineHardMs
   * (since PROTOCOL_SWM_UPDATE doesn't emit SwmShareAck).
   *
   * Direct exercise of the substrateTopUp callback baked into
   * the lazily-constructed SwmAckQuorum, with a stub messenger
   * that returns each outcome shape so we can pin the
   * classification → onAck wiring for each one.
   */
  it('PR-D #D6: substrate top-up "delivered" outcomes route through onAck (peer counts toward quorum)', async () => {
    const agent = register(await createAgent('AckQuorumD6TopUpOnAck'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWTopUpA', '12D3KooWTopUpB', '12D3KooWTopUpC'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    // Initial fan-out: all peers go to the outbox (queued).
    // Watchdog then fires; top-up gets distinct outcomes per
    // peer to pin the per-outcome wiring.
    const { calls, install } = stubMessengerSendReliable((peerId, _proto, _msgId) => {
      const isTopUp = _msgId?.startsWith('swm-topup-');
      if (!isTopUp) {
        return { delivered: false, queued: true, attempts: 1, messageId: 'q-init', error: 'transient', nextAttemptAtMs: Date.now() + 1000 };
      }
      if (peerId === '12D3KooWTopUpA') {
        // empty response → delivered → onAck
        return { delivered: true, response: new Uint8Array(), attempts: 1, messageId: _msgId! };
      }
      if (peerId === '12D3KooWTopUpB') {
        // 0x02 retryable → no onAck; next watchdog tick retries
        return { delivered: true, response: new Uint8Array([0x02]), attempts: 1, messageId: _msgId! };
      }
      // 0x01 rejected → no onAck (permanent)
      return { delivered: true, response: new Uint8Array([0x01]), attempts: 1, messageId: _msgId! };
    });
    install(agent);

    const { shareOperationId } = await agent.share('cg-d6-topup', [{
      subject: 'urn:test:d6', predicate: 'http://schema.org/name', object: '"d6"', graph: '',
    }]);

    // Pre-watchdog: all 3 peers still pending.
    const before = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(before?.acked).toEqual([]);

    // Drive the substrateTopUp path directly via the test-only
    // helper. Bypasses the watchdog's setInterval (so the test
    // isn't real-time-flaky) and hits the EXACT method the
    // watchdog wires into createSwmAckQuorum's
    // `substrateTopUp` dep.
    await agent.invokeSwmSubstrateTopUpForTests({
      shareOperationId,
      cgId: 'cg-d6-topup',
      payload: new Uint8Array([0xde, 0xad, 0xbe, 0xef]),
      missingPeers: ['12D3KooWTopUpA', '12D3KooWTopUpB', '12D3KooWTopUpC'],
    });

    // After top-up: A got `delivered` → routed through onAck;
    // B got 0x02 retryable → NOT onAck'd (will retry next
    // tick); C got 0x01 rejected → NOT onAck'd (permanent).
    const after = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(after?.acked).toEqual(['12D3KooWTopUpA']);

    // Sanity: the top-up actually invoked sendReliable for all
    // three peers (with the documented swm-topup- prefix).
    const topUpCalls = calls.filter((c) => c.messageId?.startsWith('swm-topup-'));
    expect(topUpCalls.map((c) => c.peerId).sort()).toEqual([
      '12D3KooWTopUpA',
      '12D3KooWTopUpB',
      '12D3KooWTopUpC',
    ]);
  });

  /**
   * rc.9 PR-D codex follow-up #D3 (originally) required the
   * ack-quorum tracker to run for the gossip-only-because-too-
   * many-subscribers branch (public CG above the substrate cap)
   * — pre-D3 the track gate keyed off `substrateMembers.length`,
   * which that branch intentionally empties.
   *
   * PR-K (2026-05-18 Miles<->Lex soak) reverses D3 for a new
   * reason. The soak proved that "gossip-deliverable but not
   * substrate-eligible" peers cannot reliably send
   * SwmShareAck back either — the ack travels through the same
   * `messenger.sendReliable` and hits the same limited
   * Circuit Relay V2 budget exhaustion that broke the
   * substrate fan-out. Tracking acks from peers that can't
   * actually ack just produces deadlineExpired noise.
   *
   * The post-PR-K contract:
   *   - `ackQuorumActive` is gated on `substrateMembers.length`
   *     (back to pre-D3 behaviour, but for the OPPOSITE reason).
   *   - `expectedMembers` is `substrateMembers` (the
   *     roundtrip-eligible subset), NOT `enumeratedMembers`.
   *   - Gossip-only-too-many-subscribers branch publishes via
   *     gossip and skips quorum tracking entirely. Verification
   *     for that branch falls back to cross-peer SWM-inbox
   *     SPARQL, which is the ground truth anyway.
   */
  it('PR-K (revises D3): gossip-only-large-public CGs are NOT tracked (acks unreliable)', async () => {
    const agent = register(await createAgent('AckQuorumKLargePublic'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWBig1', '12D3KooWBig2', '12D3KooWBig3'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    (agent as unknown as { swmSubstrateMaxMembers: number }).swmSubstrateMaxMembers = 1;

    const { calls, install } = stubMessengerSendReliable(new Map());
    install(agent);

    const { shareOperationId } = await agent.share('cg-k-large-public', [{
      subject: 'urn:test:k', predicate: 'http://schema.org/name', object: '"k"', graph: '',
    }]);

    const updateCalls = calls.filter((c) => c.protocolId === PROTOCOL_SWM_UPDATE);
    expect(updateCalls).toEqual([]);

    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => {
        inspect: (op: string) => { expectedMembers: readonly string[]; acked: readonly string[] } | undefined;
      };
    }).getOrCreateSwmAckQuorum();

    // PR-K: track is skipped (substrateMembers is empty in the
    // gossip-only branch). `inspect` returns undefined.
    expect(quorum.inspect(shareOperationId)).toBeUndefined();
  });

  /**
   * PR-H bug 1: when substrate top-up returns the 0x02
   * retryable sentinel, the watchdog MUST be re-armed so the
   * next tick fires another top-up. Pre-PR-H `watchdogFired`
   * stayed `true` after the first fire so the record sat
   * pending until `deadlineHardMs` (5 min) even though the
   * 0x02 sentinel was an explicit "retry later" signal.
   *
   * Exercises the production callback chain via
   * `invokeSwmSubstrateTopUpForTests` (bypassing the 5s
   * setInterval so the test isn't real-time-flaky) and asserts
   * the watchdog state transitions through fire → rearm.
   */
  it('PR-H bug 1: substrate top-up with retryable outcomes re-arms the watchdog', async () => {
    const agent = register(await createAgent('AckQuorumHBug1Rearm'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWRetry1', '12D3KooWRetry2'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable((_peerId, _proto, msgId) => {
      const isTopUp = msgId?.startsWith('swm-topup-');
      if (!isTopUp) {
        return { delivered: false, queued: true, attempts: 1, messageId: 'q-init', error: 'transient', nextAttemptAtMs: Date.now() + 1000 };
      }
      return { delivered: true, response: new Uint8Array([0x02]), attempts: 1, messageId: msgId! };
    });
    install(agent);

    const { shareOperationId } = await agent.share('cg-h-bug1-rearm', [{
      subject: 'urn:test:hbug1', predicate: 'http://schema.org/name', object: '"hbug1"', graph: '',
    }]);

    const before = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(before?.watchdogFired).toBe(false);
    const watchdogArmedAtInitially = before?.watchdogArmedAtMs ?? 0;

    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => { tick: (now?: number) => void };
    }).getOrCreateSwmAckQuorum();
    quorum.tick(Date.now() + 30_001);

    await agent.invokeSwmSubstrateTopUpForTests({
      shareOperationId,
      cgId: 'cg-h-bug1-rearm',
      payload: new Uint8Array([0xde, 0xad]),
      missingPeers: ['12D3KooWRetry1', '12D3KooWRetry2'],
    });

    const after = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(after).toBeDefined();
    expect(after?.acked).toEqual([]);
    expect(after?.watchdogFired).toBe(false);
    expect(after?.watchdogArmedAtMs).toBeGreaterThanOrEqual(watchdogArmedAtInitially);
  });

  /**
   * PR-H round 2 (codex feedback on #582 bug 2): when the top-up
   * returns ONLY terminal outcomes (delivered + rejected), the
   * rejected peer is dropped from `expectedMembers` instead of
   * lingering and forcing another watchdog top-up. With the
   * denominator shrunk, the single ack from `Final1` now
   * represents 100% of expected (1/1) so the record completes
   * cleanly — no re-arm, no five-minute deadline wait.
   *
   * Pre-round-2 behaviour (kept here for the test name to
   * provide historical context): the rejected peer stayed in
   * `expectedMembers`, `acked = ['Final1']` was only 50%, and
   * since no peer was retryable the watchdog stayed `fired =
   * true` until `deadlineHardMs` reaped the record. The new
   * behaviour is strictly an improvement (avoids the 5-min
   * wall-clock wait for a share that already has all the acks
   * it can ever get).
   */
  it('PR-H bug 2: top-up with terminal outcomes (delivered + rejected) completes via dropPeer instead of stalling until deadline', async () => {
    const agent = register(await createAgent('AckQuorumHBug2DropPeer'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWFinal1', '12D3KooWFinal2'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable((peerId, _proto, msgId) => {
      const isTopUp = msgId?.startsWith('swm-topup-');
      if (!isTopUp) {
        return { delivered: false, queued: true, attempts: 1, messageId: 'q-init', error: 'transient', nextAttemptAtMs: Date.now() + 1000 };
      }
      if (peerId === '12D3KooWFinal1') return { delivered: true, response: new Uint8Array(), attempts: 1, messageId: msgId! };
      return { delivered: true, response: new Uint8Array([0x01]), attempts: 1, messageId: msgId! };
    });
    install(agent);

    const { shareOperationId } = await agent.share('cg-h-bug2-dropper', [{
      subject: 'urn:test:hbug2', predicate: 'http://schema.org/name', object: '"hbug2"', graph: '',
    }]);

    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => { tick: (now?: number) => void; stats: () => { completed: number; pending: number } };
    }).getOrCreateSwmAckQuorum();
    quorum.tick(Date.now() + 30_001);

    await agent.invokeSwmSubstrateTopUpForTests({
      shareOperationId,
      cgId: 'cg-h-bug2-dropper',
      payload: new Uint8Array([0xde, 0xad]),
      missingPeers: ['12D3KooWFinal1', '12D3KooWFinal2'],
    });

    // The record completed and was removed from tracking. Snapshot
    // returns undefined; stats show the completion counter ticked.
    const after = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(after).toBeUndefined();
    expect(quorum.stats().completed).toBe(1);
    expect(quorum.stats().pending).toBe(0);
  });

  /**
   * PR-H bug 2: late substrate deliveries (peers that started
   * as queued/inFlight on the synchronous fan-out and were
   * delivered LATER by the outbox) MUST reach the quorum. The
   * fix is for receivers to emit SwmShareAck on substrate
   * apply too — symmetric with the gossip-apply path. Pre-PR-H
   * only the gossip path emitted, so outbox-delivered peers
   * never ack'd and stayed pending until deadlineHardMs even
   * though the share had actually applied.
   */
  it('PR-H bug 2: handleSwmUpdate emits SwmShareAck on substrate-applied shares (late deliveries reach quorum)', async () => {
    const agent = register(await createAgent('AckQuorumHBug2SubstrateAck'));
    const gossip = new CapturingGossip();
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { calls, install } = stubMessengerSendReliable(new Map());
    install(agent);

    const wh = (agent as unknown as {
      getOrCreateSharedMemoryHandler: () => { handle: (data: Uint8Array, from: string) => Promise<unknown> };
    }).getOrCreateSharedMemoryHandler();
    const origHandle = wh.handle.bind(wh);
    wh.handle = async () => ({
      applied: true,
      cgId: 'cg-h-bug2',
      shareOperationId: 'op-h-bug2-late-delivery',
      publisherPeerId: '12D3KooWPublisherLate',
    });

    try {
      const handler = (agent as unknown as {
        handleSwmUpdate: (data: Uint8Array, from: string) => Promise<Uint8Array>;
      }).handleSwmUpdate.bind(agent);

      const response = await handler(new Uint8Array([0x01, 0x02, 0x03]), '12D3KooWSenderForLate');
      expect(response).toEqual(new Uint8Array());

      await new Promise((r) => setTimeout(r, 5));

      const ackCalls = calls.filter((c) => c.protocolId === PROTOCOL_SWM_SHARE_ACK);
      expect(ackCalls).toHaveLength(1);
      expect(ackCalls[0]?.peerId).toBe('12D3KooWPublisherLate');
      expect(ackCalls[0]?.method).toBe('sendToPeer');
    } finally {
      wh.handle = origHandle;
    }
  });

  /**
   * PR-H bug 2 negative case: when `handleSwmUpdate` returns
   * `applied: false` (retryable OR permanent rejection), it
   * MUST NOT emit a SwmShareAck — only successful applies ack.
   */
  it('PR-H bug 2: handleSwmUpdate does NOT emit SwmShareAck on rejected (retryable or permanent) substrate shares', async () => {
    const agent = register(await createAgent('AckQuorumHBug2NoAckOnReject'));
    const gossip = new CapturingGossip();
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { calls, install } = stubMessengerSendReliable(new Map());
    install(agent);

    const wh = (agent as unknown as {
      getOrCreateSharedMemoryHandler: () => { handle: (data: Uint8Array, from: string) => Promise<unknown> };
    }).getOrCreateSharedMemoryHandler();
    const origHandle = wh.handle.bind(wh);

    try {
      wh.handle = async () => ({ applied: false, retryable: true, reason: 'transient apply error' });
      const handler = (agent as unknown as {
        handleSwmUpdate: (data: Uint8Array, from: string) => Promise<Uint8Array>;
      }).handleSwmUpdate.bind(agent);
      const retryResp = await handler(new Uint8Array([0x01]), '12D3KooWSenderRetry');
      expect(retryResp).toEqual(new Uint8Array([0x02]));

      wh.handle = async () => ({ applied: false, retryable: false, reason: 'bad signature' });
      const rejectResp = await handler(new Uint8Array([0x02]), '12D3KooWSenderReject');
      expect(rejectResp).toEqual(new Uint8Array([0x01]));

      await new Promise((r) => setTimeout(r, 5));

      const ackCalls = calls.filter((c) => c.protocolId === PROTOCOL_SWM_SHARE_ACK);
      expect(ackCalls).toEqual([]);
    } finally {
      wh.handle = origHandle;
    }
  });

  /**
   * Hardening — corrupt protobuf bytes arriving on the
   * PROTOCOL_SWM_SHARE_ACK handler MUST NOT crash the handler
   * or the agent. handleSwmShareAck wraps decodeSwmShareAck in
   * a try/catch; this test locks that protection so a
   * regression that removed the try/catch (DoS hazard — a
   * malformed peer could repeatedly throw and stall the
   * receive loop) would fail loudly. Asserts:
   *   1. handler resolves (no throw) for arbitrary garbage
   *   2. handler returns the documented empty-bytes ack
   *   3. quorum state is untouched (no spurious onAck)
   */
  it('hardening: handleSwmShareAck does NOT crash on corrupt protobuf bytes (DoS guard)', async () => {
    const agent = register(await createAgent('AckQuorumCorruptBytes'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWCorrupt1', '12D3KooWCorrupt2'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(() => ({
      delivered: false, queued: true, attempts: 1, messageId: 'q-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    const { shareOperationId } = await agent.share('cg-corrupt-bytes', [{
      subject: 'urn:test:corrupt', predicate: 'http://schema.org/name', object: '"corrupt"', graph: '',
    }]);

    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();

    const garbageInputs: Array<{ name: string; bytes: Uint8Array }> = [
      { name: 'empty bytes', bytes: new Uint8Array() },
      { name: 'random noise', bytes: new Uint8Array([0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff]) },
      { name: 'truncated varint', bytes: new Uint8Array([0x0a]) },
      { name: 'all zeros', bytes: new Uint8Array(64) },
      { name: 'mid-message truncation', bytes: new Uint8Array([0x0a, 0x05, 0x68]) },
    ];

    for (const { bytes } of garbageInputs) {
      const out = await ackHandler(bytes, '12D3KooWCorrupt1');
      expect(out).toEqual(new Uint8Array());
    }

    // Quorum state remains untouched — none of the garbage
    // inputs caused a spurious onAck against the live record.
    const inspect = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(inspect?.acked).toEqual([]);
    expect(agent.getSwmAckQuorumStats().completed).toBe(0);
  });

  /**
   * Hardening — an ack arriving for a shareOperationId that
   * was never tracked (or already reaped via deadline) MUST be
   * silently dropped at the agent integration layer. The unit
   * test for createSwmAckQuorum.onAck covers the no-op
   * behaviour at the component level; this test locks that the
   * agent's handler propagates the same behaviour through the
   * full decode → onAck pipeline. A regression where
   * handleSwmShareAck mutated some other state on
   * unknown-id arrival would not be caught by the unit test
   * alone.
   */
  it('hardening: handleSwmShareAck silently drops acks for unknown shareOperationId (no-op propagation)', async () => {
    const agent = register(await createAgent('AckQuorumUnknownOpId'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWUnknown1'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(new Map());
    install(agent);

    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();

    const beforeStats = agent.getSwmAckQuorumStats();
    expect(beforeStats).toEqual({
      tracked: 0, completed: 0, watchdogFired: 0, deadlineExpired: 0, pending: 0,
    });

    const out = await ackHandler(
      encodeSwmShareAck({
        shareOperationId: 'op-never-tracked-by-this-agent',
        ackPeerId: '12D3KooWUnknown1',
      }),
      '12D3KooWUnknown1',
    );
    expect(out).toEqual(new Uint8Array());

    const afterStats = agent.getSwmAckQuorumStats();
    expect(afterStats).toEqual(beforeStats);
  });

  /**
   * Hardening — concurrent shares to DIFFERENT context graphs
   * track independently. Locks that the agent's quorum state
   * is keyed strictly by shareOperationId (no Map collision,
   * no shared state) and that completing one share does NOT
   * affect the other's pending/acked/expectedMembers state.
   *
   * Failure modes this catches:
   *   - Map keyed off cgId instead of shareOperationId (acks
   *     from share A would land against share B)
   *   - completeRecord side-effects leaking into sibling
   *     records via a shared array reference
   *   - stats.tracked/.completed/.pending double-counting or
   *     under-counting when two records exist simultaneously
   */
  it('hardening: two concurrent shares to different cgs track independently (Map isolation)', async () => {
    const agent = register(await createAgent('AckQuorumConcurrentShares'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWAlpha', '12D3KooWBeta'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    const { install } = stubMessengerSendReliable(() => ({
      delivered: false, queued: true, attempts: 1, messageId: 'q-' + Math.random().toString(36).slice(2),
      error: 'transient', nextAttemptAtMs: Date.now() + 1000,
    }));
    install(agent);

    // Drive two share() calls back-to-back to two different
    // CGs. Both must register; both must reach `pending=1`
    // independently for a brief window.
    const shareA = await agent.share('cg-concurrent-A', [{
      subject: 'urn:test:concA', predicate: 'http://schema.org/name', object: '"A"', graph: '',
    }]);
    const shareB = await agent.share('cg-concurrent-B', [{
      subject: 'urn:test:concB', predicate: 'http://schema.org/name', object: '"B"', graph: '',
    }]);
    expect(shareA.shareOperationId).not.toBe(shareB.shareOperationId);

    const stats0 = agent.getSwmAckQuorumStats();
    expect(stats0.tracked).toBe(2);
    expect(stats0.pending).toBe(2);
    expect(stats0.completed).toBe(0);

    // Snapshots must be independent — different
    // shareOperationIds, distinct expectedMembers references
    // (same set of peers in this test, but the snapshot's
    // membership list is not aliased between records).
    const snapA = agent.getSwmAckQuorumRecordSnapshotForTests(shareA.shareOperationId);
    const snapB = agent.getSwmAckQuorumRecordSnapshotForTests(shareB.shareOperationId);
    expect(snapA?.shareOperationId).toBe(shareA.shareOperationId);
    expect(snapB?.shareOperationId).toBe(shareB.shareOperationId);
    expect(snapA?.acked).toEqual([]);
    expect(snapB?.acked).toEqual([]);

    // Drive A to completion. B's state must remain pending +
    // unacked.
    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();
    await ackHandler(
      encodeSwmShareAck({ shareOperationId: shareA.shareOperationId, ackPeerId: '12D3KooWAlpha' }),
      '12D3KooWAlpha',
    );
    await ackHandler(
      encodeSwmShareAck({ shareOperationId: shareA.shareOperationId, ackPeerId: '12D3KooWBeta' }),
      '12D3KooWBeta',
    );

    // A is reaped on completion; B is still pending with no
    // acks (the Alpha+Beta acks did NOT bleed into B).
    expect(agent.getSwmAckQuorumRecordSnapshotForTests(shareA.shareOperationId)).toBeUndefined();
    const snapBAfter = agent.getSwmAckQuorumRecordSnapshotForTests(shareB.shareOperationId);
    expect(snapBAfter).toBeDefined();
    expect(snapBAfter?.acked).toEqual([]);

    const statsAfter = agent.getSwmAckQuorumStats();
    expect(statsAfter.tracked).toBe(2);
    expect(statsAfter.completed).toBe(1);
    expect(statsAfter.pending).toBe(1);
  });

  /**
   * #1236 review 🔴 discriminator (otReviewAgent on the #1227
   * regression fix). PR #1236 widened the ack-quorum's
   * `expectedMembers` back to the FULL `plan.substrateMembers`
   * (not the churn-narrowed initial-push set), but the watchdog
   * top-up in `swmSubstrateTopUp` was NOT changed: for a public
   * (`topic-subscribers`) CG it still runs the fan-out selector,
   * which SKIPS any peer sitting in the selector's negative-TTL
   * cache. The bot's worry: a quorum-EXPECTED peer that
   * TERMINALLY FAILED the initial push (→ negative cache) and
   * does NOT ack via gossip might never be re-attempted by the
   * watchdog, stalling the quorum until the hard deadline.
   *
   * This test constructs EXACTLY that scenario and asserts what
   * actually happens (empirical, not theoretical):
   *
   *   - `Term` rejects the initial push (0x01 sentinel →
   *     classified `failed` → recorded into the selector's
   *     negative cache with the 2-min negative TTL). It IS part
   *     of the full quorum set (`expectedMembers`) and does NOT
   *     ack via gossip.
   *   - `Good1` / `Good2` queue on the initial push (NOT terminal
   *     → NOT cached), then ack via gossip — but 2/3 < the 0.9
   *     default quorum threshold, so the record stays pending on
   *     `Term`.
   *   - The watchdog then fires substrate top-up for the only
   *     missing peer, `Term`.
   *
   * OBSERVED OUTCOME = (b), graceful degradation:
   *   1. The top-up does NOT re-reach `Term` — the selector
   *      classifies it as a recent-negative peer and drops it
   *      from `selectedPeers` (`sendReliable` for the
   *      `swm-topup-` messageId is never called for `Term`). This
   *      confirms the bot's mechanism is real: the watchdog DOES
   *      skip a quorum-expected, negative-cached peer.
   *   2. BUT the share is NOT permanently stuck: the hard
   *      deadline reaps the record cleanly via `onDeadlineExpired`
   *      (`deadlineExpired` ticks, `pending` returns to 0, the
   *      record is removed). Offline/rejecting peers fall through
   *      to `runSyncOnConnect` — the documented catch-up safety
   *      net (ack-quorum.ts §4).
   *
   * Why (b) and not (c): the skipped peer here TERMINALLY
   * rejected the initial push. Re-blasting the identical wire
   * bytes at it on every 30s watchdog tick would reproduce the
   * exact churn-amplification soak bug #1227's selector was added
   * to stop — and would never succeed (a permanent rejection
   * rejects again). The deadline-reap + sync-on-connect recovery
   * is the correct terminal for a peer that can't accept the
   * share. So the 🔴 is a defensible graceful-degradation path,
   * NOT a correctness bug — no source change is warranted; this
   * test documents the deadline-resolution contract.
   */
  it('#1236 🔴 (b): watchdog top-up skips a terminally-failed, negative-cached quorum-expected peer; deadline reaps it cleanly (graceful degradation)', async () => {
    const agent = register(await createAgent('AckQuorum1236TerminalSkip'));
    const gossip = new CapturingGossip();
    gossip.subscribers = ['12D3KooWTerm', '12D3KooWGood1', '12D3KooWGood2'];
    (agent as unknown as { gossip: CapturingGossip }).gossip = gossip;

    // Initial fan-out: `Term` returns the 0x01 REJECTED sentinel
    // (delivered: true at the wire layer, but the receiver
    // permanently rejected) → classified `failed` →
    // terminal → fed into the fan-out selector's negative cache.
    // `Good1` / `Good2` queue (non-terminal → NOT cached) so
    // they stay pending after the initial push.
    const { calls, install } = stubMessengerSendReliable((peerId, _proto, msgId) => {
      const isTopUp = msgId?.startsWith('swm-topup-');
      if (!isTopUp) {
        if (peerId === '12D3KooWTerm') {
          // 0x01 → rejected → classifySwmFanoutPeerOutcome → 'failed'
          return { delivered: true, response: new Uint8Array([0x01]), attempts: 1, messageId: 'init-term' };
        }
        return { delivered: false, queued: true, attempts: 1, messageId: 'q-init', error: 'transient', nextAttemptAtMs: Date.now() + 1000 };
      }
      // Any top-up that DOES reach a peer succeeds (so if the
      // watchdog were to re-reach `Term`, we'd see it).
      return { delivered: true, response: new Uint8Array(), attempts: 1, messageId: msgId! };
    });
    install(agent);

    const { shareOperationId } = await agent.share('cg-1236-terminal-skip', [{
      subject: 'urn:test:1236', predicate: 'http://schema.org/name', object: '"t1236"', graph: '',
    }]);

    // The share must have run a public (topic-subscribers)
    // substrate fan-out — that's the only enumeration source the
    // selector (and thus the negative cache) applies to. If this
    // assertion ever flips, the discriminator's preconditions no
    // longer hold and the outcome below is meaningless.
    const snap0 = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(snap0).toBeDefined();
    expect([...(snap0?.expectedMembers ?? [])].sort()).toEqual([
      '12D3KooWGood1', '12D3KooWGood2', '12D3KooWTerm',
    ]);

    // Precondition 1: the initial fan-out actually attempted all 3
    // peers (so `Term`'s rejection was observed by the bookkeeper).
    const initCalls = calls.filter((c) => !c.messageId?.startsWith('swm-topup-') && c.protocolId === PROTOCOL_SWM_UPDATE);
    expect(initCalls.map((c) => c.peerId).sort()).toEqual([
      '12D3KooWGood1', '12D3KooWGood2', '12D3KooWTerm',
    ]);

    // Precondition 2: `Term` landed in the fan-out selector's
    // negative cache (this is the gate the watchdog top-up
    // re-applies). `failed` is a negative outcome, so the
    // selector will skip it inside the 2-min TTL.
    expect(agent.getSwmFanoutPeerOutcomeForTests('cg-1236-terminal-skip', '12D3KooWTerm')).toBe('failed');

    // `Good1` / `Good2` ack via gossip. 2/3 = 0.67 < 0.9 default
    // threshold, so the record is still pending on `Term`.
    const ackHandler = await agent.getOrCreateSwmShareAckHandlerForTests();
    await ackHandler(encodeSwmShareAck({ shareOperationId, ackPeerId: '12D3KooWGood1' }), '12D3KooWGood1');
    await ackHandler(encodeSwmShareAck({ shareOperationId, ackPeerId: '12D3KooWGood2' }), '12D3KooWGood2');
    const snap1 = agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId);
    expect(snap1).toBeDefined();
    expect([...(snap1?.acked ?? [])].sort()).toEqual(['12D3KooWGood1', '12D3KooWGood2']);
    expect(agent.getSwmAckQuorumStats().completed).toBe(0);
    expect(agent.getSwmAckQuorumStats().pending).toBe(1);

    // Make `Term` dialable for the top-up's `isPeerDialable`
    // filter so we isolate the negative-cache skip (not a
    // dialability skip) as the reason the top-up doesn't reach it.
    // (Short test peer ids resolve via the gossip-subscriber
    // fast path in `installAllReachableLibp2pStub`; `Term` is in
    // `gossip.subscribers`, so it's already dialable.)

    // Count top-up calls BEFORE driving the watchdog.
    const topUpCallsBefore = calls.filter((c) => c.messageId?.startsWith('swm-topup-')).length;

    // Drive the watchdog top-up for the only missing peer (`Term`).
    // This is the EXACT path the 5s tick → fireWatchdog →
    // substrateTopUp would take, with `missingPeers` =
    // expectedMembers ∖ acked = [Term].
    await agent.invokeSwmSubstrateTopUpForTests({
      shareOperationId,
      cgId: 'cg-1236-terminal-skip',
      payload: new Uint8Array([0xde, 0xad, 0xbe, 0xef]),
      missingPeers: ['12D3KooWTerm'],
    });

    // OBSERVED (1): the top-up did NOT re-reach `Term`. The
    // selector classified it as a recent-negative peer and
    // dropped it from `selectedPeers`, so no `swm-topup-`
    // sendReliable fired for it. This confirms the bot's
    // mechanism: the watchdog DOES skip a quorum-expected,
    // negative-cached peer.
    const topUpCallsAfter = calls.filter((c) => c.messageId?.startsWith('swm-topup-'));
    expect(topUpCallsAfter.filter((c) => c.peerId === '12D3KooWTerm')).toEqual([]);
    expect(topUpCallsAfter.length).toBe(topUpCallsBefore);

    // The record is still pending on `Term` (the top-up was a
    // no-op for it).
    expect(agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId)).toBeDefined();
    expect(agent.getSwmAckQuorumStats().pending).toBe(1);
    expect(agent.getSwmAckQuorumStats().completed).toBe(0);

    // OBSERVED (2): the share is NOT permanently stuck. The hard
    // deadline reaps the record cleanly via `onDeadlineExpired`.
    // Advance the quorum clock past `startedAtMs + deadlineHardMs`
    // (5 min default) and tick — the record is removed,
    // `deadlineExpired` increments, `pending` returns to 0.
    const quorum = (agent as unknown as {
      getOrCreateSwmAckQuorum: () => { tick: (now?: number) => void };
    }).getOrCreateSwmAckQuorum();
    quorum.tick(Date.now() + 5 * 60_000 + 1);

    expect(agent.getSwmAckQuorumRecordSnapshotForTests(shareOperationId)).toBeUndefined();
    const finalStats = agent.getSwmAckQuorumStats();
    expect(finalStats.pending).toBe(0);
    expect(finalStats.deadlineExpired).toBe(1);
    // Never falsely counted as completed.
    expect(finalStats.completed).toBe(0);
  });
});
