import { describe, test, expect, beforeEach } from "bun:test";
import { configureSync, getConfig, resetSync } from "@intx/log";
import { generateKeyPair, signEd25519 } from "@intx/crypto";
import { hexDecode, hexEncode, parseAgentAddress } from "@intx/types";
import { chunkPack } from "@intx/pack-transport";
import { deriveWorkflowRunRepoId } from "@intx/workflow-deploy";
import type {
  PackRejectReason,
  RepoId,
  RunGrantsFrame,
} from "@intx/types/sidecar";
import {
  createSidecarRouter,
  type SidecarAuthenticator,
  type SidecarRouterConfig,
  type WsHandle,
} from "./sidecar-handler";

// Test authenticator that accepts any presented token and echoes the
// claimed id back as the verified identity. Tests that exercise routing
// rather than auth use it so the handshake succeeds; tests that assert
// auth behavior pass their own authenticator instead.
const acceptAnySidecar: SidecarAuthenticator = async ({ sidecarId }) => ({
  kind: "shared",
  sidecarId,
});

// Build a router with a default accept-any authenticator so routing tests
// need not restate it. A test that cares about auth passes its own
// `authenticateSidecar` in `config`, which overrides the default.
function createTestRouter(
  config: Omit<SidecarRouterConfig, "authenticateSidecar"> &
    Partial<Pick<SidecarRouterConfig, "authenticateSidecar">> = {},
): ReturnType<typeof createSidecarRouter> {
  return createSidecarRouter({
    authenticateSidecar: acceptAnySidecar,
    ...config,
  });
}

async function signChallenge(
  nonce: string,
  address: string,
  privateKeyBytes: Uint8Array,
): Promise<string> {
  const nonceBytes = hexDecode(nonce);
  const addressBytes = new TextEncoder().encode(address);
  const payload = new Uint8Array(nonceBytes.length + addressBytes.length);
  payload.set(nonceBytes);
  payload.set(addressBytes, nonceBytes.length);
  const sig = await signEd25519(privateKeyBytes, payload);
  return hexEncode(new Uint8Array(sig));
}

function createMockWs(): WsHandle & { sent: string[]; closed: boolean } {
  return {
    sent: [],
    closed: false,
    send(data: string) {
      this.sent.push(data);
    },
    close() {
      this.closed = true;
    },
  };
}

function lastSent(ws: ReturnType<typeof createMockWs>) {
  const last = ws.sent[ws.sent.length - 1];
  if (last === undefined) throw new Error("No messages sent");
  return JSON.parse(last);
}

// Let the async register key-existence gate settle. A `register` frame now
// routes on a microtask (handleRegister awaits `lookupPublicKey` per address),
// so a test that reads the routing table right after sending one must await
// this first.
async function tick(): Promise<void> {
  await new Promise((resolve) => setTimeout(resolve, 0));
}

// Route the package logger's warn-level records into `sink` until the returned
// restore is called. Used to assert loud surfacing of dropped disconnect-queue
// mail, whose warn is emitted asynchronously from a TTL timer (so the capture
// must stay installed across the wait, not just a synchronous block).
function installWarningCapture(sink: string[]): () => void {
  const savedConfig = getConfig();
  configureSync({
    reset: true,
    sinks: {
      capture: (record) => {
        if (record.level !== "warning") return;
        const message = Array.isArray(record.message)
          ? record.message
              .map((part) =>
                typeof part === "string" ? part : JSON.stringify(part),
              )
              .join("")
          : String(record.message);
        sink.push(message);
      },
    },
    loggers: [{ category: [], lowestLevel: "warning", sinks: ["capture"] }],
  });
  return () => {
    if (savedConfig) {
      configureSync({ reset: true, ...savedConfig });
    } else {
      resetSync();
    }
  };
}

// Capture the package logger's warn records emitted synchronously by `fn`.
function captureWarnings(fn: () => void): string[] {
  const warnings: string[] = [];
  const restore = installWarningCapture(warnings);
  try {
    fn();
  } finally {
    restore();
  }
  return warnings;
}

