diff --git a/.changeset/clean-windows-file-handles.md b/.changeset/clean-windows-file-handles.md new file mode 100644 index 00000000000..8b1183b118a --- /dev/null +++ b/.changeset/clean-windows-file-handles.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Release project file handles immediately after reads on Windows so editors and tools can replace existing files without restarting Kilo. diff --git a/packages/opencode/src/kilocode/text-stream.ts b/packages/opencode/src/kilocode/text-stream.ts index 7dae004dd59..dc2b530a102 100644 --- a/packages/opencode/src/kilocode/text-stream.ts +++ b/packages/opencode/src/kilocode/text-stream.ts @@ -28,14 +28,38 @@ function decode(decoder: TextDecoder, bytes?: Uint8Array) { } } -async function* chunks(fs: FileSystem, filepath: string) { +function utf8(fs: FileSystem, filepath: string, signal?: AbortSignal) { + signal?.throwIfAborted() + const iterator = Stream.toAsyncIterable(fs.stream(filepath))[Symbol.asyncIterator]() const decoder = new TextDecoder("utf-8", { fatal: true }) - for await (const bytes of Stream.toAsyncIterable(fs.stream(filepath))) { - const text = decode(decoder, bytes) - if (text) yield text - } - const tail = decode(decoder) - if (tail) yield tail + const out = new Readable({ + read() { + void (async () => { + while (true) { + const next = await iterator.next() + if (next.done) { + const tail = decode(decoder) + if (tail) this.push(tail) + this.push(null) + return + } + const text = decode(decoder, next.value) + if (!text) continue + this.push(text) + return + } + })().catch((err) => this.destroy(err instanceof Error ? err : new Error(String(err)))) + }, + destroy(err, callback) { + Promise.resolve(iterator.return?.()).then( + () => callback(err), + (cause) => callback(cause instanceof Error ? cause : new Error(String(cause))), + ) + }, + }) + const closed = new Promise((resolve) => out.once("close", resolve)) + + return { stream: abortable(out, signal), closed } } export function abortable(stream: Readable, signal?: AbortSignal) { @@ -44,7 +68,7 @@ export function abortable(stream: Readable, signal?: AbortSignal) { /** UTF-8 text stream backed by the injected filesystem service. */ export function openUtf8(fs: FileSystem, filepath: string, signal?: AbortSignal): Readable { - return abortable(Readable.from(chunks(fs, filepath)), signal) + return utf8(fs, filepath, signal).stream } export function safeSlice(text: string, end: number) { @@ -69,10 +93,14 @@ export async function withFallback( fn: (input: Readable) => Promise, signal?: AbortSignal, ): Promise { + const input = utf8(fs, filepath, signal) try { - return await fn(openUtf8(fs, filepath, signal)) + return await fn(input.stream) } catch (err) { if (!(err instanceof InvalidUtf8Error)) throw err + } finally { + input.stream.destroy() + await input.closed } return fn(await openDecoded(fs, filepath, signal)) } diff --git a/packages/opencode/test/kilocode/tool-encoding.test.ts b/packages/opencode/test/kilocode/tool-encoding.test.ts index d1c2846e423..2f3606ea2b5 100644 --- a/packages/opencode/test/kilocode/tool-encoding.test.ts +++ b/packages/opencode/test/kilocode/tool-encoding.test.ts @@ -193,15 +193,16 @@ describe("tool encoding preservation", () => { }) describe("ReadTool streaming and pagination", () => { - it.live("streams UTF-8 files and stops after the output cap", () => + it.live("releases a truncated UTF-8 file before atomic replacement", () => provideTmpdirInstance((dir) => Effect.gen(function* () { const filepath = path.join(dir, "large.txt") + const temp = `${filepath}.tmp` const content = `${"x".repeat(80)}\n`.repeat(50_000) yield* Effect.promise(() => fs.writeFile(filepath, content)) const base = yield* FSUtil.Service - const counter = { bytes: 0 } + const state = { bytes: 0, closed: false } const result = yield* runRead({ filePath: filepath }).pipe( Effect.provideService( FSUtil.Service, @@ -211,7 +212,12 @@ describe("tool encoding preservation", () => { base.stream(file, options).pipe( Stream.tap((chunk) => Effect.sync(() => { - counter.bytes += chunk.length + state.bytes += chunk.length + }), + ), + Stream.ensuring( + Effect.sync(() => { + state.closed = true }), ), ), @@ -220,16 +226,23 @@ describe("tool encoding preservation", () => { ) expect(result.metadata.truncated).toBe(true) - expect(counter.bytes).toBeGreaterThan(0) - expect(counter.bytes).toBeLessThan(Buffer.byteLength(content, "utf-8") / 2) + expect(state.bytes).toBeGreaterThan(0) + expect(state.bytes).toBeLessThan(Buffer.byteLength(content, "utf-8") / 2) + yield* Effect.promise(async () => { + await fs.writeFile(temp, "replacement\n") + await fs.rename(temp, filepath) + }) + expect(yield* Effect.promise(() => fs.readFile(filepath, "utf8"))).toBe("replacement\n") + expect(state.closed).toBe(true) }), ), ) - it.live("stops the filesystem stream when the tool is aborted", () => + it.live("closes the source stream when the read is aborted", () => provideTmpdirInstance((dir) => Effect.gen(function* () { const filepath = path.join(dir, "abort.txt") + const temp = `${filepath}.tmp` yield* Effect.promise(() => fs.writeFile(filepath, `${"x".repeat(80)}\n`.repeat(50_000))) const base = yield* FSUtil.Service @@ -262,14 +275,20 @@ describe("tool encoding preservation", () => { expect(Exit.isFailure(exit)).toBe(true) expect(state.chunks).toBeGreaterThan(0) expect(state.closed).toBe(true) + yield* Effect.promise(async () => { + await fs.writeFile(temp, "replacement\n") + await fs.rename(temp, filepath) + }) + expect(yield* Effect.promise(() => fs.readFile(filepath, "utf8"))).toBe("replacement\n") }), ), ) - it.live("restarts cleanly when invalid UTF-8 appears after streamed lines", () => + it.live("releases a fallback-decoded file before atomic replacement", () => provideTmpdirInstance((dir) => Effect.gen(function* () { const filepath = path.join(dir, "legacy.txt") + const temp = `${filepath}.tmp` const lines = Array.from({ length: 1_000 }, (_, i) => `valid-${i + 1}-${"x".repeat(70)}`) const content = Buffer.concat([ Buffer.from(lines.join("\n") + "\n"), @@ -306,6 +325,11 @@ describe("tool encoding preservation", () => { expect(result.output.match(/999: valid-999-/g)?.length).toBe(1) expect(result.output).toContain(`1001: ${samples.shiftJis}`) expect(result.output).toContain("1002: last") + yield* Effect.promise(async () => { + await fs.writeFile(temp, "replacement\n") + await fs.rename(temp, filepath) + }) + expect(yield* Effect.promise(() => fs.readFile(filepath, "utf8"))).toBe("replacement\n") }), ), )