fix: harden the run relay + workspace sandbox against resource exhaustion

.changeset/eventstream-iterator-cancel.md

@@ -0,0 +1,7 @@
+---
+'@gemstack/ai-autopilot': patch
+---
+
+fix(ai-autopilot): EventStream iterators are now cancellable
+
+A consumer's async iterator gained a `return()` that drops its waiter from the stream and settles any pending `next()`. Previously a consumer that stopped iterating (e.g. a disconnected SSE client) left its waiter registered until the next `push`/`close`, so many short-lived consumers on an idle stream leaked. Live iteration and history replay are unchanged.

.changeset/relay-sandbox-hardening.md

@@ -0,0 +1,10 @@
+---
+'@gemstack/framework': patch
+---
+
+fix(framework): harden the run relay and workspace sandbox against resource exhaustion
+
+- The relay now caps how many runs it holds in memory and evicts the least-recently-used one on overflow. Because it is unauthenticated, an anonymous request to `/r/<id>/…` could previously create per-run state that was never freed; run creation is now bounded (`maxRuns`, default 200).
+- A disconnected SSE viewer now cancels its stream iterator, releasing its waiter immediately instead of lingering on the stream until the next event (which may never arrive for an idle run).
+- `snapshotWorkspace` checks a file's size before reading it, so a large asset in the workspace is skipped without ever being loaded into memory during a `--sandbox docker` sync.
+- `relayPublisher`'s POST has a timeout, so a relay that accepts a connection but never responds can no longer hang the CLI on exit (`flush()`).

README.md

