import { describe, it } from "@std/testing/bdd";
import { expect } from "@std/expect";
import { stub } from "@std/testing/mock";
import { FakeTime } from "@std/testing/time";
import {
  fabricFromRealmValue,
  realmFromFabricValue,
} from "@commonfabric/data-model/codecs";
import {
  FabricBytes,
  FabricKeyPair,
} from "@commonfabric/data-model/fabric-primitives";
import type { FabricValue } from "@commonfabric/data-model";
import type { DID } from "@commonfabric/identity";
import { toValuePath } from "@commonfabric/memory/v2";
import { getLogger } from "@commonfabric/utils/logger";

import { AUTHOR_CLAIM_SCHEMA, observeAuthorship } from "@/authorship.ts";
import { CellHandle } from "@/cell-handle.ts";
import { RuntimeConnection } from "@/client/connection.ts";
import { EventEmitter } from "@/client/emitter.ts";
import {
  type CellRef,
  type InitializationData,
  type IPCClientMessage,
  type IPCClientNotification,
  isCellUpdateNotification,
  NotificationType,
  type OperationUpdateNotification,
  RequestType,
  RuntimeErrorCode,
  type RuntimeSecurityContext,
} from "@/protocol/mod.ts";
import type {
  RuntimeTransport,
  RuntimeTransportEvents,
} from "@/client/transport.ts";
import { $conn, type RuntimeClient } from "@/runtime-client.ts";

/**
 * Transport that records everything sent and auto-acknowledges every request
 * except the types in `holdTypes`, which are left pending (simulating
 * in-flight work). A request whose type `answers` names is answered with
 * that data, and any other with none.
 */
