Add JSON Schema Compose node for building records from schema fields

bchapuis · claude · bchapuis · commit f5ddc63715eb · 2026-04-01T00:05:35.000+02:00
Inverse of JsonSchemaExtractNode — takes individual field inputs and
combines them into a single JSON record. The widget uses a single
atomic updateNodeData call to avoid race conditions between competing
input state updates.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/apps/api/src/runtime/cloudflare-node-registry.ts b/apps/api/src/runtime/cloudflare-node-registry.ts
@@ -325,6 +325,7 @@ import { JsonObjectKeysNode } from "@dafthunk/runtime/nodes/json/json-object-key
 import { JsonObjectValuesNode } from "@dafthunk/runtime/nodes/json/json-object-values-node";
 import { JsonRemoveNode } from "@dafthunk/runtime/nodes/json/json-remove-node";
 import { JsonReplaceNode } from "@dafthunk/runtime/nodes/json/json-replace-node";
+import { JsonSchemaComposeNode } from "@dafthunk/runtime/nodes/json/json-schema-compose-node";
 import { JsonSchemaExtractNode } from "@dafthunk/runtime/nodes/json/json-schema-extract-node";
 import { JsonSetNode } from "@dafthunk/runtime/nodes/json/json-set-node";
 import { JsonStripNullsNode } from "@dafthunk/runtime/nodes/json/json-strip-nulls-node";
