diff --git a/.changeset/mcp-attachment-originals.md b/.changeset/mcp-attachment-originals.md new file mode 100644 index 00000000000..218a4fa8e44 --- /dev/null +++ b/.changeset/mcp-attachment-originals.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Preserve MCP attachments that cannot be delivered directly to the model. diff --git a/docs/en/customization/mcp.md b/docs/en/customization/mcp.md index a91e22d3abe..29f08109e6f 100644 --- a/docs/en/customization/mcp.md +++ b/docs/en/customization/mcp.md @@ -4,6 +4,10 @@ MCP tool results can include text (`content`) and structured data (`structuredContent`). Kimi Code CLI makes both available to the agent and omits the structured copy only when it can confirm that a text block already contains the same complete JSON value. Text summaries and media do not replace structured records. +Kimi Code CLI preserves embedded MCP attachments that cannot be delivered directly because of format or size limits. Embedded images, audio, and video are saved even when they can be delivered unchanged, because provider conversion or later history reduction may omit them. Session-attachment readers remain available without workspace filesystem access when the model supports the corresponding content. Originals are retained in the session's media storage instead of an evictable image cache. Saved originals, including images preserved during compression, have absolute paths and stable `kimi-file://` references. Pass a reference as the `path` to `Read` or `ReadMediaFile`; bytes are read from the current session's storage even when the workspace runtime cannot access it. Pagination keeps the reference, including after a fork. For binary formats that `Read` cannot open, its error includes a server-local path when available; an external converter must have access to that filesystem. Text attachments such as CSV, HTML, JSON, and plain SVG use readable extensions. + +Attachment paths and compression details share the tool-output budget. Large lists are saved to a text file, with a short pointer that remains visible when accompanying text is shortened; the agent can pass the list’s `kimi-file://` reference to `Read` and page through it. Canceling the tool stops subsequent attachment processing and signals active writes. If decoding or saving fails, the result explicitly reports that the original could not be preserved while retaining other usable output. Resource links are not automatically downloaded. + ## Connection Methods Kimi Code CLI supports three MCP server connection methods: diff --git a/docs/zh/customization/mcp.md b/docs/zh/customization/mcp.md index 52ac7cf8ccc..bd1faf85440 100644 --- a/docs/zh/customization/mcp.md +++ b/docs/zh/customization/mcp.md @@ -4,6 +4,10 @@ MCP 工具结果可以包含文本(`content`)和结构化数据(`structuredContent`)。Kimi Code CLI 会将两者提供给 Agent,只有能够确认某个文本块已包含同一份完整 JSON 值时,才省略重复的结构化内容。文本摘要和媒体不会替代结构化记录。 +Kimi Code CLI 会保留因格式或大小限制而无法直接交付的内嵌 MCP 附件。内嵌图片、音频和视频即使能够原样交付也会保存,因为后续供应商协议转换或历史精简可能省略它们。模型支持相应内容时,即使工作区文件系统不可用,也仍可读取会话附件。原件随会话保存在媒体存储中,不会被图片缓存淘汰。保存的原件(包括图片压缩前的原图)均提供绝对路径和稳定的 `kimi-file://` 引用。将引用作为 `path` 传给 `Read` 或 `ReadMediaFile`,即使工作区 runtime 无法访问会话存储,也能直接从当前会话存储读取字节。分页续读会保留该引用,包括 fork 后的会话。对于 `Read` 无法打开的二进制格式,错误信息会在可用时提供服务端本地路径;外部转换工具必须能够访问该文件系统。CSV、HTML、JSON 和普通 SVG 等文本附件使用可读取的扩展名。 + +附件路径和压缩说明共用工具输出预算。较长的清单会保存为文本文件,结果中保留简短指针,即使伴随的文本被截短,该指针仍然可见;Agent 可将清单的 `kimi-file://` 引用传给 `Read`,分页读取完整内容。取消工具调用会停止后续附件处理,并通知正在进行的写入操作。如果解码或保存失败,结果会明确说明原件未能保留,并保留其他可用输出。资源链接不会被自动下载。 + ## 接入方式 Kimi Code CLI 支持三种 MCP server 接入方式: diff --git a/packages/agent-core-v2/src/_base/execEnv/decodeText.ts b/packages/agent-core-v2/src/_base/execEnv/decodeText.ts index 2f0a1290842..62bae9b58d1 100644 --- a/packages/agent-core-v2/src/_base/execEnv/decodeText.ts +++ b/packages/agent-core-v2/src/_base/execEnv/decodeText.ts @@ -1,5 +1,33 @@ export type TextDecodeErrors = 'strict' | 'replace' | 'ignore'; +export async function* readUtf8Lines( + source: AsyncIterable, + errors: TextDecodeErrors = 'strict', +): AsyncGenerator { + let pending: Buffer[] = []; + let offset = 0; + let pendingOffset = 0; + for await (const bytes of source) { + const chunk = Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength); + let start = 0; + for (let i = 0; i < chunk.length; i++) { + if (chunk[i] !== 0x0a) continue; + const piece = chunk.subarray(start, i + 1); + const lineOffset = pending.length === 0 ? offset + start : pendingOffset; + const line = pending.length === 0 ? piece : Buffer.concat([...pending, piece]); + yield decodeTextWithErrors(line, 'utf-8', errors, lineOffset !== 0); + pending = []; + start = i + 1; + } + if (start < chunk.length) { + if (pending.length === 0) pendingOffset = offset + start; + pending.push(Buffer.from(chunk.subarray(start))); + } + offset += chunk.length; + } + if (pending.length > 0) yield decodeTextWithErrors(Buffer.concat(pending), 'utf-8', errors, pendingOffset !== 0); +} + function isUtf8Continuation(byte: number): boolean { return byte >= 0x80 && byte <= 0xbf; } diff --git a/packages/agent-core-v2/src/_base/utils/fileMeta.ts b/packages/agent-core-v2/src/_base/utils/fileMeta.ts index 4f76835b0b7..4d3c9fcab88 100644 --- a/packages/agent-core-v2/src/_base/utils/fileMeta.ts +++ b/packages/agent-core-v2/src/_base/utils/fileMeta.ts @@ -65,6 +65,29 @@ export function guessMime(path: string, isBinary: boolean): string { return isBinary ? 'application/octet-stream' : 'text/plain'; } +const APPLICATION_TEXT_ALIASES: Readonly> = { + 'application/javascript': 'text/javascript', + 'application/x-javascript': 'text/javascript', + 'application/ecmascript': 'text/javascript', + 'application/yaml': 'text/yaml', + 'application/x-yaml': 'text/yaml', + 'application/sql': 'text/plain', + 'application/graphql': 'text/plain', + 'application/x-www-form-urlencoded': 'text/plain', +}; + +export function textExtensionForMime(mimeType: string): string | undefined { + const mime = mimeType.split(';')[0]!.trim().toLowerCase(); + if (mime === 'application/json' || mime.endsWith('+json')) return '.json'; + if (mime === 'application/xml' || mime.endsWith('+xml')) return '.xml'; + if (mime.endsWith('+yaml')) return '.yaml'; + if (mime === 'application/toml') return '.toml'; + if (mime === 'text/csv') return '.csv'; + const textMime = APPLICATION_TEXT_ALIASES[mime] ?? mime; + if (!textMime.startsWith('text/')) return undefined; + return Object.entries(EXT_TO_MIME).find(([, value]) => value === textMime)?.[0] ?? '.txt'; +} + const EXT_TO_LANGUAGE: Readonly> = { '.ts': 'typescript', '.tsx': 'typescriptreact', diff --git a/packages/agent-core-v2/src/agent/mcp/mcpService.ts b/packages/agent-core-v2/src/agent/mcp/mcpService.ts index 11bf08ca9cd..bddf19ee362 100644 --- a/packages/agent-core-v2/src/agent/mcp/mcpService.ts +++ b/packages/agent-core-v2/src/agent/mcp/mcpService.ts @@ -11,14 +11,13 @@ import { abortable } from '#/_base/utils/abort'; import { IAgentProfileService } from '#/agent/profile/profile'; import { IAgentStateService } from '#/agent/state/agentState'; import { ITelemetryService } from '#/app/telemetry/telemetry'; -import { sessionMediaOriginalsDir } from '#/agent/media/image-originals'; +import { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry'; import { IAgentLoopService } from '#/agent/loop/loop'; import { createMcpAuthTool } from '#/agent/mcp/tools/auth'; import { createMcpTool } from '#/agent/mcp/tools/mcp'; -import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { ISessionMcpHandle } from '#/session/mcp/sessionMcpHandle'; import type { McpServerEntry } from '#/mcpCore/connection-manager'; import { IAgentMcpService } from './mcp'; @@ -53,7 +52,6 @@ export class AgentMcpService extends Service implements IAgentMcpService { constructor( @ISessionMcpHandle private readonly mcpHandle: ISessionMcpHandle, - @ISessionContext private readonly sessionContext: ISessionContext, @IAgentToolRegistryService private readonly registry: IAgentToolRegistryService, @IAgentToolExecutorService toolExecutor: IAgentToolExecutorService, @IAgentLoopService loop: IAgentLoopService, @@ -62,6 +60,7 @@ export class AgentMcpService extends Service implements IAgentMcpService { @IAgentScopeContext private readonly scopeContext: IAgentScopeContext, @IAgentStateService private readonly states: IAgentStateService, @IAgentProfileService private readonly profile: IAgentProfileService, + @ISessionMediaStore private readonly attachmentStore: ISessionMediaStore, ) { super(); this.states.contributeState(mcpDiscoveryKey); @@ -286,7 +285,7 @@ export class AgentMcpService extends Service implements IAgentMcpService { const disposable = this._register( this.registry.register( createMcpTool(qualified, tool, client, { - originalsDir: sessionMediaOriginalsDir(this.sessionContext.sessionDir), + attachmentStore: this.attachmentStore, telemetry: this.telemetry, providerType: () => this.profile.getModelProviderType(), reconnect: (signal) => this.reconnectForToolCall(serverName, client, signal), diff --git a/packages/agent-core-v2/src/agent/mcp/output.ts b/packages/agent-core-v2/src/agent/mcp/output.ts index 16627b8a4a0..aa0e0f675ba 100644 --- a/packages/agent-core-v2/src/agent/mcp/output.ts +++ b/packages/agent-core-v2/src/agent/mcp/output.ts @@ -1,18 +1,28 @@ +import { createHash } from 'node:crypto'; +import { Readable } from 'node:stream'; import { isDeepStrictEqual } from 'node:util'; import type { ContentPart } from '#human/llm/message'; import type { ITelemetryService } from '#/app/telemetry/telemetry'; import type { ExecutableToolResult } from '#/tool/toolContract'; +import { textExtensionForMime } from '#/_base/utils/fileMeta'; -import { compressImageContentParts } from '#/agent/media/image-compress'; +import { compressImageContentParts, gateImageFormatParts } from '#/agent/media/image-compress'; import { buildUnsupportedImageNotice, isModelAcceptedImageMime, + parseImageDataUrl, + resolveEffectiveImageMime, + decodeBase64Prefix, } from '#/agent/media/image-format-policy'; import { persistOriginalImage } from '#/agent/media/image-originals'; +import type { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; +import { buildDaemonFileUrl, mediaExtensionForMime } from '#/agent/media/mediaRef'; import type { MCPContentBlock, MCPToolResult } from '#/mcpCore/types'; export interface McpOutputOptions { + readonly signal?: AbortSignal; + readonly attachmentStore?: ISessionMediaStore; readonly originalsDir?: string; readonly telemetry?: ITelemetryService; readonly providerType?: string; @@ -20,6 +30,7 @@ export interface McpOutputOptions { export const MCP_MAX_BINARY_PART_BYTES = 10 * 1024 * 1024; const MCP_MAX_BINARY_PART_CHARS = Math.ceil((MCP_MAX_BINARY_PART_BYTES * 4) / 3); +const MCP_MAX_INLINE_NOTICES_CHARS = 4096; function binaryPartTooLargeNotice(kind: 'image' | 'audio' | 'video', urlLength: number): string { const approxMb = ((urlLength * 3) / 4 / (1024 * 1024)).toFixed(1); @@ -115,9 +126,43 @@ export async function mcpResultToExecutableOutput( qualifiedToolName: string, options: McpOutputOptions = {}, ): Promise { + options.signal?.throwIfAborted(); const converted: ContentPart[] = []; + const attachmentNotices: string[] = []; + const preservedUrls = new Set(); + let omittedAttachment = false; + const preserveInlineMedia = async (url: string): Promise => { + options.signal?.throwIfAborted(); + if (preservedUrls.has(url)) return; + const parsed = parseImageDataUrl(url); + if (parsed === null) return; + preservedUrls.add(url); + const mime = parsed.mimeType.startsWith('image/') + ? resolveEffectiveImageMime(parsed.mimeType, decodeBase64Prefix(parsed.base64)) + : parsed.mimeType; + attachmentNotices.push(await preserveAttachment(parsed.base64, mime, options)); + }; for (const block of result.content) { - converted.push(convertMCPContentBlock(block, options.providerType)); + options.signal?.throwIfAborted(); + const part = convertMCPContentBlock(block, options.providerType); + if (part.type === 'image_url' && options.attachmentStore !== undefined) await preserveInlineMedia(part.imageUrl.url); + if (part.type === 'audio_url') await preserveInlineMedia(part.audioUrl.url); + if (part.type === 'video_url') await preserveInlineMedia(part.videoUrl.url); + const gated = gateImageFormatParts([part], options.providerType); + converted.push(...gated); + if (part.type === 'image_url' && gated[0]?.type === 'text') { + omittedAttachment = true; + await preserveInlineMedia(part.imageUrl.url); + } + if (part.type === 'text' && block.type === 'resource' && + typeof block.resource?.blob === 'string' && typeof block.resource.text !== 'string') { + omittedAttachment = true; + attachmentNotices.push(await preserveAttachment( + block.resource.blob, + block.resource.mimeType ?? 'application/octet-stream', + options, + )); + } } const wrapped = wrapMediaOnly(converted, qualifiedToolName); @@ -150,30 +195,126 @@ export async function mcpResultToExecutableOutput( } const compressed = await compressImageContentParts(wrapped, { + signal: options.signal, telemetry: options.telemetry, telemetrySource: 'mcp_tool_result', providerType: options.providerType, annotate: { - persistOriginal: (bytes, mimeType) => - persistOriginalImage( + persistOriginal: async (bytes, mimeType) => { + if (options.attachmentStore !== undefined) { + const saved = await saveAttachment(bytes, mimeType, options.attachmentStore, options.signal); + attachmentNotices.push(attachmentNotice(saved, mimeType, bytes.length)); + return saved.reference; + } + return persistOriginalImage( bytes, mimeType, options.originalsDir === undefined ? {} : { dir: options.originalsDir }, - ), + ); + }, }, }); - const capped = applyBinaryPartCap(compressed.parts); - const output = collapseSingleText(capped.parts); - const note = compressed.captions.length > 0 ? compressed.captions.join('\n') : undefined; + const capped = await applyBinaryPartCap(compressed.parts, preserveInlineMedia); + const notices = await attachmentDetails( + [...compressed.captions, ...attachmentNotices, ...capped.notices], options, + ); + const parts = [...capped.parts]; + if (notices.content.length > 0) parts.push({ type: 'text', text: notices.content }); + const output = collapseSingleText(parts); const base = { output, - note, - truncated: capped.truncated ? true : undefined, - spill: capped.notices.length > 0 ? { suffix: capped.notices.join('\n') } : undefined, + truncated: capped.truncated || omittedAttachment ? true : undefined, + spill: notices.suffix.length > 0 ? { suffix: notices.suffix } : undefined, }; return result.isError ? { ...base, isError: true } : base; } +async function attachmentDetails( + notices: readonly string[], + options: McpOutputOptions, +): Promise<{ readonly content: string; readonly suffix: string }> { + options.signal?.throwIfAborted(); + const content = [...new Set(notices)].join('\n'); + if (content.length <= MCP_MAX_INLINE_NOTICES_CHARS) return { content, suffix: content }; + try { + if (options.attachmentStore === undefined) throw new Error('Session attachment storage is unavailable'); + const saved = await saveAttachment(Buffer.from(content, 'utf8'), 'text/plain', options.attachmentStore, options.signal); + const pointer = [ + ...(saved.path === undefined ? [] : [`MCP attachment details saved at: ${JSON.stringify(saved.path)}`]), + `Attachment details reference: ${JSON.stringify(saved.reference)}`, + `Session-relative attachment details: ${JSON.stringify(saved.relativePath)}`, + 'Pass the attachment details reference to Read to retrieve all original attachment references and compression details in the current session.', + ].join('\n'); + return { content: pointer, suffix: pointer }; + } catch { + options.signal?.throwIfAborted(); + const suffix = 'The complete MCP attachment list could not be saved separately. Attachment details are included in the tool output and may be truncated. Do not repeat the MCP call automatically.'; + return { content: `${content}\n${suffix}`, suffix }; + } +} + +async function preserveAttachment( + base64: string, + mimeType: string, + options: McpOutputOptions, +): Promise { + options.signal?.throwIfAborted(); + try { + if (options.attachmentStore === undefined) throw new Error('Session attachment storage is unavailable'); + const compact = base64.replaceAll(/\s/g, ''); + const bytes = Buffer.from(compact, 'base64'); + const canonical = bytes.toString('base64'); + if (canonical !== compact && canonical.replace(/=+$/, '') !== compact) { + throw new Error('Invalid base64 attachment'); + } + const saved = await saveAttachment(bytes, mimeType, options.attachmentStore, options.signal); + return attachmentNotice(saved, mimeType, bytes.length); + } catch (error) { + options.signal?.throwIfAborted(); + return `Original attachment could not be saved (${JSON.stringify(mimeType)}): ${error instanceof Error ? error.message : String(error)}. No readable original path is available; original attachment preservation is incomplete. Do not repeat the MCP call automatically.`; + } +} + +interface SavedAttachment { + readonly path?: string; + readonly reference: string; + readonly relativePath: string; +} + +function attachmentNotice(saved: SavedAttachment, mimeType: string, size: number): string { + return [ + ...(saved.path === undefined ? [] : [`Original attachment saved at: ${JSON.stringify(saved.path)}`]), + `Attachment reference: ${JSON.stringify(saved.reference)}`, + `Session-relative attachment: ${JSON.stringify(saved.relativePath)}`, + `MIME: ${JSON.stringify(mimeType)}; size: ${String(size)} bytes. Pass the attachment reference to Read or ReadMediaFile in the current session. For other binary formats, Read reports the resolved local path for a converter.`, + ].join('\n'); +} + +async function saveAttachment( + bytes: Uint8Array, + mimeType: string, + store: ISessionMediaStore, + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted(); + const mime = mimeType.split(';')[0]!.trim().toLowerCase(); + const hash = createHash('sha256').update(mime).update('\0').update(bytes).digest('hex'); + const ext = mime === 'application/pdf' ? '.pdf' + : mime === 'image/svg+xml' ? (bytes[0] === 0x1f && bytes[1] === 0x8b ? '.svgz' : '.svg') + : textExtensionForMime(mime) ?? mediaExtensionForMime(mime) ?? '.bin'; + const fileId = `f_mcp_${hash}`; + const path = await store.materialize({ + fileId, + size: bytes.length, + name: `attachment${ext}`, + mimeType: mime, + stream: () => Readable.from([bytes]), + signal, + }); + signal?.throwIfAborted(); + return { path, reference: buildDaemonFileUrl(fileId), relativePath: `media/${fileId}${ext}` }; +} + function parseComparableJson(text: string): unknown { return JSON.parse(text, (_key: string, value: unknown, context?: { source?: string }) => { if (typeof value === 'number' && context?.source !== JSON.stringify(value)) { @@ -226,11 +367,14 @@ function wrapMediaOnly(parts: readonly ContentPart[], qualifiedToolName: string) ]; } -function applyBinaryPartCap(parts: readonly ContentPart[]): { +async function applyBinaryPartCap( + parts: readonly ContentPart[], + preserve: (url: string) => Promise, +): Promise<{ readonly parts: ContentPart[]; readonly truncated: boolean; readonly notices: string[]; -} { +}> { let truncated = false; const out: ContentPart[] = []; const notices: string[] = []; @@ -248,6 +392,7 @@ function applyBinaryPartCap(parts: readonly ContentPart[]): { ? part.audioUrl.url : part.videoUrl.url; if (url.length > MCP_MAX_BINARY_PART_CHARS) { + await preserve(url); const kind = part.type === 'image_url' ? 'image' : part.type === 'audio_url' ? 'audio' : 'video'; const notice = binaryPartTooLargeNotice(kind, url.length); diff --git a/packages/agent-core-v2/src/agent/mcp/tools/mcp.ts b/packages/agent-core-v2/src/agent/mcp/tools/mcp.ts index 974b6fbd70e..e8b2b7ffe7e 100644 --- a/packages/agent-core-v2/src/agent/mcp/tools/mcp.ts +++ b/packages/agent-core-v2/src/agent/mcp/tools/mcp.ts @@ -4,7 +4,7 @@ import { Error2, ErrorCodes, toErrorMessage } from '#/errors'; import { isAbortError } from '#/_base/utils/abort'; import type { ExecutableTool, ExecutableToolContext } from '#/tool/toolContract'; -import { mcpResultToExecutableOutput } from '#/agent/mcp/output'; +import { mcpResultToExecutableOutput, type McpOutputOptions } from '#/agent/mcp/output'; import type { MCPClient, MCPToolResult } from '#/mcpCore/types'; import { isMcpConnectionClosedError, @@ -14,6 +14,7 @@ import { } from '#/mcpCore/client-shared'; interface McpToolOptions { + readonly attachmentStore?: McpOutputOptions['attachmentStore']; readonly originalsDir?: string; readonly telemetry?: ITelemetryService; readonly providerType?: () => string | undefined; @@ -51,6 +52,8 @@ export function createMcpTool( result = await retryAfterReconnect(error, client, args, context, options, callTool); } return mcpResultToExecutableOutput(result, qualifiedName, { + signal: context.signal, + attachmentStore: options.attachmentStore, originalsDir: options.originalsDir, telemetry: options.telemetry, providerType: options.providerType?.(), diff --git a/packages/agent-core-v2/src/agent/media/image-compress.ts b/packages/agent-core-v2/src/agent/media/image-compress.ts index 552ff58c586..0c877306f50 100644 --- a/packages/agent-core-v2/src/agent/media/image-compress.ts +++ b/packages/agent-core-v2/src/agent/media/image-compress.ts @@ -327,16 +327,20 @@ export async function compressImageContentParts( options: CompressImageOptions & { readonly annotate?: CompressAnnotateOptions; readonly providerType?: string; + readonly signal?: AbortSignal; } = {}, ): Promise { - const { annotate, providerType, ...compressOptions } = options; + const { annotate, providerType, signal, ...compressOptions } = options; + signal?.throwIfAborted(); const out: ContentPart[] = []; const captions: string[] = []; for (const part of gateImageFormatParts(parts, providerType)) { + signal?.throwIfAborted(); if (part.type === 'image_url') { const parsed = parseImageDataUrl(part.imageUrl.url); if (parsed !== null) { const result = await compressBase64ForModel(parsed.base64, parsed.mimeType, compressOptions); + signal?.throwIfAborted(); if (result.changed) { if (annotate !== undefined) { let originalPath: string | null = null; @@ -347,6 +351,7 @@ export async function compressImageContentParts( parsed.mimeType, ); } catch { + signal?.throwIfAborted(); originalPath = null; } } diff --git a/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts b/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts index 12f49e90d1f..b076a36bb07 100644 --- a/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts +++ b/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts @@ -18,6 +18,7 @@ import { extendWorkspaceWithSkillRoots } from '#/tool/path-access'; import { IAgentMediaToolsRegistrar } from './mediaTools'; import { createVideoUploader, registerMediaTools } from './registerMediaTools'; +import { ISessionMediaStore } from './sessionMediaStore'; export const mediaRegisteredKeyKey = defineState( 'media.registeredKey', @@ -39,6 +40,7 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool @ITelemetryService private readonly telemetry: ITelemetryService, @IAgentStateService private readonly states: IAgentStateService, @ISessionSkillCatalog private readonly skillCatalog?: ISessionSkillCatalog, + @ISessionMediaStore private readonly attachmentStore?: ISessionMediaStore, ) { super(); this.states.contributeState(mediaRegisteredKeyKey); @@ -68,7 +70,8 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool private refresh(): void { const capabilities = this.profile.getModelCapabilities(); const modelAlias = this.profile.getModel(); - if (!this.runtime.isAvailable(['fs'])) { + const hasRuntimeFs = this.runtime.isAvailable(['fs']); + if (!hasRuntimeFs && this.attachmentStore === undefined) { const key = [ modelAlias, String(capabilities.image_in), @@ -81,8 +84,8 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool this.registration = undefined; return; } - const inspected = this.runtime.inspect(); - const identityKey = [ + const inspected = hasRuntimeFs ? this.runtime.inspect() : undefined; + const identityKey = inspected === undefined ? 'session-attachments' : [ inspected.identity.workspaceId, inspected.identity.runtimeId, inspected.identity.generation, @@ -95,9 +98,9 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool String(capabilities.image_in), String(capabilities.video_in), identityKey, - inspected.status, - inspected.environment.pathClass, - String(inspected.capabilities.has('fs')), + inspected?.status, + inspected?.environment.pathClass, + String(hasRuntimeFs), ].join('|'); if (key === this.registeredKey) return; this.registeredKey = key; @@ -105,7 +108,7 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool const workspaceCtx = this.workspaceCtx; const skillCatalog = this.skillCatalog; const runtime = this.runtime; - const pathClass = inspected.environment.pathClass; + const pathClass = inspected?.environment.pathClass; let requester: ModelRequester | undefined; if (model !== undefined) { try { @@ -115,6 +118,7 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool } } this.registration = registerMediaTools(this.toolRegistry, { + attachmentStore: this.attachmentStore, runtime, workspace: { get workspaceDir() { diff --git a/packages/agent-core-v2/src/agent/media/registerMediaTools.ts b/packages/agent-core-v2/src/agent/media/registerMediaTools.ts index 6254ad011df..b5f9d048b73 100644 --- a/packages/agent-core-v2/src/agent/media/registerMediaTools.ts +++ b/packages/agent-core-v2/src/agent/media/registerMediaTools.ts @@ -2,6 +2,7 @@ import type { ModelCapability } from '#human/llm/capability'; import type { ModelRequester } from '#/llm-adapter/model/model-requester'; import type { VideoUploadEvent } from '#/app/telemetry/events'; import type { ITelemetryService } from '#/app/telemetry/telemetry'; +import type { ISessionMediaStore } from './sessionMediaStore'; import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; import type { WorkspaceConfig } from '#/tool/path-access'; @@ -11,6 +12,7 @@ import { ReadMediaFileTool } from '#/agent/tools/read-media-file/readMediaFileTo import type { VideoUploader } from '#/agent/tools/read-media-file/read-media-file'; export interface RegisterMediaToolsDeps { + readonly attachmentStore?: ISessionMediaStore; readonly runtime: IAgentRuntimeService; readonly workspace: WorkspaceConfig; readonly capabilities: ModelCapability; @@ -25,7 +27,7 @@ export function registerMediaTools( deps: RegisterMediaToolsDeps, ): IDisposable { if ( - !deps.runtime.isAvailable(['fs']) || + (!deps.runtime.isAvailable(['fs']) && deps.attachmentStore === undefined) || (!deps.capabilities.image_in && !deps.capabilities.video_in) ) { return toDisposable(() => {}); @@ -39,6 +41,7 @@ export function registerMediaTools( deps.telemetry, deps.inlineVideoSupported, deps.providerType, + deps.attachmentStore, ), ); } diff --git a/packages/agent-core-v2/src/agent/tools/fileReadSource.ts b/packages/agent-core-v2/src/agent/tools/fileReadSource.ts new file mode 100644 index 00000000000..f6b8026d976 --- /dev/null +++ b/packages/agent-core-v2/src/agent/tools/fileReadSource.ts @@ -0,0 +1,64 @@ +import { readUtf8Lines } from '#/_base/execEnv/decodeText'; +import type { HostFileStat, IHostFileSystem } from '#/os/interface/hostFileSystem'; +import { parseDaemonFileUrl } from '#/agent/media/mediaRef'; +import type { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; +import type { ExecutableToolResult } from '#/tool/toolContract'; + +export interface FileReadSource { + readonly name: string; + readonly localPath?: string; + stat(): Promise; + readBytes(n?: number): Promise; + readLines(): AsyncIterable; +} + +export function withAttachmentLocation(result: ExecutableToolResult, source: FileReadSource): ExecutableToolResult { + if (!result.isError || source.localPath === undefined || typeof result.output !== 'string') return result; + return { ...result, output: `${result.output}\nServer-local attachment path: ${JSON.stringify(source.localPath)}` }; +} + +export function runtimeFileSource(fs: IHostFileSystem, path: string): FileReadSource { + return { + name: path, + stat: () => fs.stat(path), + readBytes: (n) => fs.readBytes(path, n), + readLines: () => fs.readLines(path, { errors: 'strict' }), + }; +} + +export async function attachmentFileSource(reference: string, store?: ISessionMediaStore): Promise { + const ref = parseDaemonFileUrl(reference); + const open = async () => { + const file = ref === undefined ? undefined : await store?.open(ref.fileId); + if (file === undefined) throw new Error(`Attachment ${JSON.stringify(reference)} is not available in the current session.`); + return file; + }; + const initial = await open(); + return { + name: initial.name, + localPath: initial.path, + stat: async () => ({ isFile: true, isDirectory: false, size: (await open()).size }), + readBytes: async (n) => { + const file = await open(); + const size = Math.min(n ?? file.size, file.size); + if (size === 0) return new Uint8Array(); + const chunks: Buffer[] = []; + for await (const chunk of file.stream({ start: 0, end: size - 1 })) chunks.push(Buffer.from(chunk)); + const bytes = Buffer.concat(chunks); + if (bytes.length !== size) throw new Error('Attachment changed or became unavailable while reading.'); + return bytes; + }, + readLines: async function* () { + const file = await open(); + const checkedStream = async function* () { + let size = 0; + for await (const chunk of file.stream()) { + size += chunk.length; + yield chunk; + } + if (size !== file.size) throw new Error('Attachment changed or became unavailable while reading.'); + }; + yield* readUtf8Lines(checkedStream()); + }, + }; +} diff --git a/packages/agent-core-v2/src/agent/tools/os/read/read.md b/packages/agent-core-v2/src/agent/tools/os/read/read.md index d3ab6c58a06..177d26441ce 100644 --- a/packages/agent-core-v2/src/agent/tools/os/read/read.md +++ b/packages/agent-core-v2/src/agent/tools/os/read/read.md @@ -1,5 +1,7 @@ Read a text file from the local filesystem. +The path may be a `kimi-file://` attachment reference. Its bytes come from the current session's storage, independently of the workspace runtime. Next Read keeps the reference so pagination also works after a fork. For a binary attachment, the error includes a server-local path when available; a converter must be able to access that filesystem. ReadMediaFile accepts the same reference for images and videos. + If the user provides a concrete file path to a text file, call Read directly. Do not `Glob`, `ls`, or otherwise pre-check known text file paths; missing or invalid file paths return errors you can handle. Do not use Read for directories; use `ls` via Bash for a known directory, or Glob when you need files matching a name pattern (Glob lists files only, never directories). Use `Grep` only when the task is to search for unknown content or locations. When you need several files, prefer to read them in parallel: emit multiple `Read` calls in a single response instead of reading one file per turn. diff --git a/packages/agent-core-v2/src/agent/tools/os/read/read.ts b/packages/agent-core-v2/src/agent/tools/os/read/read.ts index 05262569a30..060f82571c8 100644 --- a/packages/agent-core-v2/src/agent/tools/os/read/read.ts +++ b/packages/agent-core-v2/src/agent/tools/os/read/read.ts @@ -15,7 +15,7 @@ export const ReadInputSchema = z.object({ path: z .string() .describe( - 'Path to a text file. Relative paths resolve against the working directory; a path outside the working directory must be absolute. Directories are not supported; use `ls` via Bash for a known directory, or Glob for pattern search.', + 'Path to a text file or a kimi-file:// attachment reference in the current session. Relative filesystem paths resolve against the working directory; a path outside the working directory must be absolute. Directories are not supported; use `ls` via Bash for a known directory, or Glob for pattern search.', ), line_offset: z .union([PositiveLineOffsetSchema, TailLineOffsetSchema]) diff --git a/packages/agent-core-v2/src/agent/tools/os/read/readTool.ts b/packages/agent-core-v2/src/agent/tools/os/read/readTool.ts index 9c63700a94b..8f3b98b4282 100644 --- a/packages/agent-core-v2/src/agent/tools/os/read/readTool.ts +++ b/packages/agent-core-v2/src/agent/tools/os/read/readTool.ts @@ -1,5 +1,8 @@ import type { IHostFileSystem } from '#/os/interface/hostFileSystem'; import { IAgentRuntimeService, inspectAgentRuntime } from '#/agent/runtimeBinding/agentRuntime'; +import { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; +import { isDaemonFileUrl } from '#/agent/media/mediaRef'; +import { attachmentFileSource, runtimeFileSource, withAttachmentLocation, type FileReadSource } from '#/agent/tools/fileReadSource'; import { RuntimeWorkspaceView } from '#/runtime/runtimeWorkspaceView'; import { unwrapErrorCause } from '#/_base/errors/errors'; import { ISessionSkillCatalog } from '#/features/skill/session/skillCatalog'; @@ -18,7 +21,7 @@ import { } from '#/tool/path-access'; import { MEDIA_SNIFF_BYTES, detectFileType } from '#/agent/media/file-type'; import { toInputJsonSchema } from '#/tool/input-schema'; -import { literalRulePattern, matchesPathRuleSubject } from '#/tool/rule-match'; +import { literalRulePattern, matchesGlobRuleSubject, matchesPathRuleSubject } from '#/tool/rule-match'; import { makeCarriageReturnsVisible, splitLinesKeepingTerminator, type LineEndingStyle } from '#/_base/text/line-endings'; import { detectTextEncoding, type UtfTextEncoding } from '#/_base/text/encoding'; import { renderPrompt } from '#/_base/utils/render-prompt'; @@ -175,6 +178,7 @@ export class ReadTool implements IReadTool { @ISessionSkillCatalog private readonly skillCatalog: ISessionSkillCatalog, @IAgentToolResultTruncationService private readonly resultTruncation: IAgentToolResultTruncationService, @IConfigService private readonly config: IConfigService, + @ISessionMediaStore private readonly attachmentStore?: ISessionMediaStore, ) {} private limits(): { defaultMaxChars: number; maxChars: number } { @@ -190,10 +194,11 @@ export class ReadTool implements IReadTool { return { workspaceDir: view.workDir, additionalDirs: view.additionalDirs }; } - resolveExecution(args: ReadInput): ToolExecution { + resolveExecution(args: ReadInput): ToolExecution | Promise { if (args.column_offset !== undefined && (args.line_offset ?? 1) < 0) { return { isError: true, output: 'column_offset is only supported for forward reads. Use a positive line_offset or the forward Next Read arguments.' }; } + if (isDaemonFileUrl(args.path)) return this.attachmentExecution(args); const inspected = inspectAgentRuntime(this.runtime); const view = new RuntimeWorkspaceView(inspected, { workDir: this.workspaceCtx.workDir, @@ -224,7 +229,7 @@ export class ReadTool implements IReadTool { return { isError: true, output: 'Runtime changed before execution. Retry the tool call.' }; } const eventLog = this.resultTruncation.isWireJournalPath(path); - const result = await this.execution(lease.runtime.fs!, args, path, eventLog); + const result = await this.execution(runtimeFileSource(lease.runtime.fs!, path), args, eventLog); return { ...result, spillExempt: true }; } finally { lease.dispose(); @@ -233,16 +238,30 @@ export class ReadTool implements IReadTool { }; } + private async attachmentExecution(args: ReadInput): Promise { + const source = await attachmentFileSource(args.path, this.attachmentStore); + return { + accesses: ToolAccesses.readFile(source.localPath ?? args.path), + description: `Reading ${args.path}`, + display: { kind: 'file_io', operation: 'read', path: source.localPath ?? args.path }, + approvalRule: literalRulePattern(this.name, args.path), + matchesRule: (ruleArgs) => matchesGlobRuleSubject(ruleArgs, args.path), + execute: async () => ({ + ...withAttachmentLocation(await this.execution(source, args, false), source), + spillExempt: true, + }), + }; + } + private async execution( - fs: IHostFileSystem, + source: FileReadSource, args: ReadInput, - safePath: string, eventLog: boolean, ): Promise { try { let stat: Awaited>; try { - stat = await fs.stat(safePath); + stat = await source.stat(); } catch (error) { if (isFileNotFoundError(error)) { return { isError: true, output: `"${args.path}" does not exist.` }; @@ -253,8 +272,8 @@ export class ReadTool implements IReadTool { return { isError: true, output: `"${args.path}" is not a file.` }; } - const header = await fs.readBytes(safePath, MEDIA_SNIFF_BYTES); - const fileType = detectFileType(safePath, header); + const header = await source.readBytes(MEDIA_SNIFF_BYTES); + const fileType = detectFileType(source.name, header); if (fileType.kind === 'image' || fileType.kind === 'video') { return { isError: true, @@ -276,7 +295,7 @@ export class ReadTool implements IReadTool { 'Convert it to UTF-8 first (e.g. with `iconv`).', }; } - const bytes = await fs.readBytes(safePath); + const bytes = await source.readBytes(); let decoded: string; try { decoded = new TextDecoder(detection.encoding, { fatal: true }).decode(bytes); @@ -294,7 +313,7 @@ export class ReadTool implements IReadTool { output: notReadableFileOutput(args.path), }; } else { - readLines = () => fs.readLines(safePath, { errors: 'strict' }); + readLines = () => source.readLines(); } const limits = this.limits(); @@ -311,7 +330,7 @@ export class ReadTool implements IReadTool { const rereadsFile = detectedEncoding === undefined && (args.n_lines ?? Infinity) < -lineOffset; const result = await this.readTail(readLines, request); if (!result.isError && rereadsFile) { - const currentStat = await fs.stat(safePath); + const currentStat = await source.stat(); if (!currentStat.isFile || currentStat.size !== stat.size || currentStat.mtimeMs !== stat.mtimeMs || currentStat.ino !== stat.ino) { return { isError: true, output: 'File changed while reading its tail. Retry Read with the updated file.' }; @@ -579,5 +598,4 @@ export class ReadTool implements IReadTool { registerAgentToolService(IReadTool, ReadTool, { name: 'Read', domain: 'os/backends', - requiredRuntimeCapabilities: ['fs'], }); diff --git a/packages/agent-core-v2/src/agent/tools/read-media-file/read-media-file.ts b/packages/agent-core-v2/src/agent/tools/read-media-file/read-media-file.ts index a3bdda43889..f40fc0716c2 100644 --- a/packages/agent-core-v2/src/agent/tools/read-media-file/read-media-file.ts +++ b/packages/agent-core-v2/src/agent/tools/read-media-file/read-media-file.ts @@ -17,7 +17,7 @@ export const ReadMediaFileInputSchema = z.object({ path: z .string() .describe( - 'Path to an image or video file. Relative paths resolve against the working directory; ' + + 'Path to an image or video file, or a kimi-file:// attachment reference in the current session. Relative filesystem paths resolve against the working directory; ' + 'a path outside the working directory must be absolute. ' + 'Directories and text files are not supported.', ), diff --git a/packages/agent-core-v2/src/agent/tools/read-media-file/read-media.md b/packages/agent-core-v2/src/agent/tools/read-media-file/read-media.md index 2e577989c48..5a3b43552e9 100644 --- a/packages/agent-core-v2/src/agent/tools/read-media-file/read-media.md +++ b/packages/agent-core-v2/src/agent/tools/read-media-file/read-media.md @@ -1,5 +1,7 @@ Read media content from a file. +The path may be a `kimi-file://` attachment reference. Its bytes come from the current session's storage, independently of the workspace runtime, including after a fork. Any reported local attachment path belongs to the server; external converters must be able to access that filesystem. + **Tips:** - Make sure you follow the description of each tool parameter. - A `` tag accompanies the media content; it summarizes the mime type, byte size and, for images, the original pixel dimensions, and states how the image was delivered (untouched, downsampled, cropped, or native resolution). When outputting coordinates, give relative coordinates first and compute absolute coordinates from the original image size. After generating or editing media via commands or scripts, read the result back before continuing. diff --git a/packages/agent-core-v2/src/agent/tools/read-media-file/readMediaFileTool.ts b/packages/agent-core-v2/src/agent/tools/read-media-file/readMediaFileTool.ts index e5308966f04..fff7ef5bae4 100644 --- a/packages/agent-core-v2/src/agent/tools/read-media-file/readMediaFileTool.ts +++ b/packages/agent-core-v2/src/agent/tools/read-media-file/readMediaFileTool.ts @@ -3,8 +3,10 @@ import type { ContentPart } from '#human/llm/message'; import { VideoUploadUnsupportedError } from '#/llm-adapter/contract/errors'; import { inlineVideoPart, isVideoUploadAuthError } from '#/agent/media/videoUpload'; import type { ITelemetryService } from '#/app/telemetry/telemetry'; +import type { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; +import { isDaemonFileUrl } from '#/agent/media/mediaRef'; +import { attachmentFileSource, runtimeFileSource, withAttachmentLocation, type FileReadSource } from '#/agent/tools/fileReadSource'; -import type { IHostFileSystem } from '#/os/interface/hostFileSystem'; import { RuntimeWorkspaceView } from '#/runtime/runtimeWorkspaceView'; import type { HostEnvironmentInfo } from '#/os/interface/hostEnvironment'; import { inspectAgentRuntime, type IAgentRuntimeService } from '#/agent/runtimeBinding/agentRuntime'; @@ -37,7 +39,7 @@ import { } from '#/agent/media/image-format-policy'; import { providerImagePolicy } from '#human/llm/media/image-formats'; import { toInputJsonSchema } from '#/tool/input-schema'; -import { literalRulePattern, matchesPathRuleSubject } from '#/tool/rule-match'; +import { literalRulePattern, matchesGlobRuleSubject, matchesPathRuleSubject } from '#/tool/rule-match'; import { renderPrompt } from '#/_base/utils/render-prompt'; import { MAX_MEDIA_BYTES, @@ -194,6 +196,7 @@ export class ReadMediaFileTool implements AgentTool { telemetry?: ITelemetryService, inlineVideoSupported?: boolean, providerType?: string, + private readonly attachmentStore?: ISessionMediaStore, ) { this.description = buildDescription(capabilities); this.telemetry = telemetry; @@ -221,10 +224,13 @@ export class ReadMediaFileTool implements AgentTool { return inlineVideoPart(data, mimeType); } - resolveExecution(args: ReadMediaFileInput): ToolExecution { + resolveExecution(args: ReadMediaFileInput): ToolExecution | Promise { if (!args.path) { return { isError: true, output: 'File path cannot be empty.' }; } + if (isDaemonFileUrl(args.path)) { + return this.attachmentExecution(args); + } const inspected = inspectAgentRuntime(this.runtime); const env = inspected.environment; const view = new RuntimeWorkspaceView(inspected, { @@ -254,7 +260,7 @@ export class ReadMediaFileTool implements AgentTool { if (lease.runtime.identity.generation !== inspected.identity.generation) { return { isError: true, output: 'Runtime changed before execution. Retry the tool call.' }; } - return await this.execution(args, path, lease.runtime.fs!, env); + return await this.execution(args, runtimeFileSource(lease.runtime.fs!, path), env); } finally { lease.dispose(); } @@ -262,18 +268,32 @@ export class ReadMediaFileTool implements AgentTool { }; } + private async attachmentExecution(args: ReadMediaFileInput): Promise { + const source = await attachmentFileSource(args.path, this.attachmentStore); + return { + accesses: ToolAccesses.readFile(source.localPath ?? args.path), + description: `Reading media: ${args.path}`, + display: { kind: 'file_io', operation: 'read', path: source.localPath ?? args.path }, + approvalRule: literalRulePattern(this.name, args.path), + matchesRule: (ruleArgs) => matchesGlobRuleSubject(ruleArgs, args.path), + execute: async () => withAttachmentLocation( + await this.execution(args, source, { osKind: 'unknown' }), source, + ), + }; + } + private async execution( args: ReadMediaFileInput, - safePath: string, - fs: IHostFileSystem, - env: HostEnvironmentInfo, + source: FileReadSource, + env: Pick, ): Promise { if (!args.path) { return { isError: true, output: 'File path cannot be empty.' }; } try { - const header = await fs.readBytes(safePath, MEDIA_SNIFF_BYTES); + const safePath = source.name; + const header = await source.readBytes(MEDIA_SNIFF_BYTES); const fileType = detectFileType(safePath, header, 'media'); if (fileType.kind === 'text') { @@ -305,7 +325,7 @@ export class ReadMediaFileTool implements AgentTool { ) { return { isError: true, - output: buildImageConversionGuidance(args.path, fileType.mimeType, env.osKind), + output: buildImageConversionGuidance(source.localPath ?? args.path, fileType.mimeType, env.osKind), }; } if (fileType.kind === 'video' && !this.capabilities.video_in) { @@ -317,7 +337,7 @@ export class ReadMediaFileTool implements AgentTool { }; } - const stat = await fs.stat(safePath); + const stat = await source.stat(); if (stat.size === 0) { return { isError: true, output: `"${args.path}" is empty.` }; } @@ -380,7 +400,7 @@ export class ReadMediaFileTool implements AgentTool { }; } - const data = Buffer.from(await fs.readBytes(safePath)); + const data = Buffer.from(await source.readBytes()); let dimensions = fileType.kind === 'image' ? sniffImageDimensions(data) : null; let mediaPart: ContentPart; let delivery: ImageDelivery | undefined; @@ -447,7 +467,7 @@ export class ReadMediaFileTool implements AgentTool { return { isError: true, output: buildOversizedImageConversionGuidance( - args.path, + source.localPath ?? args.path, fileType.mimeType, env.osKind, compressed.finalByteLength, @@ -489,7 +509,8 @@ export class ReadMediaFileTool implements AgentTool { } const tag = fileType.kind === 'image' ? 'image' : 'video'; - const openText = `<${tag} path="${safePath}">`; + const tagPath = isDaemonFileUrl(args.path) ? args.path : safePath; + const openText = `<${tag} path="${tagPath}">`; const closeText = ``; const note = buildMediaNote({ diff --git a/packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts b/packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts index 9591a2e5b99..38555dae335 100644 --- a/packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts +++ b/packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts @@ -1,3 +1,4 @@ +import { createReadStream } from 'node:fs'; import { appendFile, lstat, @@ -12,7 +13,7 @@ import { } from 'node:fs/promises'; import { LifecycleScope } from '#/app/scopes'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; -import { decodeTextWithErrors, type TextDecodeErrors } from '#/_base/execEnv/decodeText'; +import { decodeTextWithErrors, readUtf8Lines, type TextDecodeErrors } from '#/_base/execEnv/decodeText'; import { type HostDirEntry, type HostFileStat, IHostFileSystem } from '#/os/interface/hostFileSystem'; import { toHostFsError } from '#/os/interface/hostFsErrors'; @@ -114,57 +115,12 @@ export class HostFileSystem implements IHostFileSystem { return; } - yield* this._readUtf8Lines(path, errors); + yield* readUtf8Lines(createReadStream(path, { highWaterMark: READ_CHUNK_SIZE }), errors); } catch (error) { throw toHostFsError(error, { path, op: 'read' }); } } - private async *_readUtf8Lines( - path: string, - errors: TextDecodeErrors, - ): AsyncGenerator { - const fh = await open(path, 'r'); - try { - const buf = Buffer.alloc(READ_CHUNK_SIZE); - let pending: Buffer[] = []; - let pendingOffset = 0; - let fileOffset = 0; - - while (true) { - const { bytesRead } = await fh.read(buf, 0, buf.length, null); - if (bytesRead === 0) break; - const chunk = buf.subarray(0, bytesRead); - let lineStart = 0; - - for (let i = 0; i < chunk.length; i += 1) { - const byte = chunk[i]; - if (byte !== 0x0a) continue; - const piece = chunk.subarray(lineStart, i + 1); - const lineOffset = pending.length === 0 ? fileOffset + lineStart : pendingOffset; - const line = pending.length === 0 ? piece : Buffer.concat([...pending, piece]); - yield decodeTextWithErrors(line, 'utf-8', errors, lineOffset !== 0); - pending = []; - lineStart = i + 1; - } - - if (lineStart < chunk.length) { - const tail = Buffer.from(chunk.subarray(lineStart)); - if (pending.length === 0) pendingOffset = fileOffset + lineStart; - pending.push(tail); - } - fileOffset += bytesRead; - } - - if (pending.length > 0) { - const line = Buffer.concat(pending); - yield decodeTextWithErrors(line, 'utf-8', errors, pendingOffset !== 0); - } - } finally { - await fh.close(); - } - } - async createExclusive(path: string, data: Uint8Array): Promise { try { const fh = await open(path, 'wx'); diff --git a/packages/agent-core-v2/test/agent/loop/loop.test.ts b/packages/agent-core-v2/test/agent/loop/loop.test.ts index 45102c44de9..3a10f5c824e 100644 --- a/packages/agent-core-v2/test/agent/loop/loop.test.ts +++ b/packages/agent-core-v2/test/agent/loop/loop.test.ts @@ -142,8 +142,8 @@ describe('Agent loop', () => { [emit] turn.step.started { "time": "