From b06fbad0aa4f78f923601b2bd41edca5a1aa2814 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=97=B0=E6=98=8E?= Date: Wed, 10 Jun 2026 10:32:50 +0800 Subject: [PATCH 1/2] feat(agent-runtime): support run thread metadata Co-authored-by: multica-agent --- core/agent-runtime/src/AgentRuntime.ts | 23 +++- core/agent-runtime/src/OSSAgentStore.ts | 33 +++++ .../test/AgentRuntime.metadata.test.ts | 113 +++++++++++++++++- core/agent-runtime/test/OSSAgentStore.test.ts | 48 ++++++++ core/types/agent-runtime/AgentRuntime.ts | 11 +- core/types/agent-runtime/AgentStore.ts | 5 + core/types/agent-runtime/errors.ts | 12 ++ 7 files changed, 238 insertions(+), 7 deletions(-) diff --git a/core/agent-runtime/src/AgentRuntime.ts b/core/agent-runtime/src/AgentRuntime.ts index 7e84d4b77..2fbef7c21 100644 --- a/core/agent-runtime/src/AgentRuntime.ts +++ b/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 validateThreadMetadata(value: unknown): Record | undefined { + if (value === undefined) return undefined; + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + throw new AgentInvalidRequestError("'threadMetadata' must be an object"); + } + return value as Record; +} + interface RunEventBuffer { filePath: string; lastSeq: number; @@ -141,17 +150,23 @@ 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 explicit thread metadata. + * Run metadata belongs to the run record and is never copied to the thread. */ private async ensureThread(input: CreateRunInput): Promise<{ threadId: string; input: CreateRunInput }> { + const threadMetadata = validateThreadMetadata(input.threadMetadata); if (input.threadId) { const thread = await this.store.getThread(input.threadId); + if (threadMetadata && Object.keys(threadMetadata).length > 0) { + if (!this.store.updateThreadMetadata) { + throw new Error('AgentStore does not support updating thread metadata'); + } + await this.store.updateThreadMetadata(input.threadId, threadMetadata); + } 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(threadMetadata); return { threadId: thread.id, input: { ...input, threadId: thread.id, isResume: false } }; } diff --git a/core/agent-runtime/src/OSSAgentStore.ts b/core/agent-runtime/src/OSSAgentStore.ts index ba2cf1387..4c5c43cee 100644 --- a/core/agent-runtime/src/OSSAgentStore.ts +++ b/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>(); + private readonly threadMetadataWriteTails = new Map>(); constructor(options: OSSAgentStoreOptions) { this.client = options.client; @@ -179,6 +180,38 @@ export class OSSAgentStore implements AgentStore { return { ...meta, messages }; } + async updateThreadMetadata(threadId: string, metadata: Record): Promise { + if (Object.keys(metadata).length === 0) return; + + const previous = this.threadMetadataWriteTails.get(threadId) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise(resolve => { + release = resolve; + }); + const tail = previous.then(() => current); + this.threadMetadataWriteTails.set(threadId, tail); + + await previous; + try { + 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); + } finally { + release(); + if (this.threadMetadataWriteTails.get(threadId) === tail) { + this.threadMetadataWriteTails.delete(threadId); + } + } + } + async appendMessages(threadId: string, messages: AgentMessage[]): Promise { const metaData = await this.client.get(this.threadMetaKey(threadId)); if (!metaData) { diff --git a/core/agent-runtime/test/AgentRuntime.metadata.test.ts b/core/agent-runtime/test/AgentRuntime.metadata.test.ts index 526bbe519..10bdc6205 100644 --- a/core/agent-runtime/test/AgentRuntime.metadata.test.ts +++ b/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 { const messages = input.input.messages; @@ -89,6 +91,88 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }); describe('syncRun metadata handling', () => { + it('should initialize an auto-created thread with threadMetadata', async () => { + const threadMetadata = { + 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' }] }, + threadMetadata, + }); + + const thread = await store.getThread(result.threadId); + assert.deepStrictEqual(thread.metadata, threadMetadata); + }); + + it('should shallow-merge threadMetadata into an existing thread', async () => { + const thread = await runtime.createThread({ + metadata: { + bizId: 'order_123', + source: 'customer_service', + nested: { old: true }, + }, + }); + + await runtime.syncRun({ + threadId: thread.id, + input: { messages: [{ role: 'user', content: 'Hi' }] }, + threadMetadata: { + source: 'operator_console', + nested: { replacement: true }, + }, + }); + + const stored = await store.getThread(thread.id); + assert.deepStrictEqual(stored.metadata, { + bizId: 'order_123', + source: 'operator_console', + nested: { replacement: true }, + }); + }); + + it('should leave existing thread metadata unchanged when threadMetadata 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' }] }, + threadMetadata: {}, + }); + 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 threadMetadata before creating a thread', async () => { + const assertInvalid = async (invalid: unknown): Promise => { + await assert.rejects( + () => runtime.syncRun({ + input: { messages: [{ role: 'user', content: 'Hi' }] }, + threadMetadata: invalid, + } as unknown as CreateRunInput), + (err: unknown) => { + assert.equal((err as { status?: number }).status, 400); + assert.match((err as Error).message, /threadMetadata/); + 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 keep input.metadata on the run and not copy it to an auto-created thread', async () => { const meta = { agentName: 'bar', sandboxId: 's-42' }; const result = await runtime.syncRun({ @@ -121,6 +205,17 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }); describe('asyncRun metadata handling', () => { + it('should persist threadMetadata on the thread', async () => { + const result = await runtime.asyncRun({ + input: { messages: [{ role: 'user', content: 'Hi' }] }, + threadMetadata: { bizId: 'async_123' }, + }); + await runtime.waitForPendingTasks(); + + const thread = await store.getThread(result.threadId); + assert.deepStrictEqual(thread.metadata, { bizId: 'async_123' }); + }); + it('should keep input.metadata on the run and not copy it to an auto-created thread', async () => { const meta = { agentName: 'baz' }; const result = await runtime.asyncRun({ @@ -136,6 +231,22 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }); describe('streamRun metadata handling', () => { + it('should persist threadMetadata on the thread', async () => { + const writer = new MockSSEWriter(); + + await runtime.streamRun( + { + input: { messages: [{ role: 'user', content: 'Hi' }] }, + threadMetadata: { bizId: 'stream_123' }, + }, + writer, + ); + + const runCreatedEvent = writer.events.find(e => e.event === 'run_created')!.data as StreamEvent; + const threadId = (runCreatedEvent.data as { threadId: string }).threadId; + assert.deepStrictEqual((await store.getThread(threadId)).metadata, { bizId: 'stream_123' }); + }); + it('should keep input.metadata on the run and not copy it to an auto-created thread', async () => { const meta = { agentName: 'stream', source: 'sse' }; const writer = new MockSSEWriter(); diff --git a/core/agent-runtime/test/OSSAgentStore.test.ts b/core/agent-runtime/test/OSSAgentStore.test.ts index be4c8fd63..44698b02f 100644 --- a/core/agent-runtime/test/OSSAgentStore.test.ts +++ b/core/agent-runtime/test/OSSAgentStore.test.ts @@ -149,6 +149,54 @@ describe('test/OSSAgentStore.test.ts', () => { }); }); + describe('thread metadata', () => { + it('should shallow-merge metadata and preserve omitted keys', async () => { + const thread = await store.createThread({ + bizId: 'order_123', + source: 'customer_service', + nested: { old: true }, + }); + + await store.updateThreadMetadata(thread.id, { + source: 'operator_console', + nested: { replacement: true }, + nullable: null, + }); + + assert.deepStrictEqual((await store.getThread(thread.id)).metadata, { + bizId: 'order_123', + source: 'operator_console', + nested: { replacement: true }, + nullable: null, + }); + }); + + it('should reject updating a missing thread', async () => { + await assert.rejects( + () => store.updateThreadMetadata('thread_missing', { bizId: 'order_123' }), + AgentNotFoundError, + ); + }); + + it('should serialize concurrent updates in the same process', async () => { + const client = new MapStorageClient(); + const localStore = new OSSAgentStore({ client }); + const thread = await localStore.createThread({ original: true }); + client.delayPutWhenKeyMatches(/threads\/.*\/meta\.json$/, 20); + + await Promise.all([ + localStore.updateThreadMetadata(thread.id, { first: 1 }), + localStore.updateThreadMetadata(thread.id, { second: 2 }), + ]); + + assert.deepStrictEqual((await localStore.getThread(thread.id)).metadata, { + original: true, + first: 1, + second: 2, + }); + }); + }); + describe('runs', () => { it('should create a run', async () => { const run = await store.createRun([{ role: 'user', content: 'Hello' }]); diff --git a/core/types/agent-runtime/AgentRuntime.ts b/core/types/agent-runtime/AgentRuntime.ts index 0a8014c0a..c78e1cd47 100644 --- a/core/types/agent-runtime/AgentRuntime.ts +++ b/core/types/agent-runtime/AgentRuntime.ts @@ -54,6 +54,14 @@ export interface CreateRunInput { }; config?: AgentRunConfig; metadata?: Record; + /** + * Metadata persisted on the thread rather than the run. + * + * Auto-created threads are initialized with this object. For an existing + * thread, keys are shallow-merged and new values overwrite matching keys. + * An omitted or empty object leaves the thread metadata unchanged. + */ + threadMetadata?: Record; } // ===== Thread input ===== @@ -63,8 +71,7 @@ export interface CreateRunInput { * * `metadata` is forwarded verbatim to {@link AgentStore.createThread} so callers * can persist additional business semantics on the thread record (e.g. the - * resolved agent name, owning sandbox id, trace id). It is stored once at - * creation time and never overwritten by subsequent runs on the same thread. + * resolved agent name, owning sandbox id, trace id). */ export interface CreateThreadOptions { metadata?: Record; diff --git a/core/types/agent-runtime/AgentStore.ts b/core/types/agent-runtime/AgentStore.ts index 9b8d48fb2..3951cbddc 100644 --- a/core/types/agent-runtime/AgentStore.ts +++ b/core/types/agent-runtime/AgentStore.ts @@ -77,6 +77,11 @@ export interface AgentStore { destroy?(): Promise; createThread(metadata?: Record): Promise; getThread(threadId: string, options?: GetThreadOptions): Promise; + /** + * Shallow-merge metadata into an existing thread. + * New values overwrite matching keys; omitted keys are preserved. + */ + updateThreadMetadata?(threadId: string, metadata: Record): Promise; appendMessages(threadId: string, messages: AgentMessage[]): Promise; createRun( input: InputMessage[], diff --git a/core/types/agent-runtime/errors.ts b/core/types/agent-runtime/errors.ts index 2f3447592..61dce9467 100644 --- a/core/types/agent-runtime/errors.ts +++ b/core/types/agent-runtime/errors.ts @@ -12,6 +12,18 @@ export class AgentNotFoundError extends Error { } } +/** + * Error thrown when an agent API request contains invalid input. + */ +export class AgentInvalidRequestError extends Error { + status = 400; + + constructor(message: string) { + super(message); + this.name = 'AgentInvalidRequestError'; + } +} + /** * Error thrown when an operation conflicts with the current state * (e.g., cancelling a completed run). From 2c8c8ef3a68e9330808f5fe22717f3b1824f00aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=97=B0=E6=98=8E?= Date: Wed, 10 Jun 2026 19:01:53 +0800 Subject: [PATCH 2/2] feat(agent-runtime): persist run metadata onto the thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reuse the existing `CreateRunInput.metadata` instead of a dedicated input field. On run creation, `metadata` is double-written: kept verbatim on the run record AND shallow-merged into the thread's meta.json — initializing an auto-created thread, or merging into an existing one (matching keys overwritten, others preserved; an empty or omitted object is a no-op). Non-object metadata is rejected with HTTP 400 before any thread/run write. Robustness: - Mirroring metadata onto an existing thread is best-effort: a failed thread write is logged and does not fail run creation (the run record already holds the metadata). A store lacking updateThreadMetadata still errors. - updateThreadMetadata and recordLatestRunId both read-modify-write the same meta.json; they now share a per-thread write lock so concurrent runs on one thread no longer clobber each other (cross-process stays last-writer-wins). Co-Authored-By: Claude Opus 4.8 (1M context) --- core/agent-runtime/src/AgentRuntime.ts | 28 +++-- core/agent-runtime/src/OSSAgentStore.ts | 74 +++++++----- .../test/AgentRuntime.metadata.test.ts | 112 +++++++----------- core/agent-runtime/test/OSSAgentStore.test.ts | 22 ++++ core/types/agent-runtime/AgentRuntime.ts | 14 +-- 5 files changed, 138 insertions(+), 112 deletions(-) diff --git a/core/agent-runtime/src/AgentRuntime.ts b/core/agent-runtime/src/AgentRuntime.ts index 2fbef7c21..f7e585bd0 100644 --- a/core/agent-runtime/src/AgentRuntime.ts +++ b/core/agent-runtime/src/AgentRuntime.ts @@ -33,10 +33,10 @@ const HEARTBEAT_INTERVAL_MS = 10_000; const EVENT_DIR = join(tmpdir(), 'agent-runtime-events'); const DEFAULT_CANCEL_COMMIT_TIMEOUT_MS = 30_000; -function validateThreadMetadata(value: unknown): Record | undefined { +function validateMetadata(value: unknown): Record | undefined { if (value === undefined) return undefined; if (value === null || typeof value !== 'object' || Array.isArray(value)) { - throw new AgentInvalidRequestError("'threadMetadata' must be an object"); + throw new AgentInvalidRequestError("'metadata' must be an object"); } return value as Record; } @@ -150,23 +150,35 @@ export class AgentRuntime { } /** - * Resolve the thread for a run and persist explicit thread metadata. - * Run metadata belongs to the run record and is never 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 threadMetadata = validateThreadMetadata(input.threadMetadata); + const metadata = validateMetadata(input.metadata); if (input.threadId) { const thread = await this.store.getThread(input.threadId); - if (threadMetadata && Object.keys(threadMetadata).length > 0) { + if (metadata && Object.keys(metadata).length > 0) { if (!this.store.updateThreadMetadata) { throw new Error('AgentStore does not support updating thread metadata'); } - await this.store.updateThreadMetadata(input.threadId, threadMetadata); + // 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(threadMetadata); + const thread = await this.store.createThread(metadata); return { threadId: thread.id, input: { ...input, threadId: thread.id, isResume: false } }; } diff --git a/core/agent-runtime/src/OSSAgentStore.ts b/core/agent-runtime/src/OSSAgentStore.ts index 4c5c43cee..6549d5d84 100644 --- a/core/agent-runtime/src/OSSAgentStore.ts +++ b/core/agent-runtime/src/OSSAgentStore.ts @@ -57,7 +57,7 @@ export class OSSAgentStore implements AgentStore { private readonly prefix: string; private readonly logger: OSSAgentStoreWarnLogger; private readonly pendingIndexWrites = new Set>(); - private readonly threadMetadataWriteTails = new Map>(); + private readonly threadMetaWriteTails = new Map>(); constructor(options: OSSAgentStoreOptions) { this.client = options.client; @@ -180,19 +180,42 @@ export class OSSAgentStore implements AgentStore { return { ...meta, messages }; } - async updateThreadMetadata(threadId: string, metadata: Record): Promise { - if (Object.keys(metadata).length === 0) return; - - const previous = this.threadMetadataWriteTails.get(threadId) ?? Promise.resolve(); + /** + * 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(threadId: string, fn: () => Promise): Promise { + const previous = this.threadMetaWriteTails.get(threadId) ?? Promise.resolve(); let release!: () => void; const current = new Promise(resolve => { release = resolve; }); const tail = previous.then(() => current); - this.threadMetadataWriteTails.set(threadId, tail); + 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): Promise { + 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`); @@ -204,12 +227,7 @@ export class OSSAgentStore implements AgentStore { metadata: mergedMetadata, })); this.writeThreadActivityIndex(threadId, meta.createdAt, Date.now(), mergedMetadata); - } finally { - release(); - if (this.threadMetadataWriteTails.get(threadId) === tail) { - this.threadMetadataWriteTails.delete(threadId); - } - } + }); } async appendMessages(threadId: string, messages: AgentMessage[]): Promise { @@ -282,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 { 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( diff --git a/core/agent-runtime/test/AgentRuntime.metadata.test.ts b/core/agent-runtime/test/AgentRuntime.metadata.test.ts index 10bdc6205..37c27a188 100644 --- a/core/agent-runtime/test/AgentRuntime.metadata.test.ts +++ b/core/agent-runtime/test/AgentRuntime.metadata.test.ts @@ -91,8 +91,8 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }); describe('syncRun metadata handling', () => { - it('should initialize an auto-created thread with threadMetadata', async () => { - const threadMetadata = { + 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 ], @@ -101,14 +101,18 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }; const result = await runtime.syncRun({ input: { messages: [{ role: 'user', content: 'Hi' }] }, - threadMetadata, + metadata, }); + // 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, threadMetadata); + assert.deepStrictEqual(thread.metadata, metadata); }); - it('should shallow-merge threadMetadata into an existing thread', async () => { + it('should shallow-merge metadata into an existing thread', async () => { const thread = await runtime.createThread({ metadata: { bizId: 'order_123', @@ -117,15 +121,21 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }, }); - await runtime.syncRun({ + const result = await runtime.syncRun({ threadId: thread.id, input: { messages: [{ role: 'user', content: 'Hi' }] }, - threadMetadata: { + metadata: { source: 'operator_console', nested: { replacement: true }, }, }); + // 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, { bizId: 'order_123', @@ -134,14 +144,14 @@ describe('test/AgentRuntime.metadata.test.ts', () => { }); }); - it('should leave existing thread metadata unchanged when threadMetadata is empty or omitted', async () => { + 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' }] }, - threadMetadata: {}, + metadata: {}, }); await runtime.syncRun({ threadId: thread.id, @@ -151,16 +161,16 @@ describe('test/AgentRuntime.metadata.test.ts', () => { assert.deepStrictEqual((await store.getThread(thread.id)).metadata, original); }); - it('should reject invalid threadMetadata before creating a thread', async () => { + it('should reject invalid metadata before creating a thread or run', async () => { const assertInvalid = async (invalid: unknown): Promise => { await assert.rejects( () => runtime.syncRun({ input: { messages: [{ role: 'user', content: 'Hi' }] }, - threadMetadata: invalid, + metadata: invalid, } as unknown as CreateRunInput), (err: unknown) => { assert.equal((err as { status?: number }).status, 400); - assert.match((err as Error).message, /threadMetadata/); + assert.match((err as Error).message, /metadata/); return true; }, ); @@ -173,81 +183,44 @@ describe('test/AgentRuntime.metadata.test.ts', () => { assert.deepStrictEqual(client.keysWithPrefix('runs/'), []); }); - 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' }; - const result = await runtime.syncRun({ - input: { messages: [{ role: 'user', content: 'Hi' }] }, - metadata: meta, - }); - - assert.deepStrictEqual(result.metadata, meta); - assert.equal(result.status, RunStatus.Completed); - - const thread = await store.getThread(result.threadId); - assert.deepStrictEqual(thread.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 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: { agentName: 'OVERRIDE' }, + metadata: { b: 2 }, }); - assert.equal(result.threadId, thread.id); - - const stored = await store.getThread(thread.id); - assert.deepStrictEqual(stored.metadata, original); + // 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 persist threadMetadata on the thread', async () => { - const result = await runtime.asyncRun({ - input: { messages: [{ role: 'user', content: 'Hi' }] }, - threadMetadata: { bizId: 'async_123' }, - }); - await runtime.waitForPendingTasks(); - - const thread = await store.getThread(result.threadId); - assert.deepStrictEqual(thread.metadata, { bizId: 'async_123' }); - }); - - 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 persist threadMetadata on the thread', async () => { - const writer = new MockSSEWriter(); - - await runtime.streamRun( - { - input: { messages: [{ role: 'user', content: 'Hi' }] }, - threadMetadata: { bizId: 'stream_123' }, - }, - writer, - ); - - const runCreatedEvent = writer.events.find(e => e.event === 'run_created')!.data as StreamEvent; - const threadId = (runCreatedEvent.data as { threadId: string }).threadId; - assert.deepStrictEqual((await store.getThread(threadId)).metadata, { bizId: 'stream_123' }); - }); - - 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(); @@ -262,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); }); }); }); diff --git a/core/agent-runtime/test/OSSAgentStore.test.ts b/core/agent-runtime/test/OSSAgentStore.test.ts index 44698b02f..12513ac9b 100644 --- a/core/agent-runtime/test/OSSAgentStore.test.ts +++ b/core/agent-runtime/test/OSSAgentStore.test.ts @@ -195,6 +195,28 @@ describe('test/OSSAgentStore.test.ts', () => { second: 2, }); }); + + it('should not clobber metadata when an updateThreadMetadata races a latestRunId write', async () => { + const client = new MapStorageClient(); + const localStore = new OSSAgentStore({ client }); + const thread = await localStore.createThread({ original: true }); + // Both writers do a read-modify-write on the same meta.json; the delay + // forces them to overlap so an unsynchronized pair would clobber. + client.delayPutWhenKeyMatches(/threads\/.*\/meta\.json$/, 20); + + const [ , run ] = await Promise.all([ + localStore.updateThreadMetadata(thread.id, { merged: 1 }), + // createRun fires recordLatestRunId in the background (latestRunId write). + localStore.createRun([{ role: 'user', content: 'Hi' }], thread.id), + ]); + await localStore.awaitPendingWrites(); + + const stored = await localStore.getThread(thread.id); + // The metadata merge survives... + assert.deepStrictEqual(stored.metadata, { original: true, merged: 1 }); + // ...and so does the latestRunId pointer written by createRun. + assert.equal(await localStore.getLatestRunId(thread.id), run.id); + }); }); describe('runs', () => { diff --git a/core/types/agent-runtime/AgentRuntime.ts b/core/types/agent-runtime/AgentRuntime.ts index c78e1cd47..01db15ad6 100644 --- a/core/types/agent-runtime/AgentRuntime.ts +++ b/core/types/agent-runtime/AgentRuntime.ts @@ -53,15 +53,15 @@ export interface CreateRunInput { messages: import('./AgentMessage').InputMessage[]; }; config?: AgentRunConfig; - metadata?: Record; /** - * Metadata persisted on the thread rather than the run. - * - * Auto-created threads are initialized with this object. For an existing - * thread, keys are shallow-merged and new values overwrite matching keys. - * An omitted or empty object leaves the thread metadata unchanged. + * Metadata for the run. Stored verbatim on the run record, and additionally + * shallow-merged into the thread metadata (`meta.json`): + * - For an auto-created thread, it initializes the thread metadata. + * - For an existing thread, the keys are shallow-merged: new values overwrite + * matching keys, while keys not present are preserved. + * - An omitted or empty object leaves the thread metadata unchanged. */ - threadMetadata?: Record; + metadata?: Record; } // ===== Thread input =====