@@ -614,6 +615,7 @@ export class CloudflareNodeRegistry extends BaseNodeRegistry<Bindings> {
     this.registerImplementation(JsonToBlobNode);
     this.registerImplementation(JsonToGeojsonNode);
     this.registerImplementation(JsonTypeofNode);
+    this.registerImplementation(JsonSchemaComposeNode);
     this.registerImplementation(JsonSchemaExtractNode);
     this.registerImplementation(JsonValidNode);
     this.registerImplementation(JsonStringTemplateNode);
diff --git a/apps/app/src/components/workflow/widgets/index.ts b/apps/app/src/components/workflow/widgets/index.ts
@@ -29,6 +29,7 @@ import { javascriptInputWidget } from "./input/javascript-input";
 import { jsonInputWidget } from "./input/json-input";
 import { numberInputWidget } from "./input/number-input";
 import { replicateModelInputWidget } from "./input/replicate-model-input";
+import { schemaComposeInputWidget } from "./input/schema-compose-input";
 import { schemaExtractInputWidget } from "./input/schema-extract-input";
 import { secretInputWidget } from "./input/secret-input";
 import { sliderInputWidget } from "./input/slider-input";
@@ -86,6 +87,7 @@ const widgets = [
   audioRecorderInputWidget,
   canvasInputWidget,
   replicateModelInputWidget,
+  schemaComposeInputWidget,
   schemaExtractInputWidget,
   createDynamicInputsWidget("string-concat", {
     prefix: "input",
diff --git a/apps/app/src/components/workflow/widgets/input/schema-compose-input.tsx b/apps/app/src/components/workflow/widgets/input/schema-compose-input.tsx
@@ -0,0 +1,187 @@
+import type { Field, FieldType, GetSchemaResponse } from "@dafthunk/types";
+import { useCallback, useState } from "react";
+
+import { useAuth } from "@/components/auth-context";
+import { SchemaDialog } from "@/components/schema-dialog";
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectSeparator,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select";
+import { useSchemas } from "@/services/schema-service";
+import { makeOrgRequest } from "@/services/utils";
+import { cn } from "@/utils/utils";
+
+import { useWorkflow } from "../../workflow-context";
+import type { WorkflowParameter } from "../../workflow-types";
+import type { BaseWidgetProps } from "../widget";
+import { createWidget, getInputValue } from "../widget";
+
+const CREATE_NEW = "__create_new__";
+
+const FIELD_TYPE_TO_PARAMETER_TYPE: Record<FieldType, string> = {
+  string: "string",
+  integer: "number",
+  number: "number",
+  boolean: "boolean",
+  datetime: "date",
+  json: "json",
+};
+
+interface SchemaComposeInputProps extends BaseWidgetProps {
+  nodeId: string;
+  schemaId: string;
+  hasSchemaInputs: boolean;
+}
+
+function SchemaComposeInputWidget({
+  nodeId,
+  schemaId,
+  className,
+  disabled = false,
+}: SchemaComposeInputProps) {
+  const [loading, setLoading] = useState(false);
+  const [isCreateDialogOpen, setIsCreateDialogOpen] = useState(false);
+  const { schemas, isSchemasLoading, mutateSchemas } = useSchemas();
+  const { organization } = useAuth();
+  const { updateNodeData, edges, deleteEdge } = useWorkflow();
+
+  const applySchema = useCallback(
+    async (selectedSchemaId: string) => {
+      if (!organization?.id || !updateNodeData) return;
+
+      setLoading(true);
+      try {
+        const response = await makeOrgRequest<GetSchemaResponse>(
+          organization.id,
+          "/schemas",
+          `/${selectedSchemaId}`
+        );
+
+        const fieldInputs = response.schema.fields.map(
+          (field) =>
+            ({
+              id: field.name,
+              name: field.name,
+              type: FIELD_TYPE_TO_PARAMETER_TYPE[field.type] ?? "any",
+            }) as WorkflowParameter
+        );
+
+        // Remove all edges connected to this node (schema change breaks connections)
+        if (edges && deleteEdge) {
+          for (const edge of edges) {
+            if (edge.target === nodeId || edge.source === nodeId) {
+              deleteEdge(edge.id);
+            }
+          }
+        }
+
+        // Build the schema input from scratch to avoid race conditions
+        // between competing input updates (onChange vs updateNodeData)
+        updateNodeData(nodeId, (current) => {
+          const existingSchema = current.inputs?.find(
+            (i) => i.id === "schema"
+          );
+          const schemaInput = {
+            ...(existingSchema ?? {
+              id: "schema",
+              name: "schema",
+              type: "schema" as const,
+              hidden: true,
+              required: true,
+            }),
+            value: selectedSchemaId,
+          } as WorkflowParameter;
+          return {
+            inputs: [schemaInput, ...fieldInputs],
+          };
+        });
+      } finally {
+        setLoading(false);
+      }
+    },
+    [organization?.id, updateNodeData, edges, deleteEdge, nodeId]
+  );
+
+  const handleChange = useCallback(
+    (val: string) => {
+      if (val === CREATE_NEW) {
+        setIsCreateDialogOpen(true);
+        return;
+      }
+      applySchema(val);
+    },
+    [applySchema]
+  );
+
+  const handleCreate = useCallback(
+    async (data: { name: string; description: string; fields: Field[] }) => {
+      if (!organization?.id) return;
+      const { createSchema } = await import("@/services/schema-service");
+      const response = await createSchema(data, organization.id);
+      await mutateSchemas();
+      applySchema(response.schema.id);
+    },
+    [organization?.id, mutateSchemas, applySchema]
+  );
+
+  const isLoading = loading || isSchemasLoading;
+  const selectedName = schemas?.find((s) => s.id === schemaId)?.name;
+
+  return (
+    <div className={cn("p-2", className)}>
+      <Select
+        value={schemaId || ""}
+        onValueChange={handleChange}
+        disabled={disabled || isLoading}
+      >
+        <SelectTrigger className="h-auto text-xs">
+          <SelectValue
+            placeholder={
+              isLoading
+                ? "Loading..."
+                : schemas?.length === 0
+                  ? "No schemas"
+                  : "Select schema"
+            }
+          >
+            {selectedName}
+          </SelectValue>
+        </SelectTrigger>
+        <SelectContent>
+          {schemas?.map((schema) => (
+            <SelectItem key={schema.id} value={schema.id} className="text-xs">
+              {schema.name}
+            </SelectItem>
+          ))}
+          <SelectSeparator />
+          <SelectItem value={CREATE_NEW} className="text-xs">
+            + New Schema
+          </SelectItem>
+        </SelectContent>
+      </Select>
+
+      <SchemaDialog
+        open={isCreateDialogOpen}
+        onOpenChange={setIsCreateDialogOpen}
+        onSubmit={handleCreate}
+        title="Create New Schema"
+        submitLabel="Create Schema"
+      />
+    </div>
+  );
+}
+
+export const schemaComposeInputWidget = createWidget({
+  component: SchemaComposeInputWidget,
+  nodeTypes: ["json-schema-compose"],
+  inputField: "schema",
+  extractConfig: (nodeId, inputs) => ({
+    nodeId,
+    schemaId: getInputValue(inputs, "schema", ""),
+    hasSchemaInputs: (inputs ?? []).length > 1,
+  }),
+});
diff --git a/packages/runtime/src/nodes/json/json-schema-compose-node.test.ts b/packages/runtime/src/nodes/json/json-schema-compose-node.test.ts
@@ -0,0 +1,104 @@
+import type { NodeContext } from "@dafthunk/runtime";
+import type { Node, Schema } from "@dafthunk/types";
+import { describe, expect, it } from "vitest";
+import { JsonSchemaComposeNode } from "./json-schema-compose-node";
+
+const testSchema: Schema = {
+  name: "person",
+  fields: [
+    { name: "name", type: "string", required: true },
+    { name: "age", type: "integer" },
+    { name: "active", type: "boolean" },
+  ],
+};
+
+function createNode() {
+  return new JsonSchemaComposeNode({
+    nodeId: "test",
+  } as unknown as Node);
+}
+
+function createContext(inputs: Record<string, unknown>): NodeContext {
+  return {
+    nodeId: "test",
+    organizationId: "org-1",
+    inputs,
+    getIntegration: async () => {
+      throw new Error("No integrations in test");
+    },
+  } as unknown as NodeContext;
+}
+
+describe("JsonSchemaComposeNode", () => {
+  it("should compose a record from individual field inputs", async () => {
+    const node = createNode();
+    const context = createContext({
+      schema: testSchema,
+      name: "Alice",
+      age: 30,
+      active: true,
+    });
+
+    const result = await node.execute(context);
+    expect(result.status).toBe("completed");
+    expect(result.outputs?.record).toEqual({
+      name: "Alice",
+      age: 30,
+      active: true,
+    });
+  });
+
+  it("should output null for missing field inputs", async () => {
+    const node = createNode();
+    const context = createContext({
+      schema: testSchema,
+      name: "Bob",
+    });
+
+    const result = await node.execute(context);
+    expect(result.status).toBe("completed");
+    expect(result.outputs?.record).toEqual({
+      name: "Bob",
+      age: null,
+      active: null,
+    });
+  });
+
+  it("should ignore extra inputs not in the schema", async () => {
+    const node = createNode();
+    const context = createContext({
+      schema: testSchema,
+      name: "Carol",
+      age: 25,
+      active: false,
+      email: "c@x.com",
+    });
+
+    const result = await node.execute(context);
+    expect(result.status).toBe("completed");
+    const record = result.outputs?.record as Record<string, unknown>;
+    expect(record.name).toBe("Carol");
+    expect(record.email).toBeUndefined();
+  });
+
+  it("should error when schema is missing", async () => {
+    const node = createNode();
+    const context = createContext({ name: "X" });
+
+    const result = await node.execute(context);
+    expect(result.status).toBe("error");
+    expect(result.error).toContain("schema must be selected");
+  });
+
+  it("should error when schema is not a valid object", async () => {
+    const node = createNode();
+    const context = createContext({
+      schema: "not-an-object",
+      name: "X",
+    });
+
+    const result = await node.execute(context);
+    expect(result.status).toBe("error");
+    expect(result.error).toContain("schema must be selected");
+  });
+});
diff --git a/packages/runtime/src/nodes/json/json-schema-compose-node.ts b/packages/runtime/src/nodes/json/json-schema-compose-node.ts
