fix: clean up deps / architechture

Author: harlan-zw

packages/core/src/api/http.ts

@@ -151,18 +151,37 @@ export function createHttpRouter(opts: CreateHttpRouterOptions): Router {
           const iterable = isAsyncIterable(result) ? result : await result
           setResponseHeader(event, 'Content-Type', 'application/x-ndjson')
           const res = event.node.res
-          try {
-            if (isAsyncIterable(iterable)) {
-              for await (const chunk of iterable)
-                res.write(`${JSON.stringify(chunk)}\n`)
+          const req = event.node.req
+          if (isAsyncIterable(iterable)) {
+            // Manual iteration so a client disconnect closes the iterator (its
+            // `return()` unsubscribes and unblocks a pending next()) — a `for
+            // await` alone would keep pulling live events and hold the slot open
+            // for the whole scan after the client is gone.
+            const iterator = iterable[Symbol.asyncIterator]()
+            const onClose = () => { void iterator.return?.() }
+            req.on('close', onClose)
+            try {
+              while (true) {
+                const { value, done } = await iterator.next()
+                if (done)
+                  break
+                res.write(`${JSON.stringify(value)}\n`)
+              }
             }
-            else {
-              // Handler resolved a single value instead of an iterable — emit one line.
-              res.write(`${JSON.stringify(iterable)}\n`)
+            finally {
+              req.off('close', onClose)
+              await iterator.return?.()
+              res.end()
             }
           }
-          finally {
-            res.end()
+          else {
+            // Handler resolved a single value instead of an iterable — emit one line.
+            try {
+              res.write(`${JSON.stringify(iterable)}\n`)
+            }
+            finally {
+              res.end()
+            }
           }
           return
         }

packages/core/src/core.ts

@@ -223,13 +223,15 @@ function createSession(deps: SessionDeps): CrawlSession {
 
   // ── HookEvent fan-out ──────────────────────────────────────────────────
   //
-  // Internal queue + iterator that resolves either a buffered event or the
-  // next emit. Keeping a single queue means hook subscribers and iter
-  // consumers stay in sync: every stable event is pushed exactly once.
-  const queue: HookEvent[] = []
-  let resolveNext: ((v: IteratorResult<HookEvent>) => void) | null = null
-  let iterDone = false
+  // Every stable event fans out to all registered `handlers` (multicast). Both
+  // WS broadcast and each `events` async-iterator consumer register here, so a
+  // second consumer (a second dashboard tab, a CLI tail) sees the full stream
+  // instead of stealing events from the first — the iterator is per-consumer,
+  // not a single shared queue.
   const handlers = new Set<(event: HookEvent) => void>()
+  // Live `events` iterators, closed when the scan ends so their `for await`
+  // loops terminate.
+  const iterClosers = new Set<() => void>()
 
   // In-memory ring buffer (cap 10k) for `events.subscribe.replay`.
   const RING_CAP = 10_000
@@ -248,39 +250,64 @@ function createSession(deps: SessionDeps): CrawlSession {
         logOperationalWarn('core.scan_event_subscriber_failed', err, { scanId }, deps.logger)
       }
     }
-    if (iterDone)
-      return
-    if (resolveNext) {
-      const r = resolveNext
-      resolveNext = null
-      r({ value: event, done: false })
-    }
-    else {
-      queue.push(event)
-    }
   }
 
   function closeIter(): void {
-    iterDone = true
-    if (resolveNext) {
-      const r = resolveNext
-      resolveNext = null
-      r({ value: undefined, done: true })
-    }
+    for (const close of [...iterClosers])
+      close()
   }
 
   const events: AsyncIterable<HookEvent> = {
-    [Symbol.asyncIterator]() {
+    [Symbol.asyncIterator](): AsyncIterator<HookEvent> {
+      // Per-consumer buffer + waiter, fed by its own subscription. Cleaned up
+      // on scan end (iterClosers) or when the consumer stops (`return()` — the
+      // HTTP layer calls it on client disconnect so an abandoned tail unsubs
+      // and stops holding a slot).
+      const localQueue: HookEvent[] = []
+      let localResolve: ((v: IteratorResult<HookEvent>) => void) | null = null
+      let done = false
+
+      const unsub = subscribe((event) => {
+        if (done)
+          return
+        if (localResolve) {
+          const r = localResolve
+          localResolve = null
+          r({ value: event, done: false })
+        }
+        else {
+          localQueue.push(event)
+        }
+      })
+
+      function finish(): void {
+        if (done)
+          return
+        done = true
+        unsub()
+        iterClosers.delete(finish)
+        if (localResolve) {
+          const r = localResolve
+          localResolve = null
+          r({ value: undefined, done: true })
+        }
+      }
+      iterClosers.add(finish)
+
       return {
         next(): Promise<IteratorResult<HookEvent>> {
-          if (queue.length)
-            return Promise.resolve({ value: queue.shift()!, done: false })
-          if (iterDone)
+          if (localQueue.length)
+            return Promise.resolve({ value: localQueue.shift()!, done: false })
+          if (done)
             return Promise.resolve({ value: undefined, done: true })
           return new Promise((r) => {
-            resolveNext = r
+            localResolve = r
           })
         },
+        return(): Promise<IteratorResult<HookEvent>> {
+          finish()
+          return Promise.resolve({ value: undefined, done: true })
+        },
       }
     },
   }

packages/unlighthouse/src/host.ts

@@ -37,6 +37,7 @@ import { initStorage } from './cli/storage-init'
 import { resolveConfig } from './config/resolve'
 import { historySubscriber } from './data/history/tracking'
 import { mountServer } from './server'
+import { checkWsUpgrade, isExposedHost, normaliseOrigin } from './server-guards'
 import { createServerHooks } from './server-hooks'
 import { computeConfigCacheKey, normaliseHost } from './util'
 
@@ -470,7 +471,33 @@ export async function createUnlighthouseHost(opts: CreateUnlighthouseHostOptions
     await mountServer(mountDeps, app, { handlerCtx })
 
     if (ws) {
+      // The WS handshake arrives as a Node `'upgrade'` event on the raw server,
+      // so it bypasses the h3 pipeline and its origin gate entirely. Apply the
+      // same D-043 gate here: restrict to the `/api/ws` path and run the
+      // Origin/Host check, so a cross-origin page cannot open the scan-event
+      // stream and a rebinding Host cannot reach it. Same posture inputs as the
+      // HTTP gate (server.ts).
+      const wsPath = joinURL(apiPath, 'ws')
+      const wsSiteOrigin = normaliseOrigin(typeof resolvedConfig.site === 'string' ? resolvedConfig.site : null)
+      const wsExposed = isExposedHost((resolvedConfig.server as { hostname?: string } | undefined)?.hostname)
+      const wsTrustLoopbackOrigin = !process.env.UNLIGHTHOUSE_CORS_ORIGINS && !process.env.UNLIGHTHOUSE_API_TOKEN
       server.on('upgrade', (request: IncomingMessage, socket: Socket) => {
+        const decision = checkWsUpgrade({
+          reqPath: (request.url ?? '').split('?')[0] ?? '',
+          wsPath,
+          host: request.headers.host ?? null,
+          origin: request.headers.origin ?? null,
+          referer: request.headers.referer ?? null,
+          siteOrigin: wsSiteOrigin,
+          exposed: wsExposed,
+          trustLoopbackOrigin: wsTrustLoopbackOrigin,
+        })
+        if (decision._tag === 'reject') {
+          logger.warn?.(`ws: upgrade rejected — ${decision.reason}`)
+          socket.write('HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n')
+          socket.destroy()
+          return
+        }
         ws.handleUpgrade(request, socket)
       })
     }

packages/unlighthouse/src/server-guards.ts

@@ -173,6 +173,27 @@ export function checkApiOrigin(input: ApiOriginCheckInput): GuardDecision {
   return { _tag: 'reject', reason: `cross-origin request from ${requestOrigin}` }
 }
 
+export interface WsUpgradeCheckInput extends ApiOriginCheckInput {
+  /** Request path (no query), e.g. `/api/ws`. */
+  reqPath: string
+  /** The only path the WS server accepts an upgrade on. */
+  wsPath: string
+}
+
+/**
+ * Decide whether a WebSocket upgrade may proceed. The handshake arrives as a
+ * Node `'upgrade'` event on the raw server, bypassing the h3 pipeline and its
+ * origin gate, so this applies the same D-043 protection to the WS surface:
+ * the upgrade must target the WS path exactly, and its Origin/Host must pass
+ * `checkApiOrigin`. Closes a cross-origin page opening the scan-event stream
+ * and a rebinding Host reaching it.
+ */
+export function checkWsUpgrade(input: WsUpgradeCheckInput): GuardDecision {
+  if (input.reqPath !== input.wsPath)
+    return { _tag: 'reject', reason: `path ${input.reqPath}` }
+  return checkApiOrigin(input)
+}
+
 // ── /__launch path constraint ────────────────────────────────────────────────
 
 export type LaunchPathResult

test/core.test.ts

@@ -189,6 +189,60 @@ describe('createUnlighthouseCore orchestration', () => {
     expect(session.state()).toBe('complete')
   })
 
+  it('session.events is per-consumer multicast: two concurrent tails both see the full stream', async () => {
+    const urls = ['https://example.com/a', 'https://example.com/b', 'https://example.com/c']
+    const storage: Storage = memoryStorage()
+    const core = createUnlighthouseCore({
+      config: baseConfig,
+      auditor: passingAuditor(),
+      seeds: emptySeeds,
+      crawler: discoveryCrawler(urls),
+      storage,
+    })
+
+    const session = core.run()
+    const collect = async () => {
+      const names: string[] = []
+      for await (const e of session.events as AsyncIterable<HookEvent>)
+        names.push(e.event)
+      return names
+    }
+    // Two consumers started before the scan finishes. Previously they shared a
+    // single queue and stole events from each other; now each gets its own.
+    const [a, b] = await Promise.all([collect(), collect()])
+
+    // Both terminate (the scan-end close fans out to every live iterator)...
+    expect(a.at(-1)).toBe('scan:complete')
+    expect(b.at(-1)).toBe('scan:complete')
+    // ...and both see every route-complete, none stolen by the other.
+    expect(a.filter(n => n === 'scan:route-complete')).toHaveLength(3)
+    expect(b.filter(n => n === 'scan:route-complete')).toHaveLength(3)
+    expect(a).toEqual(b)
+  })
+
+  it('abandoning a session.events iterator (return) unsubscribes it', async () => {
+    const urls = ['https://example.com/a', 'https://example.com/b']
+    const storage: Storage = memoryStorage()
+    const core = createUnlighthouseCore({
+      config: baseConfig,
+      auditor: passingAuditor(),
+      seeds: emptySeeds,
+      crawler: discoveryCrawler(urls),
+      storage,
+    })
+    const session = core.run()
+    const it = (session.events as AsyncIterable<HookEvent>)[Symbol.asyncIterator]()
+    const first = await it.next()
+    expect(first.done).toBe(false)
+    // Simulate a client disconnect: the HTTP layer calls return() on close.
+    const closed = await it.return!()
+    expect(closed.done).toBe(true)
+    // A subsequent next() stays done and the scan still completes cleanly.
+    expect((await it.next()).done).toBe(true)
+    await session.done
+    expect(session.state()).toBe('complete')
+  })
+
   it('scanner.include scopes the audited routes (config was previously ignored)', async () => {
     const urls = [
       'https://example.com/',

test/server-guards.test.ts

@@ -6,6 +6,7 @@
 
 import {
   checkApiOrigin,
+  checkWsUpgrade,
   isExposedHost,
   isLoopbackHostname,
   normaliseOrigin,
@@ -156,6 +157,36 @@ describe('host / origin classification helpers', () => {
   })
 })
 
+describe('checkWsUpgrade — WS handshake gate', () => {
+  const base = { wsPath: '/api/ws', siteOrigin: null, exposed: false, trustLoopbackOrigin: true }
+
+  it('allows the WS path with a same-origin handshake', () => {
+    const d = checkWsUpgrade({ ...base, reqPath: '/api/ws', host: 'localhost:5678', origin: LOCAL, referer: null })
+    expect(d._tag).toBe('allow')
+  })
+
+  it('allows the loopback UI dev server in the default posture', () => {
+    const d = checkWsUpgrade({ ...base, reqPath: '/api/ws', host: 'localhost:5678', origin: 'http://localhost:3002', referer: null })
+    expect(d._tag).toBe('allow')
+  })
+
+  it('rejects an upgrade on any other path', () => {
+    const d = checkWsUpgrade({ ...base, reqPath: '/api/scan/start', host: 'localhost:5678', origin: LOCAL, referer: null })
+    expect(d._tag).toBe('reject')
+    expect(d.reason).toContain('path')
+  })
+
+  it('rejects a cross-origin handshake (a remote page opening the event stream)', () => {
+    const d = checkWsUpgrade({ ...base, reqPath: '/api/ws', host: 'localhost:5678', origin: 'http://evil.com', referer: null })
+    expect(d._tag).toBe('reject')
+  })
+
+  it('rejects a loopback handshake once the deployment is locked down', () => {
+    const d = checkWsUpgrade({ ...base, trustLoopbackOrigin: false, reqPath: '/api/ws', host: 'localhost:5678', origin: 'http://localhost:3002', referer: null })
+    expect(d._tag).toBe('reject')
+  })
+})
+
 describe('resolveLaunchPath — traversal constraint', () => {
   const ROOT = '/home/user/project'