From 05fd924d0ec90b91f67afaa886fcee3750fdde3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=97=B0=E6=98=8E?= Date: Wed, 29 Jul 2026 15:16:51 +0800 Subject: [PATCH 1/3] perf(agent-runtime): avoid full thread reads for resume checks --- core/agent-runtime/src/AgentRuntime.ts | 5 +- core/agent-runtime/src/OSSAgentStore.ts | 54 +++++++++++++ .../src/OSSObjectStorageClient.ts | 35 ++++++++ core/agent-runtime/test/AgentRuntime.test.ts | 45 +++++++++++ core/agent-runtime/test/OSSAgentStore.test.ts | 81 +++++++++++++++++++ .../test/OSSObjectStorageClient.test.ts | 36 +++++++++ core/agent-runtime/test/helpers.ts | 6 ++ core/types/agent-runtime/AgentRuntime.ts | 4 +- core/types/agent-runtime/AgentStore.ts | 10 +++ .../agent-runtime/ObjectStorageClient.ts | 9 +++ 10 files changed, 281 insertions(+), 4 deletions(-) diff --git a/core/agent-runtime/src/AgentRuntime.ts b/core/agent-runtime/src/AgentRuntime.ts index bdb9bb69f..af2cea700 100644 --- a/core/agent-runtime/src/AgentRuntime.ts +++ b/core/agent-runtime/src/AgentRuntime.ts @@ -158,7 +158,9 @@ export class AgentRuntime { private async ensureThread(input: CreateRunInput): Promise<{ threadId: string; input: CreateRunInput }> { const metadata = validateMetadata(input.metadata); if (input.threadId) { - const thread = await this.store.getThread(input.threadId); + const isResume = this.store.hasMessages + ? await this.store.hasMessages(input.threadId) + : (await this.store.getThread(input.threadId)).messages.length > 0; if (metadata && Object.keys(metadata).length > 0) { if (!this.store.updateThreadMetadata) { throw new Error('AgentStore does not support updating thread metadata'); @@ -175,7 +177,6 @@ export class AgentRuntime { ); } } - const isResume = thread.messages.length > 0; return { threadId: input.threadId, input: { ...input, isResume } }; } const thread = await this.store.createThread(metadata); diff --git a/core/agent-runtime/src/OSSAgentStore.ts b/core/agent-runtime/src/OSSAgentStore.ts index 9cf21125b..f55b5a1c9 100644 --- a/core/agent-runtime/src/OSSAgentStore.ts +++ b/core/agent-runtime/src/OSSAgentStore.ts @@ -11,6 +11,28 @@ import { AgentObjectType, RunStatus, AgentNotFoundError } from '@eggjs/tegg-type import { dateBucket, newRunId, newThreadId, nowUnix, reverseMs } from './AgentStoreUtils'; +const HAS_MESSAGES_PREFIX_END = 64 * 1024 - 1; + +function isConversationMessage(message: AgentMessage): boolean { + return message.type === 'user' || message.type === 'assistant'; +} + +function containsConversationMessage(data: string, completeLinesOnly: boolean): boolean { + let parseable = data; + if (completeLinesOnly) { + const lastNewline = data.lastIndexOf('\n'); + if (lastNewline < 0) return false; + parseable = data.slice(0, lastNewline); + } else { + parseable = data.trim(); + } + + return parseable + .split('\n') + .filter(line => line.length > 0) + .some(line => isConversationMessage(JSON.parse(line) as AgentMessage)); +} + /** * Warn logger used when a background thread activity-index write fails. * `console` and `egg-logger` both satisfy this shape. @@ -249,6 +271,38 @@ export class OSSAgentStore implements AgentStore { return { ...meta, messages }; } + async hasMessages(threadId: string): Promise { + const messagesKey = this.threadMessagesKey(threadId); + const [ metaData, messagesPrefix ] = await Promise.all([ + this.client.get(this.threadMetaKey(threadId)), + this.client.getRange + ? this.client.getRange(messagesKey, 0, HAS_MESSAGES_PREFIX_END) + : Promise.resolve(undefined), + ]); + if (!metaData) { + throw new AgentNotFoundError(`Thread ${threadId} not found`); + } + // Preserve getThread's behavior for corrupt metadata even though this + // lightweight path only needs the thread's existence. + JSON.parse(metaData); + + if (messagesPrefix === null || messagesPrefix === '') return false; + if ( + messagesPrefix !== undefined && + containsConversationMessage(messagesPrefix, true) + ) { + return true; + } + + // A range can end in the middle of a JSONL record. Fall back to the full + // object when the prefix is inconclusive so large leading system events or + // legacy records without a trailing newline cannot produce false negatives. + const messagesData = await this.client.get(messagesKey); + return messagesData + ? containsConversationMessage(messagesData, false) + : false; + } + /** * Serialize read-modify-write operations on a single thread's `meta.json` * within this process. Both {@link updateThreadMetadata} (business metadata diff --git a/core/agent-runtime/src/OSSObjectStorageClient.ts b/core/agent-runtime/src/OSSObjectStorageClient.ts index 4dfbf5082..2acbb29e6 100644 --- a/core/agent-runtime/src/OSSObjectStorageClient.ts +++ b/core/agent-runtime/src/OSSObjectStorageClient.ts @@ -53,6 +53,41 @@ export class OSSObjectStorageClient implements ObjectStorageClient { } } + async getRange(key: string, start: number, end: number): Promise { + if ( + !Number.isSafeInteger(start) || + !Number.isSafeInteger(end) || + start < 0 || + end < start + ) { + throw new RangeError(`Invalid object byte range: ${start}-${end}`); + } + + try { + const result = await this.client.get(key, { + headers: { + range: `bytes=${start}-${end}`, + }, + }); + if (result.content === undefined || result.content === null) { + return ''; + } + return Buffer.isBuffer(result.content) + ? result.content.toString('utf-8') + : String(result.content); + } catch (err: unknown) { + if (isOSSError(err, 'NoSuchKey')) { + return null; + } + // OSS returns InvalidRange when reading from an empty object or when the + // requested start offset is beyond the end of the object. + if (isOSSError(err, 'InvalidRange')) { + return ''; + } + throw err; + } + } + /** * Append data to an OSS Appendable Object. * diff --git a/core/agent-runtime/test/AgentRuntime.test.ts b/core/agent-runtime/test/AgentRuntime.test.ts index 4ea8e2069..d5c6e13a2 100644 --- a/core/agent-runtime/test/AgentRuntime.test.ts +++ b/core/agent-runtime/test/AgentRuntime.test.ts @@ -293,6 +293,51 @@ describe('test/AgentRuntime.test.ts', () => { assert.equal(capturedInput!.isResume, true); }); + it('should use store.hasMessages without loading the full thread', async () => { + let hasMessagesCalls = 0; + let getThreadCalls = 0; + const originalHasMessages = store.hasMessages.bind(store); + const originalGetThread = store.getThread.bind(store); + store.hasMessages = async threadId => { + hasMessagesCalls++; + return await originalHasMessages(threadId); + }; + store.getThread = async (threadId, options) => { + getThreadCalls++; + return await originalGetThread(threadId, options); + }; + + const thread = await runtime.createThread(); + await runtime.syncRun({ + threadId: thread.id, + input: { messages: [{ role: 'user', content: 'Hi' }] }, + }); + + assert.equal(hasMessagesCalls, 1); + assert.equal(getThreadCalls, 0); + }); + + it('should fall back to getThread when store.hasMessages is unavailable', async () => { + let getThreadCalls = 0; + const originalGetThread = store.getThread.bind(store); + Object.defineProperty(store, 'hasMessages', { + configurable: true, + value: undefined, + }); + store.getThread = async (threadId, options) => { + getThreadCalls++; + return await originalGetThread(threadId, options); + }; + + const thread = await runtime.createThread(); + await runtime.syncRun({ + threadId: thread.id, + input: { messages: [{ role: 'user', content: 'Hi' }] }, + }); + + assert.equal(getThreadCalls, 1); + }); + it('should not throw when store.updateRun fails in catch block', async () => { executor.execRun = async function* (): AsyncGenerator { throw new Error('exec failed'); diff --git a/core/agent-runtime/test/OSSAgentStore.test.ts b/core/agent-runtime/test/OSSAgentStore.test.ts index 819739126..9d9ac8ade 100644 --- a/core/agent-runtime/test/OSSAgentStore.test.ts +++ b/core/agent-runtime/test/OSSAgentStore.test.ts @@ -170,6 +170,87 @@ describe('test/OSSAgentStore.test.ts', () => { assert.equal(fetched.messages[1].type, 'user'); assert.equal(fetched.messages[2].type, 'result'); }); + + it('should report whether a thread contains conversation messages', async () => { + const thread = await store.createThread(); + assert.equal(await store.hasMessages(thread.id), false); + + await store.appendMessages(thread.id, [ + { type: 'system', subtype: 'init', session_id: 'sess-1' }, + { type: 'result', subtype: 'success', usage: { input_tokens: 0, output_tokens: 0 } }, + ]); + assert.equal(await store.hasMessages(thread.id), false); + + await store.appendMessages(thread.id, [ + { type: 'user', message: { role: 'user', content: 'Hello' } }, + ]); + assert.equal(await store.hasMessages(thread.id), true); + }); + + it('should throw AgentNotFoundError when checking a missing thread', async () => { + await assert.rejects( + () => store.hasMessages('thread_non_existent'), + AgentNotFoundError, + ); + }); + + it('should use a range read without fetching the full message object when the prefix is decisive', async () => { + class TrackingStorageClient extends MapStorageClient { + readonly fullReads: string[] = []; + readonly rangeReads: string[] = []; + + override async get(key: string): Promise { + this.fullReads.push(key); + return await super.get(key); + } + + override async getRange(key: string, start: number, end: number): Promise { + this.rangeReads.push(key); + return await super.getRange(key, start, end); + } + } + + const client = new TrackingStorageClient(); + const localStore = new OSSAgentStore({ client }); + const thread = await localStore.createThread(); + await localStore.appendMessages(thread.id, [ + { type: 'user', message: { role: 'user', content: 'Hello' } }, + ]); + + client.fullReads.length = 0; + assert.equal(await localStore.hasMessages(thread.id), true); + + const messageKey = `threads/${thread.id}/messages.jsonl`; + assert.deepStrictEqual(client.rangeReads, [ messageKey ]); + assert(!client.fullReads.includes(messageKey)); + }); + + it('should fall back to a full read when the range ends inside a leading record', async () => { + class TrackingStorageClient extends MapStorageClient { + readonly fullReads: string[] = []; + + override async get(key: string): Promise { + this.fullReads.push(key); + return await super.get(key); + } + } + + const client = new TrackingStorageClient(); + const localStore = new OSSAgentStore({ client }); + const thread = await localStore.createThread(); + await localStore.appendMessages(thread.id, [ + { + type: 'system', + subtype: 'init', + session_id: 'x'.repeat(70 * 1024), + }, + { type: 'assistant', message: { role: 'assistant', content: [] } }, + ]); + + client.fullReads.length = 0; + assert.equal(await localStore.hasMessages(thread.id), true); + assert(client.fullReads.includes(`threads/${thread.id}/messages.jsonl`)); + }); }); describe('threads (without append)', () => { diff --git a/core/agent-runtime/test/OSSObjectStorageClient.test.ts b/core/agent-runtime/test/OSSObjectStorageClient.test.ts index 0f138e045..846e743f7 100644 --- a/core/agent-runtime/test/OSSObjectStorageClient.test.ts +++ b/core/agent-runtime/test/OSSObjectStorageClient.test.ts @@ -123,6 +123,42 @@ describe('test/OSSObjectStorageClient.test.ts', () => { }); }); + describe('getRange', () => { + it('should pass an inclusive byte range to the SDK', async () => { + mockOSS.get.mockResolvedValue({ + content: Buffer.from('partial', 'utf-8'), + }); + + const result = await client.getRange('threads/t1.json', 0, 63); + + assert.equal(result, 'partial'); + assert.deepStrictEqual(mockOSS.get.mock.calls[0], [ + 'threads/t1.json', + { headers: { range: 'bytes=0-63' } }, + ]); + }); + + it('should return null for NoSuchKey and an empty string for InvalidRange', async () => { + const missing = new Error('Object not exists'); + (missing as Error & { code: string }).code = 'NoSuchKey'; + mockOSS.get.mockRejectedValueOnce(missing); + assert.equal(await client.getRange('missing', 0, 63), null); + + const empty = new Error('Requested range is not satisfiable'); + (empty as Error & { code: string }).code = 'InvalidRange'; + mockOSS.get.mockRejectedValueOnce(empty); + assert.equal(await client.getRange('empty', 0, 63), ''); + }); + + it('should reject invalid byte ranges before calling the SDK', async () => { + await assert.rejects( + () => client.getRange('threads/t1.json', 10, 9), + RangeError, + ); + assert.equal(mockOSS.get.mock.calls.length, 0); + }); + }); + describe('append', () => { it('should create new object with position 0 on first append', async () => { mockOSS.append.mockResolvedValue({ nextAppendPosition: '13' }); diff --git a/core/agent-runtime/test/helpers.ts b/core/agent-runtime/test/helpers.ts index 2d461cede..8e0d204f3 100644 --- a/core/agent-runtime/test/helpers.ts +++ b/core/agent-runtime/test/helpers.ts @@ -64,6 +64,12 @@ export class MapStorageClient implements ObjectStorageClient { return this.store.get(key) ?? null; } + async getRange(key: string, start: number, end: number): Promise { + const value = this.store.get(key); + if (value === undefined) return null; + return Buffer.from(value, 'utf8').subarray(start, end + 1).toString('utf8'); + } + async append(key: string, value: string): Promise { const existing = this.store.get(key) ?? ''; this.store.set(key, existing + value); diff --git a/core/types/agent-runtime/AgentRuntime.ts b/core/types/agent-runtime/AgentRuntime.ts index 5a30f17db..b3bf68c25 100644 --- a/core/types/agent-runtime/AgentRuntime.ts +++ b/core/types/agent-runtime/AgentRuntime.ts @@ -52,8 +52,8 @@ export interface CreateRunInput { threadId?: string; /** * Populated by AgentRuntime before calling execRun. - * - true: threadId was provided (resume existing conversation) - * - false: no threadId provided, new thread created + * - true: the thread contains persisted user or assistant messages + * - false: the thread is new or does not contain conversation messages */ isResume?: boolean; input: { diff --git a/core/types/agent-runtime/AgentStore.ts b/core/types/agent-runtime/AgentStore.ts index dd9f8bdbb..5eeedacb0 100644 --- a/core/types/agent-runtime/AgentStore.ts +++ b/core/types/agent-runtime/AgentStore.ts @@ -84,6 +84,16 @@ export interface AgentStore { destroy?(): Promise; createThread(metadata?: Record): Promise; getThread(threadId: string, options?: GetThreadOptions): Promise; + /** + * Return whether the thread contains at least one conversation message + * (user or assistant), matching the default filtering semantics of + * {@link getThread}. + * + * Stores may implement this optional capability with a lightweight existence + * check. Implementations must throw `AgentNotFoundError` when the thread does + * not exist. AgentRuntime falls back to `getThread()` when it is absent. + */ + hasMessages?(threadId: string): Promise; /** * Shallow-merge metadata into an existing thread. * New values overwrite matching keys; omitted keys are preserved. diff --git a/core/types/agent-runtime/ObjectStorageClient.ts b/core/types/agent-runtime/ObjectStorageClient.ts index 3d766eb35..6055d995b 100644 --- a/core/types/agent-runtime/ObjectStorageClient.ts +++ b/core/types/agent-runtime/ObjectStorageClient.ts @@ -12,6 +12,15 @@ export interface ObjectStorageClient { /** Read the full object at `key`. Returns `null` if the object does not exist. */ get(key: string): Promise; + /** + * Read an inclusive byte range from the object at `key`. + * Returns `null` if the object does not exist. + * + * This method is optional. Callers must treat a trailing partial UTF-8 + * sequence or record as incomplete because `end` is a byte offset. + */ + getRange?(key: string, start: number, end: number): Promise; + /** * Append `value` to an existing Appendable Object. * If the object does not exist yet, create it. From 296a3444a64051d674309d99ebe115cedf6fba14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=97=B0=E6=98=8E?= Date: Wed, 29 Jul 2026 16:50:49 +0800 Subject: [PATCH 2/3] test(mcp-client): polyfill web crypto on node 18 --- .../test/fixtures/streamable-mcp-server/http.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/core/mcp-client/test/fixtures/streamable-mcp-server/http.ts b/core/mcp-client/test/fixtures/streamable-mcp-server/http.ts index 601f548f4..7284b1cb3 100644 --- a/core/mcp-client/test/fixtures/streamable-mcp-server/http.ts +++ b/core/mcp-client/test/fixtures/streamable-mcp-server/http.ts @@ -1,8 +1,18 @@ import http from 'node:http'; +import { webcrypto } from 'node:crypto'; import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; import * as z from 'zod/v4'; +// The MCP SDK uses the Web Crypto global for stream IDs. Node.js 18 exposes +// Web Crypto from node:crypto but does not install it globally in script files. +if (typeof globalThis.crypto === 'undefined') { + Object.defineProperty(globalThis, 'crypto', { + configurable: true, + value: webcrypto, + }); +} + // Create an MCP server const server = new McpServer({ name: "Demo", From 462ea7566c85d7d8f887c445ba66d020675c3850 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=97=B0=E6=98=8E?= Date: Wed, 29 Jul 2026 17:31:42 +0800 Subject: [PATCH 3/3] fix(controller): provide web crypto on node 18 --- .../controller/lib/impl/mcp/MCPControllerRegister.ts | 10 ++++++++++ .../test/fixtures/streamable-mcp-server/http.ts | 10 ++++++++++ 2 files changed, 20 insertions(+) diff --git a/plugin/controller/lib/impl/mcp/MCPControllerRegister.ts b/plugin/controller/lib/impl/mcp/MCPControllerRegister.ts index b18645e91..2c102130a 100644 --- a/plugin/controller/lib/impl/mcp/MCPControllerRegister.ts +++ b/plugin/controller/lib/impl/mcp/MCPControllerRegister.ts @@ -1,6 +1,7 @@ import type { Application, Context, Router } from 'egg'; import assert from 'node:assert'; +import { webcrypto } from 'node:crypto'; import http, { IncomingMessage, ServerResponse } from 'node:http'; import { Socket } from 'node:net'; @@ -33,6 +34,15 @@ import { MCPConfig } from './MCPConfig'; import { MCPServerHelper } from './MCPServerHelper'; import { Server } from '@modelcontextprotocol/sdk/server/index.js'; +// The MCP SDK uses the Web Crypto global for stream IDs. Node.js 18 exposes +// Web Crypto from node:crypto but does not install it globally in script files. +if (typeof globalThis.crypto === 'undefined') { + Object.defineProperty(globalThis, 'crypto', { + configurable: true, + value: webcrypto, + }); +} + export interface MCPControllerHook { // SSE preSSEInitHandle?: (ctx: Context, transport: SSEServerTransport, register: MCPControllerRegister) => Promise diff --git a/plugin/mcp-client/test/fixtures/streamable-mcp-server/http.ts b/plugin/mcp-client/test/fixtures/streamable-mcp-server/http.ts index 79e171cda..25a68e3e2 100644 --- a/plugin/mcp-client/test/fixtures/streamable-mcp-server/http.ts +++ b/plugin/mcp-client/test/fixtures/streamable-mcp-server/http.ts @@ -1,8 +1,18 @@ import http from 'node:http'; +import { webcrypto } from 'node:crypto'; import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; import * as z from 'zod/v4'; +// The MCP SDK uses the Web Crypto global for stream IDs. Node.js 18 exposes +// Web Crypto from node:crypto but does not install it globally in script files. +if (typeof globalThis.crypto === 'undefined') { + Object.defineProperty(globalThis, 'crypto', { + configurable: true, + value: webcrypto, + }); +} + // Create an MCP server const server = new McpServer({ name: 'Demo',