fix(scheduled-tasks): fix scheduled tasks schema validation (#5091)

* fix(scheduled-tasks): fix schema rejection

* fix(db): fix duplicate db query
This commit is contained in:
Siddharth Ganesan
2026-06-15 22:11:19 -07:00
committed by GitHub
parent 91666b585f
commit 6cbaf4282d
6 changed files with 132 additions and 30 deletions
+34 -21
View File
@@ -9,7 +9,7 @@ import {
} from '@sim/workflow-authz'
import { and, eq, isNull } from 'drizzle-orm'
import { type NextRequest, NextResponse } from 'next/server'
import { updateScheduleContract } from '@/lib/api/contracts/schedules'
import { getScheduleByIdContract, updateScheduleContract } from '@/lib/api/contracts/schedules'
import { parseRequest } from '@/lib/api/server'
import { getSession } from '@/lib/auth'
import { generateRequestId } from '@/lib/core/utils/request'
@@ -27,16 +27,7 @@ const logger = createLogger('ScheduleAPI')
export const dynamic = 'force-dynamic'
type ScheduleRow = {
id: string
workflowId: string | null
status: string
cronExpression: string | null
timezone: string | null
sourceType: string | null
sourceWorkspaceId: string | null
jobTitle: string | null
}
type ScheduleRow = typeof workflowSchedule.$inferSelect
async function fetchAndAuthorize(
requestId: string,
@@ -45,16 +36,7 @@ async function fetchAndAuthorize(
action: 'read' | 'write'
): Promise<{ schedule: ScheduleRow; workspaceId: string | null } | NextResponse> {
const [schedule] = await db
.select({
id: workflowSchedule.id,
workflowId: workflowSchedule.workflowId,
status: workflowSchedule.status,
cronExpression: workflowSchedule.cronExpression,
timezone: workflowSchedule.timezone,
sourceType: workflowSchedule.sourceType,
sourceWorkspaceId: workflowSchedule.sourceWorkspaceId,
jobTitle: workflowSchedule.jobTitle,
})
.select()
.from(workflowSchedule)
.where(and(eq(workflowSchedule.id, scheduleId), isNull(workflowSchedule.archivedAt)))
.limit(1)
@@ -103,6 +85,37 @@ async function fetchAndAuthorize(
return { schedule, workspaceId: authorization.workflow.workspaceId ?? null }
}
export const GET = withRouteHandler(
async (request: NextRequest, context: { params: Promise<{ id: string }> }) => {
const requestId = generateRequestId()
try {
const session = await getSession()
if (!session?.user?.id) {
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
}
const parsed = await parseRequest(getScheduleByIdContract, request, context, {
validationErrorResponse: () =>
NextResponse.json({ error: 'Invalid request' }, { status: 400 }),
})
if (!parsed.success) return parsed.response
const { id: scheduleId } = parsed.data.params
// fetchAndAuthorize already loads the full row (and 404s if missing), so
// return it directly — no second query.
const authResult = await fetchAndAuthorize(requestId, scheduleId, session.user.id, 'read')
if (authResult instanceof NextResponse) return authResult
return NextResponse.json({ schedule: authResult.schedule })
} catch (error) {
logger.error(`[${requestId}] Failed to get schedule`, { error })
return NextResponse.json({ error: 'Failed to get schedule' }, { status: 500 })
}
}
)
export const PUT = withRouteHandler(
async (request: NextRequest, context: { params: Promise<{ id: string }> }) => {
const requestId = generateRequestId()
@@ -53,7 +53,7 @@ import { useUsageLimits } from '@/app/workspace/[workspaceId]/w/[workflowId]/com
import { useWorkflowExecution } from '@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution'
import { useFolders } from '@/hooks/queries/folders'
import { useLogDetail } from '@/hooks/queries/logs'
import { useWorkspaceSchedules } from '@/hooks/queries/schedules'
import { useScheduleById } from '@/hooks/queries/schedules'
import { downloadTableExport } from '@/hooks/queries/tables'
import { useWorkflows } from '@/hooks/queries/workflows'
import { useWorkspaceFiles } from '@/hooks/queries/workspace-files'
@@ -693,12 +693,8 @@ interface EmbeddedScheduledTaskProps {
scheduleId: string
}
function EmbeddedScheduledTask({ workspaceId, scheduleId }: EmbeddedScheduledTaskProps) {
const { data: schedules = [], isLoading, isError } = useWorkspaceSchedules(workspaceId)
const schedule = useMemo(
() => schedules.find((s) => s.id === scheduleId),
[schedules, scheduleId]
)
function EmbeddedScheduledTask({ scheduleId }: EmbeddedScheduledTaskProps) {
const { data: schedule, isLoading, isError } = useScheduleById(scheduleId)
if (isLoading && !schedule) return LOADING_SKELETON
+26
View File
@@ -11,6 +11,7 @@ import {
deleteScheduleContract,
disableScheduleContract,
excludeOccurrenceContract,
getScheduleByIdContract,
getScheduleContract,
listWorkspaceSchedulesContract,
reactivateScheduleContract,
@@ -31,6 +32,7 @@ export const scheduleKeys = {
details: () => [...scheduleKeys.all, 'detail'] as const,
schedule: (workflowId: string, blockId: string) =>
[...scheduleKeys.details(), workflowId, blockId] as const,
byId: (scheduleId: string) => [...scheduleKeys.details(), scheduleId] as const,
}
export type ScheduleData = WorkflowScheduleRow
@@ -88,6 +90,30 @@ export function useWorkspaceSchedules(workspaceId?: string) {
})
}
/**
* Fetch a single schedule (job) by id. Used by the mothership resource viewer so
* opening a scheduled-task artifact does a lightweight by-id read instead of the
* whole-workspace `useWorkspaceSchedules` fetch (which contended with the chat
* stream connection and stalled start/resume).
*/
export function useScheduleById(scheduleId?: string) {
return useQuery({
queryKey: scheduleKeys.byId(scheduleId ?? ''),
queryFn: async ({ signal }) => {
if (!scheduleId) throw new Error('Schedule ID required')
const data = await requestJson(getScheduleByIdContract, {
params: { id: scheduleId },
signal,
})
return data.schedule
},
enabled: Boolean(scheduleId),
staleTime: 30 * 1000,
placeholderData: keepPreviousData,
})
}
/**
* Hook to fetch schedule data for a workflow block
*/
+17
View File
@@ -216,6 +216,23 @@ export const listWorkspaceSchedulesContract = defineRouteContract({
},
})
/**
* Single-schedule read by id. Used by the mothership resource viewer so opening
* a scheduled-task artifact does a lightweight by-id fetch instead of pulling
* the entire workspace schedule list (which contended with the chat stream).
*/
export const getScheduleByIdContract = defineRouteContract({
method: 'GET',
path: '/api/schedules/[id]',
params: scheduleIdParamsSchema,
response: {
mode: 'json',
schema: z.object({
schedule: workflowScheduleRowSchema,
}),
},
})
/**
* Newly-created job schedules emit a partial summary with the canonical fields
* the route synthesizes server-side; everything else is filled in on
+4
View File
@@ -75,6 +75,7 @@ const ResourceAttachmentSchema = z.object({
'filefolder',
'task',
'log',
'scheduledtask',
'generic',
]),
id: z.string().min(1),
@@ -91,6 +92,7 @@ const GENERIC_RESOURCE_TITLE: Record<z.infer<typeof ResourceAttachmentSchema>['t
filefolder: 'File Folder',
task: 'Task',
log: 'Log',
scheduledtask: 'Scheduled Task',
generic: 'Resource',
}
@@ -108,6 +110,7 @@ const ChatContextSchema = z.object({
'file',
'folder',
'filefolder',
'scheduledtask',
'integration',
'skill',
]),
@@ -123,6 +126,7 @@ const ChatContextSchema = z.object({
folderId: z.string().optional(),
fileFolderId: z.string().optional(),
skillId: z.string().optional(),
scheduleId: z.string().optional(),
})
const ChatMessageSchema = z.object({
+48 -2
View File
@@ -1,11 +1,12 @@
import { db, dbReplica } from '@sim/db'
import { knowledgeBase } from '@sim/db/schema'
import { knowledgeBase, workflowSchedule } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import {
authorizeWorkflowByWorkspacePermission,
getActiveWorkflowRecord,
} from '@sim/workflow-authz'
import { and, eq, isNull } from 'drizzle-orm'
import { and, eq, isNull, ne } from 'drizzle-orm'
import { normalizeVfsSegment } from '@/lib/copilot/vfs/normalize-segment'
import {
buildVfsFolderPathMap,
canonicalBlockVfsPath,
@@ -168,6 +169,16 @@ export async function processContextsServer(
path: result.path,
}
}
if (ctx.kind === 'scheduledtask' && ctx.scheduleId && currentWorkspaceId) {
const result = await resolveScheduledTaskResource(ctx.scheduleId, currentWorkspaceId)
if (!result) return null
return {
type: 'active_resource',
tag: ctx.label ? `@${ctx.label}` : '@',
content: result.content,
path: result.path,
}
}
if (ctx.kind === 'docs') {
try {
const { searchDocumentationServerTool } = await import(
@@ -695,6 +706,9 @@ export async function resolveActiveResourceContext(
case 'filefolder': {
return await resolveFileFolderResource(resourceId, workspaceId)
}
case 'scheduledtask': {
return await resolveScheduledTaskResource(resourceId, workspaceId)
}
default:
return null
}
@@ -718,6 +732,38 @@ async function resolveTableResource(
}
}
async function resolveScheduledTaskResource(
scheduleId: string,
workspaceId: string
): Promise<AgentContext | null> {
const [row] = await db
.select({ id: workflowSchedule.id, jobTitle: workflowSchedule.jobTitle })
.from(workflowSchedule)
.where(
and(
eq(workflowSchedule.id, scheduleId),
eq(workflowSchedule.sourceWorkspaceId, workspaceId),
eq(workflowSchedule.sourceType, 'job'),
isNull(workflowSchedule.archivedAt),
// Mirror the VFS materializer (workspace-vfs `materializeJobs`), which
// excludes completed jobs — otherwise we'd point at a meta.json it never
// wrote and the agent's read would dangle.
ne(workflowSchedule.status, 'completed')
)
)
.limit(1)
if (!row) return null
// The VFS materializes jobs at `jobs/{sanitized title}/meta.json` (see
// workspace-vfs `materializeJobs`); emit the same lightweight path pointer so
// the agent reads it via the VFS instead of us inlining the (heavy) row.
return {
type: 'active_resource',
tag: '@active_resource',
content: '',
path: `jobs/${normalizeVfsSegment(row.jobTitle || row.id)}/meta.json`,
}
}
async function resolveFileResource(
fileId: string,
workspaceId: string