fix(workflow): persist execution metadata and fix SQL metadata filters

.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.<key>` filters.

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,
+    ]);
+  });
+});

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

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", () => {

packages/core/src/workflow/core.ts

@@ -747,6 +747,14 @@ export function createWorkflow<
       executionId = options?.executionId || randomUUID();
     }
 
+    const mergeExecutionMetadata = async (patch: Record<string, unknown>) => {
+      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) {

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<string, unknown>;
   /**
    * The user context, this can be used to track the current user context in a workflow
    */

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

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

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",
+    });
+  });
+});

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