diff --git a/packages/simulation/src/backend/control.ts b/packages/simulation/src/backend/control.ts index 10bded13af..28bb18c009 100644 --- a/packages/simulation/src/backend/control.ts +++ b/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": diff --git a/packages/simulation/src/backend/fs-util.ts b/packages/simulation/src/backend/fs-util.ts index e24d0c9c4b..4ee267fd64 100644 --- a/packages/simulation/src/backend/fs-util.ts +++ b/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 - * `FileSystem.FileSystem`, but `readDirectoryEntries`, `glob`, and `globUp` - * reach for node `fs/promises` and the `glob` package directly, and `resolve` - * canonicalizes through the host filesystem. This wraps the real layer and - * reroutes those through the injected `FileSystem`/lexical path resolution so - * every read observes the in-memory tree. + * This implementation is intentionally self-contained and only uses the + * injected simulated `FileSystem.FileSystem`. The default FSUtil layer has a + * few helpers that reach host-node APIs directly; depending on it here makes it + * easy for mutation paths to escape or miss the in-memory project 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) { - return input + const existsSafe = Effect.fn("SimulationFSUtil.existsSafe")(function* (file: string) { + 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[2], + ) { + return yield* fs.writeFile(file, data, options) + }) + + const writeFileString = Effect.fn("SimulationFSUtil.writeFileString")(function* ( + file: string, + data: string, + options?: Parameters[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] }) diff --git a/packages/simulation/src/backend/llm-exchange.ts b/packages/simulation/src/backend/llm-exchange.ts index 759f52866c..97b7b71b1f 100644 --- a/packages/simulation/src/backend/llm-exchange.ts +++ b/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 diff --git a/packages/simulation/src/protocol/index.ts b/packages/simulation/src/protocol/index.ts index 6ff63c9bb3..b05f24d501 100644 --- a/packages/simulation/src/protocol/index.ts +++ b/packages/simulation/src/protocol/index.ts @@ -127,6 +127,9 @@ export namespace Backend { }) export interface FinishParams extends Schema.Schema.Type {} + export const DisconnectParams = Schema.Struct({ id: Schema.String }) + export interface DisconnectParams extends Schema.Schema.Type {} + export const OpenedExchange = Schema.Struct({ id: Schema.String, url: Schema.String, body: Schema.Json }) export interface OpenedExchange extends Schema.Schema.Type {} @@ -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"