feat: index json schemas on tasks and schemaTask

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -77,7 +77,7 @@ export class CreateBackgroundWorkerService extends BaseService {
           version: nextVersion,
           runtimeEnvironmentId: environment.id,
           projectId: project.id,
-          metadata: body.metadata,
+          metadata: body.metadata as any,
           contentHash: body.metadata.contentHash,
           cliVersion: body.metadata.cliPackageVersion,
           sdkVersion: body.metadata.packageVersion,
@@ -280,6 +280,7 @@ async function createWorkerTask(
         fileId: tasksToBackgroundFiles?.get(task.id) ?? null,
         maxDurationInSeconds: task.maxDuration ? clampMaxDuration(task.maxDuration) : null,
         queueId: queue.id,
+        payloadSchema: task.payloadSchema as any,
       },
     });
   } catch (error) {

apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV3.server.ts

@@ -48,7 +48,7 @@ export class CreateDeploymentBackgroundWorkerServiceV3 extends BaseService {
           version: deployment.version,
           runtimeEnvironmentId: environment.id,
           projectId: environment.projectId,
-          metadata: body.metadata,
+          metadata: body.metadata as any,
           contentHash: body.metadata.contentHash,
           cliVersion: body.metadata.cliPackageVersion,
           sdkVersion: body.metadata.packageVersion,

apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV4.server.ts

@@ -65,7 +65,7 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
           version: deployment.version,
           runtimeEnvironmentId: environment.id,
           projectId: environment.projectId,
-          metadata: body.metadata,
+          metadata: body.metadata as any,
           contentHash: body.metadata.contentHash,
           cliVersion: body.metadata.cliPackageVersion,
           sdkVersion: body.metadata.packageVersion,

internal-packages/database/prisma/migrations/20250730084611_add_payload_schema_to_background_worker_task/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "BackgroundWorkerTask" ADD COLUMN     "payloadSchema" JSONB;

internal-packages/database/prisma/schema.prisma

@@ -510,6 +510,8 @@ model BackgroundWorkerTask {
 
   triggerSource TaskTriggerSource @default(STANDARD)
 
+  payloadSchema Json?
+
   @@unique([workerId, slug])
   // Quick lookup of task identifiers
   @@index([projectId, slug])

packages/cli-v3/package.json

@@ -91,6 +91,7 @@
     "@opentelemetry/semantic-conventions": "1.36.0",
     "@trigger.dev/build": "workspace:4.0.0-v4-beta.26",
     "@trigger.dev/core": "workspace:4.0.0-v4-beta.26",
+    "@trigger.dev/schema-to-json": "workspace:4.0.0-v4-beta.25",
     "ansi-escapes": "^7.0.0",
     "braces": "^3.0.3",
     "c12": "^1.11.1",

packages/cli-v3/src/entryPoints/dev-index-worker.ts

@@ -18,6 +18,7 @@ import { registerResources } from "../indexing/registerResources.js";
 import { env } from "std-env";
 import { normalizeImportPath } from "../utilities/normalizeImportPath.js";
 import { detectRuntimeVersion } from "@trigger.dev/core/v3/build";
+import { schemaToJsonSchema, initializeSchemaConverters } from "@trigger.dev/schema-to-json";
 
 sourceMapSupport.install({
   handleUncaughtExceptions: false,
@@ -100,7 +101,7 @@ async function bootstrap() {
 
 const { buildManifest, importErrors, config, timings } = await bootstrap();
 
-let tasks = resourceCatalog.listTaskManifests();
+let tasks = await convertSchemasToJsonSchemas(resourceCatalog.listTaskManifests());
 
 // If the config has retry defaults, we need to apply them to all tasks that don't have any retry settings
 if (config.retries?.default) {
@@ -190,3 +191,20 @@ await new Promise<void>((resolve) => {
     resolve();
   }, 10);
 });
+
+async function convertSchemasToJsonSchemas(tasks: TaskManifest[]): Promise<TaskManifest[]> {
+  await initializeSchemaConverters();
+
+  const convertedTasks = tasks.map((task) => {
+    const schema = resourceCatalog.getTaskSchema(task.id);
+
+    if (schema) {
+      const result = schemaToJsonSchema(schema);
+      return { ...task, payloadSchema: result?.jsonSchema };
+    }
+
+    return task;
+  });
+
+  return convertedTasks;
+}

packages/cli-v3/src/entryPoints/managed-index-worker.ts

@@ -18,6 +18,7 @@ import { registerResources } from "../indexing/registerResources.js";
 import { env } from "std-env";
 import { normalizeImportPath } from "../utilities/normalizeImportPath.js";
 import { detectRuntimeVersion } from "@trigger.dev/core/v3/build";
+import { schemaToJsonSchema, initializeSchemaConverters } from "@trigger.dev/schema-to-json";
 
 sourceMapSupport.install({
   handleUncaughtExceptions: false,
@@ -100,7 +101,7 @@ async function bootstrap() {
 
 const { buildManifest, importErrors, config, timings } = await bootstrap();
 
-let tasks = resourceCatalog.listTaskManifests();
+let tasks = await convertSchemasToJsonSchemas(resourceCatalog.listTaskManifests());
 
 // If the config has retry defaults, we need to apply them to all tasks that don't have any retry settings
 if (config.retries?.default) {
@@ -196,3 +197,20 @@ await new Promise<void>((resolve) => {
     resolve();
   }, 10);
 });
+
+async function convertSchemasToJsonSchemas(tasks: TaskManifest[]): Promise<TaskManifest[]> {
+  await initializeSchemaConverters();
+
+  const convertedTasks = tasks.map((task) => {
+    const schema = resourceCatalog.getTaskSchema(task.id);
+
+    if (schema) {
+      const result = schemaToJsonSchema(schema);
+      return { ...task, payloadSchema: result?.jsonSchema };
+    }
+
+    return task;
+  });
+
+  return convertedTasks;
+}

packages/core/src/v3/resource-catalog/catalog.ts

@@ -1,5 +1,5 @@
 import { QueueManifest, TaskManifest, WorkerManifest } from "../schemas/index.js";
-import { TaskMetadataWithFunctions } from "../types/index.js";
+import { TaskMetadataWithFunctions, TaskSchema } from "../types/index.js";
 
 export interface ResourceCatalog {
   setCurrentFileContext(filePath: string, entryPoint: string): void;
@@ -13,4 +13,5 @@ export interface ResourceCatalog {
   registerWorkerManifest(workerManifest: WorkerManifest): void;
   registerQueueMetadata(queue: QueueManifest): void;
   listQueueManifests(): Array<QueueManifest>;
+  getTaskSchema(id: string): TaskSchema | undefined;
 }

packages/core/src/v3/resource-catalog/index.ts

@@ -1,7 +1,7 @@
 const API_NAME = "resource-catalog";
 
 import { QueueManifest, TaskManifest, WorkerManifest } from "../schemas/index.js";
-import { TaskMetadataWithFunctions } from "../types/index.js";
+import { TaskMetadataWithFunctions, TaskSchema } from "../types/index.js";
 import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js";
 import { type ResourceCatalog } from "./catalog.js";
 import { NoopResourceCatalog } from "./noopResourceCatalog.js";
@@ -65,6 +65,10 @@ export class ResourceCatalogAPI {
     return this.#getCatalog().getTask(id);
   }
 
+  public getTaskSchema(id: string): TaskSchema | undefined {
+    return this.#getCatalog().getTaskSchema(id);
+  }
+
   public taskExists(id: string): boolean {
     return this.#getCatalog().taskExists(id);
   }

packages/core/src/v3/resource-catalog/noopResourceCatalog.ts

@@ -1,5 +1,5 @@
 import { QueueManifest, TaskManifest, WorkerManifest } from "../schemas/index.js";
-import { TaskMetadataWithFunctions } from "../types/index.js";
+import { TaskMetadataWithFunctions, TaskSchema } from "../types/index.js";
 import { ResourceCatalog } from "./catalog.js";
 
 export class NoopResourceCatalog implements ResourceCatalog {
@@ -31,6 +31,10 @@ export class NoopResourceCatalog implements ResourceCatalog {
     return undefined;
   }
 
+  getTaskSchema(id: string): TaskSchema | undefined {
+    return undefined;
+  }
+
   taskExists(id: string): boolean {
     return false;
   }

packages/core/src/v3/resource-catalog/standardResourceCatalog.ts

@@ -5,10 +5,11 @@ import {
   WorkerManifest,
   QueueManifest,
 } from "../schemas/index.js";
-import { TaskMetadataWithFunctions } from "../types/index.js";
+import { TaskMetadataWithFunctions, TaskSchema } from "../types/index.js";
 import { ResourceCatalog } from "./catalog.js";
 
 export class StandardResourceCatalog implements ResourceCatalog {
+  private _taskSchemas: Map<string, TaskSchema> = new Map();
   private _taskMetadata: Map<string, TaskMetadata> = new Map();
   private _taskFunctions: Map<string, TaskMetadataWithFunctions["fns"]> = new Map();
   private _taskFileMetadata: Map<string, TaskFileMetadata> = new Map();
@@ -72,6 +73,10 @@ export class StandardResourceCatalog implements ResourceCatalog {
 
     this._taskMetadata.set(task.id, metadata);
     this._taskFunctions.set(task.id, fns);
+
+    if (task.schema) {
+      this._taskSchemas.set(task.id, task.schema);
+    }
   }
 
   updateTaskMetadata(id: string, updates: Partial<TaskMetadataWithFunctions>): void {
@@ -107,15 +112,21 @@ export class StandardResourceCatalog implements ResourceCatalog {
         continue;
       }
 
-      result.push({
+      const taskManifest = {
         ...metadata,
         ...fileMetadata,
-      });
+      };
+
+      result.push(taskManifest);
     }
 
     return result;
   }
 
+  getTaskSchema(id: string): TaskSchema | undefined {
+    return this._taskSchemas.get(id);
+  }
+
   listQueueManifests(): Array<QueueManifest> {
     return Array.from(this._queueMetadata.values());
   }

packages/core/src/v3/schemas/resources.ts

@@ -13,6 +13,8 @@ export const TaskResource = z.object({
   triggerSource: z.string().optional(),
   schedule: ScheduleMetadata.optional(),
   maxDuration: z.number().optional(),
+  // JSONSchema type - using z.unknown() for runtime validation to accept JSONSchema7
+  payloadSchema: z.unknown().optional(),
 });
 
 export type TaskResource = z.infer<typeof TaskResource>;

packages/core/src/v3/schemas/schemas.ts

@@ -189,6 +189,7 @@ const taskMetadata = {
   triggerSource: z.string().optional(),
   schedule: ScheduleMetadata.optional(),
   maxDuration: z.number().optional(),
+  payloadSchema: z.unknown().optional(),
 };
 
 export const TaskMetadata = z.object(taskMetadata);

packages/core/src/v3/types/index.ts

@@ -7,6 +7,7 @@ export * from "./tasks.js";
 export * from "./idempotencyKeys.js";
 export * from "./tools.js";
 export * from "./queues.js";
+export * from "./jsonSchema.js";
 
 type ResolveEnvironmentVariablesOptions = {
   variables: Record<string, string> | Array<{ name: string; value: string }>;

packages/core/src/v3/types/jsonSchema.ts

@@ -0,0 +1,76 @@
+/**
+ * JSON Schema type definition - compatible with JSON Schema Draft 7
+ * Based on the JSONSchema7 type from @types/json-schema but defined inline to avoid import issues
+ */
+export interface JSONSchema {
+  $id?: string;
+  $ref?: string;
+  $schema?: string;
+  $comment?: string;
+
+  type?: JSONSchemaType | JSONSchemaType[];
+  enum?: any[];
+  const?: any;
+
+  // Number/Integer validations
+  multipleOf?: number;
+  maximum?: number;
+  exclusiveMaximum?: number;
+  minimum?: number;
+  exclusiveMinimum?: number;
+
+  // String validations
+  maxLength?: number;
+  minLength?: number;
+  pattern?: string;
+  format?: string;
+
+  // Array validations
+  items?: JSONSchema | JSONSchema[];
+  additionalItems?: JSONSchema | boolean;
+  maxItems?: number;
+  minItems?: number;
+  uniqueItems?: boolean;
+  contains?: JSONSchema;
+
+  // Object validations
+  maxProperties?: number;
+  minProperties?: number;
+  required?: string[];
+  properties?: Record<string, JSONSchema>;
+  patternProperties?: Record<string, JSONSchema>;
+  additionalProperties?: JSONSchema | boolean;
+  dependencies?: Record<string, JSONSchema | string[]>;
+  propertyNames?: JSONSchema;
+
+  // Conditional schemas
+  if?: JSONSchema;
+  then?: JSONSchema;
+  else?: JSONSchema;
+
+  // Boolean logic
+  allOf?: JSONSchema[];
+  anyOf?: JSONSchema[];
+  oneOf?: JSONSchema[];
+  not?: JSONSchema;
+
+  // Metadata
+  title?: string;
+  description?: string;
+  default?: any;
+  readOnly?: boolean;
+  writeOnly?: boolean;
+  examples?: any[];
+
+  // Additional properties for extensibility
+  [key: string]: any;
+}
+
+export type JSONSchemaType =
+  | "string"
+  | "number"
+  | "integer"
+  | "boolean"
+  | "object"
+  | "array"
+  | "null";

packages/core/src/v3/types/tasks.ts

@@ -28,6 +28,7 @@ import { QueueOptions } from "./queues.js";
 import { AnySchemaParseFn, inferSchemaIn, inferSchemaOut, Schema } from "./schemas.js";
 import { inferToolParameters, ToolTaskParameters } from "./tools.js";
 import { Prettify } from "./utils.js";
+import { JSONSchema } from "./jsonSchema.js";
 
 export type Queue = QueueOptions;
 export type TaskSchema = Schema;
@@ -339,6 +340,12 @@ type CommonTaskOptions<
    * onFailure is called after a task run has failed (meaning the run function threw an error and won't be retried anymore)
    */
   onFailure?: OnFailureHookFunction<TPayload, TInitOutput>;
+
+  /**
+   * JSON Schema for the task payload. This will be synced to the server during indexing.
+   * Should be a valid JSON Schema Draft 7 object.
+   */
+  jsonSchema?: JSONSchema;
 };
 
 export type TaskOptions<
@@ -348,6 +355,15 @@ export type TaskOptions<
   TInitOutput extends InitOutput = any,
 > = CommonTaskOptions<TIdentifier, TPayload, TOutput, TInitOutput>;
 
+// Task options when payloadSchema is provided - payload should be any
+export type TaskOptionsWithSchema<
+  TIdentifier extends string,
+  TOutput = unknown,
+  TInitOutput extends InitOutput = any,
+> = CommonTaskOptions<TIdentifier, any, TOutput, TInitOutput> & {
+  jsonSchema: JSONSchema;
+};
+
 export type TaskWithSchemaOptions<
   TIdentifier extends string,
   TSchema extends TaskSchema | undefined = undefined,
@@ -881,6 +897,7 @@ export type TaskMetadataWithFunctions = TaskMetadata & {
     onStart?: (payload: any, params: StartFnParams) => Promise<void>;
     parsePayload?: AnySchemaParseFn;
   };
+  schema?: TaskSchema;
 };
 
 export type RunTypes<TTaskIdentifier extends string, TPayload, TOutput> = {

packages/schema-to-json/.gitignore

@@ -0,0 +1,6 @@
+node_modules
+dist
+.tshy
+.tshy-build
+*.log
+.DS_Store