most of the driver is tested

__mocks__/isomorphic-ws.js

···
+export {WebSocket} from 'mock-socket'

package-lock.json

···
         "ts-pattern": "^5.9.0",
         "tsx": "^4.19.2",
         "webtorrent": "^2.6.8",
-        "ws": "^8.18.2",
+        "ws": "^8.18.3",
         "zod": "^4.1.12"
       },
       "devDependencies": {

package.json

···
     "ts-pattern": "^5.9.0",
     "tsx": "^4.19.2",
     "webtorrent": "^2.6.8",
-    "ws": "^8.18.2",
+    "ws": "^8.18.3",
     "zod": "^4.1.12"
   },
   "devDependencies": {

src/lib/crypto/errors.ts

···
 /** Thrown when failing to verify a JWT signature */
 export class JWTBadSignatureError extends CryptoError {
   constructor(options?: BaseErrorOpts) {
-    super('could not verify token signature', options)
+    super('bad signature', options)
   }
 }

src/lib/crypto/jwts.spec.ts

···
 describe('jwtPayload', () => {
   it('extracts and validates payload', async () => {
     const pair = await generateSigningJwkPair()
-    const token = await generateSignableJwt({
-      sub: 'test',
-      payload: {userId: '123', name: 'Alice'},
-    }).sign(pair.privateKey)
+    const token = await generateSignableJwt({sub: 'test', userId: '123', name: 'Alice'}).sign(pair.privateKey)
 
     const schema = jwtPayload(z.object({userId: z.string(), name: z.string()}))
     const result = await schema.parseAsync(token)
···
 
   it('includes token and claims in result', async () => {
     const pair = await generateSigningJwkPair()
-    const token = await generateSignableJwt({
-      sub: 'test',
-      payload: {data: 'value'},
-    }).sign(pair.privateKey)
+    const token = await generateSignableJwt({sub: 'test', data: 'value'}).sign(pair.privateKey)
 
     const schema = jwtPayload(z.object({data: z.string()}))
     const result = await schema.parseAsync(token)
···
       throw new Error('Should have thrown')
     } catch (err) {
       expect(err).toHaveProperty('message')
-      expect((err as Error).message).toContain('could not verify token signature')
+      expect((err as Error).message).toContain('bad signature')
     }
   })
 })

src/lib/crypto/jwts.ts

···
  * schema describing a verified payload in a JWT.
  * **important** - this does no claims validation, only decoding from string to JWT!
  */
-export const jwtPayload = <T>(schema: z.ZodType<T>): z.ZodType<JWTTokenPayload<T>> => {
-  const parser = z.looseObject({payload: schema})
+export const jwtPayload = <T>(schema: z.ZodType<T>): z.ZodType<JWTTokenPayload<T>, string> => {
   return jwtSchema.transform(async (payload, ctx) => {
-    const result = await parser.safeParseAsync(payload.claims)
-    if (result.success) return {...payload, payload: result.data.payload}
+    const result = await schema.safeParseAsync(payload.claims)
+    if (result.success) return {...payload, payload: result.data}
 
     ctx.addIssue({code: 'invalid_type', expected: 'custom'})
     return z.NEVER

src/lib/errors.ts

···
  * normalizes the given error into a protocol error
  * passes through input that is already protocol errors.
  */
-export function normalizeProtocolError(cause: unknown): ProtocolError {
+export function normalizeProtocolError(cause: unknown, status = 500): ProtocolError {
   if (cause instanceof ProtocolError) return cause
   if (cause instanceof ZodError) return new ProtocolError(prettifyError(cause), 400, {cause})
 
···
     if (cause.name === 'TimeoutError') return new ProtocolError('operation timed out', 408, {cause})
     if (cause.name === 'AbortError') return new ProtocolError('operation was aborted', 499, {cause})
 
-    return new ProtocolError(cause.message, 500, {cause})
+    return new ProtocolError(cause.message, status, {cause})
   }
 
   // fallback, unknown
   const options = cause == undefined ? undefined : {cause: normalizeError(cause)}
-  return new ProtocolError(`Error! ${cause}`, 500, options)
+  return new ProtocolError(`Error! ${cause}`, status, options)
 }
 
 /** error wrapper for unknown errors (not an Error?) */

src/lib/socket.spec.ts

···
 import {afterEach, beforeEach, describe, expect, it} from '@jest/globals'
+import {WebSocket} from 'isomorphic-ws'
 import WS from 'jest-websocket-mock'
 import {z} from 'zod/v4'
 
 import {streamSocket, streamSocketSchema, takeSocket, takeSocketSchema} from '#lib/socket'
 import {jsonCodec} from './schema'
+import {assertParsed} from './utils'
 
 let server: WS
 const TEST_URL = 'ws://localhost:1234'
···
 beforeEach(() => {
   server = new WS(TEST_URL)
 })
+
 afterEach(() => {
   WS.clean()
 })
···
 
     const streamPromise = (async () => {
       for await (const msg of streamSocketSchema(ws, jsonCodec(schema))) {
-        messages.push(msg)
+        messages.push(assertParsed(msg))
         if (messages.length === 2) {
           break
         }
···
     expect(messages).toEqual([{id: 1}, {id: 2}])
   })
 
-  it('rejects invalid messages', async () => {
+  it('sends errors for invalid messages', async () => {
     const ws = await createConnectedWebSocket()
     const schema = z.object({id: z.number()})
 
     const streamPromise = (async () => {
-      for await (const _ of streamSocketSchema(ws, jsonCodec(schema))) {
-        // Should reject
+      for await (const msg of streamSocketSchema(ws, jsonCodec(schema))) {
+        expect(msg.success).toBe(false)
+        expect(msg.error?.message).toContain('id')
+        return true
       }
     })()
 
     server.send('{"id":"not a number"}')
-
-    await expect(streamPromise).rejects.toThrow()
+    await expect(streamPromise).resolves.toBe(true)
   })
 })

src/lib/socket.ts

···
+import {MessageEvent, WebSocket} from 'isomorphic-ws'
 import {z} from 'zod/v4'
 
 import {combineSignals} from '#lib/async/aborts'
···
 export interface ConfigProps {
   /** maximum size of the stream buffer */
   maxDepth: number
+
+  /** skip errors, or raise? */
+  skipErrors?: boolean
 
   /** signal for abort */
   signal?: AbortSignal
···
   ws: WebSocket,
   schema: z.ZodType<T>,
   config?: Partial<ConfigProps>,
-): AsyncGenerator<T> {
+): AsyncGenerator<z.ZodSafeParseResult<T>> {
   for await (const message of streamSocket(ws, config)) {
-    yield await schema.parseAsync(message)
+    yield await schema.safeParseAsync(message)
   }
 }

src/lib/specs/helpers-socket.ts

···
+import {afterEach, beforeEach} from '@jest/globals'
+import {WebSocket} from 'isomorphic-ws'
+import WS from 'jest-websocket-mock'
+
+export type SocketServerState = {
+  url: string
+  server?: WS
+}
+
+export function mockSocketServer(url = 'ws://localhost:1234') {
+  const state: SocketServerState = {url}
+
+  beforeEach(() => {
+    state.server = new WS(url)
+  })
+
+  afterEach(() => {
+    WS.clean()
+  })
+
+  return {
+    url,
+    state,
+
+    get nextMessage() {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      return this.state.server!.nextMessage
+    },
+
+    send(data: string | object) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.state.server!.send(data)
+    },
+
+    async connect(): Promise<WebSocket> {
+      const {promise, resolve, reject} = Promise.withResolvers<WebSocket>()
+      const client = new WebSocket(this.state.url)
+
+      const open = () => {
+        resolve(client)
+        client.removeEventListener('open', open)
+        client.removeEventListener('error', error)
+      }
+
+      const error = (e: unknown) => {
+        reject(e)
+        client.removeEventListener('open', open)
+        client.removeEventListener('error', error)
+      }
+
+      client.addEventListener('open', open)
+      client.addEventListener('error', error)
+
+      return promise
+    },
+  }
+}

src/lib/utils.ts

···
+import {z} from 'zod/v4'
+
 export function assert<V>(value: V | undefined | null): V {
   if (value == null) throw new Error('unexpected nullish!')
   return value
 }
+
+export function assertParsed<T>(value: z.ZodSafeParseResult<T>): T {
+  if (!value.success) throw value.error
+  return value.data
+}

src/realm/protocol/messages-realm.ts

···
 import {jwkSchema} from '#lib/crypto/jwks'
-import {jwtPayload} from '#lib/crypto/jwts'
 import {z} from 'zod/v4'
 
 import {makeEventSchema, makeRequestSchema, makeResponseSchema} from '#lib/schema'
···
   z.object({
     localid: IdentBrand.schema,
     remoteid: IdentBrand.schema,
-    signed: jwtPayload(
-      z.object({
-        signal: z.string(),
-        initiator: z.boolean(),
-      }),
-    ),
+    signed: z.string(),
   }),
 )
 

src/realm/server/driver-preauth.spec.ts

···
+import {describe, expect, it} from '@jest/globals'
+
+import {generateSignableJwt, generateSigningJwkPair, jwkExport} from '#lib/crypto/jwks'
+import {jsonCodec} from '#lib/schema'
+import {mockSocketServer} from '#lib/specs/helpers-socket'
+
+import {
+  PreauthAuthnRequest,
+  PreauthExchangeInviteRequest,
+  PreauthRegisterRequest,
+  preauthResSchema,
+} from '#realm/protocol/messages-preauth'
+import {IdentBrand, RealmBrand} from '#realm/schema/brands'
+
+import {drivePreauth} from './driver-preauth'
+import {ensureRealm, fetchRealm} from './storage'
+
+// Note: With jest-websocket-mock, the "server" is our test harness (WS),
+// and the "client" is what we're testing (the drivePreauth function).
+// This feels backwards because we're testing a server-side component.
+
+const socket = mockSocketServer()
+
+describe('drivePreauth', () => {
+  describe('preauth.register', () => {
+    it('should successfully register a new realm', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+
+      // Export public key as JWK
+      const pubkeyJwk = await jwkExport.parseAsync(keyPair.publicKey)
+
+      // Create and sign the registration request
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.register',
+        seq: 'sequence',
+        dat: {
+          pubkey: pubkeyJwk,
+        },
+      } satisfies PreauthRegisterRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .setJti('nonce')
+        .sign(keyPair.privateKey)
+
+      // Start the preauth driver
+      const authPromise = drivePreauth(ws)
+
+      // Send the JWT from the "server" (test harness) to "client" (code under test)
+      socket.send(jwt)
+
+      // Wait for response
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      // Verify the response
+      expect(result).not.toBeNull()
+      expect(result?.realm.realmid).toBe(realmid)
+      expect(result?.identid).toBe(identid)
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'res',
+        msg: 'preauth.authn',
+        dat: {
+          peers: ['server'],
+          identities: {
+            [identid]: pubkeyJwk,
+          },
+        },
+      })
+
+      // Verify realm was created in storage
+      const realm = await fetchRealm(realmid)
+      expect(realm).not.toBeNull()
+      expect(realm?.realmid).toBe(realmid)
+    }, 1000)
+
+    it('should return existing realm if already registered', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+
+      // Pre-create the realm
+      await ensureRealm(realmid, identid, keyPair.publicKey)
+      const pubkeyJwk = await crypto.subtle.exportKey('jwk', keyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.register',
+        seq: 'sequence',
+        dat: {
+          pubkey: pubkeyJwk,
+        },
+      } satisfies PreauthRegisterRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .setJti('nonce')
+        .sign(keyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).not.toBeNull()
+      expect(result?.realm.realmid).toBe(realmid)
+      expect(result?.identid).toBe(identid)
+    }, 1000)
+  })
+
+  describe('preauth.authn', () => {
+    it('should successfully authenticate an existing identity', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+      const pubkeyJwk = await jwkExport.parseAsync(keyPair.publicKey)
+
+      // Pre-create the realm and identity
+      await ensureRealm(realmid, identid, keyPair.publicKey)
+
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.authn',
+        seq: 'sequence',
+      } satisfies PreauthAuthnRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .setJti('nonce')
+        .sign(keyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).not.toBeNull()
+      expect(result?.realm.realmid).toBe(realmid)
+      expect(result?.identid).toBe(identid)
+      expect(result?.pubkey).toBeDefined()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'res',
+        msg: 'preauth.authn',
+        dat: expect.objectContaining({
+          peers: expect.arrayContaining(['server']),
+          identities: expect.objectContaining({
+            [identid]: pubkeyJwk,
+          }),
+        }),
+      })
+    }, 1000)
+
+    it('should reject authentication for unknown realm', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.authn',
+        seq: 'sequence',
+      } satisfies PreauthAuthnRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .setJti('nonce')
+        .sign(keyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 404,
+          detail: '404 Not Found: unknown realm',
+        },
+      })
+    }, 1000)
+
+    it('should reject authentication for unknown identity', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const ownerIdentid = IdentBrand.generate()
+      const unknownIdentid = IdentBrand.generate()
+      const ownerKeyPair = await generateSigningJwkPair()
+      const unknownKeyPair = await generateSigningJwkPair()
+
+      // Create realm with owner identity
+      await ensureRealm(realmid, ownerIdentid, ownerKeyPair.publicKey)
+
+      // Try to authenticate with unknown identity
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.authn',
+        seq: 'sequence',
+      } satisfies PreauthAuthnRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(unknownIdentid)
+        .setJti('nonce')
+        .sign(unknownKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 401,
+          detail: '401 Unauthorized: bad signature',
+        },
+      })
+    }, 1000)
+
+    it('should reject authentication with invalid signature', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+      const wrongKeyPair = await generateSigningJwkPair()
+
+      // Create realm with correct identity
+      await ensureRealm(realmid, identid, keyPair.publicKey)
+
+      // Sign with wrong private key
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.authn',
+        seq: 'sequence',
+      } satisfies PreauthAuthnRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .setJti('nonce')
+        .sign(wrongKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 401,
+          detail: '401 Unauthorized: bad signature',
+        },
+      })
+    }, 1000)
+  })
+
+  describe('preauth.exchange', () => {
+    it('should successfully exchange an invitation', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const inviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const inviterKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+
+      // create realm with inviter identity
+      const realm = await ensureRealm(realmid, inviterIdentid, inviterKeyPair.publicKey)
+      const nonce = crypto.randomUUID()
+
+      // create the invitation; payload doesn't matter, just claims
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(inviterIdentid)
+        .setJti(nonce)
+        .sign(inviterKeyPair.privateKey)
+
+      // create exchange request signed by invitee
+      const inviteePubkeyJwk = await jwkExport.parseAsync(inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const result = await authPromise
+      expect(result).not.toBeNull()
+      expect(result?.realm.realmid).toBe(realmid)
+      expect(result?.identid).toBe(inviteeIdentid)
+
+      const response = await socket.nextMessage
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'res',
+        msg: 'preauth.authn',
+        dat: expect.objectContaining({
+          peers: expect.arrayContaining(['server']),
+          identities: expect.objectContaining({
+            [inviterIdentid]: expect.anything(),
+          }),
+        }),
+      })
+
+      // verify nonce was consumed
+      const nonceValid = realm.validateNonce(nonce)
+      expect(nonceValid).toBe(false)
+    }, 1000)
+
+    it('should reject invitation without nonce', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const inviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const inviterKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+
+      await ensureRealm(realmid, inviterIdentid, inviterKeyPair.publicKey)
+
+      // create the invitation; payload doesn't matter, just claims
+      // NO NONCE FOR THIS TEST
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(inviterIdentid)
+        .sign(inviterKeyPair.privateKey)
+
+      const inviteePubkeyJwk = await crypto.subtle.exportKey('jwk', inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 400,
+          detail: '400 Bad Request: missing nonce',
+        },
+      })
+    }, 1000)
+
+    it('should reject invitation with reused nonce', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const inviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const inviterKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+
+      const realm = await ensureRealm(realmid, inviterIdentid, inviterKeyPair.publicKey)
+
+      // consume the nonce
+      const nonce = crypto.randomUUID()
+      realm.validateNonce(nonce)
+
+      // create the invitation; payload doesn't matter, just claims
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(inviterIdentid)
+        .setJti(nonce)
+        .sign(inviterKeyPair.privateKey)
+
+      const inviteePubkeyJwk = await crypto.subtle.exportKey('jwk', inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 403,
+          detail: '403 Forbidden: nonce already seen',
+        },
+      })
+    }, 1000)
+
+    it('should reject invitation from unknown identity', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const ownerIdentid = IdentBrand.generate()
+      const unknownInviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const ownerKeyPair = await generateSigningJwkPair()
+      const unknownInviterKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+      const nonce = crypto.randomUUID()
+
+      // create realm with owner only
+      await ensureRealm(realmid, ownerIdentid, ownerKeyPair.publicKey)
+
+      // create the invitation; payload doesn't matter, just claims
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(unknownInviterIdentid)
+        .setJti(nonce)
+        .sign(unknownInviterKeyPair.privateKey)
+
+      const inviteePubkeyJwk = await crypto.subtle.exportKey('jwk', inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 401,
+          detail: '401 Unauthorized: bad signature',
+        },
+      })
+    }, 1000)
+
+    it('should reject invitation with invalid signature', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const inviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const inviterKeyPair = await generateSigningJwkPair()
+      const wrongKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+
+      await ensureRealm(realmid, inviterIdentid, inviterKeyPair.publicKey)
+      const nonce = crypto.randomUUID()
+
+      // create the invitation; payload doesn't matter, just claims
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(inviterIdentid)
+        .setJti(nonce)
+        .sign(wrongKeyPair.privateKey)
+
+      const inviteePubkeyJwk = await crypto.subtle.exportKey('jwk', inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 401,
+          detail: '401 Unauthorized: bad signature',
+        },
+      })
+    }, 1000)
+
+    it('should reject exchange for unknown realm', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const inviterIdentid = IdentBrand.generate()
+      const inviteeIdentid = IdentBrand.generate()
+      const inviterKeyPair = await generateSigningJwkPair()
+      const inviteeKeyPair = await generateSigningJwkPair()
+
+      // don't create realm
+      const nonce = crypto.randomUUID()
+
+      // create the invitation; payload doesn't matter, just claims
+      const inviteJwt = await generateSignableJwt({})
+        .setAudience(realmid)
+        .setIssuer(inviterIdentid)
+        .setJti(nonce)
+        .sign(inviterKeyPair.privateKey)
+
+      const inviteePubkeyJwk = await crypto.subtle.exportKey('jwk', inviteeKeyPair.publicKey)
+      const payload = {
+        typ: 'req',
+        msg: 'preauth.exchange',
+        seq: 'sequence',
+        dat: {
+          pubkey: inviteePubkeyJwk,
+          inviteJwt: inviteJwt,
+        },
+      } satisfies PreauthExchangeInviteRequest
+
+      const jwt = await generateSignableJwt(payload)
+        .setAudience(realmid)
+        .setIssuer(inviteeIdentid)
+        .sign(inviteeKeyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      const response = await socket.nextMessage
+      const result = await authPromise
+
+      expect(result).toBeNull()
+
+      const responseData = jsonCodec(preauthResSchema).safeParse(response)
+      expect(responseData.success).toBeTruthy()
+      expect(responseData.data).toMatchObject({
+        typ: 'err',
+        msg: 'preauth.authn',
+        err: {
+          code: 404,
+          detail: '404 Not Found: unknown realm',
+        },
+      })
+    }, 1000)
+  })
+
+  describe('timeout handling', () => {
+    it('should timeout if no message received within timeout', async () => {
+      const ws = await socket.connect()
+      const authPromise = drivePreauth(ws, 250)
+      await expect(authPromise).rejects.toThrow()
+    }, 500)
+
+    it('should respect external abort signal', async () => {
+      const ws = await socket.connect()
+      const controller = new AbortController()
+      const authPromise = drivePreauth(ws, undefined, controller.signal)
+
+      setTimeout(() => {
+        controller.abort()
+      }, 100)
+      await expect(authPromise).rejects.toThrow()
+    }, 500)
+  })
+
+  describe('malformed requests', () => {
+    it('should reject non-JWT messages', async () => {
+      const ws = await socket.connect()
+      const authPromise = drivePreauth(ws)
+
+      socket.send('not a jwt')
+
+      await expect(authPromise).rejects.toThrow()
+    }, 1000)
+
+    it('should reject JWT with invalid payload schema', async () => {
+      const ws = await socket.connect()
+      const realmid = RealmBrand.generate()
+      const identid = IdentBrand.generate()
+      const keyPair = await generateSigningJwkPair()
+
+      const jwt = await generateSignableJwt({huh: 'wat'})
+        .setAudience(realmid)
+        .setIssuer(identid)
+        .sign(keyPair.privateKey)
+
+      const authPromise = drivePreauth(ws)
+      socket.send(jwt)
+
+      await expect(authPromise).rejects.toThrow()
+    }, 1000)
+  })
+})

