mirror of
https://github.com/anomalyco/opencode.git
synced 2026-05-19 19:12:58 +00:00
fix read stream byte cap
This commit is contained in:
@@ -19,6 +19,8 @@ const MAX_BYTES_LABEL = `${MAX_BYTES / 1024} KB`
|
||||
const SAMPLE_BYTES = 4096
|
||||
const SUPPORTED_IMAGE_MIMES = new Set(["image/jpeg", "image/png", "image/gif", "image/webp"])
|
||||
|
||||
class ReadStop extends Schema.TaggedErrorClass<ReadStop>()("ReadStop", {}) {}
|
||||
|
||||
// `offset` and `limit` were originally `z.coerce.number()` — the runtime
|
||||
// coercion was useful when the tool was called from a shell but serves no
|
||||
// purpose in the LLM tool-call path (the model emits typed JSON). The JSON
|
||||
@@ -111,37 +113,41 @@ export const ReadTool = Tool.define(
|
||||
// Note: prefer manual TextDecoder over Stream.decodeText — when the source stream
|
||||
// ends without flushing, decodeText drops the final unterminated line. We also
|
||||
// avoid Stream.runForEachWhile (it currently swallows the final unterminated
|
||||
// line of the upstream splitLines pipeline) and instead toggle a `done` flag
|
||||
// and ignore subsequent lines.
|
||||
// line of the upstream splitLines pipeline) and use a tagged error to stop the
|
||||
// upstream file stream as soon as the byte cap is reached.
|
||||
const decoder = new TextDecoder("utf-8")
|
||||
yield* fs.stream(filepath).pipe(
|
||||
Stream.map((bytes) => decoder.decode(bytes, { stream: true })),
|
||||
Stream.splitLines,
|
||||
Stream.runForEach((text) =>
|
||||
Effect.sync(() => {
|
||||
if (flags.done) return
|
||||
flags.count += 1
|
||||
if (flags.count <= start) return
|
||||
yield* fs
|
||||
.stream(filepath)
|
||||
.pipe(
|
||||
Stream.map((bytes) => decoder.decode(bytes, { stream: true })),
|
||||
Stream.splitLines,
|
||||
Stream.runForEach((text) =>
|
||||
Effect.gen(function* () {
|
||||
if (flags.done) return yield* new ReadStop()
|
||||
flags.count += 1
|
||||
if (flags.count <= start) return
|
||||
|
||||
if (raw.length >= opts.limit) {
|
||||
flags.more = true
|
||||
return
|
||||
}
|
||||
if (raw.length >= opts.limit) {
|
||||
flags.more = true
|
||||
return
|
||||
}
|
||||
|
||||
const line = text.length > MAX_LINE_LENGTH ? text.substring(0, MAX_LINE_LENGTH) + MAX_LINE_SUFFIX : text
|
||||
const size = Buffer.byteLength(line, "utf-8") + (raw.length > 0 ? 1 : 0)
|
||||
if (flags.bytes + size <= MAX_BYTES) {
|
||||
raw.push(line)
|
||||
flags.bytes += size
|
||||
return
|
||||
}
|
||||
|
||||
const line = text.length > MAX_LINE_LENGTH ? text.substring(0, MAX_LINE_LENGTH) + MAX_LINE_SUFFIX : text
|
||||
const size = Buffer.byteLength(line, "utf-8") + (raw.length > 0 ? 1 : 0)
|
||||
if (flags.bytes + size > MAX_BYTES) {
|
||||
flags.cut = true
|
||||
flags.more = true
|
||||
flags.done = true
|
||||
return
|
||||
}
|
||||
|
||||
raw.push(line)
|
||||
flags.bytes += size
|
||||
}),
|
||||
),
|
||||
)
|
||||
return yield* new ReadStop()
|
||||
}),
|
||||
),
|
||||
Effect.catchTag("ReadStop", () => Effect.void),
|
||||
)
|
||||
|
||||
return { raw, count: flags.count, cut: flags.cut, more: flags.more, offset: opts.offset }
|
||||
})
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { afterEach, describe, expect } from "bun:test"
|
||||
import { Cause, Effect, Exit, Layer } from "effect"
|
||||
import { Cause, Effect, Exit, Layer, Stream } from "effect"
|
||||
import path from "path"
|
||||
import { Agent } from "../../src/agent/agent"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
@@ -366,6 +366,38 @@ describe("tool.read truncation", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.instance("stops streaming after the byte cap", () =>
|
||||
Effect.gen(function* () {
|
||||
const test = yield* TestInstance
|
||||
const filepath = path.join(test.directory, "huge.txt")
|
||||
const content = `${"x".repeat(80)}\n`.repeat(50_000)
|
||||
yield* put(filepath, content)
|
||||
|
||||
const fs = yield* AppFileSystem.Service
|
||||
const counter = { bytes: 0 }
|
||||
const result = yield* run({ filePath: filepath }).pipe(
|
||||
Effect.provideService(
|
||||
AppFileSystem.Service,
|
||||
AppFileSystem.Service.of({
|
||||
...fs,
|
||||
stream: (file, options) =>
|
||||
fs.stream(file, options).pipe(
|
||||
Stream.tap((chunk) =>
|
||||
Effect.sync(() => {
|
||||
counter.bytes += chunk.length
|
||||
}),
|
||||
),
|
||||
),
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
expect(result.metadata.truncated).toBe(true)
|
||||
expect(result.output).toContain("Output capped at")
|
||||
expect(counter.bytes).toBeLessThan(Buffer.byteLength(content, "utf-8") / 2)
|
||||
}),
|
||||
)
|
||||
|
||||
it.instance("truncates by line count when limit is specified", () =>
|
||||
Effect.gen(function* () {
|
||||
const test = yield* TestInstance
|
||||
|
||||
Reference in New Issue
Block a user