@@ -17,6 +17,7 @@ Full documentation lives in [`docs/`](./docs/guide/index.md) (a hosted site is o
 | [`ai‑sdk`](./packages/ai-sdk) | The agent runtime: providers, the agent loop, tools, streaming, middleware, structured output, memory, and evals. The engine the rest of the AI family builds on. | [Guide](./docs/packages/ai-sdk/index.md) | [![npm](https://img.shields.io/npm/v/@gemstack/ai-sdk)](https://www.npmjs.com/package/@gemstack/ai-sdk) |
 | [`ai‑skills`](./packages/ai-skills) | Portable capability bundles: load `SKILL.md` skills (instructions + tools + resources) and compose them onto an agent on demand. | [Guide](./docs/packages/ai-skills.md) | [![npm](https://img.shields.io/npm/v/@gemstack/ai-skills)](https://www.npmjs.com/package/@gemstack/ai-skills) |
 | [`ai‑autopilot`](./packages/ai-autopilot) | The AI‑building framework: a Supervisor (plan → dispatch → synthesize) plus stack‑aware personas, a runner sandbox, surfaces, a decisions ledger, an event‑triggered review/QA loop with a built‑in prompt library, framework presets (Vike/Next), and a bootstrap flow that takes an app from nothing to production‑grade. | [Guide](./docs/packages/ai-autopilot.md) | [![npm](https://img.shields.io/npm/v/@gemstack/ai-autopilot)](https://www.npmjs.com/package/@gemstack/ai-autopilot) |
+| [`framework`](./packages/framework) | **The (AI) Framework**: turnkey, zero‑config orchestration built on `ai-autopilot`. Wraps a coding‑agent CLI (Claude Code) as a black box and takes an idea to a running app, with a live dashboard, Open Loop domain presets, an optional Docker sandbox, and a run relay for shared sessions. | [README](./packages/framework/README.md) | [![npm](https://img.shields.io/npm/v/@gemstack/framework)](https://www.npmjs.com/package/@gemstack/framework) |
 | [`ai‑mcp`](./packages/ai-mcp) | The agent/MCP bridge: consume a remote MCP server's tools as agent tools, and expose an agent as an MCP server. | [Guide](./docs/packages/ai-mcp.md) | [![npm](https://img.shields.io/npm/v/@gemstack/ai-mcp)](https://www.npmjs.com/package/@gemstack/ai-mcp) |
 | [`mcp`](./packages/mcp) | A standalone framework for *authoring* MCP servers: tools, resources, prompts, decorators, OAuth 2.1, a framework-neutral HTTP handler, and a test client. Agent-agnostic. | [Guide](./docs/packages/mcp.md) | [![npm](https://img.shields.io/npm/v/@gemstack/mcp)](https://www.npmjs.com/package/@gemstack/mcp) |
 | [`mcp‑connectors`](./packages/mcp-connectors) | The connector contract: define a tool connector to an external service once with `defineConnector`, and compose any number into a single MCP server with `mountConnectors`. Built on `@gemstack/mcp`; agent-agnostic. | [Guide](./docs/packages/mcp-connectors.md) | [![npm](https://img.shields.io/npm/v/@gemstack/mcp-connectors)](https://www.npmjs.com/package/@gemstack/mcp-connectors) |
@@ -32,6 +33,7 @@ graph TD
     subgraph fam["AI family · depends on the agent runtime"]
         skills["<b>ai-skills</b><br/>capability bundles"] --> sdk["<b>ai-sdk</b><br/>agent runtime"]
         autopilot["<b>ai-autopilot</b><br/>orchestration"] --> sdk
+        framework["<b>framework</b><br/>turnkey app builder"] --> autopilot
         aimcp["<b>ai-mcp</b><br/>agent ↔ MCP bridge"] --> sdk
     end
     subgraph agn["Agent-agnostic · not ai-*"]

packages/ai-autopilot/README.md

@@ -298,6 +298,29 @@ own bodies from a directory with `loadPromptsFrom(dir)`, or add one to a library
 with `library.add(...)`. The bodies are the main open-source contribution surface;
 PRs that sharpen them are welcome.
 
+## Domain presets — a loop + prompts + skills bundled per domain
+
+A **domain preset** packages a domain's review loops, prompt bodies, and skill
+pointers as one `{ loops, prompts, skills }` unit, so a run is framed for a *kind*
+of work (software, web, data, product, science) instead of the generic default.
+Each preset is a directory of markdown under `presets/` — a `preset.md` manifest
+plus `loops/`, `prompts/`, and `skills/` — so a contributor adds one by writing
+files, not code.
+
+```ts
+import { builtinDomainPresets, selectPreset } from '@gemstack/ai-autopilot'
+
+const presets = await builtinDomainPresets()             // discovers every shipped preset
+const sw = selectPreset(presets, 'software-development')  // sw.loops / sw.prompts / sw.skills
+```
+
+Five ship built in (`software-development`, `web-development`, `data-science`,
+`product-management`, `biological-science`). Load your own with
+`loadDomainPreset(dir)`, and merge several with `composeDomainPresets(...)`
+(presets-of-presets). A mode variant (a `stem.<mode>.md` sibling carrying
+`metadata.conditions`) swaps in a leaner loop under a mode like `technical`.
+`@gemstack/framework` surfaces these to end users as `--preset`.
+
 ## Guardrails
 
 - **`concurrency`** (optional, default 4) — max workers in flight; positive integer.

packages/ai-autopilot/src/surface/events.test.ts

@@ -61,6 +61,23 @@ describe('EventStream async iteration', () => {
     assert.deepEqual(await a, ['plan', 'dispatch-result'])
     assert.deepEqual(await b, ['plan', 'dispatch-result'])
   })
+
+  it('return() cancels a waiting iterator so it does not linger (SSE disconnect)', async () => {
+    const s = new EventStream()
+    const it = s[Symbol.asyncIterator]()
+    const pending = it.next() // no buffered events yet → waits on the stream
+    await Promise.resolve()
+    const ended = await it.return!() // consumer disconnected
+    assert.deepEqual(ended, { value: undefined, done: true })
+    assert.deepEqual(await pending, { value: undefined, done: true }) // the waiting next() settled, not leaked
+    // A later push must not resurrect the cancelled iterator; a fresh one still works.
+    s.push({ type: 'plan', task: 't', subtasks: [] })
+    assert.deepEqual(await it.next(), { value: undefined, done: true })
+    s.close()
+    const fresh: string[] = []
+    for await (const e of s) fresh.push(e.type)
+    assert.deepEqual(fresh, ['plan'])
+  })
 })
 
 describe('formatEvent', () => {

packages/ai-autopilot/src/surface/events.ts

@@ -58,23 +58,43 @@ export class EventStream<E = SupervisorEvent> {
    */
   [Symbol.asyncIterator](): AsyncIterableIterator<E> {
     let index = 0
+    let done = false
+    let pending: (() => void) | undefined
     const stream = this
     return {
       [Symbol.asyncIterator]() {
         return this
       },
       next(): Promise<IteratorResult<E>> {
+        if (done) return Promise.resolve({ value: undefined, done: true })
         if (index < stream.buffer.length) {
           return Promise.resolve({ value: stream.buffer[index++]!, done: false })
         }
         if (stream.closed) return Promise.resolve({ value: undefined, done: true })
         return new Promise(resolve => {
-          stream.waiters.push(() => {
-            if (index < stream.buffer.length) resolve({ value: stream.buffer[index++]!, done: false })
-            else resolve({ value: undefined, done: true })
-          })
+          const wake = (): void => {
+            pending = undefined
+            if (done || index >= stream.buffer.length) resolve({ value: undefined, done: true })
+            else resolve({ value: stream.buffer[index++]!, done: false })
+          }
+          pending = wake
+          stream.waiters.push(wake)
         })
       },
+      // Cancel this iterator (e.g. an SSE client disconnected): drop its waiter so
+      // it does not linger in `waiters` until the next push, and settle any pending
+      // `next()`. Without this, a disconnected consumer leaks until the stream ends.
+      return(): Promise<IteratorResult<E>> {
+        done = true
+        if (pending) {
+          const i = stream.waiters.indexOf(pending)
+          if (i >= 0) stream.waiters.splice(i, 1)
+          const wake = pending
+          pending = undefined
+          wake()
+        }
+        return Promise.resolve({ value: undefined, done: true })
+      },
     }
   }
 }

packages/framework/src/dashboard/sse.ts

@@ -22,10 +22,11 @@ export function serveSSE(
   })
   clients.add(res)
 
+  const iterator = stream[Symbol.asyncIterator]()
   const send = (event: FrameworkEvent) => res.write(`data: ${JSON.stringify(event)}\n\n`)
   void (async () => {
     try {
-      for await (const event of stream[Symbol.asyncIterator]()) send(event)
+      for await (const event of iterator) send(event)
     } catch {
       // client went away
     } finally {
@@ -34,8 +35,11 @@ export function serveSSE(
     }
   })()
 
+  // On disconnect, cancel the iterator so its waiter is released immediately
+  // instead of lingering on the stream until the next event (which may never come).
   req.on('close', () => {
     clients.delete(res)
     res.end()
+    void iterator.return?.()
   })
 }

packages/framework/src/relay.test.ts

@@ -1,6 +1,7 @@
 import { strict as assert } from 'node:assert'
 import { test } from 'node:test'
 import { get, request } from 'node:http'
+import { createServer as createNetServer } from 'node:net'
 import { startRelay, relayPublisher } from './relay.js'
 import type { FrameworkEvent } from './events.js'
 
@@ -123,6 +124,35 @@ test('GET /r/:id redirects to the trailing-slash page; the page loads relative S
   }
 })
 
+test('run count is bounded: least-recently-used runs are evicted at maxRuns (#230 hardening)', async () => {
+  const relay = await startRelay({ ...local, maxRuns: 2 })
+  try {
+    relay.ingest('a', { kind: 'log', message: '1' })
+    relay.ingest('b', { kind: 'log', message: '1' })
+    relay.ingest('a', { kind: 'log', message: '2' }) // touch a → a is now most-recent
+    relay.ingest('c', { kind: 'log', message: '1' }) // over cap → evict the LRU, which is b (not a)
+    assert.deepEqual(relay.runIds().sort(), ['a', 'c'])
+  } finally {
+    await relay.close()
+  }
+})
+
+test('relayPublisher does not hang flush() when the relay accepts but never responds (#230 hardening)', async () => {
+  // A server that accepts the TCP connection and the request but never replies.
+  const stall = createNetServer(() => {})
+  await new Promise<void>(r => stall.listen(0, '127.0.0.1', () => r()))
+  const port = (stall.address() as { port: number }).port
+  try {
+    const errors: unknown[] = []
+    const pub = relayPublisher(`http://127.0.0.1:${port}`, 'x', e => errors.push(e), 250)
+    pub.publish({ kind: 'log', message: 'hi' })
+    await pub.flush() // must resolve (via the fetch timeout), not hang forever
+    assert.equal(errors.length, 1) // the timed-out POST surfaced as an error, best-effort
+  } finally {
+    stall.close()
+  }
+})
+
 test('healthz is 200; a bad publish is rejected without crashing the run', async () => {
   const relay = await startRelay(local)
   try {

packages/framework/src/relay.ts

@@ -34,6 +34,13 @@ export interface RelayOptions {
   title?: string
   /** Max bytes accepted per publish request body. Default 256 KiB. */
   maxBodyBytes?: number
+  /**
+   * Max concurrent runs kept in memory. The relay is unauthenticated, so any
+   * request to `/r/<id>/…` would otherwise create a run that never frees — an
+   * anonymous caller could exhaust memory. On overflow the least-recently-touched
+   * run is evicted (its stream closed, its viewers dropped). Default 200.
+   */
+  maxRuns?: number
 }
 
 /** A running relay. Ingest events programmatically or over HTTP; browsers watch by run id. */
@@ -63,14 +70,29 @@ export function startRelay(opts: RelayOptions = {}): Promise<Relay> {
   const port = opts.port ?? 4488
   const title = opts.title ?? 'The Framework'
   const maxBody = opts.maxBodyBytes ?? 256 * 1024
+  const maxRuns = opts.maxRuns ?? 200
+  // Insertion-ordered as an LRU: touching a run re-inserts it at the end, so the
+  // first entry is always the least-recently-used and the eviction victim.
   const runs = new Map<string, Run>()
 
   const run = (id: string): Run => {
-    let r = runs.get(id)
-    if (!r) {
-      r = { stream: new EventStream<FrameworkEvent>(), clients: new Set() }
-      runs.set(id, r)
+    const existing = runs.get(id)
+    if (existing) {
+      runs.delete(id)
+      runs.set(id, existing) // touch: move to most-recently-used
+      return existing
+    }
+    // Bound memory: evict the least-recently-used run before creating a new one.
+    while (runs.size >= maxRuns) {
+      const oldest = runs.keys().next().value
+      if (oldest === undefined) break
+      const victim = runs.get(oldest)!
+      victim.stream.close()
+      for (const res of victim.clients) res.end()
+      runs.delete(oldest)
     }
+    const r: Run = { stream: new EventStream<FrameworkEvent>(), clients: new Set() }
+    runs.set(id, r)
     return r
   }
 
@@ -202,18 +224,26 @@ export interface RelayPublisher {
  * replay order matches the run, and best-effort: a failed POST is reported via
  * `onError` but never interrupts the run.
  */
-export function relayPublisher(base: string, runId: string, onError?: (err: unknown) => void): RelayPublisher {
+export function relayPublisher(
+  base: string,
+  runId: string,
+  onError?: (err: unknown) => void,
+  timeoutMs = 10_000,
+): RelayPublisher {
   const root = `${base.replace(/\/+$/, '')}/r/${encodeURIComponent(runId)}`
   let chain: Promise<void> = Promise.resolve()
   return {
     url: `${root}/`,
     publish(event) {
       chain = chain.then(async () => {
         try {
+          // Timeout so a relay that accepts but never responds can't wedge flush()
+          // (awaited on shutdown) and hang the whole CLI on exit.
           await fetch(`${root}/publish`, {
             method: 'POST',
             headers: { 'content-type': 'application/json' },
             body: JSON.stringify(event),
+            signal: AbortSignal.timeout(timeoutMs),
           })
         } catch (err) {
           onError?.(err)

packages/framework/src/sandbox.ts

@@ -1,4 +1,4 @@
-import { readdir, readFile } from 'node:fs/promises'
+import { readdir, readFile, stat } from 'node:fs/promises'
 import { join, relative, sep } from 'node:path'
 import type { FileTree } from '@gemstack/ai-autopilot'
 
@@ -55,8 +55,11 @@ export async function snapshotWorkspace(dir: string, opts: SnapshotOptions = {})
         continue
       }
       if (!entry.isFile()) continue // skip symlinks, sockets, devices
+      // Check the size before reading, so a huge asset is skipped without ever
+      // loading it into memory.
+      const st = await stat(child)
+      if (st.size > maxBytes) continue
       const buf = await readFile(child)
-      if (buf.byteLength > maxBytes) continue
       if (buf.includes(0)) continue // binary
       // Always use POSIX separators: the key is a path inside a Linux container.
       tree[relative(dir, child).split(sep).join('/')] = buf.toString('utf8')