fix(core): reduce Windows CPU under parallel sessions

packages/ai/src/protocols/anthropic-messages.ts

@@ -864,7 +864,7 @@ const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepRes
   return [{ ...state, lifecycle: Lifecycle.stepStart(state.lifecycle, events) }, [...events, result]]
 }
 
-const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(function* (
+const onContentBlockDelta = Effect.fnUntraced(function* (
   state: ParserState,
   event: AnthropicEvent,
 ) {

packages/core/src/bus.ts

@@ -67,6 +67,25 @@ const envelope = (aggregateID: string, seq: number, version: number) => ({
   version: Event.Version.make(version),
 })
 
+const encoders = new WeakMap<Event.Definition, (input: unknown) => unknown>()
+const decoders = new WeakMap<Event.Definition, (input: unknown) => unknown>()
+
+function encodeData(definition: Event.Definition, data: unknown) {
+  const cached = encoders.get(definition)
+  if (cached) return cached(data)
+  const encode = Schema.encodeUnknownSync(definition.data)
+  encoders.set(definition, encode)
+  return encode(data)
+}
+
+function decodeData(definition: Event.Definition, data: unknown) {
+  const cached = decoders.get(definition)
+  if (cached) return cached(data)
+  const decode = Schema.decodeUnknownSync(definition.data)
+  decoders.set(definition, decode)
+  return decode(data)
+}
+
 const decodeSerializedEvent = (event: SerializedEvent): Event.Payload => {
   const definition = Durable.get(event.type)
   if (!definition?.durable) {
@@ -77,7 +96,7 @@ const decodeSerializedEvent = (event: SerializedEvent): Event.Payload => {
     created: event.created ?? 0,
     type: definition.type,
     durable: envelope(event.aggregateID, event.seq, definition.durable.version),
-    data: Schema.decodeUnknownSync(definition.data)(event.data),
+    data: decodeData(definition, event.data),
   }
 }
 
@@ -260,10 +279,7 @@ export function configured(options?: Options) {
                               .get()
                               .pipe(Effect.orDie)
                             const latest = row?.seq ?? -1
-                            const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<
-                              string,
-                              unknown
-                            >
+                            const encoded = encodeData(definition, event.data) as Record<string, unknown>
                             if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
                               yield* Effect.die(
                                 new InvalidDurableEventError({
@@ -529,10 +545,7 @@ export function configured(options?: Options) {
                           const ids = new Set<Event.ID>()
                           for (const [index, item] of payloads.entries()) {
                             const seq = firstSeq + index
-                            const encoded = Schema.encodeUnknownSync(item.definition.data)(item.event.data) as Record<
-                              string,
-                              unknown
-                            >
+                            const encoded = encodeData(item.definition, item.event.data) as Record<string, unknown>
                             if (persist) {
                               if (ids.has(item.event.id))
                                 yield* Effect.die(
@@ -621,7 +634,7 @@ export function configured(options?: Options) {
                     id: event.id,
                     created: event.created ?? 0,
                     type: definition.type,
-                    data: Schema.decodeUnknownSync(definition.data)(event.data),
+                    data: decodeData(definition, event.data),
                   } as Event.Payload
                   const committed = yield* commitDurableEvent(definition, payload, {
                     seq: event.seq,

packages/core/src/filesystem/search.ts

@@ -26,16 +26,22 @@ const REFRESH_INTERVAL = Duration.toMillis("10 seconds")
 type Prepared = ReturnType<typeof fuzzysort.prepare>
 
 function emptyIndex() {
-  return { files: new Map<string, Prepared>(), directories: new Map<string, Prepared>() }
+  return {
+    files: new Map<string, Prepared>(),
+    directories: new Map<string, Prepared>(),
+    fileTargets: [] as Prepared[],
+    directoryTargets: [] as Prepared[],
+    combinedTargets: undefined as Prepared[] | undefined,
+  }
 }
 
 function search(index: ReturnType<typeof emptyIndex>, input: FileSystem.FindInput) {
   const items =
     input.type === "file"
-      ? Array.from(index.files.values())
+      ? index.fileTargets
       : input.type === "directory"
-        ? Array.from(index.directories.values())
-        : [...index.files.values(), ...index.directories.values()]
+        ? index.directoryTargets
+        : (index.combinedTargets ??= [...index.fileTargets, ...index.directoryTargets])
   const result = fuzzysort.go(input.query, items, { limit: input.limit ?? 50 })
   // Targets are owned by the current location index. The only global fuzzysort
   // state left is its query cache, which must not retain every query forever.
@@ -66,20 +72,28 @@ export const ripgrepLayer = Layer.effect(
       const next = emptyIndex()
       const previous = index
       if (!initialized) index = next
-      yield* ripgrep.find({
+      yield* ripgrep.scan({
         cwd: location.directory,
         pattern: "*",
         limit: location.vcs && !home ? Number.MAX_SAFE_INTEGER : 100_000,
         exclude: home ? [...Protected.names()].map((name) => `${name}/**`) : undefined,
         onEntry: (entry) =>
           Effect.sync(() => {
-            next.files.set(entry.path, previous.files.get(entry.path) ?? fuzzysort.prepare(entry.path))
+            const file = previous.files.get(entry.path) ?? fuzzysort.prepare(entry.path)
+            next.files.set(entry.path, file)
+            next.fileTargets.push(file)
+            next.combinedTargets = undefined
             const parts = entry.path.split("/")
-            parts.slice(0, -1).forEach((_, offset) => {
-              const directory = parts.slice(0, offset + 1).join("/") + path.sep
-              if (!next.directories.has(directory))
-                next.directories.set(directory, previous.directories.get(directory) ?? fuzzysort.prepare(directory))
-            })
+            let prefix = ""
+            for (const [offset, part] of parts.entries()) {
+              if (offset === parts.length - 1) break
+              prefix = prefix ? `${prefix}/${part}` : part
+              const directory = prefix + path.sep
+              if (next.directories.has(directory)) continue
+              const prepared = previous.directories.get(directory) ?? fuzzysort.prepare(directory)
+              next.directories.set(directory, prepared)
+              next.directoryTargets.push(prepared)
+            }
           }),
       })
       index = next

packages/core/src/git.ts

@@ -9,6 +9,10 @@ import { AppProcess } from "@opencode-ai/util/process"
 import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
 import { File } from "./file.js"
 import { KeyedMutex } from "./effect/keyed-mutex.js"
+import { which } from "./util/which.js"
+
+const resolvedGit = which("git")
+const gitExecutable = resolvedGit ? path.resolve(resolvedGit) : "git"
 
 export class Repository extends Schema.Class<Repository>("Git.Repository")({
   worktree: AbsolutePath,
@@ -314,10 +318,11 @@ const layer = Layer.effect(
     ) {
       const result = yield* proc
         .run(
-          ChildProcess.make("git", repositoryArgs(repository, args), {
+          ChildProcess.make(gitExecutable, repositoryArgs(repository, args), {
             cwd: repository.worktree,
             env: options?.env,
             extendEnv: true,
+            stdin: "ignore",
           }),
           { stdin: options?.stdin },
         )
@@ -422,17 +427,20 @@ const layer = Layer.effect(
       ignores?: Repository
       maximumUntrackedFileBytes?: number
     }) {
-      const list = (args: string[]) =>
-        repositoryOperation("refresh", input.repository, args).pipe(
-          Effect.map((result) => result.text.split("\0").filter(Boolean)),
-        )
-      const [tracked, untracked] = yield* Effect.all(
-        [
-          list(["diff-files", "--name-only", "-z", "--", input.scope]),
-          list(["ls-files", "--others", "--exclude-standard", "-z", "--", input.scope]),
-        ],
-        { concurrency: 2 },
-      )
+      const entries = (yield* repositoryOperation("refresh", input.repository, [
+        "ls-files",
+        "--modified",
+        "--others",
+        "--exclude-standard",
+        "-t",
+        "-z",
+        "--",
+        input.scope,
+      ])).text
+        .split("\0")
+        .filter(Boolean)
+      const tracked = entries.filter((entry) => !entry.startsWith("? ")).map((entry) => entry.slice(2))
+      const untracked = entries.filter((entry) => entry.startsWith("? ")).map((entry) => entry.slice(2))
       const candidates = Array.from(new Set([...tracked, ...untracked]))
       if (!candidates.length) return { skipped: [] }
       const ignored = input.ignores
@@ -444,11 +452,11 @@ const layer = Layer.effect(
               .filter(Boolean),
           )
         : new Set<string>()
-      const allowed = candidates.filter((item) => !ignored.has(item))
+      const allowed = new Set(candidates.filter((item) => !ignored.has(item)))
       const maximum = input.maximumUntrackedFileBytes
       const skipped = maximum
         ? (yield* Effect.forEach(
-            untracked.filter((item) => allowed.includes(item)),
+            untracked.filter((item) => allowed.has(item)),
             (item) =>
               fs.stat(path.join(input.repository.worktree, item)).pipe(
                 Effect.map((info) =>
@@ -459,7 +467,8 @@ const layer = Layer.effect(
             { concurrency: 8 },
           )).filter((item): item is RelativePath => item !== undefined)
         : []
-      const stage = allowed.filter((item) => !skipped.includes(RelativePath.make(item)))
+      const skippedSet = new Set(skipped)
+      const stage = Array.from(allowed).filter((item) => !skippedSet.has(RelativePath.make(item)))
       const remove = [...ignored, ...skipped]
       if (remove.length)
         yield* repositoryOperation(
@@ -485,10 +494,14 @@ const layer = Layer.effect(
       if (!input.paths.length) return new Set<RelativePath>()
       const result = yield* proc
         .run(
-          ChildProcess.make("git", repositoryArgs(input.repository, ["check-ignore", "--no-index", "--stdin", "-z"]), {
-            cwd: input.repository.worktree,
-            extendEnv: true,
-          }),
+          ChildProcess.make(
+            gitExecutable,
+            repositoryArgs(input.repository, ["check-ignore", "--no-index", "--stdin", "-z"]),
+            {
+              cwd: input.repository.worktree,
+              extendEnv: true,
+            },
+          ),
           { stdin: input.paths.join("\0") + "\0" },
         )
         .pipe(
@@ -662,7 +675,7 @@ const layer = Layer.effect(
       cwd = repository.worktree,
     ) {
       const result = yield* proc
-        .run(ChildProcess.make("git", args, { cwd, extendEnv: true, stdin: "ignore" }))
+        .run(ChildProcess.make(gitExecutable, args, { cwd, extendEnv: true, stdin: "ignore" }))
         .pipe(
           Effect.mapError(
             (cause) => new WorktreeError({ operation, directory: worktreeDirectory, message: cause.message, cause }),
@@ -759,7 +772,7 @@ function execute(cwd: string, proc: AppProcess.Interface) {
   return (args: string[]) =>
     proc
       .run(
-        ChildProcess.make("git", args, {
+        ChildProcess.make(gitExecutable, args, {
           cwd,
           extendEnv: true,
           stdin: "ignore",

packages/core/src/ripgrep.ts

@@ -60,6 +60,10 @@ export interface FindInput {
   readonly onEntry?: (entry: Entry) => Effect.Effect<void>
 }
 
+export interface ScanInput extends Omit<FindInput, "onEntry"> {
+  readonly onEntry: (entry: Entry) => Effect.Effect<void>
+}
+
 export interface GlobInput {
   readonly cwd: string
   readonly pattern: string
@@ -80,6 +84,7 @@ export interface GrepInput {
 
 export interface Interface {
   readonly find: (input: FindInput) => Effect.Effect<readonly Entry[], Error>
+  readonly scan: (input: ScanInput) => Effect.Effect<void, Error>
   readonly glob: (input: GlobInput) => Effect.Effect<readonly Entry[], Error>
   readonly grep: (input: GrepInput) => Effect.Effect<readonly Match[], Error | InvalidPatternError>
 }
@@ -105,6 +110,7 @@ const layer = Layer.effect(
       readonly parse: (line: string) => Effect.Effect<A | undefined, Error>
       readonly pattern?: string
       readonly onItem?: (item: A) => Effect.Effect<void>
+      readonly collect?: boolean
     }) => {
       const program = Effect.scoped(
         Effect.gen(function* () {
@@ -127,11 +133,17 @@ const layer = Layer.effect(
               return input.onItem(row)
             }),
             Stream.take(input.limit + 1),
-            Stream.runCollect,
-            Effect.map((chunk) => [...chunk]),
+            Stream.runFold(
+              () => ({ count: 0, items: [] as A[] }),
+              (result, row) => {
+                result.count++
+                if (input.collect !== false) result.items.push(row)
+                return result
+              },
+            ),
           )
-          const truncated = rows.length > input.limit
-          if (truncated) return { items: rows.slice(0, input.limit), truncated, partial: false }
+          const truncated = rows.count > input.limit
+          if (truncated) return { items: rows.items.slice(0, input.limit), truncated, partial: false }
 
           const code = yield* handle.exitCode
           const stderr = yield* Fiber.join(stderrFiber)
@@ -141,7 +153,7 @@ const layer = Layer.effect(
           if (code !== 0 && code !== 1 && code !== 2) {
             return yield* failure(stderr.trim() || `ripgrep failed with code ${code}`)
           }
-          return { items: code === 1 ? [] : rows, truncated: false, partial: code === 2 }
+          return { items: code === 1 ? [] : rows.items, truncated: false, partial: code === 2 }
         }),
       )
       const abortable = input.signal ? program.pipe(Effect.raceFirst(waitForAbort(input.signal))) : program
@@ -154,6 +166,40 @@ const layer = Layer.effect(
       )
     }
 
+    const find = (input: FindInput, collect = true) =>
+      run<Entry>({
+        cwd: input.cwd,
+        limit: input.limit,
+        signal: input.signal,
+        args: [
+          "--no-config",
+          "--files",
+          ...(input.hidden ? ["--hidden"] : []),
+          ...(input.follow ? ["--follow"] : []),
+          ...(input.pattern === "*" ? [] : [`--glob=${input.pattern}`]),
+          ...(input.exclude ?? []).map((pattern) => `--glob=!${pattern}`),
+          "--glob=!**/.git/**",
+          ".",
+        ],
+        parse: (line) => {
+          const relative = line
+            .replace(/^(?:\.[\\/])+/u, "")
+            .replace(/^[\\/]+/u, "")
+            .replaceAll("\\", "/")
+          return Effect.succeed(
+            Entry.make({
+              path: RelativePath.make(relative),
+              type: "file",
+            }),
+          )
+        },
+        onItem: input.onEntry,
+        collect,
+      }).pipe(
+        Effect.map((result) => result.items),
+        Effect.catchTag("Ripgrep.InvalidPatternError", (cause) => Effect.fail(failure(cause.message, cause))),
+      )
+
     return Service.of({
       glob: (input) =>
         run<string>({
@@ -187,38 +233,8 @@ const layer = Layer.effect(
           ),
           Effect.catchTag("Ripgrep.InvalidPatternError", (cause) => Effect.fail(failure(cause.message, cause))),
         ),
-      find: (input) =>
-        run<Entry>({
-          cwd: input.cwd,
-          limit: input.limit,
-          signal: input.signal,
-          args: [
-            "--no-config",
-            "--files",
-            ...(input.hidden ? ["--hidden"] : []),
-            ...(input.follow ? ["--follow"] : []),
-            ...(input.pattern === "*" ? [] : [`--glob=${input.pattern}`]),
-            ...(input.exclude ?? []).map((pattern) => `--glob=!${pattern}`),
-            "--glob=!**/.git/**",
-            ".",
-          ],
-          parse: (line) => {
-            const relative = line
-              .replace(/^(?:\.[\\/])+/u, "")
-              .replace(/^[\\/]+/u, "")
-              .replaceAll("\\", "/")
-            return Effect.succeed(
-              Entry.make({
-                path: RelativePath.make(relative),
-                type: "file",
-              }),
-            )
-          },
-          onItem: input.onEntry,
-        }).pipe(
-          Effect.map((result) => result.items),
-          Effect.catchTag("Ripgrep.InvalidPatternError", (cause) => Effect.fail(failure(cause.message, cause))),
-        ),
+      find,
+      scan: (input) => find(input, false).pipe(Effect.asVoid),
       grep: (input) =>
         run<RawMatchData>({
           ...input,