perf(agent-runtime): avoid full thread reads for resume checks

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);

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<boolean> {
+    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

core/agent-runtime/src/OSSObjectStorageClient.ts

@@ -53,6 +53,41 @@ export class OSSObjectStorageClient implements ObjectStorageClient {
     }
   }
 
+  async getRange(key: string, start: number, end: number): Promise<string | null> {
+    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.
    *

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<AgentMessage> {
         throw new Error('exec failed');

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<string | null> {
+          this.fullReads.push(key);
+          return await super.get(key);
+        }
+
+        override async getRange(key: string, start: number, end: number): Promise<string | null> {
+          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<string | null> {
+          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)', () => {

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' });

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<string | null> {
+    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<void> {
     const existing = this.store.get(key) ?? '';
     this.store.set(key, existing + value);

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",

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: {

core/types/agent-runtime/AgentStore.ts

@@ -84,6 +84,16 @@ export interface AgentStore {
   destroy?(): Promise<void>;
   createThread(metadata?: Record<string, unknown>): Promise<ThreadRecord>;
   getThread(threadId: string, options?: GetThreadOptions): Promise<ThreadRecord>;
+  /**
+   * 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<boolean>;
   /**
    * Shallow-merge metadata into an existing thread.
    * New values overwrite matching keys; omitted keys are preserved.

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<string | null>;
 
+  /**
+   * 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<string | null>;
+
   /**
    * Append `value` to an existing Appendable Object.
    * If the object does not exist yet, create it.