src/realm/server/driver-preauth.ts

···
+import WebSocket from 'isomorphic-ws'
 import {match} from 'ts-pattern'
 
 import {combineSignals, timeoutSignal} from '#lib/async/aborts'
-import {jwkImport} from '#lib/crypto/jwks'
+import {JWK, jwkExport, jwkImport} from '#lib/crypto/jwks'
 import {jwtPayload, jwtSchema, JWTTokenPayload, verifyJwtToken} from '#lib/crypto/jwts'
 import {normalizeProtocolError, ProtocolError} from '#lib/errors'
 import {putSocket, takeSocket} from '#lib/socket'
···
 import {Realm} from './realm'
 import {ensureRealm, fetchRealm} from './storage'
 
-export async function drivePreauth(ws: WebSocket, _signal?: AbortSignal) {
-  const timeout = timeoutSignal(3_000)
+export type AuthenticatedIdentity = {
+  realm: Realm
+  identid: IdentID
+  pubkey: CryptoKey
+  pubjwk: JWK
+}
+
+export type PreauthOptions = {
+  timeout?: number
+  signal?: AbortSignal
+}
+
+/**
+ * drives a websocket through the authentication state machine
+ *
+ * - within timeout (default 3 seconds)
+ * - expect an authn, invite, or registration message, and handle appriately
+ *
+ * @param ws - the socket to drive
+ * @param timeout - options
+ * @param signal - an optional abort signal to cancel the machine
+ * @returns a promise which resolves when the machine has completed (socket has disconnected)
+ */
+export async function drivePreauth(
+  ws: WebSocket,
+  timeoutms = 3_000,
+  _signal?: AbortSignal,
+): Promise<AuthenticatedIdentity | null> {
+  const timeout = timeoutSignal(timeoutms)
   const signal = combineSignals(_signal, timeout.signal)
 
   try {
···
       })
       return auth
     } catch (exc: unknown) {
-      const error = normalizeProtocolError(exc)
+      const error = normalizeProtocolError(exc, 401)
+      if (error.status >= 500) throw error
+
       putSocket<PreauthErrorResponse>(ws, {
         typ: 'err',
         msg: 'preauth.authn',
···
           detail: error.message,
         },
       })
