feat(simulation): isolate filesystem and disconnect llm (#35590)

packages/simulation/src/backend/control.ts

@@ -17,6 +17,7 @@ import { SimulationNetwork } from "./network"
  *                              as `llm.request` notifications
  * - `llm.chunk`   { id, items }   append response items to an exchange
  * - `llm.finish`  { id, reason? } finish an exchange
+ * - `llm.disconnect` { id } abruptly terminate an exchange without a finish
  * - `llm.pending`                 list open exchanges
  * - `network.log`                 simulated network request log
  */
@@ -54,6 +55,11 @@ async function handle(socket: ControlSocket, request: SimulationProtocol.JsonRpc
       await Effect.runPromise(SimulationLLMExchange.push(params.id, [{ type: "finish", reason: params.reason }]))
       return { ok: true }
     }
+    case "llm.disconnect": {
+      const params = await SimulationProtocol.Backend.decodeDisconnectParams(request.params)
+      await Effect.runPromise(SimulationLLMExchange.disconnect(params.id))
+      return { ok: true }
+    }
     case "llm.pending":
       return { exchanges: SimulationLLMExchange.pending() }
     case "network.log":

packages/simulation/src/backend/fs-util.ts

@@ -8,22 +8,100 @@ import path from "path"
 /**
  * Simulation replacement for `FSUtil`.
  *
- * The real `FSUtil` layer builds most helpers on the injected
+ * This implementation is intentionally self-contained and only uses the
- * `FileSystem.FileSystem`, but `readDirectoryEntries`, `glob`, and `globUp`
+ * injected simulated `FileSystem.FileSystem`. The default FSUtil layer has a
- * reach for node `fs/promises` and the `glob` package directly, and `resolve`
+ * few helpers that reach host-node APIs directly; depending on it here makes it
- * canonicalizes through the host filesystem. This wraps the real layer and
+ * easy for mutation paths to escape or miss the in-memory project tree.
- * reroutes those through the injected `FileSystem`/lexical path resolution so
- * every read observes the in-memory tree.
  */
 
 const layer = Layer.effect(
   FSUtil.Service,
   Effect.gen(function* () {
-    const base = yield* FSUtil.Service
     const fs = yield* FileSystem.FileSystem
 
-    const resolve = Effect.fn("SimulationFSUtil.resolve")(function* (input: string) {
+    const existsSafe = Effect.fn("SimulationFSUtil.existsSafe")(function* (file: string) {
-      return input
+      return yield* fs.exists(file).pipe(Effect.orElseSucceed(() => false))
+    })
+
+    const isDir = Effect.fn("SimulationFSUtil.isDir")(function* (file: string) {
+      const info = yield* fs.stat(file).pipe(Effect.catch(() => Effect.succeed(undefined)))
+      return info?.type === "Directory"
+    })
+
+    const isFile = Effect.fn("SimulationFSUtil.isFile")(function* (file: string) {
+      const info = yield* fs.stat(file).pipe(Effect.catch(() => Effect.succeed(undefined)))
+      return info?.type === "File"
+    })
+
+    const realPath = Effect.fn("SimulationFSUtil.realPath")(function* (file: string) {
+      return yield* fs.realPath(file)
+    })
+
+    const stat = Effect.fn("SimulationFSUtil.stat")(function* (file: string) {
+      return yield* fs.stat(file)
+    })
+
+    const readFile = Effect.fn("SimulationFSUtil.readFile")(function* (file: string) {
+      return yield* fs.readFile(file)
+    })
+
+    const readFileString = Effect.fn("SimulationFSUtil.readFileString")(function* (file: string) {
+      return yield* fs.readFileString(file)
+    })
+
+    const readFileStringSafe = Effect.fn("SimulationFSUtil.readFileStringSafe")(function* (file: string) {
+      return yield* fs
+        .readFileString(file)
+        .pipe(Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(undefined)))
+    })
+
+    const readJson = Effect.fn("SimulationFSUtil.readJson")(function* (file: string) {
+      const text = yield* readFileString(file)
+      return JSON.parse(text) as unknown
+    })
+
+    const writeFile = Effect.fn("SimulationFSUtil.writeFile")(function* (
+      file: string,
+      data: Uint8Array,
+      options?: Parameters<typeof fs.writeFile>[2],
+    ) {
+      return yield* fs.writeFile(file, data, options)
+    })
+
+    const writeFileString = Effect.fn("SimulationFSUtil.writeFileString")(function* (
+      file: string,
+      data: string,
+      options?: Parameters<typeof fs.writeFileString>[2],
+    ) {
+      return yield* fs.writeFileString(file, data, options)
+    })
+
+    const makeDirectory: FileSystem.FileSystem["makeDirectory"] = (file, options) => fs.makeDirectory(file, options)
+
+    const ensureDir = Effect.fn("SimulationFSUtil.ensureDir")(function* (file: string) {
+      yield* fs.makeDirectory(file, { recursive: true })
+    })
+
+    const writeWithDirs = Effect.fn("SimulationFSUtil.writeWithDirs")(function* (
+      file: string,
+      content: string | Uint8Array,
+      mode?: number,
+    ) {
+      const write =
+        typeof content === "string"
+          ? fs.writeFileString(file, content)
+          : fs.writeFile(file, content)
+      yield* write.pipe(
+        Effect.catchReason("PlatformError", "NotFound", () =>
+          fs.makeDirectory(path.dirname(file), { recursive: true }).pipe(Effect.andThen(write)),
+        ),
+      )
+      if (mode !== undefined) yield* fs.chmod(file, mode)
+    })
+
+    const writeJson = Effect.fn("SimulationFSUtil.writeJson")(function* (file: string, data: unknown, mode?: number) {
+      yield* writeFileString(file, JSON.stringify(data, null, 2))
+      if (mode !== undefined) yield* fs.chmod(file, mode)
     })
 
     const readDirectoryEntries = Effect.fn("SimulationFSUtil.readDirectoryEntries")(function* (dirPath: string) {
@@ -48,6 +126,10 @@ const layer = Layer.effect(
       )
     })
 
+    const resolve = Effect.fn("SimulationFSUtil.resolve")(function* (input: string) {
+      return path.resolve(input)
+    })
+
     const glob = Effect.fn("SimulationFSUtil.glob")(function* (pattern: string, options?: Glob.Options) {
       const cwd = path.resolve(options?.cwd ?? process.cwd())
       const entries = yield* fs
@@ -80,9 +162,53 @@ const layer = Layer.effect(
       return result
     })
 
-    return FSUtil.Service.of({ ...base, readDirectoryEntries, resolve, glob, globUp })
+    const up = Effect.fn("SimulationFSUtil.up")(function* (options: { targets: string[]; start: string; stop?: string }) {
+      const result: string[] = []
+      let current = path.resolve(options.start)
+      while (true) {
+        for (const target of options.targets) {
+          const search = path.join(current, target)
+          if (yield* fs.exists(search)) result.push(search)
+        }
+        if (options.stop === current) break
+        const parent = path.dirname(current)
+        if (parent === current) break
+        current = parent
+      }
+      return result
+    })
+
+    const findUp = Effect.fn("SimulationFSUtil.findUp")(function* (target: string, start: string, stop?: string) {
+      return yield* up({ targets: [target], start, stop })
+    })
+
+    return FSUtil.Service.of({
+      ...fs,
+      realPath,
+      stat,
+      readFile,
+      readFileString,
+      writeFile,
+      writeFileString,
+      makeDirectory,
+      isDir,
+      isFile,
+      existsSafe,
+      readFileStringSafe,
+      readJson,
+      writeJson,
+      ensureDir,
+      writeWithDirs,
+      readDirectoryEntries,
+      resolve,
+      findUp,
+      up,
+      globUp,
+      glob,
+      globMatch: Glob.match,
+    })
   }),
-).pipe(Layer.provide(FSUtil.layer))
+)
 
 export const node = makeGlobalNode({ service: FSUtil.Service, layer, deps: [filesystem] })
 

packages/simulation/src/backend/llm-exchange.ts

@@ -80,6 +80,14 @@ export const push = (id: string, chunks: readonly Chunk[]) =>
     yield* Queue.offerAll(exchange.queue, chunks)
   })
 
