feat(agent-runtime): support run thread metadata

core/agent-runtime/src/AgentRuntime.ts

@@ -19,6 +19,7 @@ import {
   RunStatus,
   AgentObjectType,
   AgentConflictError,
+  AgentInvalidRequestError,
   AgentNotFoundError,
   AgentTimeoutError,
 } from '@eggjs/tegg-types/agent-runtime';
@@ -32,6 +33,14 @@ const HEARTBEAT_INTERVAL_MS = 10_000;
 const EVENT_DIR = join(tmpdir(), 'agent-runtime-events');
 const DEFAULT_CANCEL_COMMIT_TIMEOUT_MS = 30_000;
 
+function validateMetadata(value: unknown): Record<string, unknown> | undefined {
+  if (value === undefined) return undefined;
+  if (value === null || typeof value !== 'object' || Array.isArray(value)) {
+    throw new AgentInvalidRequestError("'metadata' must be an object");
+  }
+  return value as Record<string, unknown>;
+}
+
 interface RunEventBuffer {
   filePath: string;
   lastSeq: number;
@@ -141,17 +150,35 @@ export class AgentRuntime {
   }
 
   /**
-   * Resolve the thread for a run. If the caller provided a `threadId` we reuse
-   * it as-is. When no `threadId` is present we auto-create an empty thread.
-   * Run metadata belongs to the run record and must not be copied to the thread.
+   * Resolve the thread for a run and persist the run's `metadata` onto the
+   * thread. The same `metadata` is also stored on the run record (see
+   * {@link AgentStore.createRun}); here it initializes an auto-created thread or
+   * is shallow-merged into an existing thread's `meta.json`.
    */
   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);
+      if (metadata && Object.keys(metadata).length > 0) {
+        if (!this.store.updateThreadMetadata) {
+          throw new Error('AgentStore does not support updating thread metadata');
+        }
+        // Best-effort: the same metadata is already persisted on the run record,
+        // so a failure to mirror it onto the thread must not fail run creation.
+        try {
+          await this.store.updateThreadMetadata(input.threadId, metadata);
+        } catch (err) {
+          this.logger.error(
+            '[AgentRuntime] failed to persist metadata onto thread threadId=%s:',
+            input.threadId,
+            err,
+          );
+        }
+      }
       const isResume = thread.messages.length > 0;
       return { threadId: input.threadId, input: { ...input, isResume } };
     }
-    const thread = await this.store.createThread();
+    const thread = await this.store.createThread(metadata);
     return { threadId: thread.id, input: { ...input, threadId: thread.id, isResume: false } };
   }
 

core/agent-runtime/src/OSSAgentStore.ts