-
-      throw error
+      return null
     }
   } finally {
     timeout.cancel()
   }
 }
 
-async function preauthAuthenticate(realm: Realm, identid: IdentID, token: string) {
+async function preauthAuthenticate(
+  realm: Realm,
+  identid: IdentID,
+  token: string,
+): Promise<AuthenticatedIdentity> {
   const pubkey = await realm.fetchIdentity(identid)
-  if (pubkey == null) throw new ProtocolError('unknown identity', 401)
+  if (pubkey == null) throw new ProtocolError('bad signature', 401)
 
   await verifyJwtToken(token, pubkey)
-  return {realm, identid, pubkey}
+
+  const pubjwk = await jwkExport.parseAsync(pubkey)
+  return {realm, identid, pubkey, pubjwk}
 }
 
 async function preauthRegister(
···
   realmid: RealmID,
   jwt: JWTTokenPayload<PreauthExchangeInviteRequest>,
 ): Promise<Realm | null> {
-  const token = jwtSchema.parse(jwt.payload.dat.inviteJwt)
   const realm = await fetchRealm(realmid)
   if (!realm) return null
 
-  // invitation has an unused nonce
-  if (!token.claims.jti) throw new ProtocolError('invitation missing nonce!', 400)
-  if (!realm.validateNonce(token.claims.jti)) throw new ProtocolError('invitation nonce already used!', 403)
+  // invitation must have an unused nonce
+  const token = jwtSchema.parse(jwt.payload.dat.inviteJwt)
+  if (!token.claims.jti) throw new ProtocolError('missing nonce', 400)
+  if (!realm.validateNonce(token.claims.jti)) throw new ProtocolError('nonce already seen', 403)
 
   // invitation is issued and signed by the inviter
   const inviter = IdentBrand.parse(token.claims.iss)
   const inviterkey = await realm.fetchIdentity(inviter)
-  if (inviterkey == null) throw new ProtocolError('invitation from unknown identity?', 403)
+  if (inviterkey == null) throw new ProtocolError('bad signature', 401)
 
   // looked everything up, but we still need check the signature
   await verifyJwtToken(jwt.payload.dat.inviteJwt, inviterkey)
+
+  // invitation is valid, add them to the realm!
+  const invitee = IdentBrand.parse(jwt.claims.iss)
+  const inviteekey = await jwkImport.parseAsync(jwt.payload.dat.pubkey)
+  await realm.admitIdentity(invitee, inviteekey)
+
+  // invitee is valid, add them to the realm!
   return realm
 }

src/realm/server/driver-realm.spec.ts

···
+import {afterEach, beforeEach, describe, expect, it} from '@jest/globals'
+import {WebSocket} from 'isomorphic-ws'
+import {z} from 'zod/v4'
+
+import {sleep} from '#lib/async/sleep'
+import {generateSignableJwt, generateSigningJwkPair, jwkExport} from '#lib/crypto/jwks'
+import {jwtPayload} from '#lib/crypto/jwts'
+import {exceptionMessageSchema, jsonCodec} from '#lib/schema'
+import {mockSocketServer} from '#lib/specs/helpers-socket'
+
+import {
+  RealmAnnounceRequest,
+  realmAnnounceResponseSchema,
+  RealmBroadcastEvent,
+  RealmPeerLeftEvent,
+  realmPeerLeftEventSchema,
+  RealmPingRequest,
+  RealmPingResponse,
+  realmPingResponseSchema,
+  RealmSignalEvent,
+  realmSignalEventSchema,
+} from '#realm/protocol/messages-realm'
+import {actionSchema, RealmAction} from '#realm/schema'
+import {IdentBrand, RealmBrand, RealmID} from '#realm/schema/brands'
+import {generate as generateTimestamp} from '#realm/schema/timestamp'
+
+import {AuthenticatedIdentity} from './driver-preauth'
+import {driveRealm} from './driver-realm'
+import {Realm} from './realm'
+
+const socket = mockSocketServer()
+
+async function createAuthenticatedConnection(
+  realm: Realm,
+  signal?: AbortSignal,
+): Promise<{ws: WebSocket; auth: AuthenticatedIdentity; promise: Promise<unknown>}> {
+  const identid = IdentBrand.generate()
+  const keyPair = await generateSigningJwkPair()
+  const pubkey = keyPair.publicKey
+  const pubjwk = await jwkExport.parseAsync(pubkey)
+  await realm.admitIdentity(identid, pubkey)
+
+  const ws = await socket.connect()
+  realm.attachSocket(identid, ws)
+
+  const auth = {realm, identid, pubkey, pubjwk}
+  const promise = driveRealm(ws, auth, signal)
+
+  return {ws, auth, promise}
+}
+
+describe('driveRealm', () => {
+  let realm: Realm
+  let realmid: RealmID
+
+  beforeEach(() => {
+    realmid = RealmBrand.generate()
+    realm = new Realm(realmid, ':memory:')
+  })
+
+  afterEach(() => {
+    realm.shutdown()
+  })
+
+  describe('connection lifecycle', () => {
+    it('should broadcast peer-left event when connection closes', async () => {
+      const {ws: ws1, auth: auth1, promise: promise1} = await createAuthenticatedConnection(realm)
+      const {ws: ws2, promise: promise2} = await createAuthenticatedConnection(realm)
+      await socket.nextMessage // peer2 -> 1
+
+      // close first connection
+      ws1.close()
+      await promise1
+
+      // second peer should receive peer-left event
+      const message = await socket.nextMessage
+      const event = jsonCodec(realmPeerLeftEventSchema).parse(message)
+      expect(event).toMatchObject({
+        typ: 'evt',
+        msg: 'realm.peer-left',
+        dat: {
+          identid: auth1.identid,
+        },
+      } satisfies RealmPeerLeftEvent)
+
+      // cleanup
+      ws2.close()
+      await promise2
+    }, 1000)
+  })
+
+  describe('realm.ping', () => {
+    it('should respond to ping request with current clocks', async () => {
+      const {ws, auth, promise} = await createAuthenticatedConnection(realm)
+
+      // record some actions to populate clocks
+      const timestamp = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 0)
+      realm.recordActions([
+        {
+          typ: 'act',
+          msg: 'test.action',
+          clk: timestamp,
+          dat: {value: 1},
+        } satisfies RealmAction,
+      ])
+
+      const pingRequest: RealmPingRequest = {
+        typ: 'req',
+        msg: 'realm.ping',
+        seq: 'ping-1',
+        dat: {
+          clocks: {},
+          requestsSync: false,
+        },
+      }
+      socket.send(JSON.stringify(pingRequest))
+
+      // check response
+
+      const message = await socket.nextMessage
+      const response = jsonCodec(realmPingResponseSchema).parse(message)
+      expect(response).toMatchObject({
+        typ: 'res',
+        msg: 'realm.ping',
+        seq: 'ping-1',
+        dat: {
+          clocks: {
+            [auth.identid]: timestamp,
+          },
+        },
+      } satisfies RealmPingResponse)
+
+      ws.close()
+      await promise
+    }, 1000)
+
+    it('should send sync delta when ping requests sync', async () => {
+      const {ws, auth, promise} = await createAuthenticatedConnection(realm)
+      const timestamp = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 0)
+      realm.recordActions([
+        {
+          typ: 'act',
+          msg: 'test.action',
+          clk: timestamp,
+          dat: {value: 1},
+        } satisfies RealmAction,
+      ])
+
+      const pingRequest: RealmPingRequest = {
+        typ: 'req',
+        msg: 'realm.ping',
+        seq: 'ping-sync',
+        dat: {
+          clocks: {},
+          requestsSync: true,
+        },
+      }
+      socket.send(JSON.stringify(pingRequest))
+
+      // should receive ping response
+      const pingResponseMessage = await socket.nextMessage
+      const pingResponse = jsonCodec(realmPingResponseSchema).safeParse(pingResponseMessage)
+      expect(pingResponse.success).toBeTruthy()
+
+      // should also receive sync delta (array of actions)
+      const syncDeltaMessage = await socket.nextMessage
+      const syncDeltaResponse = jsonCodec(z.array(actionSchema)).safeParse(syncDeltaMessage)
+      expect(syncDeltaResponse.success).toBeTruthy()
+      expect(syncDeltaResponse.data).toHaveLength(1)
+      expect(syncDeltaResponse.data).toContainEqual({
+        typ: 'act',
+        msg: 'test.action',
+        clk: timestamp,
+        dat: {value: 1},
+      })
+
+      ws.close()
+      await promise
+    }, 1000)
+  })
+
+  describe('realm.announce', () => {
+    it('should respond to announce request with server info', async () => {
+      const {ws, promise} = await createAuthenticatedConnection(realm)
+
+      // send announce request
+      const announceRequest: RealmAnnounceRequest = {
+        typ: 'req',
+        msg: 'realm.announce',
+        seq: 'announce-1',
+        dat: {
+          clocks: {},
+          requestsSync: false,
+          deviceCaps: {
+            corsFetch: false,
+            networkQuality: 3,
+          },
+          deviceInfo: {
+            name: 'test-client',
+            ua: 'test-ua',
+          },
+        },
+      }
+      socket.send(JSON.stringify(announceRequest))
+
+      // should receive announce response
+      const message = await socket.nextMessage
+      const response = jsonCodec(realmAnnounceResponseSchema).parse(message)
+
+      expect(response).toMatchObject({
+        typ: 'res',
+        msg: 'realm.announce',
+        seq: 'announce-1',
+        dat: expect.objectContaining({
+          clocks: expect.any(Object),
+          deviceCaps: {
+            corsFetch: true,
+            networkQuality: 5,
+          },
+          deviceInfo: expect.objectContaining({
+            name: expect.any(String),
+            ua: expect.any(String),
+          }),
+        }),
+      })
+
+      ws.close()
+      await promise
+    }, 1000)
+
+    it('should send sync delta when announce requests sync', async () => {
+      const {ws, auth, promise} = await createAuthenticatedConnection(realm)
+      const timestamp = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 0)
+      realm.recordActions([
+        {
+          typ: 'act',
+          msg: 'test.action',
+          clk: timestamp,
+          dat: {value: 1},
+        } satisfies RealmAction,
+      ])
+
+      const announceRequest: RealmAnnounceRequest = {
+        typ: 'req',
+        msg: 'realm.announce',
+        seq: 'announce-sync',
+        dat: {
+          clocks: {},
+          requestsSync: true,
+        },
+      }
+      socket.send(JSON.stringify(announceRequest))
+
+      // Should receive announce response
+      const announceMessage = await socket.nextMessage
+      const announceResponse = jsonCodec(realmAnnounceResponseSchema).safeParse(announceMessage)
+      expect(announceResponse.success).toBeTruthy()
+
+      // Should also receive sync delta
+      const syncMessage = await socket.nextMessage
+      const syncResponse = jsonCodec(z.array(actionSchema)).safeParse(syncMessage)
+      expect(syncResponse.success).toBeTruthy()
+      expect(syncResponse.data).toHaveLength(1)
+
+      ws.close()
+      await promise
+    }, 1000)
+  })
+
+  describe('action recording', () => {
+    it('should record actions sent by client', async () => {
+      const {ws, auth, promise} = await createAuthenticatedConnection(realm)
+
+      // send actions
+      const timestamp = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 0)
+      const actions: RealmAction[] = [
+        {
+          typ: 'act',
+          msg: 'test.action',
+          clk: timestamp,
+          dat: {value: 1},
+        },
+      ]
+      socket.send(JSON.stringify(actions))
+
+      // verify actions were recorded
+      await sleep(50) // let it flow
+      const syncState = realm.buildSyncState()
+      expect(syncState[auth.identid]).toBe(timestamp)
+
+      ws.close()
+      await promise
+    }, 1000)
+
+    it('should handle multiple actions in a batch', async () => {
+      const {ws, auth, promise} = await createAuthenticatedConnection(realm)
+
+      // send multiple actions
+      const ts1 = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 0)
+      const ts2 = generateTimestamp(auth.identid, Math.floor(Date.now() / 1000), 1)
+      const actions: RealmAction[] = [
+        {typ: 'act', msg: 'test.action1', clk: ts1, dat: {value: 1}},
+        {typ: 'act', msg: 'test.action2', clk: ts2, dat: {value: 2}},
+      ]
+      socket.send(JSON.stringify(actions))
+
+      // Give it a moment to process
+      await new Promise((resolve) => setTimeout(resolve, 50))
+
+      // verify both actions were recorded
+      const syncState = realm.buildSyncState()
+      expect(syncState[auth.identid]).toBe(ts2)
+
+      const delta = realm.buildSyncDelta({})
+      expect(delta).toHaveLength(2)
+
+      ws.close()
+      await promise
+    }, 1000)
+  })
+
+  describe('realm.broadcast', () => {
+    it('should broadcast message to other peers', async () => {
+      const {ws: ws1, promise: promise1} = await createAuthenticatedConnection(realm)
+      const {ws: ws2, promise: promise2} = await createAuthenticatedConnection(realm)
+      await socket.nextMessage // peer-joined (peer2 -> peer1)
+
+      // send broadcast from peer1
+      const broadcastEvent: RealmBroadcastEvent = {
+        typ: 'evt',
+        msg: 'realm.broadcast',
+        dat: {
+          payload: {message: 'hello world'},
+          recipients: false, // broadcast to all except sender
+        },
+      }
+      socket.send(JSON.stringify(broadcastEvent))
+      await sleep(50)
+
+      // TODO: make sure we only get one more message on the socket
+
+      ws1.close()
+      ws2.close()
+      await promise1
+      await promise2
+    }, 1000)
+
+    it('should broadcast to specific recipients', async () => {
+      const {ws: ws1, promise: promise1} = await createAuthenticatedConnection(realm)
+      const {ws: ws2, auth: auth2, promise: promise2} = await createAuthenticatedConnection(realm)
+      await socket.nextMessage // peer joined -> 1
+
+      const {ws: ws3, promise: promise3} = await createAuthenticatedConnection(realm)
+      await socket.nextMessage // peer joined -> 2
+      await socket.nextMessage // peer joined -> 1
+
+      // Send broadcast to specific recipient
+      const broadcastEvent: RealmBroadcastEvent = {
+        typ: 'evt',
+        msg: 'realm.broadcast',
+        dat: {
+          payload: {message: 'direct message'},
+          recipients: [auth2.identid],
+        },
+      }
+
+      socket.send(JSON.stringify(broadcastEvent))
+      await sleep(50)
+
+      // TODO: make sure there's only one message on the server
+
+      ws1.close()
+      ws2.close()
+      ws3.close()
+      await promise1
+      await promise2
+      await promise3
+    }, 1000)
+  })
+
+  describe('realm.signal', () => {
+    it('should relay signal to specific peer', async () => {
+      const {ws: ws1, auth: auth1, promise: promise1} = await createAuthenticatedConnection(realm)
+      const {ws: ws2, auth: auth2, promise: promise2} = await createAuthenticatedConnection(realm)
+      await socket.nextMessage // peer joined -> 1
+
+      // create a signed signal
+      const keyPair = await generateSigningJwkPair()
+      const signedPayload = {initiator: true}
+      const signedJwt = await generateSignableJwt(signedPayload)
+        .setIssuer(auth1.identid)
+        .setAudience(auth2.identid)
+        .sign(keyPair.privateKey)
+
+      // send signal event
+      const signalEvent = {
+        typ: 'evt',
+        msg: 'realm.signal',
+        dat: {
+          localid: auth1.identid,
+          remoteid: auth2.identid,
+          signed: signedJwt,
+        },
+      } satisfies RealmSignalEvent
+      socket.send(JSON.stringify(signalEvent))
+      await sleep(50)
+
+      // we got the signal
+      const signalData = await socket.nextMessage
+      const signalMessage = jsonCodec(realmSignalEventSchema).safeParse(signalData)
+      expect(signalMessage.success).toBeTruthy()
+
+      // and it's got the payload in the signed jwt
+      const message = await jwtPayload(z.object({initiator: z.boolean()})).safeParseAsync(
+        signalMessage.data?.dat.signed,
+      )
+      expect(message.success).toBeTruthy()
+      expect(message.data?.payload.initiator).toBe(true)
+
+      ws1.close()
+      ws2.close()
+      await promise1
+      await promise2
+    }, 1000)
+  })
+
+  describe('error handling', () => {
+    it('should handle malformed messages gracefully', async () => {
+      const {ws, promise} = await createAuthenticatedConnection(realm)
+
+      socket.send('not valid json')
+
+      const errorMessage = await socket.nextMessage
+      const errorResult = jsonCodec(exceptionMessageSchema).safeParse(errorMessage)
+      expect(errorResult.success).toBeTruthy()
+
+      ws.close()
+      await promise
+    }, 1000)
+
+    it('should handle unknown message types', async () => {
+      const {ws, promise} = await createAuthenticatedConnection(realm)
+
+      socket.send(
+        JSON.stringify({
+          typ: 'req',
+          msg: 'realm.unknown',
+          seq: 'unknown-1',
+          dat: {},
+        }),
+      )
+
+      const errorMessage = await socket.nextMessage
+      const errorResult = jsonCodec(exceptionMessageSchema).safeParse(errorMessage)
+      expect(errorResult.success).toBeTruthy()
+
+      ws.close()
+      await promise
+    }, 1000)
+  })
+
+  describe('abort signal handling', () => {
+    it('should stop processing when abort signal is triggered', async () => {
+      const controller = new AbortController()
+      const {promise} = await createAuthenticatedConnection(realm, controller.signal)
+
+      controller.abort()
+      await expect(promise).rejects.toThrow()
+    }, 1000)
+  })
+})