+/** Abruptly ends the provider body without a finish chunk or SSE sentinel. */
+export const disconnect = (id: string) =>
+  Effect.gen(function* () {
+    const exchange = state.exchanges.get(id)
+    if (!exchange) return yield* Effect.fail(new ExchangeNotFoundError(id))
+    yield* Queue.shutdown(exchange.queue)
+  })
+
 /**
  * Registers a listener for newly opened exchanges and immediately replays
  * currently-pending ones, so a late-attaching driver observes requests that

packages/simulation/src/protocol/index.ts

@@ -127,6 +127,9 @@ export namespace Backend {
   })
   export interface FinishParams extends Schema.Schema.Type<typeof FinishParams> {}
 
+  export const DisconnectParams = Schema.Struct({ id: Schema.String })
+  export interface DisconnectParams extends Schema.Schema.Type<typeof DisconnectParams> {}
+
   export const OpenedExchange = Schema.Struct({ id: Schema.String, url: Schema.String, body: Schema.Json })
   export interface OpenedExchange extends Schema.Schema.Type<typeof OpenedExchange> {}
 
@@ -140,6 +143,7 @@ export namespace Backend {
 
   export const decodeChunkParams = Schema.decodeUnknownPromise(ChunkParams)
   export const decodeFinishParams = Schema.decodeUnknownPromise(FinishParams)
+  export const decodeDisconnectParams = Schema.decodeUnknownPromise(DisconnectParams)
 }
 
 export * as SimulationProtocol from "./index"