class FakeTransport extends EventEmitter<RuntimeTransportEvents>
  implements RuntimeTransport {
  readonly sent: Array<IPCClientMessage | IPCClientNotification> = [];
  disposeCalls = 0;

  #holdTypes: RequestType[];
  #answers: Partial<Record<RequestType, Record<string, unknown>>>;

  constructor(
    holdTypes: RequestType[] = [],
    answers: Partial<Record<RequestType, Record<string, unknown>>> = {},
  ) {
    super();
    this.#holdTypes = holdTypes;
    this.#answers = answers;
  }

  send(original: IPCClientMessage | IPCClientNotification): void {
    // Encoded, cloned and decoded, which is what the real transport does and
    // what a bare `structuredClone` would misreport: the clone alone strips a
    // `FabricBytes` to a plain object, so a double using it would model a
    // transport that loses exactly what this connection carries. The round
    // trip also gives the capture its own copy, so what is captured cannot
    // change under later mutation by the sender.
    const message = fabricFromRealmValue(
      structuredClone(realmFromFabricValue(original)),
    ) as IPCClientMessage | IPCClientNotification;
    this.sent.push(message);
    // Notifications carry no msgId and get no reply.
    if (!("msgId" in message)) return;
    if (this.#holdTypes.includes(message.data.type)) return;
    // Acknowledge asynchronously, like a real worker round-trip.
    queueMicrotask(() => {
      this.emit("message", {
        msgId: message.msgId,
        data: this.#answers[message.data.type],
      });
    });
  }

  dispose(): Promise<void> {
    this.disposeCalls++;
    return Promise.resolve();
  }
}

async function initializedConnection(
  transport: FakeTransport,
): Promise<RuntimeConnection> {
  const connection = new RuntimeConnection(transport);
  await connection.initialize({} as InitializationData);
  return connection;
}

/**
 * Transport whose `send()` throws, as the real one does two ways: the
 * envelope's encode refuses a value that has no encoding, and `postMessage()`
 * refuses one structured cloning cannot carry.
 */
class ThrowingTransport extends EventEmitter<RuntimeTransportEvents>
  implements RuntimeTransport {
  readonly #failOn: RequestType;

  constructor(failOn: RequestType) {
    super();
    this.#failOn = failOn;
  }

  send(message: IPCClientMessage | IPCClientNotification): void {
    if (!("msgId" in message)) return;
    if (message.data.type === this.#failOn) {
      throw new Error("no encoding for value");
    }
    // Everything else is acknowledged, so the connection can initialize.
    queueMicrotask(() => {
      this.emit("message", { msgId: message.msgId, data: undefined });
    });
  }

  dispose(): Promise<void> {
    return Promise.resolve();
  }
}

describe("connection", () => {
  it("records an abandoned request exactly once even if its reply arrives late", async () => {
    // Disposal ends the wait; the worker knows nothing of that and answers
    // later. The late reply finds no pending entry, and must not record a
    // second terminal wait for the same request.
    const transport = new FakeTransport([RequestType.Idle]);
    const connection = await initializedConnection(transport);
    const logger = getLogger("runtime-client");
    logger.resetTimeStats();
    const request = connection.request({ type: RequestType.Idle });
    const rejection = expect(request).rejects.toThrow();
    await connection.dispose();
    await rejection;

    expect(connection.getPendingRequestDiagnostics()).toEqual([]);
    const idleEntry = () =>
      connection.getRequestTimelineDiagnostics().find((entry) =>
        entry.type === RequestType.Idle
      );
    expect(idleEntry()).toMatchObject({
      type: RequestType.Idle,
      error: true,
      outcome: "cancelled",
    });
    expect(logger.getTimeStats("ipc", RequestType.Idle)).toMatchObject({
      count: 1,
    });
    expect(logger.getTimeStats("ipc-outcome", "cancelled", RequestType.Idle))
      .toMatchObject({ count: 1 });

    using _warning = stub(console, "warn");
    const sent = transport.sent.find((message) =>
      "msgId" in message && message.data.type === RequestType.Idle
    ) as IPCClientMessage;
    transport.emit("message", { msgId: sent.msgId });
    expect(logger.getTimeStats("ipc", RequestType.Idle)?.count).toBe(1);
    expect(idleEntry()).toMatchObject({ outcome: "cancelled" });
  });

  it("keeps outcome statistics after the boot timeline fills", async () => {
    // The timeline is capped at the boot window, so a later request records no
    // entry in it. The outcome statistics are uncapped and must still count it.
    const transport = new FakeTransport([RequestType.Idle]);
    const connection = await initializedConnection(transport);
    for (let index = 0; index < 100; index++) {
      await connection.request({ type: RequestType.GetLoggerCounts });
    }
    const timeline = connection.getRequestTimelineDiagnostics();
    const logger = getLogger("runtime-client");
    logger.resetTimeStats();
    const request = connection.request({ type: RequestType.Idle });
    const rejection = expect(request).rejects.toThrow();
    await connection.dispose();
    await rejection;
    expect(connection.getRequestTimelineDiagnostics()).toEqual(timeline);
    expect(
      logger.getTimeStats("ipc-outcome", "cancelled", RequestType.Idle)?.count,
    )
      .toBe(1);
  });

  it("records cancellation as a terminal outcome", async () => {
    const transport = new FakeTransport([RequestType.Idle]);
    const connection = await initializedConnection(transport);
    const request = connection.request({ type: RequestType.Idle });
    const rejection = expect(request).rejects.toThrow();
    await connection.dispose();
    await rejection;
    expect(
      connection.getRequestTimelineDiagnostics().find((entry) =>
        entry.type === RequestType.Idle
      ),
    ).toMatchObject({
      doneAtMs: expect.any(Number),
      outcome: "cancelled",
      error: true,
    });
  });

  it("routes terminal event-attention notifications", async () => {
    const transport = new FakeTransport();
    const connection = await initializedConnection(transport);
    const messages: unknown[] = [];
    connection.on("eventneedsattention", (message) => messages.push(message));
    const notification = {
      type: NotificationType.EventNeedsAttention,
      space: "did:key:z6Mk-connection-attention",
      eventId: "evt-attention",
      seq: 17,
      sidecarId: "of:stream-events:connection-attention",
      reason: "delivery needs attention",
      attention: {
        phase: "dispatch-load",
        failureClass: "session-revoked",
        code: "permanent-delivery-failure",
        firstFailureAt: 10,
        lastFailureAt: 10,
        accumulatedFailureMs: 0,
        failureCount: 1,
        recovery: "explicit-retry",
      },
    } as const;

    transport.emit("message", notification);

    expect(messages).toEqual([notification]);
    await connection.dispose();
  });

  it("routes collaborative operation notifications", async () => {
    const transport = new FakeTransport();
    const connection = await initializedConnection(transport);
    const messages: unknown[] = [];
    connection.on("operationupdate", (message) => messages.push(message));

    const notification: OperationUpdateNotification = {
      type: NotificationType.OperationUpdate,
      subscriptionId: "subscription:1",
      field: {
        branch: "",
        id: "of:connection",
        scopeKey: "space",
        path: toValuePath([]),
        active: false,
        codec: null,
        cursor: null,
        baselineHash: "baseline",
        materialized: "value",
        operations: [],
      },
    };
    transport.emit("message", notification);

    expect(messages).toEqual([notification]);
    await connection.dispose();
  });

  describe("cell updates the worker refuses", () => {
    // Each case sends a refused update the way the worker does, through the
    // realm encoding, so that what arrives is what a real transport hands
    // over.

    const ref: CellRef = {
      id: "of:connection-refused-cell" as CellRef["id"],
      space: "did:key:test" as CellRef["space"],
      scope: "space",
      path: [],
    };
    const refusal = { refusedBy: "display-ceiling" } as const;
    /** The runtime a handle reaches `connection` through. */
    const runtimeOf = (connection: RuntimeConnection) =>
      ({ [$conn]: () => connection }) as unknown as RuntimeClient;
    const posted = (message: Record<string, FabricValue>) => {
      const arrived = fabricFromRealmValue(
        structuredClone(realmFromFabricValue({
          type: NotificationType.CellUpdate,
          cell: ref,
          ...message,
        })),
      );
      if (!isCellUpdateNotification(arrived)) {
        throw new Error("The encoding did not carry a cell update.");
      }
      return arrived;
    };

    it("hands each subscriber the refusal, and drops what it held", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const first = new CellHandle<string>(runtime, ref);
      const second = new CellHandle<string>(runtime, ref);
      const heard: unknown[] = [];
      first.subscribe(() => {}, { onRefused: (r) => heard.push(r) });
      second.subscribe(() => {}, { onRefused: () => {} });
      transport.emit("message", posted({ value: "shown before the seal" }));

      transport.emit("message", posted({ refused: refusal }));

      expect(heard).toEqual([refusal]);
      for (const handle of [first, second]) {
        expect(handle.lastRead()).toEqual({ refused: refusal });
      }
      await connection.dispose();
    });

    it("hands a later subscriber the refusal, once, rather than nothing", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const first = new CellHandle<string>(runtime, ref);
      first.subscribe(() => {}, { onRefused: () => {} });
      transport.emit("message", posted({ refused: refusal }));

      const later = new CellHandle<string>(runtime, ref);
      const heard: unknown[] = [];
      const values: unknown[] = [];
      later.subscribe((value) => {
        values.push(value);
      }, { onRefused: (r) => heard.push(r) });

      expect(later.refusal).toEqual(refusal);
      // One refusal stands, so the subscriber hears it once.
      expect(heard).toEqual([refusal]);
      expect(values).toEqual([]);
      await connection.dispose();
    });

    it("ends no refusal a later subscriber holds with a value another holds", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const first = new CellHandle<string>(runtime, ref);
      first.subscribe(() => {}, { onRefused: () => {} });
      transport.emit("message", posted({ value: "held by another" }));

      // What the other holds may be a write it made, or a copy it was made
      // with; only a read of the later one ends its refusal.
      const later = new CellHandle<string>(runtime, ref, { refused: refusal });
      later.subscribe(() => {}, { onRefused: () => {} });

      expect(later.lastRead()).toEqual({ refused: refusal });
      await connection.dispose();
    });
  });

  describe("label currency across handles of one cell", () => {
    // Each case delivers the worker's updates through the realm encoding, as
    // the refusal cases above do, so that a handle hears what a real
    // transport hands over.

    // The cell is bound under the claim schema, the one an authorship
    // observation reads an author cell under, so the handle an observation
    // reads through shares a subscription with the handles each case makes.
    const ref: CellRef = {
      id: "of:connection-label-currency-cell" as CellRef["id"],
      space: "did:key:test" as CellRef["space"],
      scope: "space",
      path: [],
      schema: AUTHOR_CLAIM_SCHEMA,
    };
    const label = (sender: string) => ({
      version: 1 as const,
      entries: [{
        path: [],
        label: { integrity: [{ kind: "authored-by", subject: sender }] },
      }],
    });
    /** The label of a profile `did:key:alice` owns, read through a link. */
    const profileLabel = {
      version: 1 as const,
      entries: [{
        path: [],
        label: {
          integrity: [{
            kind: "represents-principal",
            subject: "did:key:alice",
          }],
        },
      }],
    };
    const runtimeOf = (connection: RuntimeConnection) =>
      ({ [$conn]: () => connection }) as unknown as RuntimeClient;
    const posted = (message: Record<string, FabricValue>) => {
      const arrived = fabricFromRealmValue(
        structuredClone(realmFromFabricValue({
          type: NotificationType.CellUpdate,
          cell: ref,
          ...message,
        })),
      );
      if (!isCellUpdateNotification(arrived)) {
        throw new Error("The encoding did not carry a cell update.");
      }
      return arrived;
    };
    /** Subscribes label-aware, keeping each delivery's label and currency. */
    const listen = (cell: CellHandle<string>) => {
      const heard: Array<{ cfcLabel: unknown; current: boolean | undefined }> =
        [];
      const cancel = cell.subscribe((_value, cfcLabel, delivery) => {
        heard.push({ cfcLabel, current: delivery?.cfcLabelCurrent });
      }, { includeCfcLabel: true, onRefused: () => {} });
      return { heard, cancel };
    };

    it("hands a handle that rejoins a cell another handle watches without labels its retained label as not current, with each value-only update", async () => {
      // The handle heard a label, left, and came back to a subscription a
      // value-only handle had opened in the meantime: the worker's label has
      // moved on, and the connection delivers no labels on that subscription.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const labeled = new CellHandle<string>(runtime, ref);
      const first = listen(labeled);
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      expect(first.heard.at(-1)).toEqual({
        cfcLabel: label("alice"),
        current: true,
      });
      first.cancel();
      const valueOnly = new CellHandle<string>(runtime, ref);
      const valueOnlyCancel = valueOnly.subscribe(() => {}, {
        onRefused: () => {},
      });

      const again = listen(labeled);
      transport.emit("message", posted({ value: "revised by bob" }));

      expect(again.heard).toEqual([
        { cfcLabel: label("alice"), current: false },
        { cfcLabel: label("alice"), current: false },
      ]);
      expect(labeled.get()).toBe("revised by bob");
      again.cancel();
      valueOnlyCancel();
      await connection.dispose();
    });

    it("seeds a handle new to a cell with another handle's label, as current as it is there", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const holder = new CellHandle<string>(runtime, ref);
      const holding = listen(holder);
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );

      const seededCurrent = listen(new CellHandle<string>(runtime, ref));
      transport.emit("message", posted({ value: "revised by bob" }));
      const seededStale = listen(new CellHandle<string>(runtime, ref));

      expect(seededCurrent.heard[0]).toEqual({
        cfcLabel: label("alice"),
        current: true,
      });
      expect(seededStale.heard[0]).toEqual({
        cfcLabel: label("alice"),
        current: false,
      });
      holding.cancel();
      seededCurrent.cancel();
      seededStale.cancel();
      await connection.dispose();
    });

    it("seeds a handle from a value-only handle with the label a label-only update gave the cell, as current", async () => {
      // A label-only update reaches every handle on the shared subscription;
      // one whose subscriber never asked for labels keeps the new label all
      // the same, so a handle seeded from it after the label-aware one left
      // starts from the worker's label.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const labeled = new CellHandle<string>(runtime, ref);
      const valueOnly = new CellHandle<string>(runtime, ref);
      const labeledListening = listen(labeled);
      const valueOnlyCancel = valueOnly.subscribe(() => {}, {
        onRefused: () => {},
      });
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("bob") }),
      );
      labeledListening.cancel();

      const seeded = listen(new CellHandle<string>(runtime, ref));

      expect(seeded.heard).toEqual([{
        cfcLabel: label("bob"),
        current: true,
      }]);
      seeded.cancel();
      valueOnlyCancel();
      await connection.dispose();
    });

    it("reports a seeded handle's authorship from the label a label-only update gave the cell", async () => {
      // The same handoff, read to the verdict: the badge on the seeded handle
      // decides on Bob's label, the worker's, not on Alice's, which the cell
      // carried first.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const labeled = new CellHandle<string>(runtime, ref);
      const valueOnly = new CellHandle<string>(runtime, ref);
      const labeledListening = listen(labeled);
      const valueOnlyCancel = valueOnly.subscribe(() => {}, {
        onRefused: () => {},
      });
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("bob") }),
      );
      labeledListening.cancel();

      const states: string[] = [];
      const cancel = observeAuthorship(
        new CellHandle<string>(runtime, ref),
        "alice",
        (observation) => states.push(observation.state),
      );
      // The observation reads through settled promises, so one zero-delay
      // turn lets its reads finish, as the authorship tests settle theirs.
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(states).toEqual(["unverified"]);
      cancel();
      valueOnlyCancel();
      await connection.dispose();
    });

    it("seeds a handle new to a cell the worker last reported as undefined with the label the cell held, as not current", async () => {
      // The connection drops an `undefined` value, so the handle that heard
      // it keeps what it held; the label it holds is no longer the one the
      // worker read with the cell's latest value.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const holding = listen(new CellHandle<string>(runtime, ref));
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      const heardBefore = [...holding.heard];
      transport.emit(
        "message",
        posted({ value: undefined, cfcLabel: label("bob") }),
      );

      const seeded = listen(new CellHandle<string>(runtime, ref));

      expect(seeded.heard).toEqual([{
        cfcLabel: label("alice"),
        current: false,
      }]);
      // A label-aware subscriber hears what its handle holds once more, as
      // not current.
      expect(holding.heard).toEqual([
        ...heardBefore,
        { cfcLabel: label("alice"), current: false },
      ]);
      holding.cancel();
      seeded.cancel();
      await connection.dispose();
    });

    it("reports a message whose author the worker reported as undefined while its handle upgraded to labels as unknown, not from the author's former label", async () => {
      // A handle upgrading a value-only subscription to a label-aware one
      // leaves the cell's subscription until the worker answers the
      // unsubscribe, so an update the worker sends in between reaches it in
      // no form; the label it held from before is not the worker's read.
      const transport = new FakeTransport([RequestType.CellUnsubscribe], {
        [RequestType.CellGetCfcLabel]: { cfcLabel: undefined },
        [RequestType.CellResolveAsCell]: { cell: ref },
      });
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const labeled = new CellHandle<string>(runtime, ref);
      const valueOnly = new CellHandle<string>(runtime, ref);
      const labeledListening = listen(labeled);
      const valueOnlyCancel = valueOnly.subscribe(() => {}, {
        onRefused: () => {},
      });
      transport.emit(
        "message",
        posted({ value: { name: "Alice" }, cfcLabel: profileLabel }),
      );
      labeledListening.cancel();
      const upgraded = listen(valueOnly);
      transport.emit("message", posted({ value: undefined }));

      const states: string[] = [];
      const cancel = observeAuthorship(
        { getCfcLabel: () => Promise.resolve(label("did:key:alice")) },
        valueOnly,
        (observation) => states.push(observation.state),
      );
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(upgraded.heard[0]).toEqual({
        cfcLabel: profileLabel,
        current: false,
      });
      expect(states).toEqual(["unknown"]);
      cancel();
      upgraded.cancel();
      valueOnlyCancel();
      await connection.dispose();
    });

    it("reports a message whose author the worker last reported as undefined as unknown on a handle seeded after, not from the author's former label", async () => {
      // An obliterated message keeps no `authorProfile`. A handle that
      // watched that field before still holds the profile and its
      // `represents-principal` label, and a reader opening the message
      // after must not take that label as the author's claim: the worker's
      // read of the field finds no label, so no author is claimed.
      const transport = new FakeTransport([], {
        [RequestType.CellGetCfcLabel]: { cfcLabel: undefined },
        [RequestType.CellResolveAsCell]: { cell: ref },
      });
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const watching = listen(new CellHandle<string>(runtime, ref));
      transport.emit(
        "message",
        posted({ value: { name: "Alice" }, cfcLabel: profileLabel }),
      );
      transport.emit(
        "message",
        posted({ value: undefined, cfcLabel: label("did:key:alice") }),
      );

      const states: string[] = [];
      const cancel = observeAuthorship(
        { getCfcLabel: () => Promise.resolve(label("did:key:alice")) },
        new CellHandle<string>(runtime, ref),
        (observation) => states.push(observation.state),
      );
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(states).toEqual(["unknown"]);
      cancel();
      watching.cancel();
      await connection.dispose();
    });

    it("hands a label-aware subscriber what the handle holds once more, as not current, when the worker reports the cell undefined", async () => {
      // The connection drops the update, so the handle keeps the value and
      // label the worker has read past. A second such update changes nothing
      // more.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const holder = new CellHandle<string>(runtime, ref);
      const holding = listen(holder);
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );

      transport.emit("message", posted({ value: undefined }));
      transport.emit("message", posted({ value: undefined }));

      expect(holding.heard).toEqual([
        { cfcLabel: undefined, current: false },
        { cfcLabel: label("alice"), current: true },
        { cfcLabel: label("alice"), current: false },
      ]);
      expect(holder.get()).toBe("by alice");
      holding.cancel();
      await connection.dispose();
    });

    it("hands a label-aware subscriber what the handle holds again, as current, when the worker restores the value and label it held", async () => {
      // After an update holding nothing, the worker's next update brings back
      // the value and label the handle kept: nothing it holds changes but
      // that the label is the worker's read again.
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const holding = listen(new CellHandle<string>(runtime, ref));
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      transport.emit("message", posted({ value: undefined }));

      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );

      expect(holding.heard).toEqual([
        { cfcLabel: undefined, current: false },
        { cfcLabel: label("alice"), current: true },
        { cfcLabel: label("alice"), current: false },
        { cfcLabel: label("alice"), current: true },
      ]);
      holding.cancel();
      await connection.dispose();
    });

    it("calls the subscriber of a handle that never asked for labels no more when the worker reports the cell undefined and then restores it, though the currency of the label the handle holds changes", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const labeled = listen(new CellHandle<string>(runtime, ref));
      const valueOnly = new CellHandle<string>(runtime, ref);
      const values: Array<string | undefined> = [];
      const valueOnlyCancel = valueOnly.subscribe((value) => {
        values.push(value);
      }, { onRefused: () => {} });
      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );
      expect(valueOnly.cfcLabelCurrent).toBe(true);

      transport.emit("message", posted({ value: undefined }));

      expect(values).toEqual([undefined, "by alice"]);
      expect(valueOnly.cfcLabelCurrent).toBe(false);

      transport.emit(
        "message",
        posted({ value: "by alice", cfcLabel: label("alice") }),
      );

      expect(values).toEqual([undefined, "by alice"]);
      expect(valueOnly.cfcLabelCurrent).toBe(true);
      labeled.cancel();
      valueOnlyCancel();
      await connection.dispose();
    });

    it("reports a message whose author the worker reports as undefined as unknown on a handle seeded before the report arrived", async () => {
      // A view opened while the worker's update emptying `authorProfile` is
      // still on its way is seeded with the profile and its
      // `represents-principal` label, as current, and verifies. Once the
      // update arrives, the label is the worker's no longer, and the worker,
      // asked, reads none.
      const transport = new FakeTransport([], {
        [RequestType.CellGetCfcLabel]: { cfcLabel: undefined },
        [RequestType.CellResolveAsCell]: { cell: ref },
      });
      const connection = await initializedConnection(transport);
      const runtime = runtimeOf(connection);
      const watching = listen(new CellHandle<string>(runtime, ref));
      transport.emit(
        "message",
        posted({ value: { name: "Alice" }, cfcLabel: profileLabel }),
      );
      const states: string[] = [];
      const cancel = observeAuthorship(
        { getCfcLabel: () => Promise.resolve(label("did:key:alice")) },
        new CellHandle<string>(runtime, ref),
        (observation) => states.push(observation.state),
      );
      await new Promise((resolve) => setTimeout(resolve, 0));
      expect(states).toEqual(["verified"]);

      transport.emit(
        "message",
        posted({ value: undefined, cfcLabel: label("did:key:alice") }),
      );
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(states).toEqual(["verified", "unknown"]);
      cancel();
      watching.cancel();
      await connection.dispose();
    });
  });

  describe("RuntimeConnection disposal", () => {
    it("settles in-flight requests as cancellation on dispose", async () => {
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);

      const inFlight = connection.request<RequestType.Idle>({
        type: RequestType.Idle,
      });
      // Avoid an unhandled rejection between dispose() and the assertion.
      const settled = inFlight.then(() => undefined, (error) => error);

      await connection.dispose();

      const error = await settled;
      expect(connection.signal.aborted).toBe(true);
      // Cancelled with the standard abort reason, not a bespoke error.
      expect(error).toBe(connection.signal.reason);
      expect((error as { name?: string })?.name).toBe("AbortError");
    });

    it("runs registered teardown synchronously, before transport teardown", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);

      const order: string[] = [];
      connection.onDispose(() => order.push("teardown"));
      const realDispose = transport.dispose.bind(transport);
      transport.dispose = () => {
        order.push("transport");
        return realDispose();
      };

      await connection.dispose();

      expect(order).toEqual(["teardown", "transport"]);
    });

    it("does not send requests issued after dispose", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      await connection.dispose();

      const sentBefore = transport.sent.length;
      const error = await connection
        .request<RequestType.Idle>({ type: RequestType.Idle })
        .then(() => undefined, (e) => e);

      expect(connection.signal.aborted).toBe(true);
      expect(error).toBe(connection.signal.reason);
      // The request never reached the transport.
      expect(transport.sent.length).toBe(sentBefore);
    });

    it("runs teardown immediately when registered after dispose", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      await connection.dispose();

      let ran = false;
      connection.onDispose(() => {
        ran = true;
      });
      expect(ran).toBe(true);
    });

    it("is idempotent", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);

      await connection.dispose();
      await connection.dispose();

      expect(transport.disposeCalls).toBe(1);
    });

    it("waits for the worker to confirm dispose before tearing down", async () => {
      // Hold the Dispose reply: the worker has not yet confirmed its flush.
      const transport = new FakeTransport([RequestType.Dispose]);
      const connection = await initializedConnection(transport);

      let done = false;
      const disposing = connection.dispose().then(() => {
        done = true;
      });

      // Dispose stays pending until the worker confirms its flush. Draining the
      // task queue (all microtasks, then one macrotask) runs every continuation a
      // completed dispose would take; the held Dispose reply never arrives, so a
      // correct dispose is still pending and the transport is untouched.
      // `dispose()` has already aborted the lifetime by this point, and the
      // Dispose request carries no abort listener, so what is observed here is
      // a request only its reply can settle.
      await new Promise((resolve) => setTimeout(resolve, 0));
      expect(done).toBe(false);
      expect(transport.disposeCalls).toBe(0);

      // The worker finishes flushing and replies; teardown then completes.
      const disposeMsg = transport.sent.find(
        (m): m is IPCClientMessage =>
          "msgId" in m && m.data.type === RequestType.Dispose,
      );
      transport.emit("message", { msgId: disposeMsg!.msgId });

      await disposing;
      expect(done).toBe(true);
      expect(transport.disposeCalls).toBe(1);
    });

    it("ignores incoming messages without warning once dead", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      await connection.dispose();

      const warnings: unknown[][] = [];
      const original = console.warn;
      console.warn = (...args: unknown[]) => warnings.push(args);
      try {
        // A late response for an unknown request, a malformed message, and a
        // stray notification — all ignored silently in the dead state.
        transport.emit("message", { msgId: 9999 });
        transport.emit("message", "garbage" as never);
        transport.emit(
          "message",
          { type: "cell:update", cell: {}, value: 1 } as never,
        );
      } finally {
        console.warn = original;
      }

      expect(warnings.length).toBe(0);
    });

    it("leaves nothing pending when the send itself throws", async () => {
      // The throw reaches the caller, and the promise `request()` would have
      // returned is never returned at all -- so the abort hook and pending
      // entry registered before the send must not be left holding it. Were
      // they, disposal would reject a promise nobody holds, which is an
      // unhandled rejection rather than a caught one.
      const connection = new RuntimeConnection(
        new ThrowingTransport(RequestType.CellSet),
      );
      await connection.initialize({} as InitializationData);

      expect(() =>
        connection.request({
          type: RequestType.CellSet,
          cell: { id: "of:c", space: "did:key:t", scope: "space", path: [] },
          value: 1,
        } as never)
      ).toThrow("no encoding for value");

      expect(connection.getPendingRequestDiagnostics()).toHaveLength(0);

      // And the timeline says so too: a missing `doneAtMs` reads as still in
      // flight, which is the one thing this request is not.
      const entry = connection.getRequestTimelineDiagnostics().find((e) =>
        e.type === RequestType.CellSet
      );
      expect(entry?.doneAtMs).toEqual(expect.any(Number));
      expect(entry?.error).toBe(true);
      expect(entry?.outcome).toBe("send-error");

      // Would reject the orphan, if the bookkeeping still held one.
      await connection.dispose();
    });

    it("throws when a notification is sent after disposal", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const session = connection.attachVDom(() => {});
      await connection.dispose();

      const sentBefore = transport.sent.length;
      // Notifications come only from owned consumers torn down before disposal,
      // so a post-disposal send is a contract violation, not a benign race.
      expect(() => session.sendEvent(1, 2, { type: "click" } as never, 3))
        .toThrow();
      expect(() => session.ackBatch(1, 7)).toThrow();
      // Nothing reached the transport.
      expect(transport.sent.length).toBe(sentBefore);
    });
  });

  describe("RuntimeConnection.attachVDom", () => {
    it("runs the consumer teardown on disposal", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      let torn = false;
      connection.attachVDom(() => {
        torn = true;
      });
      await connection.dispose();
      expect(torn).toBe(true);
    });

    it("does not run the teardown after `detach()`", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      let torn = false;
      const session = connection.attachVDom(() => {
        torn = true;
      });
      session.detach();
      await connection.dispose();
      expect(torn).toBe(false);
    });

    it("sends events and acks through the session while alive", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const session = connection.attachVDom(() => {});
      session.sendEvent(1, 2, { type: "click" } as never, 3);
      session.ackBatch(1, 9);
      const notifications = transport.sent.filter((m) => !("msgId" in m));
      expect(notifications.length).toBe(2);
      session.detach();
    });

    it("cancels a mount whose response fails before forgetting its id", async () => {
      const transport = new FakeTransport([RequestType.VDomMount]);
      const connection = await initializedConnection(transport);
      const session = connection.attachVDom(() => {});
      try {
        const mounting = session.mount(37, {
          id: "of:mount-failure",
          space: "did:key:test",
          scope: "space",
          path: [],
        });
        const rejected = expect(mounting).rejects.toThrow("mount refused");
        const held = transport.sent.find((message) =>
          "msgId" in message && message.data.type === RequestType.VDomMount
        ) as IPCClientMessage;
        transport.emit("message", {
          msgId: held.msgId,
          error: "mount refused",
        });
        await rejected;
        const requests = transport.sent.filter(
          (message): message is IPCClientMessage => "msgId" in message,
        );
        const cancellations = requests.filter((message) =>
          message.data.type === RequestType.VDomUnmount
        );
        expect(cancellations).toHaveLength(1);
        expect(cancellations[0].data).toMatchObject({ mountId: 37 });
        expect(connection.getPendingRequestDiagnostics()).toEqual([]);

        const mounted = requests.find((message) =>
          message.data.type === RequestType.VDomMount
        )!;
        using _warning = stub(console, "warn");
        transport.emit("message", {
          msgId: mounted.msgId,
          data: { rootId: 5 },
        });
        expect(connection.getPendingRequestDiagnostics()).toEqual([]);
        expect(
          transport.sent.filter((message) =>
            "msgId" in message && message.data.type === RequestType.VDomUnmount
          ),
        ).toHaveLength(1);
      } finally {
        session.detach();
        await connection.dispose();
      }
    });

    it("mounts, unmounts, and routes batch notifications via the session", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const session = connection.attachVDom(() => {});

      const cellRef = {
        id: "of:mount-cell",
        space: "did:key:test",
        scope: "space",
        path: [],
      };
      // mount/unmount are awaited round-trips, auto-acked by the transport.
      await session.mount(1, cellRef as never);
      await session.unmount(1);
      const sentTypes = transport.sent
        .filter((m): m is IPCClientMessage => "msgId" in m)
        .map((m) => m.data.type);
      expect(sentTypes).toContain(RequestType.VDomMount);
      expect(sentTypes).toContain(RequestType.VDomUnmount);

      // onBatch/offBatch wire and unwire the vdombatch subscription.
      const batches: number[] = [];
      const handler = (n: { batchId: number }) => batches.push(n.batchId);
      session.onBatch(handler as never);
      transport.emit(
        "message",
        { type: "vdom:batch", batchId: 7, ops: [] } as never,
      );
      expect(batches).toEqual([7]);
      session.offBatch(handler as never);
      transport.emit(
        "message",
        { type: "vdom:batch", batchId: 8, ops: [] } as never,
      );
      expect(batches).toEqual([7]);

      session.detach();
    });

    it("skips the unmount round-trip once disposed", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      const session = connection.attachVDom(() => {});
      await connection.dispose();

      const sentBefore = transport.sent.length;
      // The worker tears down every mount wholesale on dispose, so a per-mount
      // unmount is redundant and is skipped.
      await session.unmount(1);
      expect(transport.sent.length).toBe(sentBefore);
    });
  });

  describe("RuntimeConnection request timeline", () => {
    it("records send and settle offsets for boot-window requests", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);
      await connection.request<RequestType.Idle>({ type: RequestType.Idle });

      const timeline = connection.getRequestTimelineDiagnostics();
      expect(timeline.map((entry) => entry.type)).toEqual([
        RequestType.Initialize,
        RequestType.Idle,
      ]);
      for (const entry of timeline) {
        expect(typeof entry.msgId).toBe("number");
        expect(typeof entry.sentAtMs).toBe("number");
        // Both requests settled, so both are stamped done — at or after send.
        expect(typeof entry.doneAtMs).toBe("number");
        expect(entry.doneAtMs!).toBeGreaterThanOrEqual(entry.sentAtMs);
        expect(entry.error).toBeUndefined();
      }
      await connection.dispose();
    });

    it("leaves doneAtMs unset while in flight and flags error replies", async () => {
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);
      const inFlight = connection.request<RequestType.Idle>({
        type: RequestType.Idle,
      });
      const settled = inFlight.then(() => undefined, (error) => error);

      const pendingEntry = connection.getRequestTimelineDiagnostics()
        .find((entry) => entry.type === RequestType.Idle);
      expect(pendingEntry).toBeDefined();
      expect(pendingEntry!.doneAtMs).toBeUndefined();
      expect(pendingEntry!.error).toBeUndefined();

      // The worker answers with an error; the entry is stamped done + error.
      const sent = transport.sent.find(
        (m): m is IPCClientMessage =>
          "msgId" in m && m.data.type === RequestType.Idle,
      );
      transport.emit("message", { msgId: sent!.msgId, error: "boom" });
      const error = await settled;
      expect((error as Error).message).toBe("boom");

      const doneEntry = connection.getRequestTimelineDiagnostics()
        .find((entry) => entry.type === RequestType.Idle);
      expect(doneEntry!.error).toBe(true);
      expect(typeof doneEntry!.doneAtMs).toBe("number");
      await connection.dispose();
    });

    it("surfaces coded request failures as lifecycle error events", async () => {
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);
      const errors: unknown[] = [];
      connection.on("error", (error) => errors.push(error));

      const request = connection.request<RequestType.Idle>({
        type: RequestType.Idle,
      });
      const settled = request.then(() => undefined, (error) => error);
      const sent = transport.sent.find(
        (message): message is IPCClientMessage =>
          "msgId" in message && message.data.type === RequestType.Idle,
      );
      transport.emit("message", {
        msgId: sent!.msgId,
        error: "Failed to load the compiler stack",
        code: RuntimeErrorCode.CompilerStackLoadFailed,
      });

      const error = await settled as Error & { code?: RuntimeErrorCode };
      expect(error.code).toBe(RuntimeErrorCode.CompilerStackLoadFailed);
      expect(errors).toEqual([{
        type: NotificationType.ErrorReport,
        message: "Failed to load the compiler stack",
        code: RuntimeErrorCode.CompilerStackLoadFailed,
      }]);
      await connection.dispose();
    });

    it("rejects a request that finds no space with its code, and emits no lifecycle error", async () => {
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);
      const errors: unknown[] = [];
      connection.on("error", (error) => errors.push(error));

      const request = connection.request<RequestType.Idle>({
        type: RequestType.Idle,
      });
      const settled = request.then(() => undefined, (error) => error);
      const sent = transport.sent.find(
        (message): message is IPCClientMessage =>
          "msgId" in message && message.data.type === RequestType.Idle,
      );
      transport.emit("message", {
        msgId: sent!.msgId,
        error: "No space answers to did:key:z6MkMissing",
        code: RuntimeErrorCode.SpaceNotFound,
      });

      const error = await settled as Error & { code?: RuntimeErrorCode };
      expect(error.code).toBe(RuntimeErrorCode.SpaceNotFound);
      expect(errors).toEqual([]);
      await connection.dispose();
    });

    it("returns a copy that does not expose the internal ledger", async () => {
      const transport = new FakeTransport();
      const connection = await initializedConnection(transport);

      const first = connection.getRequestTimelineDiagnostics();
      expect(first.length).toBe(1);
      // Mutate the returned array and its entries every way a caller could.
      first[0].type = "mutated" as RequestType;
      first[0].doneAtMs = -1;
      first.push({ msgId: 999, type: RequestType.Idle, sentAtMs: 0 });

      const second = connection.getRequestTimelineDiagnostics();
      expect(second.length).toBe(1);
      expect(second[0].type).toBe(RequestType.Initialize);
      expect(second[0].doneAtMs).not.toBe(-1);
      await connection.dispose();
    });
  });

  describe("RuntimeConnection pending-request diagnostics", () => {
    it("snapshots in-flight requests with their ages, oldest first", async () => {
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);

      // Give the two in-flight requests a deterministic age gap without sleeping:
      // freeze performance.now() and step it forward between the sends, so the
      // first request is measurably older. This is what exercises the
      // descending-age sort — with equal ages a stable sort preserves insertion
      // order whichever way the comparator runs, leaving the direction untested.
      const realNow = performance.now;
      let fakeNow = realNow.call(performance);
      Reflect.set(performance, "now", () => fakeNow);
      try {
        const first = connection.request<RequestType.Idle>({
          type: RequestType.Idle,
        });
        const firstSettled = first.then(() => undefined, (error) => error);
        fakeNow += 5;
        const second = connection.request<RequestType.Idle>({
          type: RequestType.Idle,
        });
        const secondSettled = second.then(() => undefined, (error) => error);

        const pending = connection.getPendingRequestDiagnostics();
        expect(pending.length).toBe(2);
        for (const entry of pending) {
          expect(entry.type).toBe(RequestType.Idle);
          expect(typeof entry.msgId).toBe("number");
          expect(entry.ageMs).toBeGreaterThanOrEqual(0);
        }
        // Sorted oldest (largest age) first — the request a stalled caller has
        // been waiting on longest names the wedged layer.
        expect(pending[0].ageMs).toBeGreaterThan(pending[1].ageMs);
        expect(pending[0].msgId).toBeLessThan(pending[1].msgId);

        await connection.dispose();
        await firstSettled;
        await secondSettled;
        // Disposal settled both as cancellation; nothing is pending anymore.
        expect(connection.getPendingRequestDiagnostics()).toEqual([]);
      } finally {
        Reflect.set(performance, "now", realNow);
      }
    });
  });

  describe("RuntimeConnection request lifetime", () => {
    it("leaves a request the worker has not answered pending however long it waits", async () => {
      // A request settles on its reply or on disposal, and on nothing else.
      // The wait is what a `FakeTime` makes assertable: advancing an hour of
      // fake time runs every timer a bound would have been armed on, so a
      // request still pending afterwards is one no clock is going to reject.
      // Real time does not move, so the case costs nothing to run.
      //
      // The connection is built before the clock is faked. Its loop-lag probe
      // arms an interval through `unrefTimer`, which hands the id to
      // `Deno.unrefTimer`, and a faked id names an unrelated real timer.
      // Building first also keeps that interval off the fake clock, where an
      // hour of ticks would run it thirty-six thousand times.
      const transport = new FakeTransport([RequestType.Idle]);
      const connection = await initializedConnection(transport);
      using time = new FakeTime();

      const inFlight = connection.request<RequestType.Idle>({
        type: RequestType.Idle,
      });
      let outcome: "pending" | "settled" = "pending";
      const settled = inFlight.then(
        () => (outcome = "settled"),
        (error) => {
          outcome = "settled";
          return error;
        },
      );

      await time.tickAsync(60 * 60 * 1000);

      expect(outcome).toBe("pending");
      expect(connection.getPendingRequestDiagnostics()).toHaveLength(1);

      // Disposal is what settles it, so the case leaves nothing unhandled.
      const disposing = connection.dispose();
      await time.tickAsync(0);
      await disposing;
      await settled;
      expect(outcome).toBe("settled");
      expect(connection.getPendingRequestDiagnostics()).toEqual([]);
    });
  });

  describe("RuntimeConnection loop-lag probe", () => {
    it("samples main-thread lag while alive and clears its timer on dispose", async () => {
      const logger = getLogger("runtime-client");
      const lagCountBefore = logger.getTimeStats("loop", "mainLag")?.count ?? 0;

      // Spy on clearInterval: the loop-lag interval is the only interval the
      // connection owns, so dispose must clear exactly one.
      const originalClearInterval = globalThis.clearInterval;
      const cleared: unknown[] = [];
      globalThis.clearInterval = ((id?: number) => {
        cleared.push(id);
        return originalClearInterval(id);
      }) as typeof clearInterval;

      try {
        const transport = new FakeTransport();
        const connection = await initializedConnection(transport);

        // Block the thread past the first 100ms sample: the tick due during the
        // block can only fire late, so the probe records a positive mainLag.
        // (Structure/counter assertion only — the magnitude is not asserted.)
        const end = performance.now() + 110;
        while (performance.now() < end) {
          // busy
        }
        await new Promise((resolve) => setTimeout(resolve, 5));

        const lagCountAfter = logger.getTimeStats("loop", "mainLag")?.count ??
          0;
        expect(lagCountAfter).toBeGreaterThan(lagCountBefore);

        expect(cleared.length).toBe(0);
        await connection.dispose();
        expect(cleared.length).toBe(1);
      } finally {
        globalThis.clearInterval = originalClearInterval;
      }
    });
  });

  describe("FakeTransport", () => {
    it("delivers a message unshared with the sender", () => {
      // `RuntimeTransport.send()` requires the far end to receive a value it
      // owns, which is what lets a handler cede a byte payload rather than copy
      // it. A double that passed the sender's object through would model
      // something no real transport does.
      const transport = new FakeTransport();
      const body = new FabricBytes(new Uint8Array([1, 2, 3]));
      // A plain nested container alongside it: the codec rebuilds a
      // `FabricBytes` either way, so that arm alone cannot tell whether the
      // clone is there. An encode passes a subtree needing no encoding through
      // by identity, and a decode hands that same object back -- so without the
      // clone this is the sender's own object, frozen under it.
      const nested = { note: "unshared" };
      const message: IPCClientMessage = {
        msgId: 1,
        data: {
          type: RequestType.UploadBlob,
          space: "did:key:test-space",
          contentType: "image/png",
          body,
          nested,
        } as never,
      };

      transport.send(message);

      const captured = transport.sent[0];
      expect(
        (captured as unknown as { data: { nested: unknown } }).data.nested,
      ).not.toBe(nested);
      expect(Object.isFrozen(nested)).toBe(false);
      expect(captured).not.toBe(message);
      if (
        !("msgId" in captured) || captured.data.type !== RequestType.UploadBlob
      ) {
        throw new Error("Expected the upload request to have been captured.");
      }
      expect(captured.data.body).not.toBe(body);
      // The bytes arrive as a `FabricBytes` and not as the plain object a bare
      // structured clone would leave, which is the whole point of encoding the
      // envelope. Read both rather than compare identity alone: a shallow copy
      // differs by identity while still sharing the buffer beneath, and the
      // sender's still reading after the captured one did is what says the two
      // do not share one.
      expect(captured.data.body).toBeInstanceOf(FabricBytes);
      expect(captured.data.body.slice()).toEqual(new Uint8Array([1, 2, 3]));
      expect(body.slice()).toEqual(new Uint8Array([1, 2, 3]));
    });
  });
});

