refactor(fs): share read preflight helpers
This commit is contained in:
@@ -90,6 +90,24 @@ function commandOpts(signal: AbortSignal | undefined): { envs: Record<string, st
|
||||
return { envs: e2bControlEnvs(), ...signalOpts(signal) }
|
||||
}
|
||||
|
||||
async function openReadStream(
|
||||
sandbox: Sandbox,
|
||||
target: FsTarget,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<ReadableStream<Uint8Array>> {
|
||||
try {
|
||||
// The pinned SDK's stream overload lies for empty files: content-length 0
|
||||
// returns '' instead of a ReadableStream.
|
||||
const read = await sandbox.files.read(String(target.targetKey), { format: 'stream', ...signalOpts(signal) }) as
|
||||
ReadableStream<Uint8Array> | string
|
||||
return typeof read === 'string'
|
||||
? new ReadableStream<Uint8Array>({ start(controller) { controller.close() } })
|
||||
: read
|
||||
} catch (error: unknown) {
|
||||
throw mapError(error, 'read', target.displayPath, signal)
|
||||
}
|
||||
}
|
||||
|
||||
function entryType(entry: EntryInfo): FsInfo['type'] {
|
||||
switch (entry.type) {
|
||||
case FileType.FILE:
|
||||
@@ -233,18 +251,7 @@ export class E2BFileSystem extends FileSystem {
|
||||
if (info.size !== undefined && info.size > maxBytes) {
|
||||
throw new FsError(`cannot read "${target.displayPath}": ${info.size} bytes exceeds the ${maxBytes}-byte limit`, 'FS_TOO_LARGE')
|
||||
}
|
||||
let stream: ReadableStream<Uint8Array>
|
||||
try {
|
||||
// Same pinned-SDK quirk as streamText: content-length 0 returns ''
|
||||
// instead of a ReadableStream.
|
||||
const read = await sandbox.files.read(String(target.targetKey), { format: 'stream', ...signalOpts(signal) }) as
|
||||
ReadableStream<Uint8Array> | string
|
||||
stream = typeof read === 'string'
|
||||
? new ReadableStream<Uint8Array>({ start(controller) { controller.close() } })
|
||||
: read
|
||||
} catch (error: unknown) {
|
||||
throw mapError(error, 'read', target.displayPath, signal)
|
||||
}
|
||||
const stream = await openReadStream(sandbox, target, signal)
|
||||
const reader = stream.getReader()
|
||||
const chunks: Uint8Array[] = []
|
||||
let bytes = 0
|
||||
@@ -288,18 +295,7 @@ export class E2BFileSystem extends FileSystem {
|
||||
override async streamText(target: FsTarget, signal?: AbortSignal): Promise<AsyncIterable<string>> {
|
||||
const sandbox = await this.ctx.e2b.getSandbox()
|
||||
await this.requireRegular(target, signal)
|
||||
let stream: ReadableStream<Uint8Array>
|
||||
try {
|
||||
// The pinned SDK's stream overload lies for empty files: content-length 0
|
||||
// returns '' instead of a ReadableStream.
|
||||
const read = await sandbox.files.read(String(target.targetKey), { format: 'stream', ...signalOpts(signal) }) as
|
||||
ReadableStream<Uint8Array> | string
|
||||
stream = typeof read === 'string'
|
||||
? new ReadableStream<Uint8Array>({ start(controller) { controller.close() } })
|
||||
: read
|
||||
} catch (error: unknown) {
|
||||
throw mapError(error, 'read', target.displayPath, signal)
|
||||
}
|
||||
const stream = await openReadStream(sandbox, target, signal)
|
||||
const displayPath = target.displayPath
|
||||
return {
|
||||
async *[Symbol.asyncIterator](): AsyncGenerator<string> {
|
||||
|
||||
@@ -19,9 +19,8 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import type { GenericCallView, ToolExecution } from '@deepseek-ai/dsh-tools'
|
||||
import { FsError } from '@deepseek-ai/dsh-fs'
|
||||
import type {} from '@deepseek-ai/dsh-fs'
|
||||
import { sessionResolveOptions } from './session-cwd.ts'
|
||||
import { resolveRegularReadTarget } from './read-target.ts'
|
||||
|
||||
/** Extensions `read_image` accepts; magic-byte validation at the attachment service stays authoritative. */
|
||||
const IMAGE_EXTENSIONS: Readonly<Record<string, ImageMediaType>> = {
|
||||
@@ -179,13 +178,7 @@ export function applyReadImageTool(ctx: Context): void {
|
||||
}
|
||||
await assertImageCapableRoute(ctx, exec, args.file_path)
|
||||
|
||||
const target = await ctx.fs.resolve(args.file_path, sessionResolveOptions(exec, args.file_path))
|
||||
const info = await ctx.fs.stat(target, exec.signal)
|
||||
if (!info) {
|
||||
ctx.emit('fs/observed', target, { kind: 'absent' }, exec)
|
||||
throw new FsError(`cannot read "${target.displayPath}": not found`, 'FS_NOT_FOUND')
|
||||
}
|
||||
if (info.type !== 'file') throw new FsError(`cannot read "${target.displayPath}": not a regular file`, 'FS_NOT_REGULAR_FILE')
|
||||
const { target, info } = await resolveRegularReadTarget(ctx, exec, args.file_path)
|
||||
|
||||
// The tool result is one message carrying one image, so the per-message
|
||||
// aggregate bound applies beside the per-image bound.
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
/**
|
||||
* Shared path resolution and regular-file validation for model-facing read tools.
|
||||
* @module @deepseek-ai/dsh-tool-fs/src/read-target
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { FsError } from '@deepseek-ai/dsh-fs'
|
||||
import type { FsInfo, FsTarget } from '@deepseek-ai/dsh-fs'
|
||||
import type { ToolExecution } from '@deepseek-ai/dsh-tools'
|
||||
import { sessionResolveOptions } from './session-cwd.ts'
|
||||
|
||||
/**
|
||||
* Resolve a model-supplied path, observe absence, and require a regular file.
|
||||
* @param ctx - the plugin context providing filesystem resolution and observation events.
|
||||
* @param exec - the current tool execution, including session cwd and cancellation.
|
||||
* @param requestedPath - the raw path supplied to the tool.
|
||||
* @returns the resolved target and its single stat result.
|
||||
*/
|
||||
export async function resolveRegularReadTarget(
|
||||
ctx: Context,
|
||||
exec: ToolExecution,
|
||||
requestedPath: string,
|
||||
): Promise<{ target: FsTarget; info: FsInfo }> {
|
||||
const target = await ctx.fs.resolve(requestedPath, sessionResolveOptions(exec, requestedPath))
|
||||
const info = await ctx.fs.stat(target, exec.signal)
|
||||
if (info === undefined) {
|
||||
ctx.emit('fs/observed', target, { kind: 'absent' }, exec)
|
||||
throw new FsError(`cannot read "${target.displayPath}": not found`, 'FS_NOT_FOUND')
|
||||
}
|
||||
if (info.type !== 'file') {
|
||||
throw new FsError(`cannot read "${target.displayPath}": not a regular file`, 'FS_NOT_REGULAR_FILE')
|
||||
}
|
||||
return { target, info }
|
||||
}
|
||||
@@ -7,11 +7,10 @@
|
||||
import type { Context } from 'cordis'
|
||||
import { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import type { GenericCallView, ReadResultView, ToolResult } from '@deepseek-ai/dsh-tools'
|
||||
import { FsError } from '@deepseek-ai/dsh-fs'
|
||||
import type {} from '@deepseek-ai/dsh-fs'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
import { buildWindow, formatReadOutput, langFromPath, readMetaFromMeta } from './read-render.ts'
|
||||
import { sessionResolveOptions } from './session-cwd.ts'
|
||||
import { resolveRegularReadTarget } from './read-target.ts'
|
||||
|
||||
/** Default and maximum number of lines returned by one `read` call (the `readLimit` config). */
|
||||
export const READ_LIMIT = 2000
|
||||
@@ -136,16 +135,9 @@ export function applyReadTool(ctx: Context, caps: ReadToolCaps): void {
|
||||
isConcurrencySafe: () => true,
|
||||
async execute(args, exec) {
|
||||
const input = parseReadArgs(args, caps.limit)
|
||||
const target = await ctx.fs.resolve(input.filePath, sessionResolveOptions(exec, input.filePath))
|
||||
|
||||
// One stat: absence observation OR type check + size routing + present version.
|
||||
// A concurrent write can only make a later guarded mutation fail stale and require reread.
|
||||
const info = await ctx.fs.stat(target, exec.signal)
|
||||
if (!info) {
|
||||
ctx.emit('fs/observed', target, { kind: 'absent' }, exec)
|
||||
throw new FsError(`cannot read "${target.displayPath}": not found`, 'FS_NOT_FOUND')
|
||||
}
|
||||
if (info.type !== 'file') throw new FsError(`cannot read "${target.displayPath}": not a regular file`, 'FS_NOT_REGULAR_FILE')
|
||||
const { target, info } = await resolveRegularReadTarget(ctx, exec, input.filePath)
|
||||
|
||||
// Stream when the file is large OR size is unknown, so a size-less backend
|
||||
// never buffers an arbitrarily large file.
|
||||
|
||||
Reference in New Issue
Block a user