describe("SidecarRouter", () => {
  let router: ReturnType<typeof createSidecarRouter>;

  const TEST_HUB_KEY = "a".repeat(64);

  // A minimal valid source list for exercising the requestId-correlated
  // sendRequest path (sendSourcesUpdate). The routing/reconnect invariants
  // under test are independent of the source content.
  const TEST_SOURCES = [
    {
      id: "anthropic:claude-sonnet-5",
      provider: "anthropic",
      baseURL: "https://api.anthropic.com",
      apiKey: "sk-test",
      model: "claude-sonnet-5",
    },
  ];
  const TEST_DEFAULT_SOURCE = "anthropic:claude-sonnet-5";

  beforeEach(() => {
    router = createTestRouter({
      requestTimeoutMs: 500,
      hubPublicKey: TEST_HUB_KEY,
      // Always-null lookup: no address has a stored key, so the register
      // key-existence gate classifies every registered address as a keyless
      // first-deploy and routes it. This makes the dependency explicit for
      // the pure-routing tests (the gate is fail-closed without it) while
      // preserving their "register routes the address" behavior.
      lookups: { lookupPublicKey: async () => null },
    });
  });

  describe("registration", () => {
    test("register frame populates routing table", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent-a@local", "agent-b@local"],
        }),
      );
      await tick();

      expect(router.getConnectedSidecars()).toEqual(["sc-1"]);
      expect(router.getRoutableAddresses().sort()).toEqual([
        "agent-a@local",
        "agent-b@local",
      ]);
    });

    test("re-registration adds addresses without dropping owned routes", async () => {
      // Additive re-register: a second register ADDS its addresses without
      // dropping those from the first. Removal happens via undeploy/disconnect,
      // not register-omission (the frame no longer carries the full live set).
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent-a@local"],
        }),
      );
      await tick();

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent-c@local"],
        }),
      );
      await tick();

      expect(router.getRoutableAddresses().sort()).toEqual([
        "agent-a@local",
        "agent-c@local",
      ]);
    });

    test("disconnect cleans up routing table", () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent-a@local"],
        }),
      );

      router.handleClose(ws);
      expect(router.getConnectedSidecars()).toEqual([]);
      expect(router.getRoutableAddresses()).toEqual([]);
    });

    const rejectAll: SidecarAuthenticator = async () => null;

    test("register with an invalid token closes the connection", async () => {
      const router = createTestRouter({ authenticateSidecar: rejectAll });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "bad",
          agentAddresses: [],
        }),
      );
      await tick();

      expect(ws.closed).toBe(true);
      expect(router.getConnectedSidecars()).toEqual([]);
    });

    test("reconnect with an invalid token closes the connection", async () => {
      const router = createTestRouter({
        authenticateSidecar: rejectAll,
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "bad",
          agentAddresses: ["agent-a@local"],
        }),
      );
      await tick();

      expect(ws.closed).toBe(true);
      expect(router.getConnectedSidecars()).toEqual([]);
    });

    test("a thrown authenticator fails closed and closes the connection", async () => {
      const router = createTestRouter({
        authenticateSidecar: async () => {
          throw new Error("database unavailable");
        },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      expect(ws.closed).toBe(true);
      expect(router.getConnectedSidecars()).toEqual([]);
    });

    test("the connection is keyed by the verified id, not the claimed id", async () => {
      // The claimed frame id is untrusted: the authenticator resolves the
      // token to a different verified id, and routing must key off that.
      const router = createTestRouter({
        authenticateSidecar: async () => ({
          kind: "shared",
          sidecarId: "verified-sc",
        }),
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "claimed-sc",
          token: "tok",
          agentAddresses: ["agent-a@local"],
        }),
      );
      await tick();

      expect(router.getConnectedSidecars()).toEqual(["verified-sc"]);
    });

    test("a reconnect authenticates exactly once despite its internal register", async () => {
      // handleReconnect performs an internal register; the handshake must
      // authenticate at the dispatch boundary only, never twice per frame.
      let calls = 0;
      const router = createTestRouter({
        authenticateSidecar: async ({ sidecarId }) => {
          calls += 1;
          return { kind: "shared", sidecarId };
        },
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      expect(calls).toBe(1);
    });

    test("re-registration by another sidecar cleans ghost from old connection", async () => {
      const ws1 = createMockWs();
      router.handleOpen(ws1);
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local", "other@local"],
        }),
      );
      await tick();

      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-2",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      // ws2 now owns agent@local. Closing ws1 should only remove
      // other@local (which ws1 still owns), not agent@local.
      router.handleClose(ws1);

      expect(router.getRoutableAddresses()).toContain("agent@local");
      expect(router.getRoutableAddresses()).not.toContain("other@local");
      expect(router.routeMail("agent@local", "hello")).toBe(true);
      expect(ws2.sent).toHaveLength(1);
    });
  });

  describe("register key-existence gate", () => {
    // A register frame is token-authenticated but proves no per-address
    // ownership, so it may route only a KEYLESS first-deploy address. An
    // address that already has a stored key must prove ownership through the
    // challenged reconnect path; register must neither route it to the caller
    // nor disturb its existing owner.
    const KEYED = "victim@local";

    function gatedRouter(publicKeyHex: string | null) {
      return createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === KEYED ? publicKeyHex : null,
        },
      });
    }

    // Bring KEYED up as owned by `ws` through the challenged reconnect path.
    async function reconnectVerify(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      privateKey: Uint8Array,
    ) {
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "owner",
          token: "tok",
          agentAddresses: [KEYED],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
      const challenge = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address: KEYED,
              signature: await signChallenge(
                challenge.challenges[0].nonce,
                KEYED,
                privateKey,
              ),
            },
          ],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    test("routes a keyless first-deploy address", async () => {
      const r = gatedRouter(null);
      const ws = createMockWs();
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["fresh@local"],
        }),
      );
      await tick();

      expect(r.getRoutableAddresses()).toContain("fresh@local");
      expect(r.routeMail("fresh@local", "dGVzdA==")).toBe(true);
    });

    test("refuses a keyed address and leaves its owner's route untouched", async () => {
      const kp = await generateKeyPair();
      const r = gatedRouter(hexEncode(kp.publicKey));

      // The true owner establishes KEYED via the challenged reconnect.
      const ownerWs = createMockWs();
      await reconnectVerify(r, ownerWs, kp.privateKey);
      expect(r.getRoutableAddresses()).toContain(KEYED);

      // A rogue sidecar with a valid token names KEYED in a register frame.
      const rogueWs = createMockWs();
      r.handleOpen(rogueWs);
      r.handleMessage(
        rogueWs,
        JSON.stringify({
          type: "register",
          sidecarId: "rogue",
          token: "tok",
          agentAddresses: [KEYED],
        }),
      );
      await tick();

      // KEYED is still routed to the owner, not the rogue: the gate refused to
      // route it AND -- because it runs before the ghost-cleanup -- never
      // evicted the owner. No hijack, and no downgrade to a denial of service.
      expect(r.getRoutableAddresses()).toContain(KEYED);
      ownerWs.sent.length = 0;
      rogueWs.sent.length = 0;
      expect(r.routeMail(KEYED, "dGVzdA==")).toBe(true);
      expect(ownerWs.sent).toHaveLength(1);
      expect(rogueWs.sent).toHaveLength(0);
    });

    test("a re-register does not drop an address already verified via reconnect", async () => {
      // A connection that proved a keyed address via challenged reconnect keeps
      // that route across a later register, even though the gate would refuse
      // to re-add a keyed address: additive re-register inherits the owned set
      // rather than replacing it. Removal is via undeploy/disconnect, not
      // register-omission.
      const kp = await generateKeyPair();
      const r = gatedRouter(hexEncode(kp.publicKey));
      const ws = createMockWs();
      await reconnectVerify(r, ws, kp.privateKey);
      expect(r.getRoutableAddresses()).toContain(KEYED);

      // The same ws registers a keyless first-deploy. The keyed
      // reconnect-verified route must survive alongside the new one.
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "owner",
          token: "tok",
          agentAddresses: ["fresh@local"],
        }),
      );
      await tick();

      expect(r.getRoutableAddresses().sort()).toEqual(
        [KEYED, "fresh@local"].sort(),
      );
      expect(r.routeMail(KEYED, "dGVzdA==")).toBe(true);
    });

    test("fails closed and surfaces an error when no lookup is configured", async () => {
      const captured: { level: string; message: string }[] = [];
      const savedConfig = getConfig();
      configureSync({
        reset: true,
        sinks: {
          capture: (record) => {
            const message = Array.isArray(record.message)
              ? record.message
                  .map((part) =>
                    typeof part === "string" ? part : JSON.stringify(part),
                  )
                  .join("")
              : String(record.message);
            captured.push({ level: record.level, message });
          },
        },
        loggers: [{ category: [], lowestLevel: "debug", sinks: ["capture"] }],
      });
      try {
        // No lookups configured: the gate cannot distinguish a keyed address
        // from a first-deploy, so it must route nothing rather than permit.
        const r = createTestRouter({
          requestTimeoutMs: 500,
          hubPublicKey: TEST_HUB_KEY,
        });
        const ws = createMockWs();
        r.handleOpen(ws);
        r.handleMessage(
          ws,
          JSON.stringify({
            type: "register",
            sidecarId: "sc-1",
            token: "tok",
            agentAddresses: ["x@local"],
          }),
        );
        await tick();

        expect(r.getRoutableAddresses()).toEqual([]);
        expect(r.routeMail("x@local", "dGVzdA==")).toBe(false);
        // The missing dependency surfaces as an error, not a silent permit.
        expect(
          captured.some(
            (l) =>
              l.level === "error" &&
              l.message.includes("lookupPublicKey is not configured"),
          ),
        ).toBe(true);
      } finally {
        if (savedConfig) {
          configureSync({ reset: true, ...savedConfig });
        } else {
          resetSync();
        }
      }
    });

    test("fails closed on a key-lookup error instead of crashing", async () => {
      // A rejecting lookup (e.g. a transient DB failure) must be caught and
      // surfaced, not floated out of the void-dispatched handler as an
      // unhandled rejection that could take down the hub. Fail closed: route
      // nothing and log an error. (The test completing rather than hanging on
      // an unhandled rejection is itself part of the assertion.)
      const captured: { level: string; message: string }[] = [];
      const savedConfig = getConfig();
      configureSync({
        reset: true,
        sinks: {
          capture: (record) => {
            const message = Array.isArray(record.message)
              ? record.message
                  .map((part) =>
                    typeof part === "string" ? part : JSON.stringify(part),
                  )
                  .join("")
              : String(record.message);
            captured.push({ level: record.level, message });
          },
        },
        loggers: [{ category: [], lowestLevel: "debug", sinks: ["capture"] }],
      });
      try {
        const r = createTestRouter({
          requestTimeoutMs: 500,
          hubPublicKey: TEST_HUB_KEY,
          lookups: {
            lookupPublicKey: async () => {
              throw new Error("db unavailable");
            },
          },
        });
        const ws = createMockWs();
        r.handleOpen(ws);
        r.handleMessage(
          ws,
          JSON.stringify({
            type: "register",
            sidecarId: "sc-1",
            token: "tok",
            agentAddresses: ["x@local"],
          }),
        );
        await tick();

        expect(r.getRoutableAddresses()).toEqual([]);
        expect(
          captured.some(
            (l) =>
              l.level === "error" && l.message.includes("Key lookup failed"),
          ),
        ).toBe(true);
      } finally {
        if (savedConfig) {
          configureSync({ reset: true, ...savedConfig });
        } else {
          resetSync();
        }
      }
    });
  });

  describe("frame dispatch serialization (F1)", () => {
    const STATE = {
      threadRoot: "<root@example.com>",
      lastMessageId: "<last@example.com>",
      replyTo: "user@example.com",
      cc: [],
    };

    async function driveReconnect(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      addr: string,
      privateKey: Uint8Array,
    ) {
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc",
          token: "tok",
          agentAddresses: [addr],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
      const challenge = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address: addr,
              signature: await signChallenge(
                challenge.challenges[0].nonce,
                addr,
                privateKey,
              ),
            },
          ],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    test("a dependent frame after a non-empty register is processed, not dropped", async () => {
      // The register key-existence gate awaits lookupPublicKey, so routing lands
      // asynchronously. Per-ws serialization makes a following dependent frame
      // (connector.state.changed for the just-registered keyless address) wait
      // for that routing rather than be dropped because addressIndex was empty.
      const r = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      r.handleOpen(ws);
      // Non-empty register immediately followed by a dependent frame, with NO
      // await between -- the follower must still observe the route.
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc",
          token: "tok",
          agentAddresses: ["fresh@local"],
        }),
      );
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "fresh@local",
          connectorState: STATE,
        }),
      );
      await tick();

      expect(r.getConnectorState("fresh@local")).toEqual(STATE);
    });

    test("a rogue's dependent frame for a keyed victim is dropped, not poisoned", async () => {
      // Serialization must NOT reintroduce the poisoning it exists to prevent: a
      // rogue's non-empty register naming a KEYED victim is refused by the gate,
      // so a connector.state.changed from the rogue in the same window still
      // sees addressIndex.get(victim) !== rogueWs and is dropped; the victim's
      // cached state is untouched.
      const kp = await generateKeyPair();
      const victim = "victim@local";
      const r = createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === victim ? hexEncode(kp.publicKey) : null,
        },
      });
      const ownerWs = createMockWs();
      r.handleOpen(ownerWs);
      await driveReconnect(r, ownerWs, victim, kp.privateKey);
      r.handleMessage(
        ownerWs,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: victim,
          connectorState: STATE,
        }),
      );
      await tick();
      expect(r.getConnectorState(victim)).toEqual(STATE);

      // Rogue: register naming the keyed victim + a poisoning frame, no await.
      const rogueWs = createMockWs();
      r.handleOpen(rogueWs);
      r.handleMessage(
        rogueWs,
        JSON.stringify({
          type: "register",
          sidecarId: "rogue",
          token: "tok",
          agentAddresses: [victim],
        }),
      );
      r.handleMessage(
        rogueWs,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: victim,
          connectorState: { ...STATE, replyTo: "attacker@evil" },
        }),
      );
      await tick();

      // Not poisoned: still the owner's state.
      expect(r.getConnectorState(victim)).toEqual(STATE);
    });

    test("a reconnect whose reaction awaits a later session.ack does not deadlock", async () => {
      // handleChallengeResponse awaits the agent.reconnected reaction, whose
      // subscriber issues sendSourcesUpdate -> awaits a LATER session.ack frame.
      // session.ack must BYPASS the per-ws chain, or the reconnect wedges behind
      // its own in-flight handler. (The test completing rather than timing out
      // is the assertion.)
      const kp = await generateKeyPair();
      const addr = "agent@local";
      const r = createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async (a) =>
            a === addr ? hexEncode(kp.publicKey) : null,
        },
      });
      let reactionDone = false;
      r.events.on("agent.reconnected", async () => {
        await r.sendSourcesUpdate(addr, TEST_SOURCES, TEST_DEFAULT_SOURCE);
        reactionDone = true;
      });
      const ws = createMockWs();
      r.handleOpen(ws);
      await driveReconnect(r, ws, addr, kp.privateKey);

      // The reaction has sent a sources.update request and is awaiting its ack.
      const sourcesFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "sources.update");
      expect(sourcesFrame).toBeDefined();

      // Answer with a session.ack (a later inbound frame). If it queued behind
      // the still-in-flight handleChallengeResponse it would never resolve.
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "session.ack",
          requestId: sourcesFrame.requestId,
        }),
      );
      await new Promise((res) => setTimeout(res, 50));

      expect(reactionDone).toBe(true);
      expect(r.getRoutableAddresses()).toContain(addr);
    });
  });

  describe("workflow-address reconnect (challenged, at parity with launched agents)", () => {
    // A workflow-substrate deployment address (ins_dep_...) proves ownership
    // through the SAME Ed25519 challenge as a launched agent: the hub resolves
    // the deployment's public key, issues a nonce, and routes the address only
    // after a valid signature. These tests pin that parity -- there is no
    // keyless register-field shortcut for a workflow address anymore, so a
    // token-holding sidecar cannot reclaim a deployment's route without the
    // deployment's own key.
    const WF_ADDR = "ins_dep_abc@local";

    // Router whose key lookup resolves WF_ADDR to `publicKeyHex` (a live
    // deployment) or to null (unknown / torn-down), mirroring the
    // launched-agent reconnect tests.
    function workflowReconnectRouter(publicKeyHex: string | null) {
      return createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === WF_ADDR ? publicKeyHex : null,
        },
      });
    }

    function findFrame(ws: ReturnType<typeof createMockWs>, type: string) {
      return ws.sent.map((s) => JSON.parse(s)).find((f) => f.type === type);
    }

    async function reconnect(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
    ) {
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [WF_ADDR],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    async function respondToChallenge(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      privateKey: Uint8Array,
    ) {
      const { address, nonce } = findFrame(ws, "challenge").challenges[0];
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address,
              signature: await signChallenge(nonce, address, privateKey),
            },
          ],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    test("routes a workflow address only after a passed challenge", async () => {
      const kp = await generateKeyPair();
      const r = workflowReconnectRouter(hexEncode(kp.publicKey));
      const ws = createMockWs();
      await reconnect(r, ws);

      // The reconnect frame alone earns nothing: the address is not routable
      // until the challenge is answered.
      expect(r.getRoutableAddresses()).not.toContain(WF_ADDR);
      const challenge = findFrame(ws, "challenge");
      expect(challenge).toBeDefined();
      expect(challenge.challenges[0].address).toBe(WF_ADDR);

      await respondToChallenge(r, ws, kp.privateKey);

      expect(r.getRoutableAddresses()).toContain(WF_ADDR);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(true);
    });

    test("rejects a workflow address signed with the wrong key (no hijack)", async () => {
      const kp = await generateKeyPair();
      const wrongKp = await generateKeyPair();
      const r = workflowReconnectRouter(hexEncode(kp.publicKey));
      const ws = createMockWs();
      await reconnect(r, ws);

      await respondToChallenge(r, ws, wrongKp.privateKey);

      expect(r.getRoutableAddresses()).not.toContain(WF_ADDR);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(false);
      expect(findFrame(ws, "challenge.failed").address).toBe(WF_ADDR);
    });

    test("fails closed when the deployment has no live key", async () => {
      // lookupPublicKey resolves null -- an unknown or torn-down deployment.
      // No challenge is issued and the address never routes.
      const r = workflowReconnectRouter(null);
      const ws = createMockWs();
      await reconnect(r, ws);

      expect(r.getRoutableAddresses()).not.toContain(WF_ADDR);
      expect(findFrame(ws, "challenge")).toBeUndefined();
      expect(findFrame(ws, "challenge.failed").address).toBe(WF_ADDR);
    });

    test("disconnect removes a verified workflow address from routing", async () => {
      const kp = await generateKeyPair();
      const r = workflowReconnectRouter(hexEncode(kp.publicKey));
      const ws = createMockWs();
      await reconnect(r, ws);
      await respondToChallenge(r, ws, kp.privateKey);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(true);

      r.handleClose(ws);

      expect(r.getRoutableAddresses()).toEqual([]);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(false);
    });

    test("a fresh ws reclaims a workflow address only by passing its own challenge", async () => {
      const kp = await generateKeyPair();
      const r = workflowReconnectRouter(hexEncode(kp.publicKey));

      const oldWs = createMockWs();
      await reconnect(r, oldWs);
      await respondToChallenge(r, oldWs, kp.privateKey);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(true);

      const newWs = createMockWs();
      await reconnect(r, newWs);
      await respondToChallenge(r, newWs, kp.privateKey);

      // The stale ws's later close must not clobber the new owner (the
      // ownership guard in handleClose).
      r.handleClose(oldWs);

      expect(r.getRoutableAddresses()).toContain(WF_ADDR);
      expect(r.routeMail(WF_ADDR, "dGVzdA==")).toBe(true);
    });
  });

  describe("per-step route bind/unbind", () => {
    const DEPLOYMENT_ADDR = "ins_dep_multi@local";
    const STEP_ADDR = "ins_dep_multi-step1@local";
    const ZERO_SHA = "0".repeat(40);

    async function registerDeploymentSidecar(): Promise<
      ReturnType<typeof createMockWs>
    > {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [DEPLOYMENT_ADDR],
        }),
      );
      // Register routing is async (the key-existence gate awaits the lookup);
      // settle it before assertions read the routing table.
      await new Promise((res) => setTimeout(res, 0));
      return ws;
    }

    test("bind makes a step address routable; unbind removes it", async () => {
      await registerDeploymentSidecar();

      // An unbound step address has no route: `sendPack` rejects before
      // touching the wire.
      await expect(
        router.sendPack(
          STEP_ADDR,
          new Uint8Array([1]),
          "refs/heads/main",
          ZERO_SHA,
        ),
      ).rejects.toThrow(/No sidecar connected/);

      router.bindStepRoute(STEP_ADDR);
      expect(router.getRoutableAddresses()).toContain(STEP_ADDR);
      // Routable: `routeMail` (and `sendPack`) now resolve the sidecar.
      expect(router.routeMail(STEP_ADDR, "dGVzdA==")).toBe(true);

      router.unbindStepRoute(STEP_ADDR);
      expect(router.getRoutableAddresses()).not.toContain(STEP_ADDR);
      await expect(
        router.sendPack(
          STEP_ADDR,
          new Uint8Array([1]),
          "refs/heads/main",
          ZERO_SHA,
        ),
      ).rejects.toThrow(/No sidecar connected/);
    });

    test("a reconnect after unbind does not resurrect the transient route", async () => {
      // The deployment address is re-announced on reconnect and proves
      // ownership through the challenge; the transient per-step route
      // (bind/unbind) is never persisted into that announcement, so a
      // reconnect must not bring it back.
      const kp = await generateKeyPair();
      const reconnectRouter = createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === DEPLOYMENT_ADDR ? hexEncode(kp.publicKey) : null,
        },
      });
      const ws = createMockWs();
      reconnectRouter.handleOpen(ws);

      // Bring the deployment address up through the challenged reconnect path.
      async function reconnectDeployment() {
        reconnectRouter.handleMessage(
          ws,
          JSON.stringify({
            type: "reconnect",
            sidecarId: "sc-1",
            token: "tok",
            agentAddresses: [DEPLOYMENT_ADDR],
          }),
        );
        await new Promise((r) => setTimeout(r, 50));
        const challenges = ws.sent
          .map((s) => JSON.parse(s))
          .filter((f) => f.type === "challenge");
        const { address, nonce } =
          challenges[challenges.length - 1].challenges[0];
        reconnectRouter.handleMessage(
          ws,
          JSON.stringify({
            type: "challenge.response",
            responses: [
              {
                address,
                signature: await signChallenge(nonce, address, kp.privateKey),
              },
            ],
          }),
        );
        await new Promise((r) => setTimeout(r, 50));
      }

      await reconnectDeployment();
      expect(reconnectRouter.getRoutableAddresses()).toContain(DEPLOYMENT_ADDR);

      reconnectRouter.bindStepRoute(STEP_ADDR);
      reconnectRouter.unbindStepRoute(STEP_ADDR);

      // The sidecar reconnects announcing only the deployment address -- never
      // the transient per-step address.
      await reconnectDeployment();

      // The reconnect actually ran (it did not bail and close the socket).
      expect(ws.closed).toBe(false);
      // The deployment address is routable again; the transient step route
      // stays gone.
      expect(reconnectRouter.getRoutableAddresses()).toContain(DEPLOYMENT_ADDR);
      expect(reconnectRouter.getRoutableAddresses()).not.toContain(STEP_ADDR);
    });

    test("bind throws when no sidecar is connected", () => {
      expect(() => router.bindStepRoute(STEP_ADDR)).toThrow(/No sidecar/);
    });

    test("unbind is a no-op for an address that was never bound", async () => {
      await registerDeploymentSidecar();
      expect(() => router.unbindStepRoute(STEP_ADDR)).not.toThrow();
      expect(router.getRoutableAddresses()).not.toContain(STEP_ADDR);
    });

    test("handleClose reclaims a still-bound step route", async () => {
      const ws = await registerDeploymentSidecar();
      router.bindStepRoute(STEP_ADDR);
      expect(router.getRoutableAddresses()).toContain(STEP_ADDR);

      // A sidecar drop mid-stage tears the transient binding down with the
      // rest of the connection's addresses -- no stale route survives.
      router.handleClose(ws);
      expect(router.getRoutableAddresses()).not.toContain(STEP_ADDR);
    });
  });

  describe("mail routing", () => {
    test("routes mail between two sidecars", async () => {
      const ws1 = createMockWs();
      const ws2 = createMockWs();
      router.handleOpen(ws1);
      router.handleOpen(ws2);

      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["sender@local"],
        }),
      );
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-2",
          token: "tok",
          agentAddresses: ["receiver@local"],
        }),
      );
      await tick();

      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "mail.outbound",
          rawMessage: "dGVzdA==",
          recipients: ["receiver@local"],
        }),
      );
      await tick();

      const delivered = lastSent(ws2);
      expect(delivered.type).toBe("mail.inbound");
      expect(delivered.agentAddress).toBe("receiver@local");
      expect(delivered.rawMessage).toBe("dGVzdA==");
    });

    test("routeMail returns false for unknown address", () => {
      expect(router.routeMail("nobody@local", "dGVzdA==")).toBe(false);
    });

    test("routeMail returns true for routable address", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      expect(router.routeMail("agent@local", "dGVzdA==")).toBe(true);
      const delivered = lastSent(ws);
      expect(delivered.type).toBe("mail.inbound");
    });

    test("unroutable mail emits mail.outbound.undelivered", async () => {
      const outbound: { rawMessage: string; recipients: string[] }[] = [];
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      router.events.on("mail.outbound.undelivered", (event) => {
        outbound.push({
          rawMessage: event.rawMessage,
          recipients: event.recipients,
        });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["sender@local"],
        }),
      );

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "mail.outbound",
          rawMessage: "dGVzdA==",
          recipients: ["external@remote"],
        }),
      );
      await tick();

      expect(outbound).toHaveLength(1);
      expect(outbound[0]?.recipients).toEqual(["external@remote"]);
    });

    test("agent reply to non-agent address still persists outbound record", async () => {
      // When an agent replies to a human user (usr_ address), the
      // persistMail lookup must still persist at least the outbound
      // record on the sender's session and emit mail.persisted so a
      // mail.delivered SSE event is dispatched.
      const persisted: { id: string; address: string }[] = [];
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          persistMail: async ({ senderAddress, recipients }) => {
            // Mirrors the fixed persistMail: always create the
            // outbound record for the sender, and only create
            // inbound records for recipients that are agent
            // instances (ins_ prefix).
            const results: {
              id: string;
              direction: "inbound" | "outbound";
              instanceId: string | null;
              address: string;
              createdAt: Date;
            }[] = [
              {
                id: "mail_outbound",
                direction: "outbound",
                instanceId:
                  parseAgentAddress(senderAddress)?.instanceId ?? senderAddress,
                address: senderAddress,
                createdAt: new Date(),
              },
            ];
            for (const addr of recipients) {
              if (addr.startsWith("ins_")) {
                results.push({
                  id: `mail_in_${addr}`,
                  direction: "inbound",
                  instanceId: parseAgentAddress(addr)?.instanceId ?? addr,
                  address: addr,
                  createdAt: new Date(),
                });
              }
            }
            return results;
          },
        },
      });
      router.events.on("mail.persisted", (row) => {
        persisted.push({ id: row.id, address: row.address });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["ins_sender@tenant.example"],
        }),
      );

      // Agent sends a reply to a human user address.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "mail.outbound",
          delivered: true,
          senderAddress: "ins_sender@tenant.example",
          rawMessage: btoa("test message"),
          recipients: ["usr_human@tenant.example"],
        }),
      );

      // Allow the async handleMailPersist to settle.
      await new Promise((resolve) => setTimeout(resolve, 50));

      // The outbound record for the sender must always be persisted,
      // regardless of whether recipients are agent instances. No inbound
      // record should be created for the non-agent recipient.
      expect(persisted).toHaveLength(1);
      expect(persisted[0]?.address).toBe("ins_sender@tenant.example");
    });

    test("persistMail lookup result fans out as mail.persisted events", async () => {
      const persisted: { id: string; address: string }[] = [];
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          persistMail: async ({ senderAddress, recipients }) => {
            return [
              {
                id: "mail_out",
                direction: "outbound" as const,
                instanceId:
                  parseAgentAddress(senderAddress)?.instanceId ?? senderAddress,
                address: senderAddress,
                createdAt: new Date(),
              },
              ...recipients.map((addr) => ({
                id: `mail_in_${addr}`,
                direction: "inbound" as const,
                instanceId: parseAgentAddress(addr)?.instanceId ?? addr,
                address: addr,
                createdAt: new Date(),
              })),
            ];
          },
        },
      });
      router.events.on("mail.persisted", (row) => {
        persisted.push({ id: row.id, address: row.address });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["ins_sender@tenant.example"],
        }),
      );

      // Agent sends mail to another agent (both are instances).
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "mail.outbound",
          delivered: true,
          senderAddress: "ins_sender@tenant.example",
          rawMessage: btoa("test message"),
          recipients: ["ins_receiver@tenant.example"],
        }),
      );

      await new Promise((resolve) => setTimeout(resolve, 50));

      // Both outbound (sender) and inbound (receiver) records persisted.
      expect(persisted).toHaveLength(2);
      expect(persisted[0]?.address).toBe("ins_sender@tenant.example");
      expect(persisted[1]?.address).toBe("ins_receiver@tenant.example");
    });
  });

  describe("mail-triggered run grants", () => {
    const WORKFLOW_ADDR = "ins_dep_wf1@tenant.example";
    const AGENT_ADDR = "ins_agent1@tenant.example";

    function mailWithMessageId(messageId: string): string {
      const raw = [
        "From: user@tenant.example",
        `To: ${WORKFLOW_ADDR}`,
        `Message-ID: ${messageId}`,
        "",
        "trigger body",
      ].join("\r\n");
      return btoa(raw);
    }

    const SAMPLE_GRANTS: RunGrantsFrame["stepGrants"] = [
      {
        id: "grant-1",
        resource: "tool:x",
        action: "invoke",
        effect: "allow",
        origin: "creator",
        conditions: null,
        expiresAt: null,
        roleId: null,
        principalId: "prin-run",
      },
    ];

    async function connectRecipient(
      router: ReturnType<typeof createSidecarRouter>,
      address: string,
    ): Promise<ReturnType<typeof createMockWs>> {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-recipient",
          token: "tok",
          agentAddresses: [address],
        }),
      );
      await tick();
      return ws;
    }

    async function sendOutbound(
      router: ReturnType<typeof createSidecarRouter>,
      rawMessage: string,
      recipients: string[],
    ): Promise<void> {
      const sender = createMockWs();
      router.handleOpen(sender);
      router.handleMessage(
        sender,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-outbound",
          token: "tok",
          agentAddresses: ["ins_dep_sender@tenant.example"],
        }),
      );
      await tick();
      await router.handleMessage(
        sender,
        JSON.stringify({ type: "mail.outbound", rawMessage, recipients }),
      );
      await tick();
    }

    test("materializes grants and sends them before the inbound mail for a workflow recipient", async () => {
      const calls: { agentAddress: string; runId: string }[] = [];
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async (args) => {
            calls.push(args);
            return {
              outcome: "materialized",
              stepGrants: SAMPLE_GRANTS,
            };
          },
        },
      });
      const ws = await connectRecipient(router, WORKFLOW_ADDR);

      await sendOutbound(
        router,
        mailWithMessageId("<mail-run-1@tenant.example>"),
        [WORKFLOW_ADDR],
      );

      // The lookup was invoked with the derived runId: the deployment's mail
      // address (the stable runId), not this mail's Message-ID.
      expect(calls).toEqual([
        { agentAddress: WORKFLOW_ADDR, runId: WORKFLOW_ADDR },
      ]);
      // The recipient received the run.grants frame BEFORE the mail.inbound.
      const frames = ws.sent.map((s) => JSON.parse(s));
      expect(frames[0]?.type).toBe("run.grants");
      expect(frames[0]?.runId).toBe(WORKFLOW_ADDR);
      expect(frames[0]?.stepGrants).toEqual(SAMPLE_GRANTS);
      expect(frames[1]?.type).toBe("mail.inbound");
      // The run-committing mail goes through the messageId handshake (routeMail),
      // NOT a fire-and-forget send: the mail.inbound carries a messageId, so a
      // connected-window drop before the ack is redelivered on reconnect and
      // the committed run cannot be left bodiless.
      expect(frames[1]?.messageId).toBeDefined();
    });

    test("does not materialize grants for a non-workflow recipient", async () => {
      let called = false;
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async () => {
            called = true;
            return { outcome: "skip" };
          },
        },
      });
      const ws = await connectRecipient(router, AGENT_ADDR);

      await sendOutbound(router, btoa("From: x\r\nTo: y\r\n\r\nbody"), [
        AGENT_ADDR,
      ]);

      expect(called).toBe(false);
      // The mail still forwarded, with no preceding run.grants frame.
      const frames = ws.sent.map((s) => JSON.parse(s));
      expect(frames.map((f) => f.type)).toEqual(["mail.inbound"]);
    });

    test("forwards the mail without grants when the lookup skips", async () => {
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async () => ({ outcome: "skip" }),
        },
      });
      const ws = await connectRecipient(router, WORKFLOW_ADDR);

      await sendOutbound(
        router,
        mailWithMessageId("<mail-run-skip@tenant.example>"),
        [WORKFLOW_ADDR],
      );

      const frames = ws.sent.map((s) => JSON.parse(s));
      expect(frames.map((f) => f.type)).toEqual(["mail.inbound"]);
    });

    test("fails a rejected run closed: no grants, no mail, no commit", async () => {
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async () => ({
            outcome: "rejected",
            status: 403,
            code: "grant_requirement_unsatisfied",
            message: "creator lacks secret:vault/use",
          }),
        },
      });
      const ws = await connectRecipient(router, WORKFLOW_ADDR);

      await sendOutbound(
        router,
        mailWithMessageId("<mail-run-reject@tenant.example>"),
        [WORKFLOW_ADDR],
      );

      // The run is fail-closed: the recipient sees neither the run.grants
      // frame nor the mail that would dispatch an under-authorized run.
      expect(ws.sent).toHaveLength(0);
    });

    test("isolates a rejected recipient from a co-recipient's delivery", async () => {
      // One workflow recipient is rejected; a co-recipient agent address must
      // still receive its mail. (A mail with more than one workflow-derived
      // recipient is separately rejected wholesale, so the co-recipient here
      // is a non-workflow address.)
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async (args) =>
            args.agentAddress === WORKFLOW_ADDR
              ? {
                  outcome: "rejected",
                  status: 403,
                  code: "grant_requirement_unsatisfied",
                  message: "creator lacks authority",
                }
              : { outcome: "skip" },
        },
      });
      const wfWs = await connectRecipient(router, WORKFLOW_ADDR);
      const agentWs = await connectRecipient(router, AGENT_ADDR);

      await sendOutbound(
        router,
        mailWithMessageId("<mail-run-corec@tenant.example>"),
        [WORKFLOW_ADDR, AGENT_ADDR],
      );

      // The rejected workflow recipient got nothing; the co-recipient still
      // received its mail.
      expect(wfWs.sent).toHaveLength(0);
      const agentFrames = agentWs.sent.map((s) => JSON.parse(s));
      expect(agentFrames.map((f) => f.type)).toEqual(["mail.inbound"]);
    });

    test("rejects a mail addressed to more than one workflow recipient", async () => {
      let called = false;
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async () => {
            called = true;
            return { outcome: "skip" };
          },
        },
      });
      const wfWs = await connectRecipient(router, WORKFLOW_ADDR);
      const otherWfAddr = "ins_dep_wf2@tenant.example";
      const otherWs = await connectRecipient(router, otherWfAddr);

      await sendOutbound(
        router,
        mailWithMessageId("<mail-run-multi@tenant.example>"),
        [WORKFLOW_ADDR, otherWfAddr],
      );

      // The one-workflow-recipient-per-mail restriction is refused loudly
      // before any materialization; neither workflow recipient is delivered to.
      expect(called).toBe(false);
      expect(wfWs.sent).toHaveLength(0);
      expect(otherWs.sent).toHaveLength(0);
    });

    // Connect a keyed recipient through the challenged reconnect path so a
    // later reconnect (which re-verifies ownership) can flush retained pending
    // mail -- the register path leaves a keyed address unrouted until it passes
    // a challenge, and redelivery fires only on the verified reconnect.
    async function connectRecipientViaChallenge(
      router: ReturnType<typeof createSidecarRouter>,
      address: string,
      privateKey: Uint8Array,
      sidecarId = "sc-recipient",
    ): Promise<ReturnType<typeof createMockWs>> {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId,
          token: "tok",
          agentAddresses: [address],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, privateKey),
          }),
        ),
      );
      router.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );
      await new Promise((res) => setTimeout(res, 50));
      return ws;
    }

    test("a run-committing mail-relay is retained and redelivered on reconnect after a connected-window drop", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        // Large so neither the connected-window retry nor the retention TTL
        // fires during the test; the redelivery under test is reconnect-driven.
        mailAckRetryIntervalMs: 10_000,
        disconnectQueueTTLMs: 60_000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === WORKFLOW_ADDR ? hexEncode(kp.publicKey) : null,
          materializeMailTriggeredRunGrants: async () => ({
            outcome: "materialized",
            stepGrants: SAMPLE_GRANTS,
          }),
        },
      });

      const ws1 = await connectRecipientViaChallenge(
        router,
        WORKFLOW_ADDR,
        kp.privateKey,
      );
      await sendOutbound(
        router,
        mailWithMessageId("<mail-relay-redeliver@tenant.example>"),
        [WORKFLOW_ADDR],
      );

      // Delivered over the live connection via the handshake.
      const firstInbound = ws1.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "mail.inbound");
      expect(firstInbound).toBeDefined();
      expect(firstInbound.messageId).toBeDefined();

      // Drop BEFORE any ack: the pending relay mail must be retained.
      router.handleClose(ws1);

      // Reconnect: the retained relay mail is redelivered (identical bytes,
      // same messageId) so the committed run's body is not left un-triggered.
      const ws2 = await connectRecipientViaChallenge(
        router,
        WORKFLOW_ADDR,
        kp.privateKey,
        "sc-recipient-2",
      );
      const redelivered = ws2.sent
        .map((s) => JSON.parse(s))
        .find(
          (f) =>
            f.type === "mail.inbound" && f.messageId === firstInbound.messageId,
        );
      expect(redelivered).toBeDefined();
      expect(redelivered.rawMessage).toBe(firstInbound.rawMessage);
      // No re-materialization on redelivery; downstream dedup makes the
      // replayed delivery effectively-once.
    });

    test("redelivery re-emits the run's grants ahead of the mail on reconnect", async () => {
      const kp = await generateKeyPair();
      let materializations = 0;
      const router = createTestRouter({
        requestTimeoutMs: 500,
        // Large so neither the connected-window retry nor the retention TTL
        // fires during the test; the redelivery under test is reconnect-driven.
        mailAckRetryIntervalMs: 10_000,
        disconnectQueueTTLMs: 60_000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === WORKFLOW_ADDR ? hexEncode(kp.publicKey) : null,
          materializeMailTriggeredRunGrants: async () => {
            materializations += 1;
            return {
              outcome: "materialized",
              stepGrants: SAMPLE_GRANTS,
            };
          },
        },
      });

      const ws1 = await connectRecipientViaChallenge(
        router,
        WORKFLOW_ADDR,
        kp.privateKey,
      );
      await sendOutbound(
        router,
        mailWithMessageId("<mail-grants-redeliver@tenant.example>"),
        [WORKFLOW_ADDR],
      );
      const firstInbound = ws1.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "mail.inbound");
      expect(firstInbound).toBeDefined();

      // Drop BEFORE any ack. The run.grants frame the sidecar first saw is lost
      // with the connection; the run's grants are NOT re-fetched anywhere, so
      // without replay the redelivered trigger would run with no grants and
      // fail its onRunStart barrier closed on a hub-committed run.
      router.handleClose(ws1);

      const ws2 = await connectRecipientViaChallenge(
        router,
        WORKFLOW_ADDR,
        kp.privateKey,
        "sc-recipient-2",
      );
      const frames = ws2.sent.map((s) => JSON.parse(s));
      const grantsIdx = frames.findIndex(
        (f) => f.type === "run.grants" && f.runId === WORKFLOW_ADDR,
      );
      const mailIdx = frames.findIndex(
        (f) =>
          f.type === "mail.inbound" && f.messageId === firstInbound.messageId,
      );
      expect(grantsIdx).toBeGreaterThanOrEqual(0);
      expect(mailIdx).toBeGreaterThanOrEqual(0);
      // The run.grants lands AHEAD of the redelivered mail (same-connection
      // FIFO), so the redelivered run resolves its grants instead of failing
      // closed on the barrier.
      expect(grantsIdx).toBeLessThan(mailIdx);
      // The replayed snapshot is the SAME materialized bytes, not a re-fetch:
      // materialization ran exactly once, at the original delivery.
      expect(frames[grantsIdx]?.stepGrants).toEqual(SAMPLE_GRANTS);
      expect(materializations).toBe(1);
    });

    test("a skip-path mail-relay is forwarded without the messageId handshake", async () => {
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          materializeMailTriggeredRunGrants: async () => ({ outcome: "skip" }),
        },
      });
      const ws = await connectRecipient(router, WORKFLOW_ADDR);
      await sendOutbound(
        router,
        mailWithMessageId("<mail-relay-skip@tenant.example>"),
        [WORKFLOW_ADDR],
      );

      const inbound = ws.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "mail.inbound");
      expect(inbound).toBeDefined();
      // The skip path commits no run, so it is forwarded fire-and-forget --
      // no messageId, no ack handshake, no redelivery tracking.
      expect(inbound.messageId).toBeUndefined();
    });
  });

  describe("agent lifecycle", () => {
    test("agent.deploy sends frame and resolves on ack", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "new-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const promise = router.sendAgentDeploy("new-agent@local", config);

      const frame = lastSent(ws);
      expect(frame.type).toBe("agent.deploy");
      expect(frame.config.agentAddress).toBe("new-agent@local");

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "new-agent@local",
          publicKey: "deadbeef",
        }),
      );

      await promise;
      expect(router.getRoutableAddresses()).toContain("new-agent@local");
    });

    test("agent.deploy ignores an acknowledgement from another connection", async () => {
      const primary = createMockWs();
      const other = createMockWs();
      for (const [ws, sidecarId] of [
        [primary, "sc-primary"],
        [other, "sc-other"],
      ] as const) {
        router.handleOpen(ws);
        router.handleMessage(
          ws,
          JSON.stringify({
            type: "register",
            sidecarId,
            token: "tok",
            agentAddresses: [],
          }),
        );
      }
      await tick();

      const promise = router.sendAgentDeploy("connection-bound@local", {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "connection-bound@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: TEST_SOURCES,
        defaultSource: TEST_DEFAULT_SOURCE,
      });
      expect(lastSent(primary).type).toBe("agent.deploy");

      let settled = false;
      void promise.finally(() => {
        settled = true;
      });
      router.handleMessage(
        other,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "connection-bound@local",
          publicKey: "wrong-connection-key",
        }),
      );
      await tick();
      expect(settled).toBe(false);

      router.handleMessage(
        primary,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "connection-bound@local",
          publicKey: "primary-key",
        }),
      );
      await expect(promise).resolves.toEqual({ publicKey: "primary-key" });
    });

    test("agent.deploy.ack invokes subscribers before resolving", async () => {
      const ackCalls: { address: string; publicKey: string }[] = [];
      const router = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
      });
      router.events.on("agent.deploy.ack", ({ agentAddress, publicKey }) => {
        ackCalls.push({ address: agentAddress, publicKey });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "ack-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const promise = router.sendAgentDeploy("ack-agent@local", config);

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "ack-agent@local",
          publicKey: "aabbccdd",
        }),
      );

      await promise;
      expect(ackCalls).toEqual([
        { address: "ack-agent@local", publicKey: "aabbccdd" },
      ]);
    });

    test("agent.deploy rejects when agent.deploy.ack subscriber throws", async () => {
      const router = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
      });
      router.events.on("agent.deploy.ack", () => {
        throw new Error("DB write failed");
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "fail-ack@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const promise = router.sendAgentDeploy("fail-ack@local", config);

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "fail-ack@local",
          publicKey: "aabbccdd",
        }),
      );

      await expect(promise).rejects.toThrow("Failed to store public key");
      expect(router.getRoutableAddresses()).not.toContain("fail-ack@local");
    });

    test("agent.deploy rolls back routing on error", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "fail-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const promise = router.sendAgentDeploy("fail-agent@local", config);

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.error",
          agentAddress: "fail-agent@local",
          error: "provider failed",
        }),
      );

      await expect(promise).rejects.toThrow("provider failed");
      expect(router.getRoutableAddresses()).not.toContain("fail-agent@local");
    });

    test("agent.undeploy sends frame and removes routing after ack", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      const promise = router.sendAgentUndeploy("agent@local", "session_ended");
      const frame = lastSent(ws);
      expect(frame.type).toBe("agent.undeploy");
      expect(frame.agentAddress).toBe("agent@local");
      expect(frame.reason).toBe("session_ended");

      // Routing persists until the ack arrives.
      expect(router.getRoutableAddresses()).toContain("agent@local");

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.undeploy.ack",
          agentAddress: "agent@local",
          statePushed: true,
        }),
      );

      await promise;
      expect(router.getRoutableAddresses()).not.toContain("agent@local");
    });

    test("deploy request times out", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "timeout@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      await expect(
        router.sendAgentDeploy("timeout@local", config),
      ).rejects.toThrow(/timed out/);
    });

    test("undeploy to unknown agent rejects immediately", async () => {
      await expect(
        router.sendAgentUndeploy("unknown@local", "gone"),
      ).rejects.toThrow(/No sidecar connected/);
    });

    test("disconnect during undeploy does not queue messages", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["undeploy-dc@local"],
        }),
      );
      await tick();

      // Start an undeploy but disconnect before the ack arrives.
      const promise = router.sendAgentUndeploy("undeploy-dc@local", "teardown");
      router.handleClose(ws);
      await expect(promise).rejects.toThrow(/disconnected/);

      // The address should not be routable after undeploy + disconnect.
      expect(router.getRoutableAddresses()).not.toContain("undeploy-dc@local");
    });

    test("disconnect rejects pending deploy", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "dc-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const promise = router.sendAgentDeploy("dc-agent@local", config);
      router.handleClose(ws);

      await expect(promise).rejects.toThrow(/disconnected/);
    });

    test("pack acknowledgement must come from the receiving connection", async () => {
      const owner = createMockWs();
      router.handleOpen(owner);
      router.handleMessage(
        owner,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-pack-owner",
          token: "tok",
          agentAddresses: ["pack-owner@local"],
        }),
      );
      await tick();

      const rogue = createMockWs();
      router.handleOpen(rogue);
      const promise = router.sendPack(
        "pack-owner@local",
        new Uint8Array([1, 2, 3]),
        "refs/heads/main",
        "a".repeat(40),
      );
      const done = lastSent(owner);
      expect(done.type).toBe("repo.pack.done");

      let settled = false;
      void promise.finally(() => {
        settled = true;
      });
      router.handleMessage(
        rogue,
        JSON.stringify({
          type: "repo.pack.ack",
          agentAddress: done.agentAddress,
          repoId: done.repoId,
          transferId: done.transferId,
        }),
      );
      await tick();
      expect(settled).toBe(false);

      router.handleMessage(
        owner,
        JSON.stringify({
          type: "repo.pack.ack",
          agentAddress: done.agentAddress,
          repoId: done.repoId,
          transferId: done.transferId,
        }),
      );
      await expect(promise).resolves.toBeUndefined();
    });

    test("preserves routing when address re-registered during a request await", async () => {
      const ws1 = createMockWs();
      router.handleOpen(ws1);
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      const promise = router.sendSourcesUpdate(
        "agent@local",
        TEST_SOURCES,
        TEST_DEFAULT_SOURCE,
      );
      const frame = lastSent(ws1);

      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-2",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

      router.handleMessage(
        ws1,
        JSON.stringify({ type: "session.ack", requestId: frame.requestId }),
      );
      await promise;

      expect(router.getRoutableAddresses()).toContain("agent@local");
    });

    test("closing stale sidecar after reconnect-during-request does not evict address", async () => {
      const ws1 = createMockWs();
      router.handleOpen(ws1);
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      const promise = router.sendSourcesUpdate(
        "agent@local",
        TEST_SOURCES,
        TEST_DEFAULT_SOURCE,
      );
      const frame = lastSent(ws1);

      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-2",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      router.handleMessage(
        ws1,
        JSON.stringify({ type: "session.ack", requestId: frame.requestId }),
      );
      await promise;

      router.handleClose(ws1);

      expect(router.getRoutableAddresses()).toContain("agent@local");
    });
  });

  describe("agent events", () => {
    test("agent.event frames are forwarded to subscribers", async () => {
      const events: { addr: string; sid: string; event: unknown }[] = [];
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      router.events.on("agent.event", ({ agentAddress, sessionId, event }) => {
        events.push({ addr: agentAddress, sid: sessionId, event });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(events).toHaveLength(1);
      expect(events[0]?.addr).toBe("agent@local");
      expect(events[0]?.event).toEqual({
        type: "reactor.start",
        seq: 0,
        data: {},
      });
    });

    test("agent.event frames are emitted on router.events", async () => {
      const seen: { addr: string; sid: string }[] = [];
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      router.events.on("agent.event", ({ agentAddress, sessionId }) => {
        seen.push({ addr: agentAddress, sid: sessionId });
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(seen).toEqual([{ addr: "agent@local", sid: "sess-1" }]);
    });

    test("sidecar.disconnect is emitted on router.events", async () => {
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      const seen: string[][] = [];
      router.events.on("sidecar.disconnect", ({ ownedAddresses }) => {
        seen.push(ownedAddresses);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();
      router.handleClose(ws);

      expect(seen).toEqual([["agent@local"]]);
    });

    test("connector.state.changed populates the cache and is readable via getConnectorState", async () => {
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      // Before any state frame, the cache is absent → null.
      expect(router.getConnectorState("agent@local")).toBeNull();

      const state = {
        threadRoot: "<root@example.com>",
        lastMessageId: "<last@example.com>",
        replyTo: "user@example.com",
        cc: [],
      };
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "agent@local",
          connectorState: state,
        }),
      );
      await tick();

      expect(router.getConnectorState("agent@local")).toEqual(state);

      // An explicit-null frame clears the cached state.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "agent@local",
          connectorState: null,
        }),
      );
      await tick();

      expect(router.getConnectorState("agent@local")).toBeNull();
    });

    test("connector.state.changed is emitted on router.events", async () => {
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      const seen: { addr: string; state: unknown }[] = [];
      router.events.on(
        "connector.state.changed",
        ({ agentAddress, connectorState }) => {
          seen.push({ addr: agentAddress, state: connectorState });
        },
      );

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      const state = {
        threadRoot: "<root@example.com>",
        lastMessageId: "<last@example.com>",
        replyTo: "user@example.com",
        cc: [],
        subject: "Hi",
      };
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "agent@local",
          connectorState: state,
        }),
      );
      await tick();

      expect(seen).toEqual([{ addr: "agent@local", state }]);
    });

    test("live takeover via register evicts the prior owner's cached connector state", async () => {
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });

      // First sidecar registers and reports connector state.
      const ws1 = createMockWs();
      router.handleOpen(ws1);
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "agent@local",
          connectorState: {
            threadRoot: "<root@example.com>",
            lastMessageId: "<last@example.com>",
            replyTo: "user@example.com",
            cc: [],
          },
        }),
      );
      await tick();
      expect(router.getConnectorState("agent@local")).not.toBeNull();

      // A second sidecar registers claiming the same address without
      // the first having disconnected. The prior cache must be evicted
      // so it cannot mis-thread mail in the window before the new
      // owner bootstraps.
      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-2",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      expect(router.getConnectorState("agent@local")).toBeNull();
    });

    test("disconnect clears cached connector state for affected agents", async () => {
      const router = createTestRouter({
        lookups: { lookupPublicKey: async () => null },
      });
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await tick();

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "agent@local",
          connectorState: {
            threadRoot: "<root@example.com>",
            lastMessageId: "<last@example.com>",
            replyTo: "user@example.com",
            cc: [],
          },
        }),
      );
      await tick();

      expect(router.getConnectorState("agent@local")).not.toBeNull();

      router.handleClose(ws);

      expect(router.getConnectorState("agent@local")).toBeNull();
    });
  });

  describe("session subscriptions", () => {
    test("subscriber receives events for its session", async () => {
      const received: unknown[] = [];
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      router.subscribeAgent("agent@local", (event) => received.push(event));

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(received).toHaveLength(1);
      expect(received[0]).toEqual({ type: "reactor.start", seq: 0, data: {} });
    });

    test("subscriber does not receive events for other agents", async () => {
      const received: unknown[] = [];
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      router.subscribeAgent("agent@local", (event) => received.push(event));

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "other-agent@local",
          sessionId: "sess-2",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(received).toHaveLength(0);
    });

    test("unsubscribe stops delivery", async () => {
      const received: unknown[] = [];
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      const unsub = router.subscribeAgent("agent@local", (event) =>
        received.push(event),
      );

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      unsub();

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.end", seq: 1, data: {} },
        }),
      );
      await tick();

      expect(received).toHaveLength(1);
      expect(received[0]).toEqual({ type: "reactor.start", seq: 0, data: {} });
    });

    test("multiple subscribers receive the same event", async () => {
      const received1: unknown[] = [];
      const received2: unknown[] = [];
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      router.subscribeAgent("agent@local", (event) => received1.push(event));
      router.subscribeAgent("agent@local", (event) => received2.push(event));

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(received1).toHaveLength(1);
      expect(received2).toHaveLength(1);
    });

    test("stale unsubscribe does not evict a later subscriber", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      // eslint-disable-next-line @typescript-eslint/no-empty-function
      const unsub = router.subscribeAgent("agent@local", () => {});
      unsub();

      const received: unknown[] = [];
      router.subscribeAgent("agent@local", (event) => received.push(event));

      // Double-unsubscribe with the stale closure
      unsub();

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(received).toHaveLength(1);
    });

    test("subscriber that unsubscribes mid-dispatch does not drop later subscribers", async () => {
      const received: unknown[] = [];
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      const unsub1Ref: { current: (() => void) | null } = { current: null };
      unsub1Ref.current = router.subscribeAgent("agent@local", () => {
        unsub1Ref.current?.();
      });
      router.subscribeAgent("agent@local", (event) => received.push(event));

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.event",
          agentAddress: "agent@local",
          sessionId: "sess-1",
          event: { type: "reactor.start", seq: 0, data: {} },
        }),
      );
      await tick();

      expect(received).toHaveLength(1);
      expect(received[0]).toEqual({ type: "reactor.start", seq: 0, data: {} });
    });
  });

  describe("challenge/response reconnect", () => {
    test("a provision routed during the restore window survives a reconnect", async () => {
      // An agent provisioned while the sidecar is restoring must stay
      // routable after the reconnect frame for the disk-restored agents
      // lands. lookupPublicKey returns null so the challenge
      // short-circuits; the eviction under test happens in
      // handleReconnect's internal register before any challenge work.
      const router = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async () => null,
        },
      });

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "window-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const ws = createMockWs();
      router.handleOpen(ws);

      // Empty register on socket open establishes routability before restore.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      // A fresh provision routes to this sidecar during the restore window.
      const deployPromise = router.sendAgentDeploy(
        "window-agent@local",
        config,
      );
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "window-agent@local",
          publicKey: "deadbeef",
        }),
      );
      await deployPromise;
      expect(router.getRoutableAddresses()).toContain("window-agent@local");

      // Restore finishes; the sidecar reconnects with only its
      // disk-restored addresses.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["restored@local"],
          deployRefs: {},
        }),
      );
      await new Promise((r) => setTimeout(r, 50));

      // The window-provisioned agent stays routable.
      expect(router.getRoutableAddresses()).toContain("window-agent@local");
    });

    test("a window-provisioned agent's connector state survives a reconnect", async () => {
      // The same eviction that drops routing also drops the cached
      // connector thread state. A window agent that established a thread
      // during the restore window must keep it across the reconnect, or a
      // following no-history user message forks a new thread instead of
      // continuing the agent's active one.
      const router = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async () => null,
        },
      });

      const config = {
        sessionId: "ses_test",
        agentId: "a1",
        tenantId: "t1",
        principalId: "prin_test",
        agentAddress: "window-agent@local",
        systemPrompt: "test",
        tools: [],
        grants: [],
        sources: [
          {
            id: "anthropic:claude-sonnet-5",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-test",
            model: "claude-sonnet-5",
          },
        ],
        defaultSource: "anthropic:claude-sonnet-5",
      };

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );
      await tick();

      const deployPromise = router.sendAgentDeploy(
        "window-agent@local",
        config,
      );
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "window-agent@local",
          publicKey: "deadbeef",
        }),
      );
      await deployPromise;

      // The window agent establishes a connector thread during the window.
      const connectorState = {
        threadRoot: "<root@example.com>",
        lastMessageId: "<last@example.com>",
        replyTo: "user@example.com",
        cc: [],
      };
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "connector.state.changed",
          agentAddress: "window-agent@local",
          connectorState,
        }),
      );
      await tick();
      expect(router.getConnectorState("window-agent@local")).toEqual(
        connectorState,
      );

      // Restore finishes; the sidecar reconnects with only its
      // disk-restored addresses.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["restored@local"],
          deployRefs: {},
        }),
      );
      await new Promise((r) => setTimeout(r, 50));

      // The window agent's connector thread state survives.
      expect(router.getConnectorState("window-agent@local")).toEqual(
        connectorState,
      );
    });

    test("reconnect issues challenge and verifies signature", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

      // Wait for async handleReconnect to complete.
      await new Promise((r) => setTimeout(r, 50));

      // Should have received a challenge frame.
      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      expect(challengeFrame).toBeDefined();
      expect(challengeFrame.challenges).toHaveLength(1);

      const { address, nonce } = challengeFrame.challenges[0];
      expect(address).toBe("agent@local");

      // Sign and respond.
      const signature = await signChallenge(nonce, address, kp.privateKey);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [{ address, signature }],
        }),
      );

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

      expect(router.getRoutableAddresses()).toContain("agent@local");
    });

    test("reconnect rejects invalid signature", async () => {
      const kp = await generateKeyPair();
      const wrongKp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");

      const { address, nonce } = challengeFrame.challenges[0];

      // Sign with wrong key.
      const badSig = await signChallenge(nonce, address, wrongKp.privateKey);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [{ address, signature: badSig }],
        }),
      );

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

      expect(router.getRoutableAddresses()).not.toContain("agent@local");

      const failedFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge.failed");
      expect(failedFrame).toBeDefined();
      expect(failedFrame.address).toBe("agent@local");
    });

    test("reconnect sends challenge.failed for unknown address", async () => {
      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async () => null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["unknown@local"],
        }),
      );

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

      const failedFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge.failed");
      expect(failedFrame).toBeDefined();
      expect(failedFrame.address).toBe("unknown@local");
      expect(failedFrame.reason).toBe("Unknown agent address");
    });

    test("fails closed on a key-lookup error during reconnect instead of crashing", async () => {
      // A rejecting lookup (e.g. a transient DB failure) on the production
      // reconnect path must be caught and surfaced, not floated out of the
      // void-dispatched handler as an unhandled rejection that could take down
      // the hub. Fail closed: the address is treated as unverifiable, fails
      // its challenge, and stays unrouted. (The test completing rather than
      // hanging on an unhandled rejection is itself part of the assertion.)
      const captured: { level: string; message: string }[] = [];
      const savedConfig = getConfig();
      configureSync({
        reset: true,
        sinks: {
          capture: (record) => {
            const message = Array.isArray(record.message)
              ? record.message
                  .map((part) =>
                    typeof part === "string" ? part : JSON.stringify(part),
                  )
                  .join("")
              : String(record.message);
            captured.push({ level: record.level, message });
          },
        },
        loggers: [{ category: [], lowestLevel: "debug", sinks: ["capture"] }],
      });
      try {
        const router = createTestRouter({
          requestTimeoutMs: 5000,
          lookups: {
            lookupPublicKey: async () => {
              throw new Error("db unavailable");
            },
          },
        });
        const ws = createMockWs();
        router.handleOpen(ws);
        router.handleMessage(
          ws,
          JSON.stringify({
            type: "reconnect",
            sidecarId: "sc-1",
            token: "tok",
            agentAddresses: ["agent@local"],
          }),
        );

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

        expect(router.getRoutableAddresses()).not.toContain("agent@local");
        const failedFrame = ws.sent
          .map((s) => JSON.parse(s))
          .find((f: { type: string }) => f.type === "challenge.failed");
        expect(failedFrame?.address).toBe("agent@local");
        expect(
          captured.some(
            (l) =>
              l.level === "error" && l.message.includes("Key lookup failed"),
          ),
        ).toBe(true);
      } finally {
        if (savedConfig) {
          configureSync({ reset: true, ...savedConfig });
        } else {
          resetSync();
        }
      }
    });

    test("partial success routes verified addresses only", async () => {
      const kp1 = await generateKeyPair();
      const kp2 = await generateKeyPair();
      const wrongKp = await generateKeyPair();

      const keys = new Map([
        ["agent-a@local", hexEncode(kp1.publicKey)],
        ["agent-b@local", hexEncode(kp2.publicKey)],
      ]);

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) => keys.get(addr) ?? null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent-a@local", "agent-b@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      expect(challengeFrame.challenges).toHaveLength(2);

      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => {
            const key =
              c.address === "agent-a@local"
                ? kp1.privateKey
                : wrongKp.privateKey;
            return {
              address: c.address,
              signature: await signChallenge(c.nonce, c.address, key),
            };
          },
        ),
      );

      router.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );

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

      expect(router.getRoutableAddresses()).toContain("agent-a@local");
      expect(router.getRoutableAddresses()).not.toContain("agent-b@local");
    });

    test("reconnect with stale deployRef emits deploy.ref.stale", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const staleAddresses: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
          lookupDeployRef: async () => "aaaa",
        },
      });
      router.events.on("deploy.ref.stale", ({ agentAddress }) => {
        staleAddresses.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
          deployRefs: { "agent@local": "bbbb" },
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address,
              signature: await signChallenge(nonce, address, kp.privateKey),
            },
          ],
        }),
      );

      await new Promise((r) => setTimeout(r, 50));
      expect(staleAddresses).toContain("agent@local");
    });

    test("reconnect with matching deployRef skips deploy.ref.stale", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const staleAddresses: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
          lookupDeployRef: async () => "aaaa",
        },
      });
      router.events.on("deploy.ref.stale", ({ agentAddress }) => {
        staleAddresses.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
          deployRefs: { "agent@local": "aaaa" },
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address,
              signature: await signChallenge(nonce, address, kp.privateKey),
            },
          ],
        }),
      );

      await new Promise((r) => setTimeout(r, 50));
      expect(staleAddresses).toEqual([]);
    });

    test("reconnect with absent deployRef emits deploy.ref.stale", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const staleAddresses: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
          lookupDeployRef: async () => "aaaa",
        },
      });
      router.events.on("deploy.ref.stale", ({ agentAddress }) => {
        staleAddresses.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address,
              signature: await signChallenge(nonce, address, kp.privateKey),
            },
          ],
        }),
      );

      await new Promise((r) => setTimeout(r, 50));
      expect(staleAddresses).toContain("agent@local");
    });

    test("reconnect skips re-deploy when hub has no deploy ref", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const staleAddresses: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
          lookupDeployRef: async () => null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
          deployRefs: { "agent@local": "bbbb" },
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [
            {
              address,
              signature: await signChallenge(nonce, address, kp.privateKey),
            },
          ],
        }),
      );

      await new Promise((r) => setTimeout(r, 50));
      expect(staleAddresses).toEqual([]);
    });

    // Pins the pinned-forever workflow-definition decision at the wire layer:
    // a workflow deployment keeps its deploy-time definition until an explicit
    // undeploy/redeploy, so the reconnect deploy-ref freshness catch-up must
    // fire for a launched-agent address whose ref went stale but must NOT fire
    // for a workflow-derived (ins_dep_...) address on the same stale-ref
    // reconnect. A definition change on the hub during the disconnect does not
    // reconcile onto the reconnected deployment.
    test("reconnect fires deploy.ref.stale for a launched agent but not a workflow-derived address", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const launchedAddr = "agent@local";
      const workflowAddr = "ins_dep_abc@local";
      const staleAddresses: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          // Both addresses are keyed (deployment addresses carry the
          // deployment's own key), so both reach the challenged reconnect path.
          lookupPublicKey: async () => publicKeyHex,
          // Hub ref differs from every sidecar ref below, so the catch-up would
          // fire for any address it is NOT deliberately skipping.
          lookupDeployRef: async () => "aaaa",
        },
      });
      router.events.on("deploy.ref.stale", ({ agentAddress }) => {
        staleAddresses.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [launchedAddr, workflowAddr],
          deployRefs: { [launchedAddr]: "bbbb", [workflowAddr]: "bbbb" },
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, kp.privateKey),
          }),
        ),
      );
      router.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );

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

      expect(staleAddresses).toContain(launchedAddr);
      expect(staleAddresses).not.toContain(workflowAddr);
    });

    test("disconnect cleans up pending challenge", async () => {
      const kp = await generateKeyPair();

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? hexEncode(kp.publicKey) : null,
        },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      // Disconnect before responding.
      router.handleClose(ws);

      expect(router.getConnectedSidecars()).toEqual([]);
      expect(router.getRoutableAddresses()).toEqual([]);
    });
  });

  describe("disconnect message queuing", () => {
    // Establish an initial route for a KEYED address through the challenged
    // reconnect path. A keyed address cannot be routed by a plain register
    // (the key-existence gate rejects it), so the disconnect-queue tests --
    // whose agents have a stored key so the reconnect challenge can verify --
    // bring the address up the same way production does.
    async function connectAgentViaChallenge(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      addr: string,
      privateKey: Uint8Array,
    ) {
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [addr],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, privateKey),
          }),
        ),
      );
      r.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    test("mail queued during disconnect is flushed on reconnect", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });

      // Initial connection with one agent, via the challenged reconnect path
      // (a keyed address cannot be routed by a plain register).
      const ws1 = createMockWs();
      await connectAgentViaChallenge(router, ws1, "agent@local", kp.privateKey);

      // Disconnect — creates a queue entry.
      router.handleClose(ws1);

      // Send mail while disconnected.
      const queued = router.routeMail("agent@local", "queued-message");
      expect(queued).toBe(true);

      // Reconnect with challenge/response.
      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws2.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");

      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, kp.privateKey),
          }),
        ),
      );

      router.handleMessage(
        ws2,
        JSON.stringify({ type: "challenge.response", responses }),
      );

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

      // The queued mail should have been flushed to the new connection.
      const flushed = ws2.sent
        .map((s) => JSON.parse(s))
        .filter((f: { type: string }) => f.type === "mail.inbound");
      expect(flushed).toHaveLength(1);
      expect(flushed[0].rawMessage).toBe("queued-message");
    });

    test("run.grants queued during a pre-first-reconnect disconnect flushes on reconnect", async () => {
      // Phase-1 shared-fate with the trigger mail: a freshly deployed workflow
      // address enters `agentAddresses` via the keyless register path (no
      // stored key yet), so a disconnect BEFORE its first challenged reconnect
      // queues it exactly as routeMail's does. A run.grants issued in that
      // window must ride the same queue (return true, not false) and flush on
      // reconnect. This pins the enqueueForDisconnected fallback: deleting it
      // makes sendRunGrants return false here and drops the grants.
      const kp = await generateKeyPair();
      // The address has no stored key until the deployment acks, so the first
      // register routes it keyless; the reconnect challenge then resolves the
      // key. Flip the flag once the register has landed.
      let keyStored = false;
      const router = createTestRouter({
        requestTimeoutMs: 500,
        lookups: {
          async lookupPublicKey() {
            return keyStored ? hexEncode(kp.publicKey) : null;
          },
        },
      });

      const WF_ADDR = "ins_dep_phase1@local";

      // First deploy: register the keyless workflow address onto agentAddresses.
      const ws1 = createMockWs();
      router.handleOpen(ws1);
      router.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [WF_ADDR],
        }),
      );
      await tick();
      expect(router.getRoutableAddresses()).toContain(WF_ADDR);

      // The deployment's key is now known; a later reconnect is challenged.
      keyStored = true;

      // Disconnect before any reconnect -- the address is still on
      // agentAddresses, so handleClose queues it.
      router.handleClose(ws1);

      // A run.grants in this window must queue (ride the fallback), not drop.
      const stepGrants: RunGrantsFrame["stepGrants"] = [
        {
          id: "grant-phase1",
          resource: "tool:send-mail",
          action: "invoke",
          effect: "allow",
          origin: "creator",
          conditions: null,
          expiresAt: null,
          roleId: null,
          principalId: "prn_deployment",
        },
      ];
      expect(router.sendRunGrants(WF_ADDR, "run-phase1", stepGrants)).toBe(
        true,
      );

      // Reconnect with challenge/response to flush the queued frame.
      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [WF_ADDR],
        }),
      );
      await new Promise((r) => setTimeout(r, 50));

      const challengeFrame = ws2.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, kp.privateKey),
          }),
        ),
      );
      router.handleMessage(
        ws2,
        JSON.stringify({ type: "challenge.response", responses }),
      );
      await new Promise((r) => setTimeout(r, 50));

      // The queued run.grants was flushed to the reconnected sidecar.
      const flushed = ws2.sent
        .map((s) => JSON.parse(s))
        .filter((f: { type: string }) => f.type === "run.grants");
      expect(flushed).toHaveLength(1);
      expect(flushed[0].runId).toBe("run-phase1");
      expect(flushed[0].stepGrants).toEqual(stepGrants);
    });

    test("mail to unknown address returns false", () => {
      expect(router.routeMail("unknown@local", "msg")).toBe(false);
    });

    test("queue evicts oldest when full", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        disconnectQueueMaxSize: 2,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });

      const ws = createMockWs();
      await connectAgentViaChallenge(router, ws, "agent@local", kp.privateKey);
      router.handleClose(ws);

      // Queue 3 messages with max size 2 — oldest should be evicted.
      router.routeMail("agent@local", "msg-0");
      router.routeMail("agent@local", "msg-1");
      router.routeMail("agent@local", "msg-2");

      // Reconnect with challenge/response to flush.
      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws2.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");

      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, kp.privateKey),
          }),
        ),
      );

      router.handleMessage(
        ws2,
        JSON.stringify({ type: "challenge.response", responses }),
      );

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

      const flushed = ws2.sent
        .map((s) => JSON.parse(s))
        .filter((f: { type: string }) => f.type === "mail.inbound");

      // Only the 2 newest messages should have been flushed.
      expect(flushed).toHaveLength(2);
      expect(flushed[0].rawMessage).toBe("msg-1");
      expect(flushed[1].rawMessage).toBe("msg-2");
    });

    test("size-cap eviction surfaces the dropped message loudly", async () => {
      // A size-cap eviction must not silently discard the evicted message: it
      // fires mail.outbound.undelivered for the dropped frame and warns with
      // the recipient, so an operator can see mail was lost.
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        disconnectQueueMaxSize: 2,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });

      const undelivered: { rawMessage: string; recipients: string[] }[] = [];
      router.events.on("mail.outbound.undelivered", (event) => {
        undelivered.push({
          rawMessage: event.rawMessage,
          recipients: event.recipients,
        });
      });

      const ws = createMockWs();
      await connectAgentViaChallenge(router, ws, "agent@local", kp.privateKey);
      router.handleClose(ws);

      const warnings = captureWarnings(() => {
        // Queue 3 with max size 2 — the oldest (msg-0) is evicted.
        router.routeMail("agent@local", "msg-0");
        router.routeMail("agent@local", "msg-1");
        router.routeMail("agent@local", "msg-2");
      });

      // The evicted frame surfaces through the existing undelivered channel,
      // carrying its rawMessage and the recipient it was queued for.
      expect(undelivered).toEqual([
        { rawMessage: "msg-0", recipients: ["agent@local"] },
      ]);
      // And a warn names the recipient and the drop count.
      expect(
        warnings.some(
          (w) => w.includes("agent@local") && w.includes("Dropping"),
        ),
      ).toBe(true);
    });

    test("TTL expiry after disconnect surfaces the whole dropped queue loudly", async () => {
      // When the disconnect-queue TTL fires on a still-full queue, every
      // undelivered frame must surface -- not vanish. Each queued frame fires
      // mail.outbound.undelivered and a single warn reports the recipient and
      // count.
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        disconnectQueueTTLMs: 20,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });

      const undelivered: { rawMessage: string; recipients: string[] }[] = [];
      router.events.on("mail.outbound.undelivered", (event) => {
        undelivered.push({
          rawMessage: event.rawMessage,
          recipients: event.recipients,
        });
      });

      const ws = createMockWs();
      await connectAgentViaChallenge(router, ws, "agent@local", kp.privateKey);
      router.handleClose(ws);

      router.routeMail("agent@local", "msg-a");
      router.routeMail("agent@local", "msg-b");

      const warnings: string[] = [];
      const restore = installWarningCapture(warnings);
      try {
        // Wait past the TTL so the handleClose timer fires.
        await new Promise((r) => setTimeout(r, 60));
      } finally {
        restore();
      }

      expect(undelivered).toEqual([
        { rawMessage: "msg-a", recipients: ["agent@local"] },
        { rawMessage: "msg-b", recipients: ["agent@local"] },
      ]);
      expect(
        warnings.some(
          (w) => w.includes("agent@local") && w.includes("Dropping"),
        ),
      ).toBe(true);
    });

    test("TTL expiry after a failed reconnect surfaces the whole dropped queue loudly", async () => {
      // A reconnect whose governance reaction rejects the address resets the
      // queue TTL. When that reset timer fires on a still-full queue, the
      // dropped frames must surface the same way -- via
      // mail.outbound.undelivered per frame and a warn naming the recipient.
      // TTL long enough that the original handleClose timer does not fire
      // during the ~100ms reconnect handshake -- the failed reconnect must be
      // the thing that re-arms it, so this test exercises the reset timer's
      // drop path rather than handleClose's.
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 5000,
        disconnectQueueTTLMs: 250,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? hexEncode(kp.publicKey) : null,
        },
      });
      const undelivered: { rawMessage: string; recipients: string[] }[] = [];
      router.events.on("mail.outbound.undelivered", (event) => {
        undelivered.push({
          rawMessage: event.rawMessage,
          recipients: event.recipients,
        });
      });

      // Establish the address (initial reconnect succeeds -- no governance
      // reaction yet), disconnect to create a queue, and fill it.
      const ws1 = createMockWs();
      await connectAgentViaChallenge(router, ws1, "agent@local", kp.privateKey);
      router.handleClose(ws1);
      expect(router.routeMail("agent@local", "msg-x")).toBe(true);
      expect(router.routeMail("agent@local", "msg-y")).toBe(true);

      // Now arm a governance reaction that always rejects, so the SECOND
      // reconnect forces the address into the failed set and exercises the
      // TTL-reset timer rather than a flush.
      router.events.on("agent.reconnected", () => {
        throw new Error("governance rejected");
      });

      // Reconnect and answer the challenge; the governance reaction rejects the
      // address, resetting the queue TTL rather than flushing.
      const ws2 = createMockWs();
      router.handleOpen(ws2);
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      await new Promise((r) => setTimeout(r, 50));
      const challengeFrame = ws2.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, kp.privateKey),
          }),
        ),
      );
      router.handleMessage(
        ws2,
        JSON.stringify({ type: "challenge.response", responses }),
      );
      await new Promise((r) => setTimeout(r, 50));

      // The address failed governance, so the queue was preserved with a reset
      // TTL rather than flushed.
      expect(undelivered).toHaveLength(0);

      const warnings: string[] = [];
      const restore = installWarningCapture(warnings);
      try {
        // Wait past the reset TTL so the failed-reconnect timer fires.
        await new Promise((r) => setTimeout(r, 320));
      } finally {
        restore();
      }

      expect(undelivered).toEqual([
        { rawMessage: "msg-x", recipients: ["agent@local"] },
        { rawMessage: "msg-y", recipients: ["agent@local"] },
      ]);
      expect(
        warnings.some(
          (w) => w.includes("agent@local") && w.includes("Dropping"),
        ),
      ).toBe(true);
    });

    test("sendSourcesUpdate rejects when agent is disconnected", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );
      router.handleClose(ws);

      await expect(
        router.sendSourcesUpdate(
          "agent@local",
          TEST_SOURCES,
          TEST_DEFAULT_SOURCE,
        ),
      ).rejects.toThrow("No sidecar connected");
    });
  });

  describe("ping/pong keepalive", () => {
    test("hub responds to ping with pong", () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      router.handleMessage(ws, JSON.stringify({ type: "ping" }));

      const pong = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "pong");
      expect(pong).toEqual({ type: "pong" });
    });

    test("connection closed after ping timeout", async () => {
      const router = createTestRouter({
        requestTimeoutMs: 500,
        pingTimeoutMs: 100,
        lookups: { lookupPublicKey: async () => null },
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

      // Wait for the ping timeout to fire.
      await new Promise((r) => setTimeout(r, 150));

      expect(ws.closed).toBe(true);
    });

    test("ping resets the liveness timer", async () => {
      const router = createTestRouter({
        requestTimeoutMs: 500,
        pingTimeoutMs: 100,
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      // Send pings to keep the connection alive past the timeout.
      await new Promise((r) => setTimeout(r, 60));
      router.handleMessage(ws, JSON.stringify({ type: "ping" }));
      await new Promise((r) => setTimeout(r, 60));
      router.handleMessage(ws, JSON.stringify({ type: "ping" }));
      await new Promise((r) => setTimeout(r, 60));

      expect(ws.closed).toBe(false);
    });
  });

  describe("agent.reconnected event", () => {
    test("fires for each verified address on reconnect", async () => {
      const kp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const reconnected: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
        },
      });
      router.events.on("agent.reconnected", ({ agentAddress }) => {
        reconnected.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      const signature = await signChallenge(nonce, address, kp.privateKey);

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [{ address, signature }],
        }),
      );

      // Wait for async handleChallengeResponse to complete.
      await new Promise((r) => setTimeout(r, 50));

      expect(reconnected).toEqual(["agent@local"]);
    });

    test("does not fire for unverified addresses", async () => {
      const kp = await generateKeyPair();
      const wrongKp = await generateKeyPair();
      const publicKeyHex = hexEncode(kp.publicKey);
      const reconnected: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) =>
            addr === "agent@local" ? publicKeyHex : null,
        },
      });
      router.events.on("agent.reconnected", ({ agentAddress }) => {
        reconnected.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const { address, nonce } = challengeFrame.challenges[0];
      const signature = await signChallenge(nonce, address, wrongKp.privateKey);

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "challenge.response",
          responses: [{ address, signature }],
        }),
      );

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

      expect(reconnected).toEqual([]);
    });

    test("listener error does not prevent other addresses from reconnecting", async () => {
      const kp1 = await generateKeyPair();
      const kp2 = await generateKeyPair();
      const reconnected: string[] = [];

      const router = createTestRouter({
        requestTimeoutMs: 5000,
        lookups: {
          lookupPublicKey: async (addr) => {
            if (addr === "agent1@local") return hexEncode(kp1.publicKey);
            if (addr === "agent2@local") return hexEncode(kp2.publicKey);
            return null;
          },
        },
      });
      router.events.on("agent.reconnected", ({ agentAddress }) => {
        if (agentAddress === "agent1@local") throw new Error("DB failure");
        reconnected.push(agentAddress);
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: ["agent1@local", "agent2@local"],
        }),
      );

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

      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");

      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(
              c.nonce,
              c.address,
              c.address === "agent1@local" ? kp1.privateKey : kp2.privateKey,
            ),
          }),
        ),
      );

      router.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );

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

      // agent2 should still be reconnected despite agent1's callback failure.
      expect(reconnected).toEqual(["agent2@local"]);
      expect(router.getRoutableAddresses()).not.toContain("agent1@local");
      expect(router.getRoutableAddresses()).toContain("agent2@local");

      // agent1 should receive a challenge.failed frame indicating governance rejection.
      const failedFrames = ws.sent
        .map((s) => JSON.parse(s))
        .filter(
          (f: { type: string; address?: string }) =>
            f.type === "challenge.failed" && f.address === "agent1@local",
        );
      expect(failedFrames).toHaveLength(1);
      expect(failedFrames[0].reason).toContain("governance");
    });
  });

  describe("configuration guards", () => {
    test("sendAgentDeploy without hub key throws without mutating routing table", async () => {
      const router = createTestRouter({
        requestTimeoutMs: 500,
      });

      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [],
        }),
      );

      await expect(
        router.sendAgentDeploy("new-agent@local", {
          agentId: "a1",
          agentAddress: "new-agent@local",
          sessionId: "s1",
          principalId: "p1",
          tenantId: "t1",
          systemPrompt: "test",
          tools: [],
          grants: [],
          sources: [
            {
              id: "test:m",
              provider: "test",
              apiKey: "k",
              baseURL: "http://localhost",
              model: "m",
            },
          ],
          defaultSource: "test:m",
        }),
      ).rejects.toThrow("Hub signing key is required");

      expect(router.getRoutableAddresses()).not.toContain("new-agent@local");
    });
  });

  describe("pack receive dispatch", () => {
    type RecordedReceive = {
      method: "receiveAgentStatePack" | "receiveWorkflowRunPack";
      repoId: RepoId;
      pack: Uint8Array;
      ref: string;
      commitSha: string;
    };

    function buildPackRouter(
      verdicts: {
        agentState?:
          | { accepted: true }
          | { accepted: false; reason: PackRejectReason };
        workflowRun?:
          | { accepted: true }
          | { accepted: false; reason: PackRejectReason };
      } = {},
    ) {
      const calls: RecordedReceive[] = [];
      const stateVerdict = verdicts.agentState ?? ({ accepted: true } as const);
      const wfrVerdict = verdicts.workflowRun ?? ({ accepted: true } as const);
      const packRouter = createTestRouter({
        requestTimeoutMs: 500,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          // Keyless lookup: the pack tests register the deployment address as a
          // first-deploy so the key-existence gate routes it (the gate is
          // fail-closed without a lookup configured).
          lookupPublicKey: async () => null,
          async receiveAgentStatePack(repoId, pack, ref, commitSha) {
            calls.push({
              method: "receiveAgentStatePack",
              repoId,
              pack,
              ref,
              commitSha,
            });
            return stateVerdict;
          },
          async receiveWorkflowRunPack(repoId, pack, ref, commitSha) {
            calls.push({
              method: "receiveWorkflowRunPack",
              repoId,
              pack,
              ref,
              commitSha,
            });
            return wfrVerdict;
          },
        },
      });
      return { router: packRouter, calls };
    }

    async function registerAddr(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      sidecarId: string,
      addr: string,
    ) {
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId,
          token: "tok",
          agentAddresses: [addr],
        }),
      );
      // Register routing is async (the key-existence gate awaits the lookup);
      // settle it before assertions read the routing table.
      await new Promise((res) => setTimeout(res, 0));
    }

    function pushPack(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      args: {
        agentAddress: string;
        repoId: RepoId;
        transferId: string;
        pack: Uint8Array;
        ref: string;
        commitSha: string;
      },
    ) {
      for (const chunk of chunkPack(args.pack)) {
        r.handleMessage(
          ws,
          JSON.stringify({
            type: "repo.pack.push",
            agentAddress: args.agentAddress,
            repoId: args.repoId,
            transferId: args.transferId,
            seq: chunk.seq,
            data: chunk.data,
          }),
        );
      }
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.done",
          agentAddress: args.agentAddress,
          repoId: args.repoId,
          transferId: args.transferId,
          ref: args.ref,
          commitSha: args.commitSha,
        }),
      );
    }

    test("workflow-run pack frames invoke receiveWorkflowRunPack and ack the sidecar", async () => {
      const { router: r, calls } = buildPackRouter();
      const ws = createMockWs();
      const addr = "ins_dep-wfr-1@local";
      await registerAddr(r, ws, "sc-wfr", addr);

      const transferId = "t-wfr-1";
      const repoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };
      const ref = "refs/heads/events";
      const commitSha = "f".repeat(40);
      const pack = new Uint8Array([1, 2, 3, 4, 5]);

      pushPack(r, ws, {
        agentAddress: addr,
        repoId,
        transferId,
        pack,
        ref,
        commitSha,
      });

      // The receiveWorkflowRunPack lookup is async; wait a tick.
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(calls.length).toBe(1);
      const [call] = calls;
      if (call === undefined) throw new Error("expected one recorded call");
      expect(call.method).toBe("receiveWorkflowRunPack");
      expect(call.repoId).toEqual(repoId);
      expect(call.ref).toBe(ref);
      expect(call.commitSha).toBe(commitSha);
      expect(Array.from(call.pack)).toEqual(Array.from(pack));

      const ack = lastSent(ws);
      expect(ack.type).toBe("repo.pack.ack");
      expect(ack.transferId).toBe(transferId);
      expect(ack.repoId).toEqual(repoId);
    });

    test("a workflow deployment owned by its connection can push workflow-run packs", async () => {
      // Pack-push authorization keys on `connOwnsAddress`, which unions the
      // connection's launched-agent and workflow-substrate address sets. A
      // workflow deployment address owned by the connection must authorize a
      // workflow-run pack -- otherwise the hub's workflow-run observation
      // mirror silently stops updating (mail routing resumes, but pack push
      // rejects the address as unrouted). The challenged-reconnect ownership
      // path itself is covered by the "workflow-address reconnect" block; here
      // the address is registered directly to keep the test on pack routing.
      const { router: r, calls } = buildPackRouter();
      const ws = createMockWs();
      const addr = "ins_dep_wfr@local";
      await registerAddr(r, ws, "sc-wfr", addr);

      const repoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };
      pushPack(r, ws, {
        agentAddress: addr,
        repoId,
        transferId: "t-wfr-2",
        pack: new Uint8Array([9, 8, 7]),
        ref: "refs/heads/events",
        commitSha: "e".repeat(40),
      });

      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(calls.length).toBe(1);
      expect(calls[0]?.method).toBe("receiveWorkflowRunPack");
      const ack = lastSent(ws);
      expect(ack.type).toBe("repo.pack.ack");
    });

    test("a superseded connection's close does not cancel the new owner's workflow-run transfer", async () => {
      // Abrupt restart: a fresh ws reclaims a workflow address a still-open
      // stale ws owns, then starts a pack transfer. When the stale ws finally
      // closes, its teardown must not cancel the new owner's in-flight
      // transfer (cancelByAgent is keyed by address, not connection). The
      // ghost cleanup on the fresh register evicts the address from the stale
      // connection so its close leaves the new owner alone.
      const { router: r, calls } = buildPackRouter();
      const addr = "ins_dep_reclaim@local";
      const repoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };
      const pack = new Uint8Array([4, 5, 6, 7]);
      const transferId = "t-reclaim";

      const oldWs = createMockWs();
      await registerAddr(r, oldWs, "sc", addr);

      const newWs = createMockWs();
      await registerAddr(r, newWs, "sc", addr);

      // The new owner starts (but does not finish) a workflow-run transfer.
      for (const chunk of chunkPack(pack)) {
        r.handleMessage(
          newWs,
          JSON.stringify({
            type: "repo.pack.push",
            agentAddress: addr,
            repoId,
            transferId,
            seq: chunk.seq,
            data: chunk.data,
          }),
        );
      }

      // The superseded connection closes mid-transfer.
      r.handleClose(oldWs);

      // The new owner completes the transfer; it must not have been cancelled.
      r.handleMessage(
        newWs,
        JSON.stringify({
          type: "repo.pack.done",
          agentAddress: addr,
          repoId,
          transferId,
          ref: "refs/heads/events",
          commitSha: "a".repeat(40),
        }),
      );

      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(calls.length).toBe(1);
      expect(calls[0]?.method).toBe("receiveWorkflowRunPack");
    });

    test("a workflow address reclaimed via challenged reconnect does not cancel the new owner's transfer", async () => {
      // The register-reclaim variant above covers the handleRegister ghost
      // cleanup. This covers the CHALLENGED RECONNECT reclaim path, where the
      // deployment lives on the connection's workflow set: the verified reclaim
      // must evict the address from the superseded connection, or that
      // connection's close runs cancelByAgent and kills the new owner's
      // in-flight transfer.
      const kp = await generateKeyPair();
      const calls: { method: string }[] = [];
      const r = createTestRouter({
        requestTimeoutMs: 5000,
        hubPublicKey: TEST_HUB_KEY,
        lookups: {
          lookupPublicKey: async () => hexEncode(kp.publicKey),
          async receiveWorkflowRunPack() {
            calls.push({ method: "receiveWorkflowRunPack" });
            return { accepted: true };
          },
        },
      });
      const addr = "ins_dep_reclaim_rc@local";
      const repoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };
      const pack = new Uint8Array([4, 5, 6, 7]);
      const transferId = "t-reclaim-rc";

      async function reconnectVerify(ws: ReturnType<typeof createMockWs>) {
        r.handleOpen(ws);
        r.handleMessage(
          ws,
          JSON.stringify({
            type: "reconnect",
            sidecarId: "sc",
            token: "tok",
            agentAddresses: [addr],
          }),
        );
        await new Promise((resolve) => setTimeout(resolve, 50));
        const challenge = ws.sent
          .map((s) => JSON.parse(s))
          .find((f: { type: string }) => f.type === "challenge");
        r.handleMessage(
          ws,
          JSON.stringify({
            type: "challenge.response",
            responses: [
              {
                address: addr,
                signature: await signChallenge(
                  challenge.challenges[0].nonce,
                  addr,
                  kp.privateKey,
                ),
              },
            ],
          }),
        );
        await new Promise((resolve) => setTimeout(resolve, 50));
      }

      const oldWs = createMockWs();
      await reconnectVerify(oldWs);

      const newWs = createMockWs();
      await reconnectVerify(newWs);

      // The new owner starts (but does not finish) a workflow-run transfer.
      for (const chunk of chunkPack(pack)) {
        r.handleMessage(
          newWs,
          JSON.stringify({
            type: "repo.pack.push",
            agentAddress: addr,
            repoId,
            transferId,
            seq: chunk.seq,
            data: chunk.data,
          }),
        );
      }

      // The superseded connection closes mid-transfer.
      r.handleClose(oldWs);

      // The new owner completes the transfer; it must not have been cancelled.
      r.handleMessage(
        newWs,
        JSON.stringify({
          type: "repo.pack.done",
          agentAddress: addr,
          repoId,
          transferId,
          ref: "refs/heads/events",
          commitSha: "a".repeat(40),
        }),
      );

      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(calls.length).toBe(1);
      expect(calls[0]?.method).toBe("receiveWorkflowRunPack");
    });

    test("agent-state and workflow-run packs use independent receivers (concurrent transferIds)", async () => {
      const { router: r, calls } = buildPackRouter();
      const ws = createMockWs();
      const addr = "agent-mix@local";
      await registerAddr(r, ws, "sc-mix", addr);

      // Reuse the same transferId across kinds. The two receivers'
      // in-flight state must be independent, so this must NOT collide.
      const transferId = "shared-transfer";

      const statePack = new Uint8Array([10, 11, 12]);
      const stateRepoId: RepoId = { kind: "agent-state", id: addr };

      const wfrPack = new Uint8Array([20, 21, 22]);
      const wfrRepoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };

      // Push the agent-state chunk first, then a workflow-run chunk
      // sharing the same transferId. If state were shared, the
      // workflow-run push would either evict the state transfer or get
      // rejected as a duplicate.
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.push",
          agentAddress: addr,
          repoId: stateRepoId,
          transferId,
          seq: 0,
          data: btoa(String.fromCharCode(...statePack)),
        }),
      );
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.push",
          agentAddress: addr,
          repoId: wfrRepoId,
          transferId,
          seq: 0,
          data: btoa(String.fromCharCode(...wfrPack)),
        }),
      );

      // Verify no rejection was sent before the done frames arrive.
      for (const sent of ws.sent) {
        const parsed: { type: string } = JSON.parse(sent);
        expect(parsed.type).not.toBe("repo.pack.reject");
      }

      // Complete both transfers.
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.done",
          agentAddress: addr,
          repoId: stateRepoId,
          transferId,
          ref: "refs/instances/test",
          commitSha: "a".repeat(40),
        }),
      );
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.done",
          agentAddress: addr,
          repoId: wfrRepoId,
          transferId,
          ref: "refs/heads/events",
          commitSha: "b".repeat(40),
        }),
      );

      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(calls.map((c) => c.method).sort()).toEqual([
        "receiveAgentStatePack",
        "receiveWorkflowRunPack",
      ]);

      const stateCall = calls.find((c) => c.method === "receiveAgentStatePack");
      const wfrCall = calls.find((c) => c.method === "receiveWorkflowRunPack");
      if (stateCall === undefined) {
        throw new Error("expected an agent-state receive call");
      }
      if (wfrCall === undefined) {
        throw new Error("expected a workflow-run receive call");
      }
      expect(Array.from(stateCall.pack)).toEqual(Array.from(statePack));
      expect(Array.from(wfrCall.pack)).toEqual(Array.from(wfrPack));
    });

    test("an allocated connection can push only its authenticated workflow repository", async () => {
      const addr = "ins_dep-exclusive-pack@tenant.example";
      const identity = {
        kind: "allocated" as const,
        sidecarId: "sc-exclusive-pack",
        allocationId: "allocation-exclusive-pack",
        tenantId: "tenant-1",
        anchorRunId: "dep-exclusive-pack",
        workflowRunAddress: addr,
        generation: 3,
      };
      const sources: unknown[] = [];
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => identity,
        validateSidecarIdentity: async () => true,
        lookups: {
          async receiveWorkflowRunPack(
            _repoId,
            _pack,
            _ref,
            _commitSha,
            source,
          ) {
            sources.push(source);
            return { accepted: true };
          },
        },
      });
      allocatedRouter.fenceAllocation(identity.allocationId, 3);
      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: identity.sidecarId,
          token: "token",
          agentAddresses: [addr],
        }),
      );
      await tick();

      pushPack(allocatedRouter, ws, {
        agentAddress: addr,
        repoId: { kind: "workflow-run", id: "another-deployment" },
        transferId: "allocated-wrong-repo",
        pack: new Uint8Array([1]),
        ref: "refs/heads/main",
        commitSha: "a".repeat(40),
      });
      await tick();
      expect(lastSent(ws)).toMatchObject({
        type: "repo.pack.reject",
        reason: "path_violation",
      });
      expect(sources).toEqual([]);

      pushPack(allocatedRouter, ws, {
        agentAddress: addr,
        repoId: {
          kind: "workflow-run",
          id: deriveWorkflowRunRepoId(addr),
        },
        transferId: "allocated-owned-repo",
        pack: new Uint8Array([2]),
        ref: "refs/heads/main",
        commitSha: "b".repeat(40),
      });
      await tick();
      expect(lastSent(ws).type).toBe("repo.pack.ack");
      expect(sources).toEqual([
        {
          kind: "allocated",
          agentAddress: addr,
          allocationId: identity.allocationId,
          anchorRunId: identity.anchorRunId,
          generation: 3,
        },
      ]);
    });

    test("workflow-run pack receive rejection is forwarded to the sidecar", async () => {
      const { router: r } = buildPackRouter({
        workflowRun: { accepted: false, reason: "path_violation" },
      });
      const ws = createMockWs();
      const addr = "ins_dep-wfr-rej@local";
      await registerAddr(r, ws, "sc-wfr-rej", addr);

      const transferId = "t-wfr-rej";
      const repoId: RepoId = {
        kind: "workflow-run",
        id: deriveWorkflowRunRepoId(addr),
      };
      pushPack(r, ws, {
        agentAddress: addr,
        repoId,
        transferId,
        pack: new Uint8Array([9, 9, 9]),
        ref: "refs/heads/events",
        commitSha: "c".repeat(40),
      });

      await new Promise((resolve) => setTimeout(resolve, 0));

      const last = lastSent(ws);
      expect(last.type).toBe("repo.pack.reject");
      expect(last.reason).toBe("path_violation");
      expect(last.transferId).toBe(transferId);
      expect(last.repoId).toEqual(repoId);
    });
  });

  describe("sendDrain", () => {
    test("ships a drain.deliver frame to the sidecar holding the deployment", async () => {
      const ws = createMockWs();
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-drain-1",
          token: "tok",
          agentAddresses: ["dep@integration.interchange"],
        }),
      );
      await tick();
      ws.sent.length = 0;

      router.sendDrain({
        agentAddress: "dep@integration.interchange",
        deadlineMs: 7_500,
      });

      const last = lastSent(ws);
      expect(last.type).toBe("drain.deliver");
      expect(last.agentAddress).toBe("dep@integration.interchange");
      expect(last.deadlineMs).toBe(7_500);
    });

    test("throws when no sidecar is registered for the address", () => {
      expect(() =>
        router.sendDrain({
          agentAddress: "absent@integration.interchange",
          deadlineMs: 1_000,
        }),
      ).toThrow(/No sidecar connected/);
    });
  });

  describe("signal.correlation.register ack", () => {
    const snapshot = {
      name: "charge_card",
      description: "Charge the customer's card",
      inputSchema: { type: "object" },
      arguments: { amount: 100 },
    };

    async function ownAddress(
      router: ReturnType<typeof createTestRouter>,
      ws: ReturnType<typeof createMockWs>,
      address: string,
    ): Promise<void> {
      router.handleOpen(ws);
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-1",
          token: "tok",
          agentAddresses: [address],
        }),
      );
      await tick();
    }

    test("acks the sender after the co-write resolves", async () => {
      const registered: string[] = [];
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          registerSignalCorrelation: async (reg) => {
            registered.push(reg.correlationId);
          },
        },
      });
      const ws = createMockWs();
      await ownAddress(router, ws, "agent-a@local");

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "signal.correlation.register",
          correlationId: "corr-1",
          runId: "run-1",
          deploymentId: "dep-1",
          agentAddress: "agent-a@local",
          kind: "approval",
          snapshot,
        }),
      );
      await tick();

      expect(registered).toEqual(["corr-1"]);
      const ack = ws.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "signal.correlation.register.ack");
      expect(ack).toBeDefined();
      expect(ack.correlationId).toBe("corr-1");
      expect(ack.agentAddress).toBe("agent-a@local");
    });

    test("does not ack when the co-write throws", async () => {
      const router = createTestRouter({
        lookups: {
          lookupPublicKey: async () => null,
          registerSignalCorrelation: async () => {
            throw new Error("no deployed deployment for address");
          },
        },
      });
      const ws = createMockWs();
      await ownAddress(router, ws, "agent-a@local");

      router.handleMessage(
        ws,
        JSON.stringify({
          type: "signal.correlation.register",
          correlationId: "corr-1",
          runId: "run-1",
          deploymentId: "dep-1",
          agentAddress: "agent-a@local",
          kind: "approval",
          snapshot,
        }),
      );
      await tick();

      const ack = ws.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "signal.correlation.register.ack");
      expect(ack).toBeUndefined();
    });
  });

  describe("connected-window mail redelivery (mail.inbound.ack)", () => {
    // Establish an address over the challenged reconnect path so it enters the
    // routing table under a verified connection -- the same path a real
    // sidecar hosting a keyed address takes. Routing a keyed address via a
    // plain register is refused by the key-existence gate.
    async function connectViaChallenge(
      r: ReturnType<typeof createSidecarRouter>,
      ws: ReturnType<typeof createMockWs>,
      addr: string,
      privateKey: Uint8Array,
      sidecarId = "sc-1",
    ) {
      r.handleOpen(ws);
      r.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId,
          token: "tok",
          agentAddresses: [addr],
        }),
      );
      await new Promise((res) => setTimeout(res, 50));
      const challengeFrame = ws.sent
        .map((s) => JSON.parse(s))
        .find((f: { type: string }) => f.type === "challenge");
      const responses = await Promise.all(
        challengeFrame.challenges.map(
          async (c: { address: string; nonce: string }) => ({
            address: c.address,
            signature: await signChallenge(c.nonce, c.address, privateKey),
          }),
        ),
      );
      r.handleMessage(
        ws,
        JSON.stringify({ type: "challenge.response", responses }),
      );
      await new Promise((res) => setTimeout(res, 50));
    }

    // Count `mail.inbound` frames the ws received carrying `messageId`. Each
    // (re)delivery is a fresh send of identical bytes, so the count is the
    // original delivery plus every redelivery attempt.
    function inboundCount(
      ws: ReturnType<typeof createMockWs>,
      messageId: string,
    ): number {
      return ws.sent
        .map((s) => JSON.parse(s))
        .filter((f) => f.type === "mail.inbound" && f.messageId === messageId)
        .length;
    }

    test("redelivers identical bytes until the sidecar acks", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 5,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws = createMockWs();
      await connectViaChallenge(router, ws, "agent@local", kp.privateKey);

      expect(router.routeMail("agent@local", "aGVsbG8=", "mid-1")).toBe(true);
      // Delivered once immediately, carrying the messageId.
      expect(inboundCount(ws, "mid-1")).toBe(1);
      const delivered = ws.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "mail.inbound" && f.messageId === "mid-1");
      expect(delivered.rawMessage).toBe("aGVsbG8=");

      // Withhold the ack: the retry timer redelivers identical bytes.
      await new Promise((res) => setTimeout(res, 50));
      expect(inboundCount(ws, "mid-1")).toBeGreaterThanOrEqual(2);
      const redelivered = ws.sent
        .map((s) => JSON.parse(s))
        .filter((f) => f.type === "mail.inbound" && f.messageId === "mid-1");
      // Every redelivery replays the same bytes and messageId.
      for (const f of redelivered) {
        expect(f.rawMessage).toBe("aGVsbG8=");
      }
    });

    test("an ack stops redelivery", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 5,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws = createMockWs();
      await connectViaChallenge(router, ws, "agent@local", kp.privateKey);

      expect(router.routeMail("agent@local", "aGk=", "mid-2")).toBe(true);
      expect(inboundCount(ws, "mid-2")).toBe(1);

      // The sidecar acks its durable inbox write.
      router.handleMessage(
        ws,
        JSON.stringify({
          type: "mail.inbound.ack",
          agentAddress: "agent@local",
          messageId: "mid-2",
        }),
      );
      await tick();

      // No redelivery fires after the ack clears the pending entry.
      await new Promise((res) => setTimeout(res, 60));
      expect(inboundCount(ws, "mid-2")).toBe(1);
    });

    test("redelivery is bounded by the retry budget", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 3,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws = createMockWs();
      await connectViaChallenge(router, ws, "agent@local", kp.privateKey);

      const warnings: string[] = [];
      const restore = installWarningCapture(warnings);
      try {
        expect(router.routeMail("agent@local", "eA==", "mid-3")).toBe(true);
        // Wait well past the full budget (interval * (maxRetries + 1)).
        await new Promise((res) => setTimeout(res, 200));
      } finally {
        restore();
      }

      // One original delivery plus exactly maxRetries redeliveries, then stop.
      expect(inboundCount(ws, "mid-3")).toBe(4);
      expect(
        warnings.some(
          (w) => w.includes("mid-3") && w.includes("Gave up redelivering"),
        ),
      ).toBe(true);
    });

    test("give-up surfaces the un-acked mail as undelivered on a still-live connection", async () => {
      const kp = await generateKeyPair();
      const undelivered: { rawMessage: string; recipients: string[] }[] = [];
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 3,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      // The only channel through which a dropped mail is recovered externally.
      router.events.on("mail.outbound.undelivered", (e) => {
        undelivered.push(e);
      });

      const ws = createMockWs();
      await connectViaChallenge(router, ws, "agent@local", kp.privateKey);

      // A tracked mail carrying a hub-minted messageId. Durable delivery is
      // what keeps the accepted bytes available through the retry window.
      expect(router.routeMail("agent@local", "eA==", "mid-drop")).toBe(true);

      const warnings: string[] = [];
      const restore = installWarningCapture(warnings);
      try {
        // The sidecar never acks (its local substrate write persistently
        // fails), but the WS stays open the whole time -- no ping timeout, no
        // close. Wait well past the full budget.
        await new Promise((res) => setTimeout(res, 200));
      } finally {
        restore();
      }

      // 1 original + 3 retries, then give up.
      expect(inboundCount(ws, "mid-drop")).toBe(4);
      // Give-up surfaces the mail for external relay exactly once, rather than
      // silently dropping a committed run's trigger.
      const forMidDrop = undelivered.filter((e) => e.rawMessage === "eA==");
      expect(forMidDrop).toHaveLength(1);
      expect(forMidDrop[0]?.recipients).toEqual(["agent@local"]);
      expect(
        warnings.some(
          (w) => w.includes("mid-drop") && w.includes("Gave up redelivering"),
        ),
      ).toBe(true);
    });

    test("mail without a messageId is not tracked for redelivery", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 5,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws = createMockWs();
      await connectViaChallenge(router, ws, "agent@local", kp.privateKey);

      expect(router.routeMail("agent@local", "eXk=")).toBe(true);
      const before = ws.sent.filter(
        (s) => JSON.parse(s).type === "mail.inbound",
      ).length;
      expect(before).toBe(1);

      await new Promise((res) => setTimeout(res, 60));
      const after = ws.sent.filter(
        (s) => JSON.parse(s).type === "mail.inbound",
      ).length;
      // No retry fired: the delivery carried no messageId, so no ack handshake.
      expect(after).toBe(1);
    });

    test("an ack from a non-owning sidecar does not clear pending mail", async () => {
      const kp = await generateKeyPair();
      const other = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 5,
        lookups: {
          lookupPublicKey: async (addr: string) =>
            addr === "agent@local"
              ? hexEncode(kp.publicKey)
              : hexEncode(other.publicKey),
        },
      });
      const ownerWs = createMockWs();
      await connectViaChallenge(router, ownerWs, "agent@local", kp.privateKey);
      const rogueWs = createMockWs();
      await connectViaChallenge(
        router,
        rogueWs,
        "other@local",
        other.privateKey,
        "sc-2",
      );

      expect(router.routeMail("agent@local", "aGV5", "mid-4")).toBe(true);
      expect(inboundCount(ownerWs, "mid-4")).toBe(1);

      // A sidecar that does not own agent@local acks its messageId. The gate
      // drops it, so the pending entry survives and redelivery continues.
      router.handleMessage(
        rogueWs,
        JSON.stringify({
          type: "mail.inbound.ack",
          agentAddress: "agent@local",
          messageId: "mid-4",
        }),
      );
      await tick();

      await new Promise((res) => setTimeout(res, 50));
      expect(inboundCount(ownerWs, "mid-4")).toBeGreaterThanOrEqual(2);
    });

    test("un-acked mail is retained across a disconnect and redelivered on reconnect", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        // Large intervals so neither the connected-window retry nor the
        // retention TTL fires during the test window; the redelivery under
        // test is driven by the reconnect, not a timer.
        mailAckRetryIntervalMs: 10_000,
        mailAckMaxRetries: 5,
        disconnectQueueTTLMs: 60_000,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws1 = createMockWs();
      await connectViaChallenge(router, ws1, "agent@local", kp.privateKey);
      expect(router.routeMail("agent@local", "aGVsbG8=", "mid-r1")).toBe(true);
      expect(inboundCount(ws1, "mid-r1")).toBe(1);

      // Disconnect BEFORE any ack: the pending entry must be retained.
      router.handleClose(ws1);

      // Reconnect on a fresh connection; the retained mail is redelivered with
      // identical bytes and the same messageId.
      const ws2 = createMockWs();
      await connectViaChallenge(router, ws2, "agent@local", kp.privateKey);
      expect(inboundCount(ws2, "mid-r1")).toBe(1);
      const redelivered = ws2.sent
        .map((s) => JSON.parse(s))
        .find((f) => f.type === "mail.inbound" && f.messageId === "mid-r1");
      expect(redelivered.rawMessage).toBe("aGVsbG8=");
    });

    test("an ack after reconnect redelivery clears the retained mail", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 20,
        mailAckMaxRetries: 5,
        disconnectQueueTTLMs: 60_000,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws1 = createMockWs();
      await connectViaChallenge(router, ws1, "agent@local", kp.privateKey);
      expect(router.routeMail("agent@local", "aGk=", "mid-r2")).toBe(true);
      router.handleClose(ws1);

      const ws2 = createMockWs();
      await connectViaChallenge(router, ws2, "agent@local", kp.privateKey);
      expect(inboundCount(ws2, "mid-r2")).toBeGreaterThanOrEqual(1);

      // Ack over the reconnected connection; retries must stop afterward.
      router.handleMessage(
        ws2,
        JSON.stringify({
          type: "mail.inbound.ack",
          agentAddress: "agent@local",
          messageId: "mid-r2",
        }),
      );
      await tick();
      const countAtAck = inboundCount(ws2, "mid-r2");
      await new Promise((res) => setTimeout(res, 60));
      expect(inboundCount(ws2, "mid-r2")).toBe(countAtAck);
    });

    test("retained mail is dropped after the retention TTL and not redelivered", async () => {
      const kp = await generateKeyPair();
      const router = createTestRouter({
        requestTimeoutMs: 500,
        mailAckRetryIntervalMs: 10_000,
        mailAckMaxRetries: 5,
        disconnectQueueTTLMs: 30,
        lookups: {
          async lookupPublicKey() {
            return hexEncode(kp.publicKey);
          },
        },
      });
      const ws1 = createMockWs();
      await connectViaChallenge(router, ws1, "agent@local", kp.privateKey);
      expect(router.routeMail("agent@local", "eA==", "mid-r3")).toBe(true);

      const warnings: string[] = [];
      const restore = installWarningCapture(warnings);
      try {
        // Disconnect arms the retention TTL; wait past it so the entry drops.
        router.handleClose(ws1);
        await new Promise((res) => setTimeout(res, 80));
      } finally {
        restore();
      }
      expect(
        warnings.some(
          (w) =>
            w.includes("agent@local") && w.includes("retention TTL expired"),
        ),
      ).toBe(true);

      // A later reconnect finds nothing to redeliver.
      const ws2 = createMockWs();
      await connectViaChallenge(router, ws2, "agent@local", kp.privateKey);
      expect(inboundCount(ws2, "mid-r3")).toBe(0);
    });
  });

  describe("exclusive allocation routing", () => {
    const allocationIdentity = {
      kind: "allocated" as const,
      sidecarId: "sc-allocated",
      allocationId: "alloc-1",
      tenantId: "tenant-1",
      anchorRunId: "run-anchor",
      workflowRunAddress: "workflow@exclusive",
      generation: 1,
    };

    const allocationConfig = {
      sessionId: "ses-exclusive",
      agentId: "workflow",
      tenantId: "tenant-1",
      principalId: "principal-1",
      agentAddress: "workflow@exclusive",
      systemPrompt: "test",
      tools: [],
      grants: [],
      sources: TEST_SOURCES,
      defaultSource: TEST_DEFAULT_SOURCE,
    };

    test("redelivers retained mail once when an allocated generation reconnects", async () => {
      const mailCount = (
        ws: ReturnType<typeof createMockWs>,
        messageId: string,
      ) =>
        ws.sent
          .map((sent) => JSON.parse(sent))
          .filter(
            (frame) =>
              frame.type === "mail.inbound" && frame.messageId === messageId,
          ).length;
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        mailAckRetryIntervalMs: 10_000,
        disconnectQueueTTLMs: 60_000,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);

      const ws1 = createMockWs();
      allocatedRouter.handleOpen(ws1);
      allocatedRouter.handleMessage(
        ws1,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [allocationIdentity.workflowRunAddress],
        }),
      );
      await tick();
      expect(
        allocatedRouter.routeMail(
          allocationIdentity.workflowRunAddress,
          "aGVsbG8=",
          "mid-allocated-reconnect",
        ),
      ).toBe(true);

      allocatedRouter.handleClose(ws1);

      const ws2 = createMockWs();
      allocatedRouter.handleOpen(ws2);
      const reconnectFrame = JSON.stringify({
        type: "reconnect",
        sidecarId: "sc-allocated",
        token: "token",
        agentAddresses: [allocationIdentity.workflowRunAddress],
      });
      allocatedRouter.handleMessage(ws2, reconnectFrame);
      await tick();

      expect(mailCount(ws2, "mid-allocated-reconnect")).toBe(1);
      const redelivered = ws2.sent
        .map((sent) => JSON.parse(sent))
        .find(
          (frame) =>
            frame.type === "mail.inbound" &&
            frame.messageId === "mid-allocated-reconnect",
        );
      expect(redelivered.rawMessage).toBe("aGVsbG8=");

      allocatedRouter.handleMessage(ws2, reconnectFrame);
      await tick();
      expect(mailCount(ws2, "mid-allocated-reconnect")).toBe(1);
    });

    test("reports an active workflow advertised by the allocated generation", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);

      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: ["workflow@exclusive"],
        }),
      );
      await tick();

      expect(
        await allocatedRouter.isAllocatedWorkflowActive({
          allocationId: "alloc-1",
          generation: 1,
        }),
      ).toBe(true);
    });

    test("routes only allocation-targeted deploys to an allocated worker", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        hubPublicKey: TEST_HUB_KEY,
        requestTimeoutMs: 500,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      let deployAck: unknown;
      allocatedRouter.events.on("agent.deploy.ack", (event) => {
        deployAck = event;
      });

      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      expect(
        await allocatedRouter.isAllocatedSidecarReady({
          allocationId: "alloc-1",
          generation: 1,
        }),
      ).toBe(true);
      await expect(
        allocatedRouter.sendAgentDeploy("ordinary@shared", allocationConfig),
      ).rejects.toThrow("No sidecar available");

      const deployed = allocatedRouter.sendAgentDeployToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        "workflow@exclusive",
        allocationConfig,
      );
      await tick();
      expect(lastSent(ws).type).toBe("agent.deploy");
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "workflow@exclusive",
          publicKey: "b".repeat(64),
        }),
      );
      await expect(deployed).resolves.toEqual({ publicKey: "b".repeat(64) });
      expect(deployAck).toMatchObject({
        agentAddress: "workflow@exclusive",
        allocated: {
          allocationId: "alloc-1",
          anchorRunId: "run-anchor",
          generation: 1,
        },
      });
    });

    test("disconnect rejects an allocation-targeted deploy", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        hubPublicKey: TEST_HUB_KEY,
        requestTimeoutMs: 500,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);

      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      const deployed = allocatedRouter.sendAgentDeployToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        "workflow@exclusive",
        allocationConfig,
      );
      await tick();
      allocatedRouter.handleClose(ws);

      await expect(deployed).rejects.toThrow("disconnected");
    });

    test("restores a workflow-run pack before the deployment address is routed", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        hubPublicKey: TEST_HUB_KEY,
        requestTimeoutMs: 500,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);

      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      expect(allocatedRouter.getRoutableAddresses()).not.toContain(
        allocationIdentity.workflowRunAddress,
      );
      const restored = allocatedRouter.sendWorkflowRunPackToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        allocationIdentity.workflowRunAddress,
        new Uint8Array([1, 2, 3]),
        "refs/heads/events",
        "d".repeat(40),
      );
      await tick();

      const done = lastSent(ws);
      expect(done).toMatchObject({
        type: "repo.pack.done",
        agentAddress: allocationIdentity.workflowRunAddress,
        repoId: {
          kind: "workflow-run",
          id: deriveWorkflowRunRepoId(allocationIdentity.workflowRunAddress),
        },
        ref: "refs/heads/events",
        commitSha: "d".repeat(40),
      });
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "repo.pack.ack",
          agentAddress: allocationIdentity.workflowRunAddress,
          repoId: done.repoId,
          transferId: done.transferId,
        }),
      );

      await expect(restored).resolves.toBeUndefined();
      expect(allocatedRouter.getRoutableAddresses()).not.toContain(
        allocationIdentity.workflowRunAddress,
      );
    });

    test("refuses to restore Hub history over an already-active workflow", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        hubPublicKey: TEST_HUB_KEY,
        requestTimeoutMs: 500,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      const routableWhenConnected: string[][] = [];
      allocatedRouter.events.on("sidecar.allocated.connected", () => {
        routableWhenConnected.push(allocatedRouter.getRoutableAddresses());
      });

      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [allocationIdentity.workflowRunAddress],
          deployRefs: {},
        }),
      );
      await tick();
      const sentBeforeRestore = ws.sent.length;

      expect(routableWhenConnected).toEqual([
        [allocationIdentity.workflowRunAddress],
      ]);

      await expect(
        allocatedRouter.sendWorkflowRunPackToAllocation(
          { allocationId: "alloc-1", generation: 1 },
          allocationIdentity.workflowRunAddress,
          new Uint8Array([1, 2, 3]),
          "refs/heads/main",
          "d".repeat(40),
        ),
      ).rejects.toThrow("refusing to overwrite its run history");
      expect(ws.sent).toHaveLength(sentBeforeRestore);
    });

    test("delivers grants and durable mail to the exact generation and emits its ack", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
        hubPublicKey: TEST_HUB_KEY,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      const acknowledgements: unknown[] = [];
      allocatedRouter.events.on("mail.inbound.acknowledged", (event) => {
        acknowledgements.push(event);
      });
      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      const deployed = allocatedRouter.sendAgentDeployToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        "workflow@exclusive",
        allocationConfig,
      );
      await tick();
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "agent.deploy.ack",
          agentAddress: "workflow@exclusive",
          publicKey: "b".repeat(64),
        }),
      );
      await deployed;

      await allocatedRouter.sendWorkflowRunDispatchToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        "workflow@exclusive",
        "workflow@exclusive",
        [],
        "cmF3LW1haWw=",
        "message-1",
      );

      const frames = ws.sent.map((raw) => JSON.parse(raw));
      expect(frames.slice(-2).map((frame) => frame.type)).toEqual([
        "run.grants",
        "mail.inbound",
      ]);
      expect(frames.at(-1)).toMatchObject({
        agentAddress: "workflow@exclusive",
        messageId: "message-1",
        rawMessage: "cmF3LW1haWw=",
      });

      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "mail.inbound.ack",
          agentAddress: "workflow@exclusive",
          messageId: "message-1",
        }),
      );
      await tick();
      expect(acknowledgements).toEqual([
        {
          agentAddress: "workflow@exclusive",
          messageId: "message-1",
          allocated: {
            allocationId: "alloc-1",
            anchorRunId: "run-anchor",
            generation: 1,
          },
        },
      ]);

      await allocatedRouter.sendSignalDeliverToAllocation(
        { allocationId: "alloc-1", generation: 1 },
        {
          agentAddress: "workflow@exclusive",
          runId: "workflow@exclusive",
          signalName: "continue",
          signalId: "signal-1",
          payload: { approved: true },
        },
      );
      expect(lastSent(ws)).toMatchObject({
        type: "signal.deliver",
        signalId: "signal-1",
        payload: { approved: true },
      });
    });

    test("emits exact allocation lifecycle events only for the current socket", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      const connected: { allocationId: string; generation: number }[] = [];
      const disconnected: (
        | {
            allocationId: string;
            generation: number;
          }
        | undefined
      )[] = [];
      allocatedRouter.events.on("sidecar.allocated.connected", (target) => {
        connected.push(target);
      });
      allocatedRouter.events.on("sidecar.disconnect", ({ allocated }) => {
        disconnected.push(allocated);
      });

      const oldWs = createMockWs();
      allocatedRouter.handleOpen(oldWs);
      allocatedRouter.handleMessage(
        oldWs,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      const currentWs = createMockWs();
      allocatedRouter.handleOpen(currentWs);
      allocatedRouter.handleMessage(
        currentWs,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      // The same-generation takeover closes the old socket but must not be
      // interpreted as capacity loss. Closing the new current socket does.
      expect(disconnected).toEqual([undefined]);
      allocatedRouter.handleClose(currentWs);

      expect(connected).toEqual([
        { allocationId: "alloc-1", generation: 1 },
        { allocationId: "alloc-1", generation: 1 },
      ]);
      expect(disconnected).toEqual([
        undefined,
        { allocationId: "alloc-1", generation: 1 },
      ]);
    });

    test("advancing the fence closes the previous generation", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "register",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: [],
        }),
      );
      await tick();

      allocatedRouter.fenceAllocation("alloc-1", 2);

      expect(ws.closed).toBe(true);
      expect(
        await allocatedRouter.isAllocatedSidecarReady({
          allocationId: "alloc-1",
          generation: 1,
        }),
      ).toBe(false);
    });

    test("rejects an allocated credential claiming an unrelated address", async () => {
      const allocatedRouter = createTestRouter({
        authenticateSidecar: async () => allocationIdentity,
        validateSidecarIdentity: async () => true,
      });
      allocatedRouter.fenceAllocation("alloc-1", 1);
      const ws = createMockWs();
      allocatedRouter.handleOpen(ws);
      allocatedRouter.handleMessage(
        ws,
        JSON.stringify({
          type: "reconnect",
          sidecarId: "sc-allocated",
          token: "token",
          agentAddresses: ["other@tenant"],
          deployRefs: {},
        }),
      );
      await tick();

      expect(ws.closed).toBe(true);
      expect(allocatedRouter.getRoutableAddresses()).toEqual([]);
    });
  });
});
