fix: replace all mutable Stream fold/collect patterns with immutable runForEach, revert effect to beta.57

package.json

@@ -53,7 +53,7 @@
       "dompurify": "3.3.1",
       "drizzle-kit": "1.0.0-beta.19-d95b7a4",
       "drizzle-orm": "1.0.0-beta.19-d95b7a4",
-      "effect": "4.0.0-beta.59",
+      "effect": "4.0.0-beta.57",
       "ai": "6.0.168",
       "cross-spawn": "7.0.6",
       "hono": "4.10.7",

packages/opencode/src/cli/cmd/debug/ripgrep.ts

@@ -48,15 +48,15 @@ const FilesCommand = effectCmd({
     const ctx = yield* InstanceRef
     if (!ctx) return
     const rg = yield* Ripgrep.Service
-    const files = yield* rg
+    const files: string[] = []
+    yield* rg
       .files({
         cwd: ctx.directory,
         glob: args.glob ? [args.glob] : undefined,
       })
       .pipe(
         Stream.take(args.limit ?? Infinity),
-        Stream.runCollect,
-        Effect.map((c) => [...c]),
+        Stream.runForEach((file) => Effect.sync(() => { files.push(file) })),
         Effect.orDie,
       )
     process.stdout.write(files.join(EOL) + EOL)

packages/opencode/src/file/index.ts

@@ -379,10 +379,8 @@ export const layer = Layer.effect(
 
         next.dirs = Array.from(dirs).toSorted()
       } else {
-        const files = yield* rg.files({ cwd: ctx.directory }).pipe(
-          Stream.runCollect,
-          Effect.map((chunk) => [...chunk]),
-        )
+        const files: string[] = []
+        yield* Stream.runForEach(rg.files({ cwd: ctx.directory }), (file) => Effect.sync(() => { files.push(file) }))
         const seen = new Set<string>()
         for (const file of files) {
           next.files.push(file)

packages/opencode/src/file/ripgrep.ts

@@ -383,17 +383,20 @@ export const layer: Layer.Layer<Service, never, AppFileSystem.Service | ChildPro
           Effect.gen(function* () {
             const handle = yield* spawner.spawn(yield* command(input.cwd, searchArgs(input)))
 
-            const [items, stderr, code] = yield* Effect.all(
+              const rowStream = Stream.decodeText(handle.stdout).pipe(
+                Stream.splitLines,
+                Stream.filter((line) => line.length > 0),
+                Stream.mapEffect(parse),
+                Stream.filter((item): item is Match => item.type === "match"),
+                Stream.map((item) => row(item.data)),
+              )
+              const [items, stderr, code] = yield* Effect.all(
               [
-                Stream.decodeText(handle.stdout).pipe(
-                  Stream.splitLines,
-                  Stream.filter((line) => line.length > 0),
-                  Stream.mapEffect(parse),
-                  Stream.filter((item): item is Match => item.type === "match"),
-                  Stream.map((item) => row(item.data)),
-                  Stream.runCollect,
-                  Effect.map((chunk) => [...chunk]),
-                ),
+                Effect.gen(function* () {
+                  const acc: ReturnType<typeof row>[] = []
+                  yield* Stream.runForEach(rowStream, (item) => Effect.sync(() => { acc.push(item) }))
+                  return acc
+                }),
                 Stream.mkString(Stream.decodeText(handle.stderr)),
                 handle.exitCode,
               ],
@@ -416,7 +419,8 @@ export const layer: Layer.Layer<Service, never, AppFileSystem.Service | ChildPro
 
       const tree: Interface["tree"] = Effect.fn("Ripgrep.tree")(function* (input: TreeInput) {
         log.info("tree", input)
-        const list = Array.from(yield* files({ cwd: input.cwd, signal: input.signal }).pipe(Stream.runCollect))
+        const list: string[] = []
+        yield* Stream.runForEach(files({ cwd: input.cwd, signal: input.signal }), (file) => Effect.sync(() => { list.push(file) }))
 
         interface Node {
           name: string

packages/opencode/src/git/index.ts

@@ -114,23 +114,25 @@ export const layer = Layer.effect(
         })
         const handle = yield* spawner.spawn(proc)
         const collect = (stream: typeof handle.stdout) =>
-          Stream.runFold(
-            stream,
-            () => ({ chunks: [] as Uint8Array[], bytes: 0, truncated: false }),
-            (acc, chunk) => {
-              if (opts.maxOutputBytes === undefined) {
-                acc.chunks.push(chunk)
-                acc.bytes += chunk.length
-                return acc
-              }
-
-              const remaining = opts.maxOutputBytes - acc.bytes
-              if (remaining > 0) acc.chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
-              acc.bytes += chunk.length
-              acc.truncated = acc.truncated || acc.bytes > opts.maxOutputBytes
-              return acc
-            },
-          ).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated })))
+          Effect.gen(function* () {
+            const chunks: Uint8Array[] = []
+            let bytes = 0
+            let truncated = false
+            yield* Stream.runForEach(stream, (chunk) =>
+              Effect.sync(() => {
+                if (opts.maxOutputBytes === undefined) {
+                  chunks.push(chunk)
+                  bytes += chunk.length
+                } else {
+                  const remaining = opts.maxOutputBytes - bytes
+                  if (remaining > 0) chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
+                  bytes += chunk.length
+                  truncated = truncated || bytes > opts.maxOutputBytes
+                }
+              }),
+            )
+            return { buffer: Buffer.concat(chunks), truncated }
+          })
         const [stdout, stderr] = yield* Effect.all([collect(handle.stdout), collect(handle.stderr)], { concurrency: 2 })
         return {
           exitCode: yield* handle.exitCode,

packages/opencode/src/snapshot/index.ts

@@ -573,8 +573,28 @@ export const layer: Layer.Layer<
                     stdin: Stream.make(new TextEncoder().encode(refs.map((item) => item.ref).join("\n") + "\n")),
                   })
                   const handle = yield* spawner.spawn(proc)
+                  const collectUint8 = (stream: typeof handle.stdout) => {
+                    const chunks: Uint8Array[] = []
+                    let bytes = 0
+                    return Stream.runForEach(stream, (chunk) =>
+                      Effect.sync(() => {
+                        chunks.push(chunk)
+                        bytes += chunk.length
+                      }),
+                    ).pipe(
+                      Effect.map(() => {
+                        const result = new Uint8Array(bytes)
+                        let offset = 0
+                        for (const chunk of chunks) {
+                          result.set(chunk, offset)
+                          offset += chunk.length
+                        }
+                        return result
+                      }),
+                    )
+                  }
                   const [out, err] = yield* Effect.all(
-                    [Stream.mkUint8Array(handle.stdout), Stream.mkString(Stream.decodeText(handle.stderr))],
+                    [collectUint8(handle.stdout), Stream.mkString(Stream.decodeText(handle.stderr))],
                     { concurrency: 2 },
                   )
                   const code = yield* handle.exitCode

packages/opencode/src/tool/glob.ts

@@ -47,22 +47,24 @@ export const GlobTool = Tool.define(
 
           const limit = 100
           let truncated = false
-          const files = yield* rg.files({ cwd: search, glob: [params.pattern], signal: ctx.abort }).pipe(
-            Stream.mapEffect((file) =>
-              Effect.gen(function* () {
-                const full = path.resolve(search, file)
-                const info = yield* fs.stat(full).pipe(Effect.catch(() => Effect.succeed(undefined)))
-                const mtime =
-                  info?.mtime.pipe(
-                    Option.map((date) => date.getTime()),
-                    Option.getOrElse(() => 0),
-                  ) ?? 0
-                return { path: full, mtime }
-              }),
+          const files: { path: string; mtime: number }[] = []
+          yield* Stream.runForEach(
+            rg.files({ cwd: search, glob: [params.pattern], signal: ctx.abort }).pipe(
+              Stream.mapEffect((file) =>
+                Effect.gen(function* () {
+                  const full = path.resolve(search, file)
+                  const info = yield* fs.stat(full).pipe(Effect.catch(() => Effect.succeed(undefined)))
+                  const mtime =
+                    info?.mtime.pipe(
+                      Option.map((date) => date.getTime()),
+                      Option.getOrElse(() => 0),
+                    ) ?? 0
+                  return { path: full, mtime }
+                }),
+              ),
+              Stream.take(limit + 1),
             ),
-            Stream.take(limit + 1),
-            Stream.runCollect,
-            Effect.map((chunk) => [...chunk]),
+            (item) => Effect.sync(() => { files.push(item) }),
           )
 
           if (files.length > limit) {

packages/opencode/src/tool/skill.ts

@@ -39,13 +39,16 @@ export const SkillTool = Tool.define(
           const dir = path.dirname(info.location)
           const base = pathToFileURL(dir).href
           const limit = 10
-          const files = yield* rg.files({ cwd: dir, follow: false, hidden: true, signal: ctx.abort }).pipe(
-            Stream.filter((file) => !file.includes("SKILL.md")),
-            Stream.map((file) => path.resolve(dir, file)),
-            Stream.take(limit),
-            Stream.runCollect,
-            Effect.map((chunk) => [...chunk].map((file) => `<file>${file}</file>`).join("\n")),
+          const fileList: string[] = []
+          yield* Stream.runForEach(
+            rg.files({ cwd: dir, follow: false, hidden: true, signal: ctx.abort }).pipe(
+              Stream.filter((file) => !file.includes("SKILL.md")),
+              Stream.map((file) => path.resolve(dir, file)),
+              Stream.take(limit),
+            ),
+            (file) => Effect.sync(() => { fileList.push(file) }),
           )
+          const files = fileList.map((file) => `<file>${file}</file>`).join("\n")
 
           return {
             title: `Loaded skill: ${info.name}`,