diff --git a/.changeset/workflow-execution-metadata-filters.md b/.changeset/workflow-execution-metadata-filters.md new file mode 100644 index 000000000..cb7b874a5 --- /dev/null +++ b/.changeset/workflow-execution-metadata-filters.md @@ -0,0 +1,13 @@ +--- +"@voltagent/core": patch +"@voltagent/server-core": patch +"@voltagent/libsql": patch +"@voltagent/cloudflare-d1": patch +--- + +Fix workflow execution filtering by persisted metadata across adapters. + +- Persist `options.metadata` on workflow execution state so `/workflows/executions` filters can match tenant/user metadata. +- Preserve existing execution metadata when updating cancelled/error workflow states. +- Accept `options.metadata` in server workflow execution request schema. +- Fix LibSQL and Cloudflare D1 JSON metadata query comparisons for `metadata` and `metadata.` filters. diff --git a/packages/cloudflare-d1/src/memory-adapter.spec.ts b/packages/cloudflare-d1/src/memory-adapter.spec.ts new file mode 100644 index 000000000..36ad60ac6 --- /dev/null +++ b/packages/cloudflare-d1/src/memory-adapter.spec.ts @@ -0,0 +1,83 @@ +import type { D1Database } from "@cloudflare/workers-types"; +import { describe, expect, it, vi } from "vitest"; +import { D1MemoryAdapter } from "./memory-adapter"; + +function createMockBinding(): D1Database { + return { + prepare: vi.fn(() => ({ + bind: vi.fn().mockReturnThis(), + run: vi.fn().mockResolvedValue({}), + all: vi.fn().mockResolvedValue({ results: [] }), + })), + batch: vi.fn().mockResolvedValue([]), + } as unknown as D1Database; +} + +describe("D1MemoryAdapter queryWorkflowRuns", () => { + it("builds metadata filters with JSON-aware comparisons", async () => { + vi.spyOn(D1MemoryAdapter.prototype as any, "ensureInitialized").mockResolvedValue(undefined); + + const adapter = new D1MemoryAdapter({ + binding: createMockBinding(), + tablePrefix: "test", + }); + + const allSpy = vi.spyOn(adapter as any, "all").mockResolvedValue([ + { + id: "exec-1", + workflow_id: "workflow-1", + workflow_name: "Workflow 1", + status: "completed", + input: '{"requestId":"req-1"}', + context: '[["tenantId","acme"]]', + workflow_state: '{"phase":"done"}', + suspension: null, + events: null, + output: null, + cancellation: null, + user_id: "user-1", + conversation_id: null, + metadata: '{"tenantId":"acme"}', + created_at: "2024-01-02T00:00:00Z", + updated_at: "2024-01-02T00:00:00Z", + }, + ]); + + const result = await adapter.queryWorkflowRuns({ + workflowId: "workflow-1", + status: "completed", + from: new Date("2024-01-01T00:00:00Z"), + to: new Date("2024-01-03T00:00:00Z"), + userId: "user-1", + metadata: { tenantId: "acme" }, + limit: 5, + offset: 2, + }); + + expect(result).toHaveLength(1); + expect(result[0]?.input).toEqual({ requestId: "req-1" }); + expect(result[0]?.context).toEqual([["tenantId", "acme"]]); + expect(result[0]?.workflowState).toEqual({ phase: "done" }); + + expect(allSpy).toHaveBeenCalledTimes(1); + const [sql, args] = allSpy.mock.calls[0]; + + expect(sql).toContain("workflow_id = ?"); + expect(sql).toContain("status = ?"); + expect(sql).toContain("created_at >="); + expect(sql).toContain("user_id = ?"); + expect(sql).toContain("json_extract(metadata, ?) = json_extract(json(?), '$')"); + expect(sql).toContain("ORDER BY created_at DESC"); + expect(args).toEqual([ + "workflow-1", + "completed", + "2024-01-01T00:00:00.000Z", + "2024-01-03T00:00:00.000Z", + "user-1", + '$."tenantId"', + '"acme"', + 5, + 2, + ]); + }); +}); diff --git a/packages/cloudflare-d1/src/memory-adapter.ts b/packages/cloudflare-d1/src/memory-adapter.ts index 0bafafe94..dd0fe9204 100644 --- a/packages/cloudflare-d1/src/memory-adapter.ts +++ b/packages/cloudflare-d1/src/memory-adapter.ts @@ -1375,7 +1375,7 @@ export class D1MemoryAdapter implements StorageAdapter { continue; } - conditions.push("json_extract(metadata, ?) = json(?)"); + conditions.push("json_extract(metadata, ?) = json_extract(json(?), '$')"); args.push(metadataPath, safeStringify(value)); } } diff --git a/packages/core/src/workflow/core.spec.ts b/packages/core/src/workflow/core.spec.ts index 9115e95a8..1bfa94eae 100644 --- a/packages/core/src/workflow/core.spec.ts +++ b/packages/core/src/workflow/core.spec.ts @@ -284,6 +284,55 @@ describe.sequential("workflow.run", () => { expect(persistedState?.userId).toBe("user-test-1"); expect(persistedState?.conversationId).toBe("conv-test-1"); }); + + it("should persist custom metadata in workflow state", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "workflow-metadata-context", + name: "Workflow Metadata Context", + input: z.object({ + value: z.string(), + }), + result: z.object({ + value: z.string(), + }), + memory, + }, + andThen({ + id: "echo", + execute: async ({ data }) => data, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run( + { value: "ok" }, + { + userId: "user-test-1", + metadata: { + tenantId: "acme", + region: "us-east-1", + flags: { plan: "pro" }, + }, + }, + ); + + const persistedState = await memory.getWorkflowState(result.executionId); + + expect(persistedState?.metadata).toEqual( + expect.objectContaining({ + tenantId: "acme", + region: "us-east-1", + flags: { plan: "pro" }, + traceId: expect.any(String), + spanId: expect.any(String), + }), + ); + }); }); describe.sequential("workflow streaming", () => { diff --git a/packages/core/src/workflow/core.ts b/packages/core/src/workflow/core.ts index d9a486ad6..5202646a6 100644 --- a/packages/core/src/workflow/core.ts +++ b/packages/core/src/workflow/core.ts @@ -747,6 +747,14 @@ export function createWorkflow< executionId = options?.executionId || randomUUID(); } + const mergeExecutionMetadata = async (patch: Record) => { + const existingState = await executionMemory.getWorkflowState(executionId); + return { + ...(existingState?.metadata ?? {}), + ...patch, + }; + }; + // Only create stream controller if one is provided (for streaming execution) // For normal run, we don't need a stream controller const streamController = externalStreamController || null; @@ -813,6 +821,10 @@ export function createWorkflow< : options?.context ? new Map(Object.entries(options.context)) : new Map(); + const optionMetadata = + options?.metadata && typeof options.metadata === "object" && !Array.isArray(options.metadata) + ? options.metadata + : undefined; const workflowStateStore = options?.workflowState ?? {}; // Get previous trace IDs if resuming @@ -928,6 +940,7 @@ export function createWorkflow< userId: options?.userId, conversationId: options?.conversationId, metadata: { + ...(optionMetadata ?? {}), traceId: rootSpan.spanContext().traceId, spanId: rootSpan.spanContext().spanId, }, @@ -1219,10 +1232,10 @@ export function createWorkflow< cancelledAt: new Date(), reason, }, - metadata: { + metadata: await mergeExecutionMetadata({ ...(stateManager.state?.usage ? { usage: stateManager.state.usage } : {}), cancellationReason: reason, - }, + }), updatedAt: new Date(), }); } catch (memoryError) { @@ -1943,10 +1956,10 @@ export function createWorkflow< await executionMemory.updateWorkflowState(executionId, { status: "cancelled", workflowState: stateManager.state.workflowState, - metadata: { + metadata: await mergeExecutionMetadata({ ...(stateManager.state?.usage ? { usage: stateManager.state.usage } : {}), cancellationReason, - }, + }), updatedAt: new Date(), }); } catch (memoryError) { @@ -2041,10 +2054,10 @@ export function createWorkflow< workflowState: stateManager.state.workflowState, events: collectedEvents, // Store a lightweight error summary in metadata for debugging - metadata: { + metadata: await mergeExecutionMetadata({ ...(stateManager.state?.usage ? { usage: stateManager.state.usage } : {}), errorMessage: error instanceof Error ? error.message : String(error), - }, + }), updatedAt: new Date(), }); } catch (memoryError) { diff --git a/packages/core/src/workflow/types.ts b/packages/core/src/workflow/types.ts index 7ad407532..d81d2a48b 100644 --- a/packages/core/src/workflow/types.ts +++ b/packages/core/src/workflow/types.ts @@ -240,6 +240,10 @@ export interface WorkflowRunOptions { * The user ID, this can be used to track the current user in a workflow */ userId?: string; + /** + * Additional execution metadata persisted with workflow state + */ + metadata?: Record; /** * The user context, this can be used to track the current user context in a workflow */ diff --git a/packages/libsql/src/memory-core.ts b/packages/libsql/src/memory-core.ts index 019936f3d..c2695a03b 100644 --- a/packages/libsql/src/memory-core.ts +++ b/packages/libsql/src/memory-core.ts @@ -1214,7 +1214,7 @@ export class LibSQLMemoryCore implements StorageAdapter { continue; } - conditions.push("json_extract(metadata, ?) = json(?)"); + conditions.push("json_extract(metadata, ?) = json_extract(json(?), '$')"); args.push(metadataPath, safeStringify(value)); } } diff --git a/packages/libsql/src/memory-v2-adapter.spec.ts b/packages/libsql/src/memory-v2-adapter.spec.ts index fb4fb138b..2e2a429b9 100644 --- a/packages/libsql/src/memory-v2-adapter.spec.ts +++ b/packages/libsql/src/memory-v2-adapter.spec.ts @@ -85,7 +85,7 @@ describe.sequential("LibSQLMemoryAdapter - Advanced Behavior", () => { expect(sql).toContain("status = ?"); expect(sql).toContain("created_at >="); expect(sql).toContain("user_id = ?"); - expect(sql).toContain("json_extract(metadata, ?) = json(?)"); + expect(sql).toContain("json_extract(metadata, ?) = json_extract(json(?), '$')"); expect(sql).toContain("ORDER BY created_at DESC"); expect(args).toEqual([ "workflow-1", diff --git a/packages/server-core/src/schemas/agent.schemas.spec.ts b/packages/server-core/src/schemas/agent.schemas.spec.ts new file mode 100644 index 000000000..7521ae1c1 --- /dev/null +++ b/packages/server-core/src/schemas/agent.schemas.spec.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from "vitest"; +import { WorkflowExecutionRequestSchema } from "./agent.schemas"; + +describe("WorkflowExecutionRequestSchema", () => { + it("accepts options.metadata payload", () => { + const parsed = WorkflowExecutionRequestSchema.parse({ + input: { value: 1 }, + options: { + userId: "user-1", + metadata: { + tenantId: "acme", + region: "us-east-1", + }, + }, + }); + + expect(parsed.options?.metadata).toEqual({ + tenantId: "acme", + region: "us-east-1", + }); + }); +}); diff --git a/packages/server-core/src/schemas/agent.schemas.ts b/packages/server-core/src/schemas/agent.schemas.ts index ee78c72d8..058dd1fd6 100644 --- a/packages/server-core/src/schemas/agent.schemas.ts +++ b/packages/server-core/src/schemas/agent.schemas.ts @@ -359,6 +359,7 @@ export const WorkflowExecutionRequestSchema = z.object({ executionId: z.string().optional(), context: z.any().optional(), workflowState: z.record(z.any()).optional(), + metadata: z.record(z.any()).optional(), }) .optional() .describe("Optional execution options"),