src/realm/server/driver-realm.ts

···
-import {JWK, jwkExport} from '#lib/crypto/jwks'
+import WebSocket from 'isomorphic-ws'
+
+import {JWK} from '#lib/crypto/jwks'
 import {normalizeProtocolError, ProtocolError} from '#lib/errors'
-import {makeExceptionMessage} from '#lib/schema'
+import {jsonCodec, makeExceptionMessage} from '#lib/schema'
 import {putSocket, streamSocketSchema} from '#lib/socket'
 
 import {
···
 import {IdentID} from '#realm/schema/brands'
 import {PeerClocks} from '#realm/schema/timestamp'
 
+import {AuthenticatedIdentity} from './driver-preauth'
 import {Realm} from './realm'
 
-export async function driveRealm(
-  ws: WebSocket,
-  realm: Realm,
-  identid: IdentID,
-  pubkey: CryptoKey,
-  signal?: AbortSignal,
-) {
-  const pubjwk = await jwkExport.parseAsync(pubkey)
-  broadcastPeerJoined(realm, identid, pubjwk)
+const realmServerParser = jsonCodec(realmServerMessageSchema)
 
+export async function driveRealm(ws: WebSocket, auth: AuthenticatedIdentity, signal?: AbortSignal) {
   try {
-    const stream = streamSocketSchema(ws, realmServerMessageSchema, {signal})
+    broadcastPeerJoined(auth.realm, auth.identid, auth.pubjwk)
+
+    const stream = streamSocketSchema(ws, realmServerParser, {signal, skipErrors: true})
     for await (const data of stream) {
-      driveMessage(ws, realm, identid, data, signal)
+      if (data.success) {
+        driveMessage(ws, auth.realm, auth.identid, data.data)
+      } else {
+        putError(ws, data.error)
+      }
     }
   } catch (exc: unknown) {
     putError(ws, exc) // error in socket stream
+    throw exc
   } finally {
-    broadcastPeerLeft(realm, identid)
+    broadcastPeerLeft(auth.realm, auth.identid)
   }
 }
 
···
           return
 
         case 'realm.signal':
-          realmBroadcast(realm, identid, data.dat, [data.dat.remoteid])
+          realmBroadcast(realm, identid, data, [data.dat.remoteid])
           return
 
         case 'realm.announce':
-          realmAnnounce(ws, realm, identid, data)
+          realmAnnounce(ws, realm, data)
           return
 
         case 'realm.ping':
-          realmPing(ws, realm, identid, data)
+          realmPing(ws, realm, data)
           return
 
         default:
···
   }
 }
 