describe("RuntimeConnection.attach()", () => {
  // The choke point: the last thing before a frame reaches a transport. A
  // caller holding a `RuntimeConnection` directly does not pass through
  // `RuntimeClient.attach`, so the guarantee that no key material crosses a
  // port has to live here rather than only there.

  const context: RuntimeSecurityContext = {
    identity: "did:key:z6Mk-connection-attach" as DID,
    apiUrl: "http://connection.test/",
    spaceDid: "did:key:z6Mk-connection-attach" as DID,
  };

  it("sends the asserted context and marks the connection usable", async () => {
    const transport = new FakeTransport();
    const conn = new RuntimeConnection(transport);
    await conn.attach(context);
    expect(transport.sent).toHaveLength(1);
    expect(transport.sent[0]).toMatchObject({
      data: { type: RequestType.Attach, data: context },
    });
  });

  it("refuses a context holding key material, sending nothing", async () => {
    const transport = new FakeTransport();
    const conn = new RuntimeConnection(transport);
    const keyPair = new FabricKeyPair(
      "Ed25519",
      new Uint8Array(32),
      new Uint8Array(32),
    );
    await expect(conn.attach({
      ...context,
      trustSnapshot: {
        id: "principal",
        signer: keyPair,
      } as unknown as RuntimeSecurityContext["trustSnapshot"],
    })).rejects.toThrow("no key material");
    expect(transport.sent).toEqual([]);
  });
});