diff --git a/server/session-broker.ts b/server/session-broker.ts
new file mode 100644
index 0000000..8c0d56b
--- /dev/null
+++ b/server/session-broker.ts
@@ -0,0 +1,465 @@
+import { randomUUID } from "node:crypto";
+
+export type RuntimeState = "live" | "deleting" | "disposed";
+
+export interface SessionPromptOptions {
+  preflightResult?: (accepted: boolean) => void;
+  [key: string]: unknown;
+}
+
+export interface BrokerSession {
+  readonly sessionId: string;
+  readonly isIdle: boolean;
+  subscribe(listener: (event: unknown) => void): () => void;
+  prompt(text: string, options?: SessionPromptOptions): Promise<void>;
+  steer(text: string): Promise<void>;
+  followUp(text: string): Promise<void>;
+  abort(): Promise<void>;
+  dispose(): void;
+}
+
+export interface BrokerFactory<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  create(key: string, context: Readonly<C>): Promise<S>;
+}
+
+export interface BrokerSerializer<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  snapshot(runtime: BrokerRuntime<C, S>): unknown;
+}
+
+export interface SocketSink {
+  isOpen(): boolean;
+  send(message: unknown): void;
+}
+
+export interface BrokerClock {
+  now(): number;
+}
+
+export interface PromptRequest {
+  requestId: string;
+  text: string;
+  [key: string]: unknown;
+}
+
+export interface ChoiceRequest {
+  choiceId: string;
+  payload: unknown;
+}
+
+export interface BrokerMessageHooks<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  ready(runtime: BrokerRuntime<C, S>): unknown;
+  stateSync(runtime: BrokerRuntime<C, S>, snapshot: unknown): unknown;
+  event(event: unknown): unknown;
+  choiceRequest(request: ChoiceRequest): unknown;
+  promptAccepted(requestId: string): unknown;
+  promptRejected(requestId: string): unknown;
+}
+
+export interface BrokerOptions<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  factory: BrokerFactory<C, S>;
+  serializer: BrokerSerializer<C, S>;
+  messages: BrokerMessageHooks<C, S>;
+  clock?: BrokerClock;
+  idleTtlMs?: number;
+  activeMaxAgeMs?: number;
+  maxRuntimes?: number;
+  sweepAbortGraceMs?: number;
+  promptStarter?: (session: S, request: PromptRequest, preflightResult: (accepted: boolean) => void) => void | Promise<void>;
+  onDelete: (key: string, runtime: BrokerRuntime<C, S>) => Promise<void>;
+  onError: (error: unknown) => void;
+}
+
+interface PromptRecord {
+  fingerprint: string;
+  state: "starting" | "accepted" | "completed" | "rejected" | "failed";
+  acceptance: Promise<void>;
+  accept: () => void;
+  reject: (error: Error) => void;
+}
+
+interface Attachment {
+  readonly token: symbol;
+  readonly generation: number;
+  readonly sink: SocketSink;
+  syncing: boolean;
+  attached: boolean;
+}
+
+interface PendingChoice {
+  request: ChoiceRequest;
+  resolve: (value: unknown) => void;
+  reject: (error: Error) => void;
+}
+
+const noop = (): void => {};
+
+export class BrokerRuntime<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  readonly runtimeId = randomUUID();
+  readonly createdAt: number;
+  readonly context: Readonly<C>;
+  state: RuntimeState = "live";
+  private readonly attachments = new Map<symbol, Attachment>();
+  private readonly prompts = new Map<string, PromptRecord>();
+  private readonly choices = new Map<string, PendingChoice>();
+  private unsubscribe: (() => void) | undefined;
+  private invocationTail: Promise<void> = Promise.resolve();
+  private generation = 0;
+  private lastActivity: number;
+  private ageAbortRequested = false;
+  private ageAbortDeadline = 0;
+  private abortRequested = false;
+  private disposed = false;
+  private queuedInvocations = 0;
+
+  constructor(
+    readonly key: string,
+    readonly session: S,
+    context: Readonly<Record<string, unknown>>,
+    private readonly serializer: BrokerSerializer<C, S>,
+    private readonly clock: BrokerClock,
+    private readonly onError: (error: unknown) => void,
+    private readonly onDelete: (key: string, runtime: BrokerRuntime<C, S>) => Promise<void>,
+    private readonly promptStarter: ((session: S, request: PromptRequest, preflightResult: (accepted: boolean) => void) => void | Promise<void>) | undefined,
+    private readonly messages: BrokerMessageHooks<C, S>,
+  ) {
+    this.context = Object.freeze({ ...(context as object) }) as Readonly<C>;
+    this.createdAt = clock.now();
+    this.lastActivity = this.createdAt;
+    this.unsubscribe = session.subscribe((event) => this.broadcast(event));
+  }
+
+  get attachmentCount(): number { return this.attachments.size; }
+  get hasPendingChoices(): boolean { return this.choices.size > 0; }
+  get hasQueuedInvocation(): boolean { return this.queuedInvocations > 0; }
+  get lastUsedAt(): number { return this.lastActivity; }
+  get isDetached(): boolean { return this.attachments.size === 0; }
+  get isActive(): boolean { return !this.session.isIdle || this.hasPendingChoices || this.hasQueuedInvocation || this.promptsInFlight; }
+  private get promptsInFlight(): boolean { return [...this.prompts.values()].some(({ state }) => state === "starting" || state === "accepted"); }
+
+  attach(sink: SocketSink): AttachmentHandle<C, S> {
+    if (this.state !== "live") throw new Error(`Runtime ${this.key} is ${this.state}`);
+    const attachment: Attachment = { token: Symbol("attachment"), generation: ++this.generation, sink, syncing: true, attached: true };
+    this.attachments.set(attachment.token, attachment);
+    try {
+      if (!sink.isOpen()) throw new Error("Socket is closed");
+      this.send(attachment, this.message(() => this.messages?.ready(this)));
+      this.send(attachment, this.message(() => this.messages?.stateSync(this, this.serializer.snapshot(this))));
+      for (const choice of this.choices.values()) this.send(attachment, this.message(() => this.messages?.choiceRequest(choice.request)));
+      if (!attachment.attached) throw new Error("Socket closed during attach");
+      attachment.syncing = false;
+      this.lastActivity = this.clock.now();
+      return new AttachmentHandle(this, attachment);
+    } catch (error) {
+      this.detachAttachment(attachment);
+      throw error;
+    }
+  }
+
+  detach(handle: AttachmentHandle<C, S>): void { this.detachAttachment(handle.attachment); }
+
+  isAttachmentValid(handle: AttachmentHandle<C, S>, generation = handle.generation): boolean {
+    return this.state === "live" && handle.attachment.attached && handle.generation === generation && this.attachments.get(handle.token) === handle.attachment;
+  }
+
+  private detachAttachment(attachment: Attachment): void {
+    if (!attachment.attached || this.attachments.get(attachment.token) !== attachment) return;
+    attachment.attached = false;
+    this.attachments.delete(attachment.token);
+    this.lastActivity = this.clock.now();
+  }
+
+  private message(create: () => unknown): unknown {
+    try { return create(); } catch (error) { this.onError(error); return undefined; }
+  }
+
+  private send(attachment: Attachment, message: unknown): void {
+    if (message === undefined || !attachment.attached) return;
+    if (!attachment.sink.isOpen()) { this.detachAttachment(attachment); return; }
+    try { attachment.sink.send(message); } catch (error) { this.onError(error); this.detachAttachment(attachment); }
+  }
+
+  private broadcast(event: unknown): void {
+    if ((event as { type?: string })?.type === "agent_settled") this.lastActivity = this.clock.now();
+    const message = this.message(() => this.messages?.event(event));
+    if (message === undefined) return;
+    for (const attachment of this.attachments.values()) {
+      if (!attachment.syncing) this.send(attachment, message);
+    }
+  }
+
+  private enqueue(handle: AttachmentHandle<C, S>, operation: () => Promise<void>): Promise<void> {
+    if (!this.isAttachmentValid(handle)) return Promise.reject(new Error("Attachment is no longer valid"));
+    this.queuedInvocations++;
+    const run = this.invocationTail.then(() => {
+      if (!this.isAttachmentValid(handle)) throw new Error("Attachment is no longer valid");
+      this.lastActivity = this.clock.now();
+      return operation();
+    });
+    this.invocationTail = run.catch(noop).finally(() => { this.queuedInvocations--; });
+    return run;
+  }
+
+  invoke(handle: AttachmentHandle<C, S>, operation: () => Promise<void>): Promise<void> { return this.enqueue(handle, operation); }
+  steer(handle: AttachmentHandle<C, S>, text: string): Promise<void> { return this.enqueue(handle, () => this.session.steer(text)); }
+  followUp(handle: AttachmentHandle<C, S>, text: string): Promise<void> { return this.enqueue(handle, () => this.session.followUp(text)); }
+  abort(handle: AttachmentHandle<C, S>): Promise<void> { return this.enqueue(handle, () => this.abortOnce()); }
+  prompt(handle: AttachmentHandle<C, S>, request: PromptRequest, requestFingerprint: string): Promise<void> {
+    return this.enqueue(handle, () => this.startPrompt(request, requestFingerprint, handle));
+  }
+
+  private startPrompt(request: PromptRequest, fingerprint: string, handle: AttachmentHandle<C, S>): Promise<void> {
+    const existing = this.prompts.get(request.requestId);
+    if (existing) {
+      if (existing.fingerprint !== fingerprint) return Promise.reject(new Error(`Prompt request ${request.requestId} has a different payload`));
+      this.send(handle.attachment, this.message(() => existing.state === "rejected" ? this.messages?.promptRejected(request.requestId) : this.messages?.promptAccepted(request.requestId)));
+      return existing.acceptance;
+    }
+    let accept!: () => void;
+    let reject!: (error: Error) => void;
+    const acceptance = new Promise<void>((resolve, rejectPromise) => { accept = resolve; reject = rejectPromise; });
+    const record: PromptRecord = { fingerprint, state: "starting", acceptance, accept, reject };
+    this.prompts.set(request.requestId, record);
+    let preflightCalled = false;
+    const preflightResult = (accepted: boolean): void => {
+      if (preflightCalled || record.state !== "starting") return;
+      preflightCalled = true;
+      if (this.state !== "live") {
+        this.rejectPrompt(record, new Error(`Runtime ${this.key} is ${this.state}`));
+        return;
+      }
+      if (accepted) {
+        record.state = "accepted";
+        record.accept();
+        this.send(handle.attachment, this.message(() => this.messages?.promptAccepted(request.requestId)));
+      } else {
+        this.rejectPrompt(record, new Error(`Prompt request ${request.requestId} was rejected`));
+        this.send(handle.attachment, this.message(() => this.messages?.promptRejected(request.requestId)));
+      }
+    };
+    try {
+      const started = this.promptStarter
+        ? this.promptStarter(this.session, request, preflightResult)
+        : this.session.prompt(request.text, { ...request, preflightResult });
+      void Promise.resolve(started).then(() => {
+        if (!preflightCalled) preflightResult(true);
+        if (record.state === "accepted") record.state = "completed";
+        this.trimPrompts();
+      }, (error: unknown) => {
+        if (!preflightCalled) {
+          this.rejectPrompt(record, error instanceof Error ? error : new Error(String(error)));
+          this.send(handle.attachment, this.message(() => this.messages.promptRejected(request.requestId)));
+        } else if (record.state === "accepted") record.state = "failed";
+        this.onError(error);
+        this.trimPrompts();
+      });
+    } catch (error) {
+      this.rejectPrompt(record, error instanceof Error ? error : new Error(String(error)));
+      this.send(handle.attachment, this.message(() => this.messages.promptRejected(request.requestId)));
+      this.onError(error);
+    }
+    return acceptance;
+  }
+
+  private rejectPrompt(record: PromptRecord, error: Error): void {
+    if (record.state !== "starting") return;
+    record.state = "rejected";
+    record.reject(error);
+    this.trimPrompts();
+  }
+
+  private trimPrompts(): void {
+    const settled = [...this.prompts].filter(([, record]) => record.state === "completed" || record.state === "rejected" || record.state === "failed");
+    while (settled.length > 1000) this.prompts.delete(settled.shift()![0]);
+  }
+
+  requestChoice(request: ChoiceRequest): Promise<unknown> {
+    if (this.state !== "live") return Promise.reject(new Error("Runtime is not live"));
+    if (this.choices.has(request.choiceId)) return Promise.reject(new Error(`Choice ${request.choiceId} already exists`));
+    const promise = new Promise<unknown>((resolve, reject) => this.choices.set(request.choiceId, { request, resolve, reject }));
+    const message = this.message(() => this.messages?.choiceRequest(request));
+    if (message !== undefined) {
+      for (const attachment of this.attachments.values()) {
+        if (!attachment.syncing) this.send(attachment, message);
+      }
+    }
+    return promise;
+  }
+
+  resolveChoice(handle: AttachmentHandle<C, S>, choiceId: string, value: unknown): Promise<void> {
+    return this.enqueue(handle, async () => {
+      const choice = this.choices.get(choiceId);
+      if (!choice) throw new Error(`Choice ${choiceId} is not pending`);
+      this.choices.delete(choiceId);
+      choice.resolve(value);
+    });
+  }
+
+  rejectChoice(choiceId: string, reason = "Choice cancelled"): void {
+    const choice = this.choices.get(choiceId);
+    if (!choice) return;
+    this.choices.delete(choiceId);
+    choice.reject(new Error(reason));
+  }
+
+  async delete(): Promise<void> {
+    if (this.state !== "live") throw new Error(`Runtime ${this.key} is ${this.state}`);
+    if (this.isActive || this.attachmentCount > 1) throw new Error("Runtime is active or has multiple attachments");
+    this.state = "deleting";
+    await this.dispose();
+    await this.onDelete(this.key, this);
+  }
+
+  async requestAgeAbort(graceUntil: number): Promise<void> {
+    if (!this.isDetached || this.state !== "live") return;
+    if (!this.ageAbortRequested) {
+      if (this.session.isIdle) return;
+      this.ageAbortRequested = true;
+      this.ageAbortDeadline = graceUntil;
+      void this.abortOnce().catch((error) => this.onError(error));
+    }
+    if (this.clock.now() >= this.ageAbortDeadline && !this.session.isIdle) await this.dispose();
+  }
+
+  async abortOnce(): Promise<void> {
+    if (this.abortRequested) return;
+    this.abortRequested = true;
+    await this.session.abort();
+  }
+
+  async shutdown(): Promise<void> {
+    if (this.isActive) void this.abortOnce().catch((error) => this.onError(error));
+    await this.dispose();
+  }
+
+  async dispose(): Promise<void> {
+    if (this.disposed) return;
+    this.disposed = true;
+    this.state = "disposed";
+    this.unsubscribe?.();
+    this.unsubscribe = undefined;
+    for (const choice of this.choices.values()) choice.reject(new Error("Runtime disposed"));
+    this.choices.clear();
+    for (const record of this.prompts.values()) {
+      if (record.state === "starting") this.rejectPrompt(record, new Error("Runtime disposed"));
+    }
+    for (const attachment of this.attachments.values()) attachment.attached = false;
+    this.attachments.clear();
+    try { this.session.dispose(); } catch (error) { this.onError(error); }
+  }
+}
+
+export class AttachmentHandle<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  constructor(readonly runtime: BrokerRuntime<C, S>, readonly attachment: Attachment) {}
+  detach(): void { this.runtime.detach(this); }
+  get generation(): number { return this.attachment.generation; }
+  get token(): symbol { return this.attachment.token; }
+}
+
+export class SessionBroker<C extends object = Record<string, unknown>, S extends BrokerSession = BrokerSession> {
+  private readonly runtimes = new Map<string, BrokerRuntime<C, S>>();
+  private readonly creating = new Map<string, Promise<BrokerRuntime<C, S>>>();
+  private readonly deleting = new Map<string, Promise<void>>();
+  private reservations = 0;
+  private readonly clock: BrokerClock;
+  private shuttingDown = false;
+  private shutdownPromise: Promise<void> | undefined;
+
+  constructor(private readonly options: BrokerOptions<C, S>) { this.clock = options.clock ?? { now: () => Date.now() }; }
+  get(key: string): BrokerRuntime<C, S> | undefined {
+    key = String(key);
+    const runtime = this.runtimes.get(key);
+    if (runtime?.state === "disposed") this.runtimes.delete(key);
+    return runtime?.state === "disposed" ? undefined : runtime;
+  }
+
+  async getOrCreate(key: string, context: C): Promise<BrokerRuntime<C, S>> {
+    key = String(key);
+    if (this.shuttingDown) throw new Error("Broker is shutting down");
+    const deleting = this.deleting.get(key);
+    if (deleting) { await deleting; return this.getOrCreate(key, context); }
+    const live = this.get(key);
+    if (live) {
+      if (live.state === "live") return live;
+      throw new Error(`Runtime ${key} is ${live.state}`);
+    }
+    const existing = this.creating.get(key);
+    if (existing) return existing;
+    const promise = this.createWithCapacity(key, context);
+    this.creating.set(key, promise);
+    try { return await promise; } finally { if (this.creating.get(key) === promise) this.creating.delete(key); }
+  }
+
+  private async createWithCapacity(key: string, context: C): Promise<BrokerRuntime<C, S>> {
+    this.reserveCapacity(key);
+    try { return await this.create(key, context); } finally { this.reservations--; }
+  }
+
+  private async create(key: string, context: C): Promise<BrokerRuntime<C, S>> {
+    const immutableContext = Object.freeze({ ...context }) as Readonly<C>;
+    const session = await this.options.factory.create(key, immutableContext);
+    const runtime = new BrokerRuntime<C, S>(key, session, immutableContext, this.options.serializer, this.clock, this.options.onError, this.options.onDelete, this.options.promptStarter, this.options.messages);
+    if (this.shuttingDown) { await runtime.dispose(); throw new Error("Broker shut down during runtime creation"); }
+    this.runtimes.set(key, runtime);
+    return runtime;
+  }
+
+  async sweep(): Promise<void> {
+    const now = this.clock.now();
+    for (const [key, runtime] of this.runtimes) {
+      if (!runtime.isDetached || runtime.state !== "live") continue;
+      if (!runtime.isActive && now - runtime.lastUsedAt >= (this.options.idleTtlMs ?? 15 * 60_000)) {
+        await runtime.dispose();
+        if (this.runtimes.get(key) === runtime) this.runtimes.delete(key);
+      } else if (runtime.isActive && now - runtime.createdAt >= (this.options.activeMaxAgeMs ?? 24 * 60 * 60_000)) {
+        await runtime.requestAgeAbort(now + (this.options.sweepAbortGraceMs ?? 30_000));
+        if (runtime.state !== "live" && this.runtimes.get(key) === runtime) this.runtimes.delete(key);
+      }
+    }
+  }
+
+  async delete(key: string): Promise<void> {
+    key = String(key);
+    const runtime = this.runtimes.get(key);
+    if (!runtime) throw new Error(`Runtime ${key} does not exist`);
+    const existing = this.deleting.get(key);
+    if (existing) return existing;
+    const deletion = runtime.delete();
+    this.deleting.set(key, deletion);
+    try { await deletion; } finally {
+      if (this.deleting.get(key) === deletion) this.deleting.delete(key);
+      if (this.runtimes.get(key) === runtime) this.runtimes.delete(key);
+    }
+  }
+
+  private reserveCapacity(key: string): void {
+    const limit = this.options.maxRuntimes ?? 20;
+    if (this.runtimes.size + this.reservations >= limit) {
+      const candidate = [...this.runtimes.values()]
+        .filter((runtime) => runtime.isDetached && !runtime.isActive)
+        .sort((a, b) => a.lastUsedAt - b.lastUsedAt)[0];
+      if (!candidate) throw new Error(`Runtime capacity reached for ${key}`);
+      void candidate.dispose();
+      if (this.runtimes.get(candidate.key) === candidate) this.runtimes.delete(candidate.key);
+    }
+    this.reservations++;
+  }
+
+  async shutdown(timeoutMs = 30_000): Promise<void> {
+    if (this.shutdownPromise) return this.shutdownPromise;
+    this.shuttingDown = true;
+    this.shutdownPromise = (async () => {
+      const work = [
+        ...this.creating.values(),
+        ...this.runtimes.values(),
+      ].map((item) => item instanceof Promise
+        ? item.then(async (runtime) => { await runtime.shutdown(); }).catch(() => {})
+        : item.shutdown());
+      await Promise.race([
+        Promise.all(work).then(() => undefined),
+        new Promise<void>((resolve) => setTimeout(resolve, timeoutMs).unref()),
+      ]);
+      this.runtimes.clear();
+    })();
+    return this.shutdownPromise;
+  }
+}
diff --git a/tests/session-broker.test.ts b/tests/session-broker.test.ts
new file mode 100644
index 0000000..75ac487
--- /dev/null
+++ b/tests/session-broker.test.ts
@@ -0,0 +1,340 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { BrokerRuntime, SessionBroker, type BrokerMessageHooks, type BrokerSession, type SocketSink } from "../server/session-broker.ts";
+
+type Message = { type: string };
+const messagesOf = (socket: FakeSocket): Message[] => socket.messages as Message[];
+
+class FakeClock {
+  time = 0;
+  now = (): number => this.time;
+}
+
+class FakeSocket implements SocketSink {
+  open = true;
+  messages: unknown[] = [];
+  isOpen = (): boolean => this.open;
+  send = (message: unknown): void => { this.messages.push(message); };
+}
+
+class FakeSession implements BrokerSession {
+  readonly sessionId: string;
+  isStreaming = false;
+  isIdle = true;
+  abortCount = 0;
+  disposeCount = 0;
+  promptCalls: Array<{ text: string; preflight: (accepted: boolean) => void }> = [];
+  private listeners = new Set<(event: unknown) => void>();
+  constructor(sessionId: string) { this.sessionId = sessionId; }
+  subscribe = (listener: (event: unknown) => void): (() => void) => { this.listeners.add(listener); return () => this.listeners.delete(listener); };
+  prompt = async (text: string, options?: { preflightResult?: (accepted: boolean) => void }): Promise<void> => {
+    this.isStreaming = true;
+    this.isIdle = false;
+    this.promptCalls.push({ text, preflight: options?.preflightResult ?? (() => {}) });
+  };
+  steer = async (): Promise<void> => {};
+  followUp = async (): Promise<void> => {};
+  abort = async (): Promise<void> => { this.abortCount++; this.isStreaming = false; this.isIdle = true; };
+  dispose = (): void => { this.disposeCount++; };
+  emit(event: unknown): void { for (const listener of this.listeners) listener(event); }
+  finish(): void { this.isStreaming = false; this.isIdle = true; this.emit({ type: "agent_settled" }); }
+}
+
+const testMessages: BrokerMessageHooks = {
+  ready: (runtime) => ({ type: "ready", runtimeId: runtime.runtimeId, sessionId: runtime.session.sessionId }),
+  stateSync: (_runtime, snapshot) => ({ type: "stateSync", state: snapshot }),
+  event: (event) => ({ type: "event", event }),
+  choiceRequest: (request) => ({ type: "choiceRequest", ...request }),
+  promptAccepted: (requestId) => ({ type: "promptAccepted", requestId }),
+  promptRejected: (requestId) => ({ type: "promptRejected", requestId }),
+};
+
+function setup(options: Partial<ConstructorParameters<typeof SessionBroker>[0]> = {}) {
+  const clock = new FakeClock();
+  const sessions = new Map<string, FakeSession>();
+  const broker = new SessionBroker({
+    clock,
+    factory: { create: async (key) => { const session = new FakeSession(key); sessions.set(key, session); return session; } },
+    serializer: { snapshot: (runtime) => ({ sessionId: runtime.session.sessionId }) },
+    onDelete: async () => {},
+    onError: () => {},
+    messages: testMessages,
+    ...options,
+  });
+  return { clock, sessions, broker };
+}
+
+async function runtimeAndSocket(broker: SessionBroker, key = "session-1") {
+  const runtime = await broker.getOrCreate(key, { owner: "test" });
+  const socket = new FakeSocket();
+  const attachment = runtime.attach(socket);
+  return { runtime, socket, attachment, session: runtime.session as FakeSession };
+}
+
+test("canonical keys share one in-flight creation and stable runtime identity", async () => {
+  let creates = 0;
+  let resolve!: (session: FakeSession) => void;
+  const broker = new SessionBroker({
+    factory: { create: async () => { creates++; return new Promise<FakeSession>((done) => { resolve = done; }); } },
+    serializer: { snapshot: () => ({}) },
+    messages: testMessages,
+    onDelete: async () => {},
+    onError: () => {},
+  });
+  const first = broker.getOrCreate("42", { value: 1 });
+  const second = broker.getOrCreate(String(42), { value: 2 });
+  await Promise.resolve();
+  assert.equal(creates, 1);
+  const session = new FakeSession("42");
+  resolve(session);
+  const [a, b] = await Promise.all([first, second]);
+  assert.equal(a, b);
+  assert.equal(a.runtimeId, b.runtimeId);
+  assert.equal(a.context.value, 1);
+});
+
+test("attach sends a synchronous snapshot barrier, fanout isolates failures, and detach does not abort", async () => {
+  const { broker } = setup();
+  const { runtime, socket, attachment, session } = await runtimeAndSocket(broker);
+  const second = new FakeSocket();
+  runtime.attach(second);
+  assert.deepEqual(messagesOf(socket).slice(0, 2).map((m) => m.type), ["ready", "stateSync"]);
+  session.emit({ type: "delta", value: 1 });
+  assert.equal(messagesOf(socket)[2].type, "event");
+  attachment.detach();
+  assert.equal(session.abortCount, 0);
+  session.emit({ type: "delta", value: 2 });
+  assert.equal(messagesOf(second).filter((m) => m.type === "event").length, 2);
+  const broken = new FakeSocket();
+  broken.send = () => { throw new Error("broken socket"); };
+  assert.throws(() => runtime.attach(broken), /Socket closed during attach/);
+  session.emit({ type: "safe" });
+  assert.equal(messagesOf(second).filter((m) => m.type === "event").length, 3);
+});
+
+test("explicit abort is ordered and idempotent", async () => {
+  const { broker } = setup();
+  const { runtime, attachment, session } = await runtimeAndSocket(broker);
+  await runtime.abort(attachment);
+  await runtime.abort(attachment);
+  assert.equal(session.abortCount, 1);
+});
+
+test("mutating commands preserve order and reject stale attachments", async () => {
+  const { broker } = setup();
+  const { runtime, attachment } = await runtimeAndSocket(broker);
+  const order: string[] = [];
+  const first = runtime.invoke(attachment, async () => { order.push("first"); await new Promise((resolve) => setTimeout(resolve, 2)); order.push("first-done"); });
+  const second = runtime.invoke(attachment, async () => { order.push("second"); });
+  await Promise.all([first, second]);
+  assert.deepEqual(order, ["first", "first-done", "second"]);
+  attachment.detach();
+  await assert.rejects(runtime.invoke(attachment, async () => {}), /Attachment/);
+});
+
+test("prompt records prevent duplicates and detect fingerprint conflicts", async () => {
+  const { broker } = setup();
+  const { runtime, attachment, session, socket } = await runtimeAndSocket(broker);
+  const request = { requestId: "req-1", text: "hello" };
+  const first = runtime.prompt(attachment, request, "same");
+  await Promise.resolve();
+  assert.equal(session.promptCalls.length, 1);
+  const duplicate = runtime.prompt(attachment, request, "same");
+  session.promptCalls[0].preflight(true);
+  await Promise.all([first, duplicate]);
+  assert.equal(session.promptCalls.length, 1);
+  assert.equal(messagesOf(socket).filter((m) => m.type === "promptAccepted").length, 2);
+  await assert.rejects(runtime.prompt(attachment, { ...request, text: "other" }, "different"), /different payload/);
+});
+
+test("prompt preflight rejection releases the chain and records the rejection", async () => {
+  const { broker } = setup();
+  const { runtime, attachment, session, socket } = await runtimeAndSocket(broker);
+  const first = runtime.prompt(attachment, { requestId: "reject", text: "no" }, "x");
+  await Promise.resolve();
+  session.promptCalls[0].preflight(false);
+  await assert.rejects(first, /rejected/);
+  const duplicate = runtime.prompt(attachment, { requestId: "reject", text: "no" }, "x");
+  await assert.rejects(duplicate, /rejected/);
+  assert.equal(messagesOf(socket).filter((m) => m.type === "promptRejected").length, 2);
+});
+
+test("pending choices replay on reconnect and resolve once", async () => {
+  const { broker } = setup();
+  const { runtime, attachment } = await runtimeAndSocket(broker);
+  const choice = runtime.requestChoice({ choiceId: "choice-1", payload: { question: "Continue?" } });
+  attachment.detach();
+  const socket = new FakeSocket();
+  const reconnected = runtime.attach(socket);
+  assert.equal(messagesOf(socket).filter((m) => m.type === "choiceRequest").length, 1);
+  await runtime.resolveChoice(reconnected, "choice-1", "yes");
+  assert.equal(await choice, "yes");
+  await assert.rejects(runtime.resolveChoice(reconnected, "choice-1", "no"), /not pending/);
+});
+
+test("deletion is conservative and removes an idle runtime once", async () => {
+  const deleted: string[] = [];
+  const { broker } = setup({ onDelete: async (key) => { deleted.push(key); } });
+  const { runtime, attachment, session } = await runtimeAndSocket(broker);
+  const other = new FakeSocket();
+  const otherAttachment = runtime.attach(other);
+  await assert.rejects(runtime.delete(), /active or has multiple/);
+  other.open = false;
+  otherAttachment.detach();
+  await broker.delete("session-1");
+  assert.equal(session.disposeCount, 1);
+  assert.deepEqual(deleted, ["session-1"]);
+  const replacement = await broker.getOrCreate("session-1", {});
+  assert.notEqual(replacement, runtime);
+});
+
+test("sweep retains active work, evicts idle runtimes, and enforces capacity", async () => {
+  const { broker, clock, sessions } = setup({ idleTtlMs: 10, activeMaxAgeMs: 100, maxRuntimes: 2 });
+  const a = await runtimeAndSocket(broker, "a");
+  a.attachment.detach();
+  clock.time = 20;
+  await broker.sweep();
+  assert.equal(a.session.disposeCount, 1);
+  const b = await runtimeAndSocket(broker, "b");
+  b.attachment.detach();
+  const c = await runtimeAndSocket(broker, "c");
+  assert.equal(sessions.has("b"), true);
+  assert.equal(sessions.has("c"), true);
+  assert.equal(b.session.disposeCount, 0);
+  c.attachment.detach();
+  const d = await runtimeAndSocket(broker, "d");
+  assert.equal(b.session.disposeCount, 1);
+  d.session.isIdle = false;
+  d.session.isStreaming = true;
+  d.attachment.detach();
+  clock.time = 1000;
+  await broker.sweep();
+  assert.equal(d.session.abortCount, 1);
+});
+
+test("shutdown is bounded, repeated, and disposes late factory results", async () => {
+  let resolve!: (session: FakeSession) => void;
+  const broker = new SessionBroker({
+    factory: { create: async () => new Promise<FakeSession>((done) => { resolve = done; }) },
+    serializer: { snapshot: () => ({}) },
+  });
+  const pending = broker.getOrCreate("late", {});
+  const shutting = broker.shutdown(20);
+  await Promise.resolve();
+  resolve(new FakeSession("late"));
+  await assert.rejects(pending, /shut down/);
+  await shutting;
+  await broker.shutdown(20);
+});
+
+test("deletion reservation stays visible until file removal completes", async () => {
+  let release!: () => void;
+  const { broker } = setup({ onDelete: async () => new Promise<void>((resolve) => { release = resolve; }) });
+  const { runtime } = await runtimeAndSocket(broker, "delete-race");
+  const deleting = broker.delete("delete-race");
+  await Promise.resolve();
+  const replacement = broker.getOrCreate("delete-race", {});
+  let created = false;
+  void replacement.then(() => { created = true; });
+  await Promise.resolve();
+  assert.equal(created, false);
+  assert.equal(runtime.state, "disposed");
+  release();
+  assert.notEqual(await replacement, runtime);
+  await deleting;
+});
+
+test("capacity reservation rejects concurrent over-limit creation before factory await", async () => {
+  let release!: () => void;
+  let creates = 0;
+  const broker = new SessionBroker({
+    maxRuntimes: 1,
+    factory: { create: async (key) => { creates++; return new Promise<FakeSession>((resolve) => { release = () => resolve(new FakeSession(key)); }); } },
+    serializer: { snapshot: () => ({}) },
+  });
+  const first = broker.getOrCreate("first", {});
+  await assert.rejects(broker.getOrCreate("second", {}), /capacity/);
+  assert.equal(creates, 1);
+  release();
+  await first;
+});
+
+test("synchronous prompt starter failure settles the request record", async () => {
+  const { broker } = setup({ promptStarter: () => { throw new Error("starter failed"); } });
+  const { runtime, attachment } = await runtimeAndSocket(broker);
+  const request = { requestId: "sync-failure", text: "hello" };
+  await assert.rejects(runtime.prompt(attachment, request, "throw"), /starter failed/);
+  await assert.rejects(runtime.prompt(attachment, request, "throw"), /starter failed/);
+});
+
+test("late preflight cannot accept a terminal runtime", async () => {
+  let preflight!: (accepted: boolean) => void;
+  const { broker } = setup({ promptStarter: (_session, _request, callback) => {
+    preflight = callback;
+    return new Promise<void>(() => {});
+  } });
+  const { runtime, attachment } = await runtimeAndSocket(broker);
+  const pending = runtime.prompt(attachment, { requestId: "late", text: "hello" }, "late");
+  await Promise.resolve();
+  await runtime.dispose();
+  preflight(true);
+  await assert.rejects(pending, /disposed/);
+});
+
+test("closed attachments detach and delayed responses require the current generation", async () => {
+  const { broker } = setup();
+  const { runtime, attachment, session } = await runtimeAndSocket(broker);
+  const generation = attachment.generation;
+  attachment.attachment.sink.isOpen = () => false;
+  session.emit({ type: "delta" });
+  assert.equal(runtime.attachmentCount, 0);
+  const socket = new FakeSocket();
+  const next = runtime.attach(socket);
+  assert.equal(runtime.isAttachmentValid(next, generation), false);
+  assert.equal(runtime.isAttachmentValid(next, next.generation), true);
+  next.detach();
+});
+
+test("shutdown disposes a runtime even when abort never settles", async () => {
+  let session!: FakeSession;
+  const broker = new SessionBroker({
+    factory: { create: async (key) => { session = new FakeSession(key); session.isIdle = false; session.isStreaming = true; session.abort = async () => new Promise<void>(() => {}); return session; } },
+    serializer: { snapshot: () => ({}) },
+  });
+  await broker.getOrCreate("hang", {});
+  await broker.shutdown(10);
+  assert.equal(session.abortCount, 0);
+  assert.equal(session.disposeCount, 1);
+  await broker.shutdown(10);
+  assert.equal(session.disposeCount, 1);
+});
+
+test("a throwing event hook does not stop fanout", async () => {
+  const errors: unknown[] = [];
+  const { broker } = setup({
+    onError: (error) => errors.push(error),
+    messages: {
+      ready: () => ({ type: "ready" }),
+      stateSync: () => ({ type: "stateSync" }),
+      event: () => { throw new Error("event hook failed"); },
+      choiceRequest: (request) => ({ type: "choiceRequest", ...request }),
+      promptAccepted: (requestId) => ({ type: "promptAccepted", requestId }),
+      promptRejected: (requestId) => ({ type: "promptRejected", requestId }),
+    },
+  });
+  const { runtime, session } = await runtimeAndSocket(broker);
+  const second = new FakeSocket();
+  runtime.attach(second);
+  session.emit({ type: "event" });
+  assert.equal(runtime.attachmentCount, 2);
+  assert.equal(errors.length, 1);
+});
+
+test("direct runtime deletion does not leave a disposed registry entry", async () => {
+  const { broker } = setup();
+  const { runtime } = await runtimeAndSocket(broker, "direct-delete");
+  await runtime.delete();
+  assert.equal(broker.get("direct-delete"), undefined);
+  const replacement = await broker.getOrCreate("direct-delete", {});
+  assert.notEqual(replacement, runtime);
+});