fix(cli): bound snapshot stalls during turns

.changeset/bound-snapshot-turns.md

@@ -0,0 +1,5 @@
+---
+"@kilocode/cli": patch
+---
+
+Keep turns responsive when snapshot infrastructure stalls and prevent transient snapshot progress from appearing in forked sessions.

packages/opencode/src/kilocode/question/index.ts

@@ -48,6 +48,17 @@ export namespace KiloQuestion {
         }
       })
 
+  /** Publishes the terminal event when a pending question effect is interrupted. */
+  export const finalize = <ID, Value>(input: {
+    pending: Map<ID, Value>
+    id: ID
+    publishRejected: () => Effect.Effect<void>
+  }) =>
+    Effect.gen(function* () {
+      if (!input.pending.delete(input.id)) return
+      yield* input.publishRejected()
+    })
+
   /**
    * Auto-dismiss when a newer prompt is already queued on this session — a
    * tool that calls `Question.ask` after the queue event would otherwise block

packages/opencode/src/kilocode/session/fork.ts

@@ -3,6 +3,7 @@ import { SessionID } from "@/session/schema"
 import { Database } from "@/storage/db"
 import { SyncEvent } from "@/sync"
 import { Effect } from "effect"
+import { KiloPartLifecycle } from "./part-lifecycle"
 
 const task = "task"
 const stale = /^[ \t]*task_id:[^\r\n]*(?:(?:\r?\n){1,2}|$)/m
@@ -17,6 +18,7 @@ export function writer(sessionID: SessionID, sync: SyncEvent.Interface) {
       return info
     },
     part(part: MessageV2.Part) {
+      if (KiloPartLifecycle.transient(part)) return
       items.push({ type: "part", part: structuredClone(detachPart(part)), time: Date.now() })
     },
     commit() {

packages/opencode/src/kilocode/session/part-lifecycle.ts

@@ -0,0 +1,9 @@
+import { MessageV2 } from "@/session/message-v2"
+
+export namespace KiloPartLifecycle {
+  export const key = "kilocode.lifecycle"
+
+  export function transient(part: MessageV2.Part) {
+    return part.type === "text" && part.metadata?.[key] === "transient"
+  }
+}

packages/opencode/src/kilocode/snapshot/track.ts

@@ -40,13 +40,14 @@
 // All of this is Kilo-specific — the upstream snapshot module remains a thin
 // shim that calls into here.
 
-import { Duration, Effect, Fiber } from "effect"
+import { Duration, Effect, Fiber, Option } from "effect"
 import { applyEdits, modify } from "jsonc-parser"
 import { AppFileSystem } from "@opencode-ai/core/filesystem"
 import { Question } from "@/question"
 import type { MessageID, PartID, SessionID } from "@/session/schema"
 import { PartID as PartIDSchema } from "@/session/schema"
 import type { MessageV2 } from "@/session/message-v2"
+import { KiloPartLifecycle } from "@/kilocode/session/part-lifecycle"
 import { KilocodeConfig } from "@/kilocode/config/config"
 import { ConfigParse } from "@/config/parse"
 import * as Log from "@opencode-ai/core/util/log"
@@ -83,14 +84,18 @@ type SessionRuntime = {
 export namespace KiloSnapshotTrack {
   const log = Log.create({ service: "snapshot.track" })
 
-  export const TIMEOUT_MS = iife(() => {
-    const raw = process.env["KILO_SNAPSHOT_TRACK_TIMEOUT_MS"]
-    if (raw) {
-      const parsed = Number(raw)
-      if (Number.isFinite(parsed) && parsed > 0) return parsed
-    }
-    return 10_000
-  })
+  const duration = (name: string, fallback: number) =>
+    iife(() => {
+      const raw = process.env[name]
+      if (raw) {
+        const parsed = Number(raw)
+        if (Number.isFinite(parsed) && parsed > 0) return parsed
+      }
+      return fallback
+    })
+
+  export const TIMEOUT_MS = duration("KILO_SNAPSHOT_TRACK_TIMEOUT_MS", 10_000)
+  export const TURN_TIMEOUT_MS = duration("KILO_SNAPSHOT_TURN_TIMEOUT_MS", 120_000)
 
   // Wire values — also function as i18n keys via `labelKey`/`headerKey`.
   // The backend matches replies on `label`, so the canonical English strings
@@ -140,6 +145,65 @@ export namespace KiloSnapshotTrack {
     asked: false,
   })
 
+  export const makeStates = () => {
+    const states = new Map<string, State>()
+    return (directory: string) => {
+      const found = states.get(directory)
+      if (found) return found
+      const state = makeState()
+      states.set(directory, state)
+      return state
+    }
+  }
+
+  export interface ProtectInput<A> {
+    readonly inner: Effect.Effect<A>
+    readonly state: State
+    readonly fallback: A
+    readonly operation: "track" | "patch"
+    readonly timeoutMs?: number
+  }
+
+  /**
+   * Enforces the turn-facing snapshot availability budget without waiting for
+   * cancellation. Snapshot tracking and patching are optional metadata work;
+   * once either exceeds this budget, later calls in the same directory bypass
+   * the potentially poisoned lock owner for the lifetime of this service.
+   */
+  export const protect = <A>(input: ProtectInput<A>): Effect.Effect<A> =>
+    Effect.gen(function* () {
+      if (input.state.disabledForSession) return input.fallback
+      const timeoutMs = input.timeoutMs ?? TURN_TIMEOUT_MS
+      return yield* Effect.acquireUseRelease(
+        Effect.forkDetach(input.inner, { startImmediately: true }),
+        (fiber) =>
+          Effect.gen(function* () {
+            const result = yield* Fiber.join(fiber).pipe(
+              Effect.timeoutOption(Duration.millis(timeoutMs)),
+              Effect.catchCause((cause) => {
+                input.state.disabledForSession = true
+                log.error("snapshot turn operation failed; bypassing snapshots for this directory", {
+                  cause,
+                  operation: input.operation,
+                })
+                return Effect.succeed(Option.some(input.fallback))
+              }),
+            )
+            if (Option.isSome(result)) return result.value
+            input.state.disabledForSession = true
+            log.warn("snapshot turn operation exceeded availability budget; bypassing snapshots for this directory", {
+              operation: input.operation,
+              timeoutMs,
+            })
+            return input.fallback
+          }),
+        (fiber) =>
+          Effect.sync(() => {
+            setTimeout(() => Effect.runFork(Fiber.interrupt(fiber)), 0)
+          }),
+      )
+    })
+
   /** Answer shape returned by `askUser`. Three-valued because dismiss !== disable. */
   export type Answer = "continue" | "disable" | "dismissed"
 
