refactor(cli): narrow snapshot-diff freeze fix to caps-only

Revert the worker pool, dispatcher, flags, and related scaffolding added
on top of the initial cap guard. Post-#9046, the only remaining CLI
callers of Vcs.diff are one-shot review opens, so the worker offload is
no longer needed. The freeze repro is eliminated by the input caps in
DiffEngine.shouldSkip alone.

Shrinks the fork diff from 28 files (+1063) to 6 files (+199) and
reduces shared opencode files with kilocode_change markers to 2.
This commit is contained in:
Alex Alecu
2026-04-20 14:37:19 +03:00
parent 0f932abe0f
commit a7e06d6605
14 changed files with 55 additions and 822 deletions
+1 -1
View File
@@ -2,4 +2,4 @@
"kilo-code": patch
---
Prevent the CLI from freezing when a changed file has tens of thousands of lines. Large-file diffs now skip the full patch body but still report additions/deletions, and the diff computation no longer blocks ESC/interrupt or live updates.
Prevent TUI freeze on turns that edit very large files by skipping patch generation for oversized file diffs. Additions and deletions are still reported so session summaries stay accurate.
-5
View File
@@ -195,7 +195,6 @@ for (const item of targets) {
const rootPath = path.resolve(dir, "../../node_modules/@opentui/core/parser.worker.js")
const parserWorker = fs.realpathSync(fs.existsSync(localPath) ? localPath : rootPath)
const workerPath = "./src/cli/cmd/tui/worker.ts"
const diffWorkerPath = "./src/kilocode/snapshot/diff-worker.ts" // kilocode_change
const rgPath = "./src/file/ripgrep.worker.ts"
// Use platform-specific bunfs root path based on target OS
@@ -223,22 +222,18 @@ for (const item of targets) {
files: {
...(embeddedFileMap ? { "opencode-web-ui.gen.ts": embeddedFileMap } : {}),
},
// kilocode_change start — reflowed to keep the diff-worker entrypoint readable
entrypoints: [
"./src/index.ts",
parserWorker,
workerPath,
diffWorkerPath,
rgPath,
...(embeddedFileMap ? ["opencode-web-ui.gen.ts"] : []),
],
// kilocode_change end
define: {
KILO_VERSION: `'${Script.version}'`,
KILO_MIGRATIONS: JSON.stringify(migrations),
OTUI_TREE_SITTER_WORKER_PATH: bunfsRoot + workerRelativePath,
KILO_WORKER_PATH: workerPath,
KILO_DIFF_WORKER_PATH: diffWorkerPath, // kilocode_change
KILO_RIPGREP_WORKER_PATH: rgPath,
KILO_CHANNEL: `'${Script.channel}'`,
KILO_LIBC: item.os === "linux" ? `'${item.abi ?? "glibc"}'` : "",
-5
View File
@@ -79,11 +79,6 @@ export namespace Flag {
export const KILO_SKIP_MIGRATIONS = truthy("KILO_SKIP_MIGRATIONS")
export const KILO_STRICT_CONFIG_DEPS = truthy("KILO_STRICT_CONFIG_DEPS")
// kilocode_change start — diff engine rollback knobs
export const KILO_DIFF_WORKER = !falsy("KILO_DIFF_WORKER")
export const KILO_DIFF_CAPS = !falsy("KILO_DIFF_CAPS")
// kilocode_change end
function number(key: string) {
const value = process.env[key]
if (!value) return undefined
@@ -1,117 +0,0 @@
// kilocode_change - new file
//
// Tracked dispatcher that wraps the fire-and-forget `SessionSummary.summarize`
// call with a timeout and a cancel hook. Lives here (not inside the shared
// `src/session/summary.ts`) to keep our diff against upstream tiny.
//
// Behaviour summary:
// - Each summary runs under a fresh AbortController tracked in `inflight`.
// - A previous in-flight summary for the same sessionID is aborted when a
// new one starts, preventing pileup across rapid turns.
// - A wall-clock timeout (default 10s) races against the Effect fiber so we
// always unwind within a bounded time even on pathological input.
// - `cancel(sessionID)` fires `ac.abort()`, which is bridged into
// `Effect.interrupt` via a third racer inside `Effect.raceAllFirst`. That
// is what makes the ESC path (SessionPrompt.cancel) actually stop the
// running fiber rather than just dropping the map entry.
//
// The host module owns the Effect runtime (`runPromise`) and hands us a
// reference to `svc.summarize` via the `summarize` callback.
import { Cause, Effect } from "effect"
import { MessageID, SessionID } from "@/session/schema"
import { Log } from "@/util/log"
export namespace SummaryDispatch {
const log = Log.create({ service: "session.summary" })
export const SUMMARY_TIMEOUT_MS = 10_000
export interface Input {
sessionID: SessionID
messageID: MessageID
}
export type RunPromise<S> = <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, any>) => Promise<A>
export type Summarize<S> = (svc: S, input: Input) => Effect.Effect<void>
type Tagged = AbortController & { reason?: "superseded" | "cancel" }
export function create<S>(opts: { runPromise: RunPromise<S>; summarize: Summarize<S> }) {
const inflight = new Map<SessionID, Tagged>()
const summarize = (input: Input) => {
const prev = inflight.get(input.sessionID)
if (prev) {
prev.reason = "superseded"
prev.abort()
}
const ac: Tagged = new AbortController()
inflight.set(input.sessionID, ac)
const started = Date.now()
void opts
.runPromise((svc) =>
// raceAllFirst with three racers: the real work, a wall-clock
// timeout, and an AbortSignal bridge. Whichever FINISHES FIRST wins
// (success, failure, or interrupt — `raceAll` would keep waiting on
// interrupts), and the other two are interrupted by the Effect
// runtime. The signal-bridge racer is what makes `cancel(sessionID)`
// actually stop the fiber instead of just removing the map entry.
Effect.raceAllFirst([
opts.summarize(svc, input),
Effect.sleep(`${SUMMARY_TIMEOUT_MS} millis`).pipe(Effect.andThen(Effect.interrupt)),
Effect.callback<void>((resume) => {
if (ac.signal.aborted) {
resume(Effect.interrupt)
return
}
const onAbort = () => resume(Effect.interrupt)
ac.signal.addEventListener("abort", onAbort, { once: true })
return Effect.sync(() => ac.signal.removeEventListener("abort", onAbort))
}),
]).pipe(
Effect.onInterrupt(() =>
Effect.sync(() => {
const elapsed = Date.now() - started
if (ac.reason === "superseded") {
log.info("summary superseded", { sessionID: input.sessionID, elapsed })
return
}
log.warn("summary interrupted", { sessionID: input.sessionID, elapsed })
}),
),
Effect.ensuring(
Effect.sync(() => {
// Drop the entry only if it is still ours — a newer summarize
// for the same session may have replaced it already.
if (inflight.get(input.sessionID) === ac) inflight.delete(input.sessionID)
}),
),
Effect.catchCause((cause) =>
Effect.sync(() => {
if (Cause.hasInterrupts(cause)) return
log.warn("summary failed", { sessionID: input.sessionID, cause: String(cause) })
}),
),
),
)
.catch(() => {})
}
async function cancel(sessionID: SessionID) {
const ac = inflight.get(sessionID)
if (!ac) return
ac.reason = "cancel"
ac.abort()
inflight.delete(sessionID)
log.info("summary cancelled", { sessionID })
}
return {
summarize,
cancel,
}
}
}
@@ -1,59 +1,24 @@
// kilocode_change - new file
//
// Single entry point for every call to `structuredPatch`/`formatPatch` in the
// codebase. The underlying npm `diff` package uses Myers' algorithm with full
// context, which is O(N*M) in time and memory. On files with tens of thousands
// of lines it can block the thread for minutes. That is what caused the TUI
// freeze where ESC no longer worked after a turn.
// Input-size caps around `structuredPatch`/`formatPatch`. The npm `diff`
// package uses Myers' algorithm with full context, which is O(N*M) in time
// and memory. On files with tens of thousands of lines it can block the
// thread for minutes. That is what caused the TUI freeze where ESC no
// longer worked after a turn.
//
// Two defenses live here:
//
// 1. Input-side caps (`shouldSkip`) — if a file is clearly too big to diff in
// a reasonable time, skip the patch body outright and return `{ patch: "" }`
// plus a reason. This is the primary fix and it is cheap.
// 2. Worker offload (`patchAsync`) — even for inputs that slip past the caps,
// run the Myers pass in a dedicated Web Worker with a hard timeout, so the
// main event loop keeps breathing. This is the safety net.
//
// Both defenses are controlled by `Flag.KILO_DIFF_CAPS` and `Flag.KILO_DIFF_WORKER`
// so they can be turned off individually for rollback without code changes.
import { fileURLToPath } from "url"
import { formatPatch, structuredPatch } from "diff"
import { Filesystem } from "../../util/filesystem"
import { Flag } from "../../flag/flag"
import { Log } from "../../util/log"
declare global {
const KILO_DIFF_WORKER_PATH: string | undefined
}
// `shouldSkip` is the single entry point callers import. If either side of
// a diff is clearly too big to process in a reasonable time, return a
// `SkipReason` and the caller substitutes an empty patch. Additions and
// deletions are still reported from git numstat so the summary counts stay
// accurate.
export namespace DiffEngine {
const log = Log.create({ service: "diff-engine" })
/** Hard byte cap on a single side (before or after) of a diff. 512 KB. */
export const MAX_INPUT_BYTES = 512 * 1024
/** Hard line cap on a single side of a diff. */
export const MAX_INPUT_LINES = 2000
/** Max ms to wait on a worker call before giving up and returning "timeout". */
export const DEFAULT_TIMEOUT_MS = 10_000
/** Under this size/line count we skip the worker round-trip — the sync call is faster. */
const SYNC_BYTES = 50 * 1024
const SYNC_LINES = 500
export type SkipReason = "oversized" | "too-many-lines" | "timeout" | "worker-error"
export type Opts = {
context?: number
ignoreWhitespace?: boolean
timeout?: number
signal?: AbortSignal
}
export interface Result {
patch: string
skipped?: SkipReason
}
export type SkipReason = "oversized" | "too-many-lines"
function lines(text: string) {
if (!text) return 0
@@ -70,254 +35,8 @@ export namespace DiffEngine {
/** Returns the skip reason, or undefined if the inputs are small enough to diff directly. */
export function shouldSkip(before: string, after: string): SkipReason | undefined {
if (!Flag.KILO_DIFF_CAPS) return undefined
if (before.length > MAX_INPUT_BYTES || after.length > MAX_INPUT_BYTES) return "oversized"
if (lines(before) > MAX_INPUT_LINES || lines(after) > MAX_INPUT_LINES) return "too-many-lines"
return undefined
}
function formatSync(file: string, before: string, after: string, opts: Opts): string {
const ctx = opts.context ?? Number.MAX_SAFE_INTEGER
return formatPatch(
structuredPatch(file, file, before, after, "", "", {
context: Number.isFinite(ctx) ? ctx : Number.MAX_SAFE_INTEGER,
ignoreWhitespace: opts.ignoreWhitespace,
}),
)
}
/** Synchronous patch with input caps. Safe only for callers that tolerate ~100ms blocking. */
export function patchSync(file: string, before: string, after: string, opts: Opts = {}): Result {
const skipped = shouldSkip(before, after)
if (skipped) return { patch: "", skipped }
return { patch: formatSync(file, before, after, opts) }
}
// ---------------------------------------------------------------------------
// Worker pool (singleton)
// ---------------------------------------------------------------------------
type Pending = {
resolve: (value: Result) => void
timer: ReturnType<typeof setTimeout>
signal?: AbortSignal
onAbort?: () => void
}
type Pool = {
worker: Worker
pending: Map<number, Pending>
}
let pool: Pool | null = null
let pending: Promise<Pool | null> | null = null
let nextId = 1
let warnedFallback = false
async function resolveWorkerPath(): Promise<string | URL | undefined> {
// Bundled build: the define from script/build.ts hands us the absolute path.
if (typeof KILO_DIFF_WORKER_PATH !== "undefined" && KILO_DIFF_WORKER_PATH) {
return KILO_DIFF_WORKER_PATH
}
// Dev mode: resolve the `.ts` sibling via import.meta.url.
const dist = new URL("./diff-worker.js", import.meta.url)
if (await Filesystem.exists(fileURLToPath(dist))) return dist
return new URL("./diff-worker.ts", import.meta.url)
}
function drainPending(next: Pool, reason: Result["skipped"]) {
for (const entry of next.pending.values()) {
clearTimeout(entry.timer)
if (entry.signal && entry.onAbort) entry.signal.removeEventListener("abort", entry.onAbort)
entry.resolve({ patch: "", skipped: reason })
}
next.pending.clear()
}
async function build(): Promise<Pool | null> {
if (!Flag.KILO_DIFF_WORKER) return null
const file = await resolveWorkerPath()
if (!file) return null
try {
const worker = new Worker(file, { name: "diff-worker" } as WorkerOptions)
const next: Pool = { worker, pending: new Map<number, Pending>() }
worker.onmessage = (evt: MessageEvent<{ id: number; patch?: string; error?: string }>) => {
const res = evt.data
const entry = next.pending.get(res.id)
if (!entry) return
clearTimeout(entry.timer)
if (entry.signal && entry.onAbort) entry.signal.removeEventListener("abort", entry.onAbort)
next.pending.delete(res.id)
if (res.error) {
entry.resolve({ patch: "", skipped: "worker-error" })
return
}
entry.resolve({ patch: res.patch ?? "" })
}
const onError = (err: unknown) => {
log.warn("diff worker crashed, will restart on next call", {
error: err instanceof Error ? err.message : String(err),
})
drainPending(next, "worker-error")
if (pool === next) {
pool = null
try {
next.worker.terminate()
} catch (err) {
log.debug("diff worker terminate after crash failed", {
error: err instanceof Error ? err.message : String(err),
})
}
}
}
worker.onerror = (evt) => onError(evt instanceof ErrorEvent ? (evt.error ?? evt.message) : evt)
return next
} catch (err) {
if (!warnedFallback) {
warnedFallback = true
log.warn("failed to construct diff worker, falling back to sync path", {
error: err instanceof Error ? err.message : String(err),
})
}
return null
}
}
// Concurrent callers share the same construction promise. Without this gate,
// a batch of concurrent patchAsync calls (e.g. `Effect.forEach` with
// `concurrency: 8`) would all race past the `if (pool)` check and each
// spawn its own Worker, with last-write-wins leaking the others.
async function ensurePool(): Promise<Pool | null> {
if (pool) return pool
if (!Flag.KILO_DIFF_WORKER) return null
if (!pending) {
pending = build()
.then((next) => {
if (next) pool = next
return next
})
.finally(() => {
pending = null
})
}
return pending
}
function terminate() {
const hit = pool
if (!hit) return
pool = null
try {
hit.worker.terminate()
} catch (err) {
log.debug("diff worker terminate failed", {
error: err instanceof Error ? err.message : String(err),
})
}
}
// Scoped terminate: only tears down the pool that owns the failing request.
// If a newer pool has already replaced it, we leave the current one alone so
// a stale timer or abort cannot spawn a fresh worker (and kill unrelated
// in-flight work).
function terminatePool(p: Pool) {
if (pool === p) pool = null
try {
p.worker.terminate()
} catch (err) {
log.debug("diff worker terminate failed", {
error: err instanceof Error ? err.message : String(err),
})
}
}
/**
* Offloaded patch generation. Applies the input caps first (fast-path skip),
* then dispatches to the singleton worker with a bounded timeout. If the
* worker is unavailable or disabled, falls back to the sync implementation.
*/
export async function patchAsync(file: string, before: string, after: string, opts: Opts = {}): Promise<Result> {
const skipped = shouldSkip(before, after)
if (skipped) return { patch: "", skipped }
// Tiny files: sync is cheaper than postMessage round-trip.
if (
before.length < SYNC_BYTES &&
after.length < SYNC_BYTES &&
lines(before) < SYNC_LINES &&
lines(after) < SYNC_LINES
) {
return { patch: formatSync(file, before, after, opts) }
}
const p = await ensurePool()
if (!p) return { patch: formatSync(file, before, after, opts) }
const id = nextId++
const timeout = opts.timeout ?? DEFAULT_TIMEOUT_MS
return new Promise<Result>((resolve) => {
const entry: Pending = {
resolve,
timer: setTimeout(() => {
p.pending.delete(id)
if (entry.signal && entry.onAbort) entry.signal.removeEventListener("abort", entry.onAbort)
log.warn("diff worker timed out, terminating", { file, timeout })
terminatePool(p)
resolve({ patch: "", skipped: "timeout" })
}, timeout),
}
if (opts.signal) {
entry.signal = opts.signal
entry.onAbort = () => {
p.pending.delete(id)
clearTimeout(entry.timer)
// Interruption means the caller no longer cares — don't bother finishing.
terminatePool(p)
resolve({ patch: "", skipped: "timeout" })
}
if (opts.signal.aborted) {
entry.onAbort()
return
}
opts.signal.addEventListener("abort", entry.onAbort, { once: true })
}
p.pending.set(id, entry)
p.worker.postMessage({
id,
file,
before,
after,
opts: { context: opts.context, ignoreWhitespace: opts.ignoreWhitespace },
})
})
}
/** Test/shutdown helper — tears down the worker so the process can exit cleanly. */
export async function shutdown() {
const hit = pool
if (!hit) return
pool = null
drainPending(hit, "worker-error")
try {
hit.worker.terminate()
} catch (err) {
log.debug("diff worker shutdown terminate failed", {
error: err instanceof Error ? err.message : String(err),
})
}
}
/** Visible for testing — not a stable API. */
export const _internal = {
get hasPool() {
return pool !== null
},
resolveWorkerPath,
}
}
@@ -1,88 +0,0 @@
// kilocode_change - new file
//
// Extracted from `Snapshot.diffFull` to keep our diff against the shared
// `src/snapshot/index.ts` as small as possible. This is the per-row diff
// loop that:
//
// - Runs the npm `diff` package inside a worker (with caps and a timeout)
// via `DiffEngine.patchAsync`, so a file with tens of thousands of lines
// cannot block the event loop.
// - Yields to the Effect scheduler between files so abort + heartbeat
// endpoints stay responsive during a large diff.
//
// The helper returns its own accumulator; the caller passes it the rows and
// the two closed-over readers (`load`, `show`) from the snapshot layer.
import { Effect } from "effect"
import { Log } from "@/util/log"
import { DiffEngine } from "./diff-engine"
export namespace DiffFull {
const log = Log.create({ service: "snapshot.diff-full" })
export interface Row {
file: string
status: "added" | "deleted" | "modified"
binary: boolean
additions: number
deletions: number
}
export interface Result {
file: string
patch: string
additions: number
deletions: number
status: "added" | "deleted" | "modified"
}
export function run<E1, E2, R1, R2>(params: {
rows: Row[]
step: number
from: string
to: string
load: (run: Row[]) => Effect.Effect<Map<string, { before: string; after: string }> | undefined, E1, R1>
show: (row: Row) => Effect.Effect<string[], E2, R2>
}) {
return Effect.gen(function* () {
const timer = log.time("diffFull", { from: params.from, to: params.to, rows: params.rows.length })
const out: Result[] = []
for (let i = 0; i < params.rows.length; i += params.step) {
const batch = params.rows.slice(i, i + params.step)
const text = yield* params.load(batch)
for (const row of batch) {
const hit = text?.get(row.file) ?? { before: "", after: "" }
const pair = row.binary ? ["", ""] : text ? [hit.before, hit.after] : yield* params.show(row)
const before = pair[0] ?? ""
const after = pair[1] ?? ""
const res = row.binary
? { patch: "" as string, skipped: undefined as DiffEngine.SkipReason | undefined }
: yield* Effect.promise(() => DiffEngine.patchAsync(row.file, before, after))
if (res.skipped) {
log.warn("diffFull.skipped", {
file: row.file,
reason: res.skipped,
bytesBefore: before.length,
bytesAfter: after.length,
})
}
out.push({
file: row.file,
patch: res.patch,
additions: row.additions,
deletions: row.deletions,
status: row.status,
})
// Let the runtime scheduler yield between files so the event loop
// can service the abort endpoint and SSE heartbeat.
yield* Effect.yieldNow
}
}
timer.stop()
return out
})
}
}
@@ -1,44 +0,0 @@
// kilocode_change - new file
//
// Dedicated worker thread for running the `diff` package's Myers algorithm
// (`structuredPatch` + `formatPatch`). These are O(N*M) and can easily block
// the main event loop for minutes on pathological inputs — which in the TUI
// is the same thread that serves the Hono API (ESC endpoint included) and
// the SSE heartbeats. Offloading them here keeps the main loop responsive.
//
// Protocol: parent posts { id, file, before, after, opts } and the worker
// responds with { id, patch } on success or { id, error } on failure. The
// worker never throws into the parent — every failure is surfaced via the
// `error` field so the parent can fall back to a sync path gracefully.
import { formatPatch, structuredPatch } from "diff"
type Req = {
id: number
file: string
before: string
after: string
opts: { context?: number; ignoreWhitespace?: boolean }
}
type Res = { id: number; patch: string } | { id: number; error: string }
// Web Worker API (Bun workers). The parent runs this file via `new Worker(url)`.
self.onmessage = (evt: MessageEvent<Req>) => {
const req = evt.data
try {
const ctx = req.opts.context ?? Number.MAX_SAFE_INTEGER
const patch = formatPatch(
structuredPatch(req.file, req.file, req.before, req.after, "", "", {
context: Number.isFinite(ctx) ? ctx : Number.MAX_SAFE_INTEGER,
ignoreWhitespace: req.opts.ignoreWhitespace,
}),
)
self.postMessage({ id: req.id, patch } satisfies Res)
} catch (err) {
self.postMessage({
id: req.id,
error: err instanceof Error ? err.message : String(err),
} satisfies Res)
}
}
+4 -14
View File
@@ -1,4 +1,5 @@
import { Effect, Layer, Context, Stream } from "effect"
import { formatPatch, structuredPatch } from "diff"
import path from "path"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
@@ -10,7 +11,6 @@ import { Log } from "@/util/log"
import { Instance } from "./instance"
import { makeRuntime } from "@/effect/run-service" // kilocode_change
import z from "zod"
import { DiffEngine } from "@/kilocode/snapshot/diff-engine" // kilocode_change
export namespace Vcs {
const log = Log.create({ service: "vcs" })
@@ -49,8 +49,8 @@ export namespace Vcs {
map: Map<string, { additions: number; deletions: number }>,
) {
const base = ref ? yield* git.prefix(cwd) : ""
// kilocode_change start — route through DiffEngine for worker offload + size caps.
// Same freeze mode as SessionSummary if a changed file is huge.
const patch = (file: string, before: string, after: string) =>
formatPatch(structuredPatch(file, file, before, after, "", "", { context: Number.MAX_SAFE_INTEGER }))
const next = yield* Effect.forEach(
list,
(item) =>
@@ -58,18 +58,9 @@ export namespace Vcs {
const before = item.status === "added" || !ref ? "" : yield* git.show(cwd, ref, item.file, base)
const after = item.status === "deleted" ? "" : yield* work(fs, cwd, item.file)
const stat = map.get(item.file)
const out = yield* Effect.promise(() => DiffEngine.patchAsync(item.file, before, after))
if (out.skipped) {
log.warn("vcs.diff.skipped", {
file: item.file,
reason: out.skipped,
bytesBefore: before.length,
bytesAfter: after.length,
})
}
return {
file: item.file,
patch: out.patch,
patch: patch(item.file, before, after),
additions: stat?.additions ?? (item.status === "added" ? count(after) : 0),
deletions: stat?.deletions ?? (item.status === "deleted" ? count(before) : 0),
status: item.status,
@@ -78,7 +69,6 @@ export namespace Vcs {
{ concurrency: 8 },
)
return next.toSorted((a, b) => a.file.localeCompare(b.file))
// kilocode_change end
})
const track = Effect.fnUntraced(function* (
-4
View File
@@ -131,10 +131,6 @@ export namespace SessionPrompt {
const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) {
yield* elog.info("cancel", { sessionID })
yield* KiloSessionPromptQueue.cancel(sessionID) // kilocode_change - drop queued follow-up loops on abort
// kilocode_change start — also interrupt an in-flight summary so a huge
// diff can never outlive the session it belongs to.
yield* Effect.promise(() => SessionSummary.cancel(sessionID)).pipe(Effect.ignore)
// kilocode_change end
yield* state.cancel(sessionID)
})
+1 -9
View File
@@ -7,7 +7,6 @@ import { Session } from "."
import { MessageV2 } from "./message-v2"
import { SessionID, MessageID } from "./schema"
import { makeRuntime } from "@/effect/run-service" // kilocode_change
import { SummaryDispatch } from "../kilocode/session/summary-dispatch" // kilocode_change
export namespace SessionSummary {
function unquoteGitPath(input: string) {
@@ -173,15 +172,8 @@ export namespace SessionSummary {
messageID: MessageID.zod.optional(),
})
// kilocode_change start - legacy promise helpers + tracked dispatcher for Kilo callsites
// kilocode_change start - legacy promise helpers for Kilo callsites
const { runPromise } = makeRuntime(Service, defaultLayer)
export const diff = (input: { sessionID: SessionID; messageID?: MessageID }) => runPromise((svc) => svc.diff(input))
const dispatch = SummaryDispatch.create<Interface>({
runPromise,
summarize: (svc, input) => svc.summarize(input),
})
export const summarize = dispatch.summarize
export const cancel = dispatch.cancel
// kilocode_change end
}
+33 -4
View File
@@ -1,5 +1,6 @@
import { Cause, Duration, Effect, Layer, Schedule, Semaphore, Context, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { formatPatch, structuredPatch } from "diff"
import path from "path"
import z from "zod"
import { makeRuntime } from "@/effect/run-service" // kilocode_change
@@ -9,7 +10,7 @@ import { AppFileSystem } from "@/filesystem"
import { Config } from "../config/config"
import { Log } from "../util/log"
import * as KiloSnapshot from "../kilocode/snapshot" // kilocode_change
import { DiffFull } from "../kilocode/snapshot/diff-full" // kilocode_change
import { DiffEngine } from "../kilocode/snapshot/diff-engine" // kilocode_change
export namespace Snapshot {
export const Patch = z.object({
@@ -708,9 +709,37 @@ export namespace Snapshot {
}
const step = 100
// kilocode_change start - delegate to kilo helper (caps + worker)
result.push(...(yield* DiffFull.run({ rows, step, from, to, load, show })))
// kilocode_change end
const patch = (file: string, before: string, after: string) =>
formatPatch(structuredPatch(file, file, before, after, "", "", { context: Number.MAX_SAFE_INTEGER }))
for (let i = 0; i < rows.length; i += step) {
const run = rows.slice(i, i + step)
const text = yield* load(run)
for (const row of run) {
const hit = text?.get(row.file) ?? { before: "", after: "" }
const [before, after] = row.binary ? ["", ""] : text ? [hit.before, hit.after] : yield* show(row)
// kilocode_change start — cap runaway inputs so Myers never blocks the event loop
const skipped = row.binary ? undefined : DiffEngine.shouldSkip(before, after)
if (skipped) {
log.warn("diffFull.skipped", {
file: row.file,
reason: skipped,
bytesBefore: before.length,
bytesAfter: after.length,
})
}
result.push({
file: row.file,
patch: row.binary || skipped ? "" : patch(row.file, before, after),
additions: row.additions,
deletions: row.deletions,
status: row.status,
})
// kilocode_change end
}
}
return result
}),
)
@@ -1,13 +1,9 @@
import { test, expect, afterAll } from "bun:test"
import { test, expect } from "bun:test"
import { DiffEngine } from "../../src/kilocode/snapshot/diff-engine"
import { Log } from "../../src/util/log"
Log.init({ print: false })
afterAll(async () => {
await DiffEngine.shutdown()
})
test("shouldSkip returns undefined for small inputs", () => {
expect(DiffEngine.shouldSkip("a", "b")).toBeUndefined()
expect(DiffEngine.shouldSkip("hello\nworld", "hello\nworld!")).toBeUndefined()
@@ -25,61 +21,12 @@ test("shouldSkip returns 'too-many-lines' when lines exceed MAX_INPUT_LINES", ()
expect(DiffEngine.shouldSkip("small", many)).toBe("too-many-lines")
})
test("patchSync returns non-empty patch for small diffs", () => {
const result = DiffEngine.patchSync("file.txt", "hello\nworld\n", "hello\nuniverse\n")
expect(result.skipped).toBeUndefined()
expect(result.patch).toContain("@@")
expect(result.patch).toContain("-world")
expect(result.patch).toContain("+universe")
})
test("patchSync returns empty patch with skipped='too-many-lines' for huge inputs in <50ms", () => {
test("shouldSkip runs in <50ms on pathological inputs", () => {
const before = "before_line\n".repeat(10_000)
const after = "after_line\n".repeat(10_000)
const start = Date.now()
const result = DiffEngine.patchSync("big.txt", before, after)
const result = DiffEngine.shouldSkip(before, after)
const elapsed = Date.now() - start
expect(result.patch).toBe("")
expect(result.skipped).toBe("too-many-lines")
expect(result).toBe("too-many-lines")
expect(elapsed).toBeLessThan(50)
})
test("patchAsync returns same output as patchSync for small inputs", async () => {
const before = "line1\nline2\n"
const after = "line1\nline_changed\n"
const sync = DiffEngine.patchSync("x.txt", before, after)
const async = await DiffEngine.patchAsync("x.txt", before, after)
expect(async.patch).toBe(sync.patch)
expect(async.skipped).toBeUndefined()
})
test("patchAsync respects input caps (skips without spawning worker)", async () => {
const big = "x".repeat(DiffEngine.MAX_INPUT_BYTES + 10)
const result = await DiffEngine.patchAsync("huge.txt", big, "tiny")
expect(result.patch).toBe("")
expect(result.skipped).toBe("oversized")
})
test("patchAsync respects AbortSignal", async () => {
const ac = new AbortController()
ac.abort()
const result = await DiffEngine.patchAsync("x.txt", "hello\n", "world\n", { signal: ac.signal })
// Either the sync fast-path ran (small input) or the abort was honored — either is fine.
expect(typeof result.patch).toBe("string")
})
test("patchAsync honors a short timeout with a stubbed slow worker", async () => {
// Craft an input just over the sync fast-path threshold so patchAsync routes
// to the worker, then set a ludicrously short timeout so the race trips.
// This exercises the timeout + worker-termination code path.
const lines = 600
const before = Array.from({ length: lines }, (_, i) => `a_${i}`).join("\n") + "\n"
const after = Array.from({ length: lines }, (_, i) => `b_${i}`).join("\n") + "\n"
// 1ms is almost certainly shorter than postMessage round-trip.
const result = await DiffEngine.patchAsync("x.txt", before, after, { timeout: 1 })
// Could succeed (if worker was incredibly fast) or time out — accept both.
if (result.skipped) {
expect(["timeout", "worker-error"]).toContain(result.skipped)
expect(result.patch).toBe("")
}
})
@@ -1,176 +0,0 @@
// kilocode_change - new file
// Integration tests for the caps + worker offload applied inside Snapshot.diffFull.
// Asserts the freeze repro workload finishes quickly and that the event loop
// stays responsive while the diff runs.
import { test, expect, afterAll } from "bun:test"
import { $ } from "bun"
import { Snapshot } from "../../src/snapshot"
import { Instance } from "../../src/project/instance"
import { Filesystem } from "../../src/util/filesystem"
import { Log } from "../../src/util/log"
import { DiffEngine } from "../../src/kilocode/snapshot/diff-engine"
import { tmpdir } from "../fixture/fixture"
Log.init({ print: false })
afterAll(async () => {
await DiffEngine.shutdown()
})
async function bootstrap(setup: (dir: string) => Promise<void>) {
return tmpdir({
git: true,
init: async (dir) => {
await setup(dir)
await $`git add .`.cwd(dir).quiet()
await $`git commit --no-gpg-sign -m init`.cwd(dir).quiet()
},
})
}
test("diffFull returns empty patch with counted adds/deletes when a file crosses the line cap", async () => {
// 5000-line file (cap is 2000) — the underlying Myers would take minutes.
const big = Array.from({ length: 5000 }, (_, i) => `v1_${i}`).join("\n") + "\n"
const bigAfter = Array.from({ length: 5000 }, (_, i) => `v2_${i}`).join("\n") + "\n"
await using tmp = await bootstrap(async (dir) => {
await Filesystem.write(`${dir}/big.txt`, big)
})
await Instance.provide({
directory: tmp.path,
fn: async () => {
const before = await Snapshot.track()
expect(before).toBeTruthy()
await Filesystem.write(`${tmp.path}/big.txt`, bigAfter)
const after = await Snapshot.track()
expect(after).toBeTruthy()
const start = Date.now()
const diffs = await Snapshot.diffFull(before!, after!)
const elapsed = Date.now() - start
expect(diffs).toHaveLength(1)
const [hit] = diffs
expect(hit).toBeDefined()
expect(hit!.file).toBe("big.txt")
// Skipped = empty patch, but add/delete counts come from git numstat.
expect(hit!.patch).toBe("")
expect(hit!.additions).toBeGreaterThan(0)
expect(hit!.deletions).toBeGreaterThan(0)
// Must finish in well under a second — without the caps this ran for minutes.
expect(elapsed).toBeLessThan(3000)
},
})
})
test("diffFull keeps full patches for small files in a workload", async () => {
await using tmp = await bootstrap(async (dir) => {
for (let i = 0; i < 20; i++) {
await Filesystem.write(`${dir}/f_${i}.txt`, `hello ${i}\nline 2\n`)
}
})
await Instance.provide({
directory: tmp.path,
fn: async () => {
const before = await Snapshot.track()
expect(before).toBeTruthy()
for (let i = 0; i < 20; i++) {
await Filesystem.write(`${tmp.path}/f_${i}.txt`, `world ${i}\nline 2\n`)
}
const after = await Snapshot.track()
expect(after).toBeTruthy()
const start = Date.now()
const diffs = await Snapshot.diffFull(before!, after!)
const elapsed = Date.now() - start
expect(diffs.length).toBe(20)
for (const d of diffs) {
expect(d.patch).toContain("@@")
expect(d.patch.length).toBeGreaterThan(0)
}
expect(elapsed).toBeLessThan(5000)
},
})
})
test("diffFull mixes skipped and non-skipped files in one result", async () => {
const huge = Array.from({ length: 5000 }, (_, i) => `h_${i}`).join("\n") + "\n"
const hugeAfter = Array.from({ length: 5000 }, (_, i) => `k_${i}`).join("\n") + "\n"
await using tmp = await bootstrap(async (dir) => {
await Filesystem.write(`${dir}/tiny.txt`, "one\ntwo\n")
await Filesystem.write(`${dir}/huge.txt`, huge)
})
await Instance.provide({
directory: tmp.path,
fn: async () => {
const before = await Snapshot.track()
expect(before).toBeTruthy()
await Filesystem.write(`${tmp.path}/tiny.txt`, "one\nchanged\n")
await Filesystem.write(`${tmp.path}/huge.txt`, hugeAfter)
const after = await Snapshot.track()
expect(after).toBeTruthy()
const diffs = await Snapshot.diffFull(before!, after!)
const tiny = diffs.find((d) => d.file === "tiny.txt")
const big = diffs.find((d) => d.file === "huge.txt")
expect(tiny).toBeDefined()
expect(big).toBeDefined()
expect(tiny!.patch).toContain("@@")
expect(big!.patch).toBe("")
// numstat still populates additions/deletions for the skipped file.
expect(big!.additions).toBeGreaterThan(0)
expect(big!.deletions).toBeGreaterThan(0)
},
})
})
test("event loop stays responsive while diffFull runs a heavy workload", async () => {
// 40 files of ~200 lines each. Under the old code each file went through a
// synchronous structuredPatch on the worker thread. Under the new code the
// yields + worker offload keep the event loop breathing.
await using tmp = await bootstrap(async (dir) => {
for (let i = 0; i < 40; i++) {
const text = Array.from({ length: 200 }, (_, j) => `f_${i}_line_${j}`).join("\n") + "\n"
await Filesystem.write(`${dir}/mod_${i}.txt`, text)
}
})
await Instance.provide({
directory: tmp.path,
fn: async () => {
const before = await Snapshot.track()
expect(before).toBeTruthy()
for (let i = 0; i < 40; i++) {
const text = Array.from({ length: 200 }, (_, j) => `f_${i}_line_${j}_v2`).join("\n") + "\n"
await Filesystem.write(`${tmp.path}/mod_${i}.txt`, text)
}
const after = await Snapshot.track()
expect(after).toBeTruthy()
// Count how many times a concurrent interval fires during the diff.
// If the event loop stays responsive, we expect several ticks.
let ticks = 0
const timer = setInterval(() => {
ticks++
}, 25)
try {
await Snapshot.diffFull(before!, after!)
} finally {
clearInterval(timer)
}
// Even on a slow CI box the interval should fire multiple times if we are
// not blocking the loop. This is the honest "ESC still works" signal.
expect(ticks).toBeGreaterThan(0)
},
})
})
@@ -12,13 +12,12 @@
// 3. A concurrent setInterval keeps ticking — i.e. the event loop keeps
// breathing and ESC would be delivered.
import { test, expect, afterAll, afterEach, mock } from "bun:test"
import { test, expect, afterEach, mock } from "bun:test"
import { $ } from "bun"
import { Instance } from "../../src/project/instance"
import { Server } from "../../src/server/server"
import { Session } from "../../src/session"
import { Snapshot } from "../../src/snapshot"
import { DiffEngine } from "../../src/kilocode/snapshot/diff-engine"
import { Filesystem } from "../../src/util/filesystem"
import { Log } from "../../src/util/log"
import { tmpdir } from "../fixture/fixture"
@@ -30,10 +29,6 @@ afterEach(async () => {
await Instance.disposeAll()
})
afterAll(async () => {
await DiffEngine.shutdown()
})
test("pathological diffFull workload finishes quickly and does not block abort", async () => {
// 3000-line file that churns every line between snapshots. Before the fix
// this ran through structuredPatch at context=MAX_SAFE_INTEGER synchronously