@@ -57,6 +57,7 @@ export class OSSAgentStore implements AgentStore {
   private readonly prefix: string;
   private readonly logger: OSSAgentStoreWarnLogger;
   private readonly pendingIndexWrites = new Set<Promise<void>>();
+  private readonly threadMetaWriteTails = new Map<string, Promise<void>>();
 
   constructor(options: OSSAgentStoreOptions) {
     this.client = options.client;
@@ -179,6 +180,56 @@ export class OSSAgentStore implements AgentStore {
     return { ...meta, messages };
   }
 
+  /**
+   * Serialize read-modify-write operations on a single thread's `meta.json`
+   * within this process. Both {@link updateThreadMetadata} (business metadata
+   * merge) and {@link recordLatestRunId} (the `latestRunId` pointer) mutate the
+   * same `meta.json` with no compare-and-swap, so without a shared per-thread
+   * lock a concurrent run could clobber the other's write. Cross-process writes
+   * remain last-writer-wins.
+   *
+   * The lock is a per-thread promise chain: `current` is always resolved in the
+   * `finally` (decoupled from `fn`'s success/failure), so a rejecting `fn` never
+   * leaves the chain stuck for subsequent waiters.
+   */
+  private async runExclusiveThreadMetaWrite<T>(threadId: string, fn: () => Promise<T>): Promise<T> {
+    const previous = this.threadMetaWriteTails.get(threadId) ?? Promise.resolve();
+    let release!: () => void;
+    const current = new Promise<void>(resolve => {
+      release = resolve;
+    });
+    const tail = previous.then(() => current);
+    this.threadMetaWriteTails.set(threadId, tail);
+
+    await previous;
+    try {
+      return await fn();
+    } finally {
+      release();
+      if (this.threadMetaWriteTails.get(threadId) === tail) {
+        this.threadMetaWriteTails.delete(threadId);
+      }
+    }
+  }
+
+  async updateThreadMetadata(threadId: string, metadata: Record<string, unknown>): Promise<void> {
+    if (Object.keys(metadata).length === 0) return;
+
+    await this.runExclusiveThreadMetaWrite(threadId, async () => {
+      const metaData = await this.client.get(this.threadMetaKey(threadId));
+      if (!metaData) {
+        throw new AgentNotFoundError(`Thread ${threadId} not found`);
+      }
+      const meta = JSON.parse(metaData) as ThreadMetadata;
+      const mergedMetadata = { ...meta.metadata, ...metadata };
+      await this.client.put(this.threadMetaKey(threadId), JSON.stringify({
+        ...meta,
+        metadata: mergedMetadata,
+      }));
+      this.writeThreadActivityIndex(threadId, meta.createdAt, Date.now(), mergedMetadata);
+    });
+  }
+
   async appendMessages(threadId: string, messages: AgentMessage[]): Promise<void> {
     const metaData = await this.client.get(this.threadMetaKey(threadId));
     if (!metaData) {
@@ -249,24 +300,26 @@ export class OSSAgentStore implements AgentStore {
    * Weak consistency: because the write is asynchronous and an unconditional
    * read-modify-write with no compare-and-swap, `getLatestRunId` is eventually
    * consistent — a read racing a just-created run may briefly see the prior
-   * value, and concurrent runs on the same thread are last-writer-wins. Only
-   * the `latestRunId` field can race; every writer carries the same
-   * `id/object/metadata/createdAt`, so no other thread metadata is lost.
+   * value, and concurrent runs in different processes are last-writer-wins.
+   * Within one process it shares {@link runExclusiveThreadMetaWrite} with
+   * `updateThreadMetadata`, so the two never clobber each other's `meta.json`.
    */
   private async recordLatestRunId(threadId: string, runId: string): Promise<void> {
     try {
-      const metaData = await this.client.get(this.threadMetaKey(threadId));
-      if (!metaData) {
-        this.logger.warn(
-          '[OSSAgentStore] skip latestRunId write: thread meta not found threadId=%s runId=%s',
-          threadId,
-          runId,
-        );
-        return;
-      }
-      const meta = JSON.parse(metaData) as ThreadMetadata;
-      meta.latestRunId = runId;
-      await this.client.put(this.threadMetaKey(threadId), JSON.stringify(meta));
+      await this.runExclusiveThreadMetaWrite(threadId, async () => {
+        const metaData = await this.client.get(this.threadMetaKey(threadId));
+        if (!metaData) {
+          this.logger.warn(
+            '[OSSAgentStore] skip latestRunId write: thread meta not found threadId=%s runId=%s',
+            threadId,
+            runId,
+          );
+          return;
+        }
+        const meta = JSON.parse(metaData) as ThreadMetadata;
+        meta.latestRunId = runId;
+        await this.client.put(this.threadMetaKey(threadId), JSON.stringify(meta));
+      });
     } catch (err: unknown) {
       const errForLog: Error = err instanceof Error ? err : new Error(String(err));
       this.logger.warn(

core/agent-runtime/test/AgentRuntime.metadata.test.ts

@@ -40,10 +40,12 @@ class MockSSEWriter implements SSEWriter {
 describe('test/AgentRuntime.metadata.test.ts', () => {
   let runtime: AgentRuntime;
   let store: OSSAgentStore;
+  let client: MapStorageClient;
   let executor: AgentExecutor;
 
   beforeEach(() => {
-    store = new OSSAgentStore({ client: new MapStorageClient() });
+    client = new MapStorageClient();
+    store = new OSSAgentStore({ client });
     executor = {
       async* execRun(input: CreateRunInput): AsyncGenerator<AgentMessage> {
         const messages = input.input.messages;
@@ -89,54 +91,136 @@ describe('test/AgentRuntime.metadata.test.ts', () => {
   });
 
   describe('syncRun metadata handling', () => {
-    it('should keep input.metadata on the run and not copy it to an auto-created thread', async () => {
-      const meta = { agentName: 'bar', sandboxId: 's-42' };
+    it('should store metadata on the run and initialize an auto-created thread', async () => {
+      const metadata = {
+        bizId: 'order_123',
+        nested: { source: 'customer_service' },
+        tags: [ 'vip', 7 ],
+        enabled: true,
+        nullable: null,
+      };
       const result = await runtime.syncRun({
         input: { messages: [{ role: 'user', content: 'Hi' }] },
-        metadata: meta,
+        metadata,
       });
 
-      assert.deepStrictEqual(result.metadata, meta);
+      // The run record keeps the metadata verbatim...
+      assert.deepStrictEqual(result.metadata, metadata);
       assert.equal(result.status, RunStatus.Completed);
-
+      // ...and the auto-created thread is initialized with the same metadata.
       const thread = await store.getThread(result.threadId);
-      assert.deepStrictEqual(thread.metadata, {});
+      assert.deepStrictEqual(thread.metadata, metadata);
     });
 
-    it('should NOT overwrite metadata of an existing thread (resume path)', async () => {
-      const original = { agentName: 'orig', createdBy: 'user-1' };
-      const thread = await runtime.createThread({ metadata: original });
+    it('should shallow-merge metadata into an existing thread', async () => {
+      const thread = await runtime.createThread({
+        metadata: {
+          bizId: 'order_123',
+          source: 'customer_service',
+          nested: { old: true },
+        },
+      });
 
       const result = await runtime.syncRun({
         threadId: thread.id,
         input: { messages: [{ role: 'user', content: 'Hi' }] },
-        metadata: { agentName: 'OVERRIDE' },
+        metadata: {
+          source: 'operator_console',
+          nested: { replacement: true },
+        },
       });
 
-      assert.equal(result.threadId, thread.id);
-
+      // The run keeps exactly what was passed in...
+      assert.deepStrictEqual(result.metadata, {
+        source: 'operator_console',
+        nested: { replacement: true },
+      });
+      // ...while the thread metadata is shallow-merged (bizId preserved).
       const stored = await store.getThread(thread.id);
-      assert.deepStrictEqual(stored.metadata, original);
+      assert.deepStrictEqual(stored.metadata, {
+        bizId: 'order_123',
+        source: 'operator_console',
+        nested: { replacement: true },
+      });
+    });
+
+    it('should leave existing thread metadata unchanged when metadata is empty or omitted', async () => {
+      const original = { bizId: 'order_123', source: 'customer_service' };
+      const thread = await runtime.createThread({ metadata: original });
+
+      await runtime.syncRun({
+        threadId: thread.id,
+        input: { messages: [{ role: 'user', content: 'First' }] },
+        metadata: {},
+      });
+      await runtime.syncRun({
+        threadId: thread.id,
+        input: { messages: [{ role: 'user', content: 'Second' }] },
+      });
+
+      assert.deepStrictEqual((await store.getThread(thread.id)).metadata, original);
+    });
+
+    it('should reject invalid metadata before creating a thread or run', async () => {
+      const assertInvalid = async (invalid: unknown): Promise<void> => {
+        await assert.rejects(
+          () => runtime.syncRun({
+            input: { messages: [{ role: 'user', content: 'Hi' }] },
+            metadata: invalid,
+          } as unknown as CreateRunInput),
+          (err: unknown) => {
+            assert.equal((err as { status?: number }).status, 400);
+            assert.match((err as Error).message, /metadata/);
+            return true;
+          },
+        );
+      };
+
+      for (const invalid of [ null, [], 'invalid', 1, true ]) {
+        await assertInvalid(invalid);
+      }
+      assert.deepStrictEqual(client.keysWithPrefix('threads/'), []);
+      assert.deepStrictEqual(client.keysWithPrefix('runs/'), []);
+    });
+
+    it('should not fail run creation when persisting metadata onto an existing thread fails', async () => {
+      const thread = await runtime.createThread({ metadata: { a: 1 } });
+      // Force the thread-side metadata write to fail.
+      store.updateThreadMetadata = async () => {
+        throw new Error('boom');
+      };
+
+      const result = await runtime.syncRun({
+        threadId: thread.id,
+        input: { messages: [{ role: 'user', content: 'Hi' }] },
+        metadata: { b: 2 },
+      });
+
+      // The run still completes and keeps its metadata on the run record...
+      assert.equal(result.status, RunStatus.Completed);
+      assert.deepStrictEqual(result.metadata, { b: 2 });
+      // ...while the thread metadata was left untouched by the failed write.
+      assert.deepStrictEqual((await store.getThread(thread.id)).metadata, { a: 1 });
     });
   });
 
   describe('asyncRun metadata handling', () => {
-    it('should keep input.metadata on the run and not copy it to an auto-created thread', async () => {
-      const meta = { agentName: 'baz' };
+    it('should store metadata on the run and persist it on the thread', async () => {
+      const metadata = { bizId: 'async_123' };
       const result = await runtime.asyncRun({
         input: { messages: [{ role: 'user', content: 'Hi' }] },
-        metadata: meta,
+        metadata,
       });
       await runtime.waitForPendingTasks();
 
-      assert.deepStrictEqual(result.metadata, meta);
+      assert.deepStrictEqual(result.metadata, metadata);
       const thread = await store.getThread(result.threadId);
-      assert.deepStrictEqual(thread.metadata, {});
+      assert.deepStrictEqual(thread.metadata, metadata);
     });
   });
 
   describe('streamRun metadata handling', () => {
-    it('should keep input.metadata on the run and not copy it to an auto-created thread', async () => {
+    it('should store metadata on the run and persist it on the thread', async () => {
       const meta = { agentName: 'stream', source: 'sse' };
       const writer = new MockSSEWriter();
 
@@ -151,11 +235,10 @@ describe('test/AgentRuntime.metadata.test.ts', () => {
       const runCreatedEvent = writer.events.find(e => e.event === 'run_created')!.data as StreamEvent;
       const runId = (runCreatedEvent.data as { runId: string }).runId;
       const threadId = (runCreatedEvent.data as { threadId: string }).threadId;
-      const run = await store.getRun(runId);
 
+      const run = await store.getRun(runId);
       assert.deepStrictEqual(run.metadata, meta);
-      const thread = await store.getThread(threadId);
-      assert.deepStrictEqual(thread.metadata, {});
+      assert.deepStrictEqual((await store.getThread(threadId)).metadata, meta);
     });
   });
 });