-function realmPing(ws: WebSocket, realm: Realm, identid: IdentID, req: RealmPingRequest) {
-  console.debug('ping from', identid, req.dat)
-
-  // respond immediately
+function realmPing(ws: WebSocket, realm: Realm, req: RealmPingRequest) {
   const clocks = realm.buildSyncState()
   putSocket<RealmPingResponse>(ws, {
     typ: 'res',
···
   }
 }
 
-function realmAnnounce(ws: WebSocket, realm: Realm, identid: IdentID, req: RealmAnnounceRequest) {
-  console.debug('announce from', identid, req.dat)
+function realmAnnounce(ws: WebSocket, realm: Realm, req: RealmAnnounceRequest) {
   // TODO: store device caps in the identities table?
 
-  // respond immediately
   const clocks = realm.buildSyncState()
   putSocket<RealmAnnounceResponse>(ws, {
     typ: 'res',

src/realm/server/driver.ts

···
 import {normalizeProtocolError} from '#lib/errors'
 import {makeExceptionMessage} from '#lib/schema'
 import {putSocket} from '#lib/socket'
+import {WebSocket} from 'isomorphic-ws'
 
 import {drivePreauth} from './driver-preauth'
 import {driveRealm} from './driver-realm'
···
  * @param ws - the socket to drive through the machine
  * @returns a promise which resolves when the socket session completes
  */
-export async function machine(ws: WebSocket, signal?: AbortSignal) {
+export async function driveSocket(ws: WebSocket, signal?: AbortSignal) {
   try {
-    const {realm, identid, pubkey} = await drivePreauth(ws, signal)
+    const auth = await drivePreauth(ws, 3_000, signal)
+    if (auth == null) return
+
+    const {realm} = auth
     try {
-      realm.attachSocket(identid, ws)
-      await driveRealm(ws, realm, identid, pubkey, signal)
+      realm.attachSocket(auth.identid, ws)
+      console.log('attached socket:', auth.identid)
+      await driveRealm(ws, auth, signal)
     } finally {
-      realm.detachSocket(identid, ws)
+      console.log('detached socket:', auth.identid)
+      realm.detachSocket(auth.identid, ws)
     }
   } catch (exc: unknown) {
     const error = normalizeProtocolError(exc)

src/realm/server/realm.spec.ts

···
+import {beforeEach, describe, expect, it} from '@jest/globals'
+import {WebSocket} from 'isomorphic-ws'
+
 import {RealmAction} from '#realm/schema'
 import {IdentBrand, RealmBrand} from '#realm/schema/brands'
 import {generate as generateTimestamp} from '#realm/schema/timestamp'
-import {beforeEach, describe, expect, it} from '@jest/globals'
 
 import {Realm} from './realm'
 

src/realm/server/realm.ts

···
+import {WebSocket} from 'isomorphic-ws'
 import {DatabaseSync} from 'node:sqlite'
 
 import {JWK, jwkExport, jwkImport, jwkSchema} from '#lib/crypto/jwks'