fix(snapshot): recover index contention and early stdin closure

packages/core/src/cross-spawn-spawner.ts

@@ -228,12 +228,42 @@ export const make = Effect.gen(function* () {
     Effect.suspend(() => {
       let sink: Sink.Sink<void, unknown, never, PlatformError.PlatformError> = Sink.drain
       if (Predicate.isNotNull(proc.stdin)) {
+        const writable = proc.stdin
+        const failed = Deferred.makeUnsafe<never, PlatformError.PlatformError>()
+        const onError = (err: Error) =>
+          Deferred.doneUnsafe(failed, Effect.fail(toPlatformError("fromWritable(stdin)", err, command)))
+        // NodeSink's input listener ends before writable.end(); keep late pipe errors owned until close.
+        writable.on("error", onError)
+        writable.once("close", () => writable.off("error", onError))
         sink = NodeSink.fromWritable({
-          evaluate: () => proc.stdin!,
+          evaluate: () => writable,
           onError: (err) => toPlatformError("fromWritable(stdin)", toError(err), command),
-          endOnDone: cfg.endOnDone,
+          endOnDone: false,
           encoding: cfg.encoding,
         })
+        if (cfg.endOnDone) {
+          sink = sink.pipe(
+            Sink.mapEffect(() =>
+              Effect.callback<void, PlatformError.PlatformError>((resume) => {
+                if (writable.errored)
+                  return resume(Effect.fail(toPlatformError("end(stdin)", writable.errored, command)))
+                if (writable.writableFinished || writable.closed) return resume(Effect.void)
+                const finish = () => resume(Effect.void)
+                const close = () =>
+                  resume(
+                    Effect.fail(toPlatformError("end(stdin)", new Error("stdin closed before finishing"), command)),
+                  )
+                writable.once("finish", finish)
+                writable.once("close", close)
+                writable.end()
+                return Effect.sync(() => {
+                  writable.off("finish", finish)
+                  writable.off("close", close)
+                })
+              }).pipe(Effect.raceFirst(Deferred.await(failed))),
+            ),
+          )
+        }
       }
       if (Stream.isStream(cfg.stream)) return Effect.as(Effect.forkScoped(Stream.run(cfg.stream, sink)), sink)
       return Effect.succeed(sink)

packages/core/test/process/process.test.ts

@@ -292,6 +292,89 @@ describe("AppProcess", () => {
     )
   })
 
+  describe("stdin lifecycle", () => {
+    it.live(
+      "preserves early child exit diagnostics when a small stdin pipe closes",
+      Effect.gen(function* () {
+        const svc = yield* AppProcess.Service
+        const command =
+          process.platform === "win32"
+            ? ChildProcess.make("cmd.exe", ["/d", "/s", "/c", "echo rejected 1>&2 & exit /b 128"])
+            : ChildProcess.make("/bin/sh", ["-c", "printf rejected >&2; exit 128"])
+        for (let attempt = 0; attempt < 50; attempt++) {
+          const result = yield* svc.run(command, { stdin: "x" })
+          expect(result.exitCode).toBe(128)
+          expect(result.stderr.toString("utf8").trim()).toBe("rejected")
+        }
+      }),
+    )
+
+    it.live(
+      "keeps caller-managed stdin open across multiple writes",
+      Effect.gen(function* () {
+        const svc = yield* AppProcess.Service
+        const handle = yield* svc.spawn(
+          ChildProcess.make(
+            NODE,
+            [
+              "-e",
+              `
+          let text = ""
+          process.stdin.on("data", (chunk) => {
+            text += chunk
+            if (text.length === 2) {
+              process.stdout.write(text)
+              process.exit(0)
+            }
+          })
+          process.stdin.on("end", () => process.exit(1))
+        `,
+            ],
+            { stdin: { stream: "pipe", endOnDone: false } },
+          ),
+        )
+        yield* Stream.run(Stream.make(Buffer.from("a")), handle.stdin)
+        yield* Stream.run(Stream.make(Buffer.from("b")), handle.stdin)
+        const out = yield* Stream.runCollect(handle.stdout)
+        expect(Buffer.concat(Array.from(out)).toString()).toBe("ab")
+        expect(Number(yield* handle.exitCode)).toBe(0)
+      }),
+    )
+
+    it.live(
+      "reports a closed child stdin as a typed write failure without losing its diagnostics",
+      Effect.gen(function* () {
+        const svc = yield* AppProcess.Service
+        const handle = yield* svc.spawn(cmd("-e", "process.stderr.write('rejected'); process.exit(128)"))
+        const exit = yield* Stream.run(Stream.make(Buffer.alloc(1024 * 1024)), handle.stdin).pipe(Effect.exit)
+        expect(Exit.isFailure(exit)).toBe(true)
+        if (Exit.isFailure(exit)) {
+          expect(exit.cause.reasons[0]).toMatchObject({
+            _tag: "Fail",
+            error: { _tag: "PlatformError", reason: { module: "ChildProcess", method: "fromWritable(stdin)" } },
+          })
+        }
+        const stderr = yield* Stream.runCollect(handle.stderr)
+        expect(Buffer.concat(Array.from(stderr)).toString()).toBe("rejected")
+        expect(Number(yield* handle.exitCode)).toBe(128)
+      }),
+    )
+
+    it.live(
+      "preserves failures from the stdin producer",
+      Effect.gen(function* () {
+        const svc = yield* AppProcess.Service
+        const handle = yield* svc.spawn(cmd("-e", "process.stdin.resume()"))
+        const error = new Error("stdin producer failed")
+        const exit = yield* Stream.run(Stream.fail(error), handle.stdin).pipe(Effect.exit)
+        expect(Exit.isFailure(exit)).toBe(true)
+        if (Exit.isFailure(exit)) {
+          expect(exit.cause.reasons[0]).toMatchObject({ _tag: "Fail", error })
+        }
+      }),
+    )
+  })
+
   describe("runStream", () => {
     it.live(
       "emits lines incrementally and ends cleanly on exit 0",

packages/opencode/src/snapshot/index.ts

@@ -80,15 +80,31 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
 
         const git = Effect.fnUntraced(
           function* (cmd: string[], opts?: { cwd?: string; env?: Record<string, string>; stdin?: string }) {
-            const result = yield* appProcess.run(
-              ChildProcess.make("git", cmd, { cwd: opts?.cwd, env: opts?.env, extendEnv: true }),
-              { stdin: opts?.stdin },
-            )
-            return {
-              code: ChildProcessSpawner.ExitCode(result.exitCode),
-              text: result.stdout.toString("utf8"),
-              stderr: result.stderr.toString("utf8"),
-            } satisfies GitResult
+            for (let attempt = 0; ; attempt++) {
+              const result = yield* appProcess.run(
+                ChildProcess.make("git", cmd, {
+                  cwd: opts?.cwd,
+                  env: { ...opts?.env, LC_ALL: "C" },
+                  extendEnv: true,
+                }),
+                { stdin: opts?.stdin },
+              )
+              const stderr = result.stderr.toString("utf8")
+              // Other OpenCode processes share this index; their locks must be released by their owners.
+              if (
+                result.exitCode !== 128 ||
+                !stderr.includes("index.lock") ||
+                !stderr.includes("File exists") ||
+                attempt === 6
+              ) {
+                return {
+                  code: ChildProcessSpawner.ExitCode(result.exitCode),
+                  text: result.stdout.toString("utf8"),
+                  stderr,
+                } satisfies GitResult
+              }
+              yield* Effect.sleep(Duration.millis(50 * 2 ** attempt))
+            }
           },
           Effect.catch((err) =>
             Effect.succeed({
@@ -144,19 +160,20 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
         })
 
         const stage = Effect.fnUntraced(function* (files: string[]) {
-          if (!files.length) return
+          if (!files.length) return true
           const result = yield* git(
             [...cfg, ...args(["add", "--all", "--sparse", "--pathspec-from-file=-", "--pathspec-file-nul"])],
             {
               cwd: state.worktree,
               stdin: encodeTopLevelLiteralPathspecs(files),
             },
           )
-          if (result.code === 0) return
+          if (result.code === 0) return true
           yield* Effect.logWarning("failed to add snapshot files", {
             exitCode: result.code,
             stderr: result.stderr,
           })
+          return false
         })
 
         const exists = (file: string) => fs.exists(file).pipe(Effect.orDie)
@@ -252,13 +269,13 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
               otherCode: other.code,
               otherStderr: other.stderr,
             })
-            return
+            return false
           }
 
           const tracked = diff.text.split("\0").filter(Boolean)
           const untracked = other.text.split("\0").filter(Boolean)
           const all = Array.from(new Set([...tracked, ...untracked]))
-          if (!all.length) return
+          if (!all.length) return true
 
           // Resolve source-repo ignore rules against the exact candidate set.
           // --no-index keeps this pattern-based even when a path is already tracked.
@@ -272,7 +289,7 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
           }
 
           const allow = all.filter((item) => !ignored.has(item))
-          if (!allow.length) return
+          if (!allow.length) return true
 
           const large = new Set(
             (yield* Effect.all(
@@ -294,7 +311,7 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
           const block = new Set(untracked.filter((item) => large.has(item)))
           yield* sync(Array.from(block))
           // Stage only the allowed candidate paths so snapshot updates stay scoped.
-          yield* stage(allow.filter((item) => !block.has(item)))
+          return yield* stage(allow.filter((item) => !block.has(item)))
         })
 
         const cleanup = Effect.fnUntraced(function* () {
@@ -337,8 +354,15 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
                 yield* seed()
                 yield* Effect.logInfo("initialized")
               }
-              yield* add()
+              if (!(yield* add())) return
               const result = yield* git(args(["write-tree"]), { cwd: state.directory })
+              if (result.code !== 0) {
+                yield* Effect.logWarning("failed to write snapshot tree", {
+                  exitCode: result.code,
+                  stderr: result.stderr,
+                })
+                return
+              }
               const hash = result.text.trim()
               yield* Effect.logInfo("tracking", { hash, cwd: state.directory, git: state.gitdir })
               return hash

packages/opencode/test/snapshot/snapshot.test.ts

@@ -3,6 +3,9 @@ import { $ } from "bun"
 import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { FSUtil } from "@opencode-ai/core/fs-util"
+import { Global } from "@opencode-ai/core/global"
+import { Hash } from "@opencode-ai/core/util/hash"
+import { InstanceState } from "@/effect/instance-state"
 import fs from "fs/promises"
 import path from "path"
 import { Effect, Fiber, Layer } from "effect"
@@ -1216,3 +1219,99 @@ it.instance(
   }),
   { git: true },
 )
+
+const holdIndex = Effect.fn("SnapshotTest.holdIndex")(function* (gitdir: string, cwd: string) {
+  const holder = yield* Effect.acquireRelease(
+    Effect.sync(() =>
+      Bun.spawn(["git", "--git-dir", gitdir, "update-index", "--index-info"], {
+        cwd,
+        stdin: "pipe",
+        stdout: "ignore",
+        stderr: "pipe",
+      }),
+    ),
+    (proc) =>
+      Effect.promise(async () => {
+        if (proc.exitCode === null) proc.stdin.end()
+        await proc.exited
+      }),
+  )
+  yield* Effect.promise(async () => {
+    const deadline = Date.now() + 2000
+    while (!(await Bun.file(path.join(gitdir, "index.lock")).exists())) {
+      if (Date.now() > deadline) throw new Error("git did not acquire the snapshot index lock")
+      await Bun.sleep(10)
+    }
+  })
+  return holder
+})
+
+it.instance(
+  "track retries an index lock held by another git process without losing new files",
+  Effect.gen(function* () {
+    const tmp = yield* bootstrap()
+    const snapshot = yield* Snapshot.Service
+    const before = yield* snapshot.track()
+    const ctx = yield* InstanceState.context
+    const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree))
+    yield* write(`${tmp.path}/contended.txt`, "captured after contention")
+
+    const holder = yield* holdIndex(gitdir, tmp.path)
+    const pending = yield* snapshot.track().pipe(Effect.forkChild)
+    yield* Effect.sleep("250 millis")
+    holder.stdin.end()
+    expect(yield* Effect.promise(() => holder.exited)).toBe(0)
+    const after = yield* Fiber.join(pending)
+    expect(after).toMatch(/^[0-9a-f]{40}$/)
+    expect(after).not.toBe(before)
+    const proc = Bun.spawn(["git", "--git-dir", gitdir, "show", `${after}:contended.txt`], {
+      stdout: "pipe",
+      stderr: "pipe",
+    })
+    expect(yield* Effect.promise(() => new Response(proc.stdout).text())).toBe("captured after contention")
+    expect(yield* Effect.promise(() => proc.exited)).toBe(0)
+  }),
+  { git: true },
+)
+
+it.instance(
+  "track stops retrying a persistent index collision and leaves the other owner alive",
+  Effect.gen(function* () {
+    const tmp = yield* bootstrap()
+    const snapshot = yield* Snapshot.Service
+    yield* snapshot.track()
+    const ctx = yield* InstanceState.context
+    const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree))
+    yield* write(`${tmp.path}/contended.txt`, "not snapshotted while the index is locked")
+    const holder = yield* holdIndex(gitdir, tmp.path)
+    expect(yield* snapshot.track().pipe(Effect.timeout("12 seconds"))).toBeUndefined()
+    expect(yield* exists(path.join(gitdir, "index.lock"))).toBe(true)
+    expect(holder.exitCode).toBeNull()
+  }),
+  { git: true },
+)
+
+it.instance(
+  "track does not return a stale tree or retry an unrelated fatal git staging error",
+  Effect.gen(function* () {
+    const tmp = yield* bootstrap()
+    const snapshot = yield* Snapshot.Service
+    yield* snapshot.track()
+    const ctx = yield* InstanceState.context
+    const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree))
+    yield* exec(tmp.path, [
+      "git",
+      "--git-dir",
+      gitdir,
+      "config",
+      "filter.reject.clean",
+      "printf x >> filter-count; false",
+    ])
+    yield* exec(tmp.path, ["git", "--git-dir", gitdir, "config", "filter.reject.required", "true"])
+    yield* write(`${tmp.path}/.gitattributes`, "*.fail filter=reject\n")
+    yield* write(`${tmp.path}/new.fail`, "must not be claimed as captured")
+    expect(yield* snapshot.track()).toBeUndefined()
+    expect(yield* readText(`${tmp.path}/filter-count`)).toBe("x")
+  }),
+  { git: true },
+)