@@ -152,7 +216,7 @@ export namespace KiloSnapshotTrack {
    */
   export interface Hooks {
     /** Ask the user. Returns "dismissed" if the question is rejected. */
-    readonly ask: (input: { sessionID: SessionID }) => Promise<Answer>
+    readonly ask: (input: { sessionID: SessionID }, signal?: AbortSignal) => Promise<Answer>
     /** Persist `"snapshot": false` to the project config without disposing the instance. */
     readonly persistDisable: () => Promise<void>
     /** Publish the synthetic progress part allocated by the wrapper. */
@@ -383,7 +447,7 @@ export namespace KiloSnapshotTrack {
             input.state.owner = owner
 
             const sessionID = input.sessionID
-            const answer = yield* Effect.promise(() => hooks.ask({ sessionID }))
+            const answer = yield* Effect.promise((signal) => hooks.ask({ sessionID }, signal))
 
             if (answer === "continue") {
               log.info("user chose to keep waiting for snapshot; joining fiber")
@@ -443,7 +507,7 @@ export namespace KiloSnapshotTrack {
   }
 
   /** Build the synthetic progress part payload so both start/update share one shape. */
-  const progressPart = (input: {
+  export const progressPart = (input: {
     sessionID: SessionID
     messageID: MessageID
     partID: PartID
@@ -455,6 +519,7 @@ export namespace KiloSnapshotTrack {
     type: "text",
     text: input.text,
     synthetic: true,
+    metadata: { [KiloPartLifecycle.key]: "transient" },
   })
 
   export const defaultHooks: Hooks = {
@@ -503,46 +568,55 @@ export namespace KiloSnapshotTrack {
       )
     },
 
-    async ask(input) {
-      const answers = await questionRt
-        .runPromise((svc) =>
-          svc.ask({
-            sessionID: input.sessionID,
-            blocking: true,
-            questions: [
-              {
-                header: "Snapshot is slow",
-                headerKey: "snapshot.slowRepo.header",
-                question:
-                  "It is taking a long time to initialize the snapshot system, likely due to the size of the repository.\n\n" +
-                  "Do you want to disable Snapshots for this repository?",
-                questionKey: "snapshot.slowRepo.question",
-                custom: false,
-                options: [
-                  {
-                    label: ANSWER_CONTINUE,
-                    labelKey: "snapshot.slowRepo.answer.continue",
-                    description:
-                      "Keep waiting for the snapshot to complete. Subsequent turns are fast once the initial snapshot is built.",
-                    descriptionKey: "snapshot.slowRepo.answer.continue.description",
-                  },
-                  {
-                    label: ANSWER_DISABLE,
-                    labelKey: "snapshot.slowRepo.answer.disable",
-                    description:
-                      "Turn off Kilo's snapshots for this project. You will lose undo/redo of Kilo file changes, but git still tracks everything.",
-                    descriptionKey: "snapshot.slowRepo.answer.disable.description",
-                  },
-                ],
-              },
-            ],
-          }),
+    async ask(input, signal) {
+      return questionRt
+        .runPromise(
+          (svc) =>
+            svc.ask({
+              sessionID: input.sessionID,
+              blocking: true,
+              questions: [
+                {
+                  header: "Snapshot is slow",
+                  headerKey: "snapshot.slowRepo.header",
+                  question:
+                    "It is taking a long time to initialize the snapshot system, likely due to the size of the repository.\n\n" +
+                    "Do you want to disable Snapshots for this repository?",
+                  questionKey: "snapshot.slowRepo.question",
+                  custom: false,
+                  options: [
+                    {
+                      label: ANSWER_CONTINUE,
+                      labelKey: "snapshot.slowRepo.answer.continue",
+                      description:
+                        "Keep waiting for the snapshot to complete. Subsequent turns are fast once the initial snapshot is built.",
+                      descriptionKey: "snapshot.slowRepo.answer.continue.description",
+                    },
+                    {
+                      label: ANSWER_DISABLE,
+                      labelKey: "snapshot.slowRepo.answer.disable",
+                      description:
+                        "Turn off Kilo's snapshots for this project. You will lose undo/redo of Kilo file changes, but git still tracks everything.",
+                      descriptionKey: "snapshot.slowRepo.answer.disable.description",
+                    },
+                  ],
+                },
+              ],
+            }),
+          { signal },
         )
-        .catch(() => undefined)
-      const pick = answers?.[0]?.[0]
-      if (pick === ANSWER_CONTINUE) return "continue"
-      if (pick === ANSWER_DISABLE) return "disable"
-      return "dismissed"
+        .then((answers): Answer => {
+          const pick = answers[0]?.[0]
+          if (pick === ANSWER_CONTINUE) return "continue"
+          if (pick === ANSWER_DISABLE) return "disable"
+          return "dismissed"
+        })
+        .catch((err): Answer => {
+          if (!signal?.aborted && !(err instanceof Question.RejectedError)) {
+            log.warn("snapshot question failed; treating as dismissed", { err })
+          }
+          return "dismissed"
+        })
     },
 
     async persistDisable() {

packages/opencode/src/question/index.ts

@@ -210,9 +210,13 @@ export const layer = Layer.effect(
 
       return yield* Effect.ensuring(
         Deferred.await(deferred),
-        Effect.sync(() => {
-          pending.delete(id)
+        // kilocode_change start - every asked question gets a terminal event when its waiter is interrupted
+        KiloQuestion.finalize({
+          pending,
+          id,
+          publishRejected: () => bus.publish(Event.Rejected, { sessionID: info.sessionID, requestID: info.id }),
         }),
+        // kilocode_change end
       )
     })
 

packages/opencode/src/snapshot/index.ts

@@ -873,7 +873,7 @@ export const layer: Layer.Layer<Service, never, Requirements> =
       )
 
       // kilocode_change start - service-local state and cache avoid leaking across Snapshot layer instances
-      const trackState = KiloSnapshotTrack.makeState()
+      const trackState = KiloSnapshotTrack.makeStates()
       const cache = new Map<string, Promise<FileDiff[]>>()
       const max = 100
       // kilocode_change end
@@ -885,20 +885,34 @@ export const layer: Layer.Layer<Service, never, Requirements> =
         cleanup: Effect.fn("Snapshot.cleanup")(function* () {
           return yield* InstanceState.useEffect(state, (s) => s.cleanup())
         }),
-        // kilocode_change start - guard slow snapshots and surface progress to the active session
+        // kilocode_change start - isolate turn-facing snapshot work from poisoned locks
         track: Effect.fn("Snapshot.track")(function* (opts) {
-          return yield* KiloSnapshotTrack.wrap({
-            inner: InstanceState.useEffect(state, (s) => s.track(opts)),
-            state: trackState,
-            snapshotInitialization: opts?.snapshotInitialization,
-            sessionID: opts?.sessionID,
-            messageID: opts?.messageID,
+          const ctx = yield* InstanceState.context
+          const guard = trackState(ctx.worktree)
+          return yield* KiloSnapshotTrack.protect({
+            inner: KiloSnapshotTrack.wrap({
+              inner: InstanceState.useEffect(state, (s) => s.track(opts)),
+              state: guard,
+              snapshotInitialization: opts?.snapshotInitialization,
+              sessionID: opts?.sessionID,
+              messageID: opts?.messageID,
+            }),
+            state: guard,
+            fallback: undefined,
+            operation: "track",
+          })
+        }),
+        patch: Effect.fn("Snapshot.patch")(function* (hash: string) {
+          const ctx = yield* InstanceState.context
+          const guard = trackState(ctx.worktree)
+          return yield* KiloSnapshotTrack.protect({
+            inner: InstanceState.useEffect(state, (s) => s.patch(hash)),
+            state: guard,
+            fallback: { hash, files: [] },
+            operation: "patch",
           })
         }),
         // kilocode_change end
-        patch: Effect.fn("Snapshot.patch")(function* (hash: string) {
-          return yield* InstanceState.useEffect(state, (s) => s.patch(hash))
-        }),
         restore: Effect.fn("Snapshot.restore")(function* (snapshot: string) {
           return yield* InstanceState.useEffect(state, (s) => s.restore(snapshot))
         }),

packages/opencode/test/kilocode/question-cancel.test.ts

@@ -0,0 +1,59 @@
+import { afterEach, expect } from "bun:test"
+import { Effect, Fiber, Layer, Queue } from "effect"
+import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { Question } from "../../src/question"
+import { Bus } from "../../src/bus"
+import { QuestionID } from "../../src/question/schema"
+import { SessionID } from "../../src/session/schema"
+import { disposeAllInstances } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+const it = testEffect(
+  Layer.mergeAll(Question.layer.pipe(Layer.provideMerge(Bus.layer)), CrossSpawnSpawner.defaultLayer),
+)
+
+afterEach(async () => {
+  await disposeAllInstances()
+})
+
+it.instance(
+  "publishes rejection when a pending question is interrupted",
+  () =>
+    Effect.gen(function* () {
+      const question = yield* Question.Service
+      const bus = yield* Bus.Service
+      const asked = yield* Queue.unbounded<{ properties: Question.Request }>()
+      const rejected = yield* Queue.unbounded<{
+        properties: { sessionID: SessionID; requestID: QuestionID }
+      }>()
+      const offAsked = yield* bus.subscribeCallback(Question.Event.Asked, (event) => Queue.offerUnsafe(asked, event))
+      const offRejected = yield* bus.subscribeCallback(Question.Event.Rejected, (event) =>
+        Queue.offerUnsafe(rejected, event),
+      )
+      yield* Effect.addFinalizer(() => Effect.sync(() => [offAsked(), offRejected()]))
+
+      const fiber = yield* question
+        .ask({
+          sessionID: SessionID.make("ses_test"),
+          questions: [
+            {
+              header: "Snapshot",
+              question: "Keep waiting?",
+              options: [{ label: "Continue", description: "Keep waiting" }],
+            },
+          ],
+        })
+        .pipe(Effect.forkChild)
+      const request = yield* Queue.take(asked).pipe(Effect.timeout("2 seconds"))
+
+      yield* Fiber.interrupt(fiber)
+
+      const event = yield* Queue.take(rejected).pipe(Effect.timeout("2 seconds"))
+      expect(event.properties).toEqual({
+        sessionID: request.properties.sessionID,
+        requestID: request.properties.id,
+      })
+      expect(yield* question.list()).toEqual([])
+    }),
+  { git: true },
+)

packages/opencode/test/kilocode/session-fork-remap.test.ts

@@ -11,6 +11,7 @@ import { disposeAllInstances, tmpdir } from "../fixture/fixture"
 import { Database, eq } from "../../src/storage/db"
 import { EventSequenceTable, EventTable } from "../../src/sync/event.sql"
 import { Flag } from "@opencode-ai/core/flag/flag"
+import { KiloPartLifecycle } from "../../src/kilocode/session/part-lifecycle"
 
 Log.init({ print: false })
 
@@ -381,4 +382,48 @@ describe("Session.fork task detachment", () => {
     },
     { timeout: 30000 },
   )
+
+  test(
+    "drops transient UI parts while preserving durable synthetic context",
+    async () => {
+      await using tmp = await tmpdir({ git: true })
+      await provideTestInstance({
+        directory: tmp.path,
+        fn: async () => {
+          const parent = await sessions.create({ title: "parent" })
+          const user = await userMsg(parent.id)
+          const parts = [
+            { text: "Initializing snapshot... but durable", synthetic: true },
+            { text: "<system-reminder>durable context</system-reminder>", synthetic: true },
+            {
+              text: "arbitrary live status",
+              synthetic: true,
+              metadata: { [KiloPartLifecycle.key]: "transient" },
+            },
+          ]
+          for (const part of parts) {
+            await sessions.updatePart({
+              id: PartID.ascending(),
+              messageID: user,
+              sessionID: parent.id,
+              type: "text",
+              ...part,
+            } as MessageV2.TextPart)
+          }
+
+          const forked = await Session.fork({ sessionID: parent.id })
+          const source = await sessions.messages({ sessionID: parent.id })
+          const copy = await sessions.messages({ sessionID: forked.id })
+          const texts = copy.flatMap((msg) => msg.parts).flatMap((part) => (part.type === "text" ? [part.text] : []))
+
+          expect(source.flatMap((msg) => msg.parts)).toHaveLength(3)
+          expect(texts).toEqual([
+            "Initializing snapshot... but durable",
+            "<system-reminder>durable context</system-reminder>",
+          ])
+        },
+      })
+    },
+    { timeout: 30000 },
+  )
 })

packages/opencode/test/kilocode/snapshot-track-timeout.test.ts

@@ -5,10 +5,12 @@
 // touch the real Question module or write to the filesystem.
 
 import { describe, expect, test } from "bun:test"
-import { Duration, Effect } from "effect"
-import type { MessageID, SessionID } from "../../src/session/schema"
+import { Deferred, Duration, Effect, Fiber } from "effect"
+import * as TestClock from "effect/testing/TestClock"
+import { PartID, type MessageID, type SessionID } from "../../src/session/schema"
 import { KiloSnapshotTrack } from "../../src/kilocode/snapshot/track"
-import { awaitWithTimeout } from "../lib/effect"
+import { KiloPartLifecycle } from "../../src/kilocode/session/part-lifecycle"
+import { awaitWithTimeout, it } from "../lib/effect"
 
 const SESSION = "ses_test" as SessionID
 const MESSAGE = "msg_test" as MessageID
@@ -64,6 +66,81 @@ const makeHooks = (
   return { hooks, calls }
 }
 
+describe("KiloSnapshotTrack.protect", () => {
+  it.effect("returns at the availability deadline without waiting for cancellation", () =>
+    Effect.gen(function* () {
+      const state = KiloSnapshotTrack.makeState()
+      const started = yield* Deferred.make<void>()
+      const fallback = { hash: "base", files: [] as string[] }
+      let finalized = false
+      const inner = Deferred.succeed(started, undefined).pipe(
+        Effect.andThen(Effect.never),
+        Effect.ensuring(
+          Effect.sync(() => {
+            finalized = true
+          }),
+        ),
+      )
+      const fiber = yield* KiloSnapshotTrack.protect({
+        inner,
+        state,
+        fallback,
+        operation: "patch",
+        timeoutMs: 100,
+      }).pipe(Effect.forkChild)
+
+      yield* Deferred.await(started)
+      yield* TestClock.adjust(100)
+
+      expect(yield* Fiber.join(fiber)).toEqual(fallback)
+      expect(finalized).toBe(false)
+      expect(state.disabledForSession).toBe(true)
+    }),
+  )
+
+  it.effect("bypasses later operations after a deadline opens the directory circuit", () =>
+    Effect.gen(function* () {
+      const state = KiloSnapshotTrack.makeState()
+      const started = yield* Deferred.make<void>()
+      let calls = 0
+      const first = yield* KiloSnapshotTrack.protect({
+        inner: Effect.sync(() => {
+          calls += 1
+        }).pipe(Effect.andThen(Deferred.succeed(started, undefined)), Effect.andThen(Effect.never)),
+        state,
+        fallback: undefined,
+        operation: "track",
+        timeoutMs: 100,
+      }).pipe(Effect.forkChild)
+
+      yield* Deferred.await(started)
+      yield* TestClock.adjust(100)
+      expect(yield* Fiber.join(first)).toBeUndefined()
+
+      const second = yield* KiloSnapshotTrack.protect({
+        inner: Effect.sync(() => {
+          calls += 1
+          return "unexpected"
+        }),
+        state,
+        fallback: undefined,
+        operation: "track",
+      })
+      expect(second).toBeUndefined()
+      expect(calls).toBe(1)
+    }),
+  )
+
+  test("keeps circuit state isolated by directory", () => {
+    const states = KiloSnapshotTrack.makeStates()
+    const first = states("/repo/a")
+    first.disabledForSession = true
+
+    expect(states("/repo/a")).toBe(first)
+    expect(states("/repo/b").disabledForSession).toBe(false)
+  })
+})
+
 describe("KiloSnapshotTrack.wrap", () => {
   test("returns the hash when inner resolves before the timeout", async () => {
     const state = KiloSnapshotTrack.makeState()
@@ -483,6 +560,18 @@ describe("KiloSnapshotTrack.wrap", () => {
 })
 
 describe("KiloSnapshotTrack progress indicator", () => {
+  test("classifies persisted progress as transient", () => {
+    const part = KiloSnapshotTrack.progressPart({
+      sessionID: SESSION,
+      messageID: MESSAGE,
+      partID: PartID.make("prt_test"),
+      text: "arbitrary status",
+    })
+
+    expect(part.synthetic).toBe(true)
+    expect(KiloPartLifecycle.transient(part)).toBe(true)
+  })
+
   // Strip the braille spinner frame (first Unicode codepoint, plus the
   // trailing space) so tests can assert on the stable descriptive text
   // without caring which animation frame landed.