diff --git a/.agents/skills/trigger-agents/SKILL.md b/.agents/skills/trigger-agents/SKILL.md new file mode 100644 index 0000000..81666a9 --- /dev/null +++ b/.agents/skills/trigger-agents/SKILL.md @@ -0,0 +1,287 @@ +--- +name: trigger-agents +description: AI agent patterns with Trigger.dev - orchestration, parallelization, routing, evaluator-optimizer, and human-in-the-loop. Use when building LLM-powered tasks that need parallel workers, approval gates, tool calling, or multi-step agent workflows. +--- + +# AI Agent Patterns with Trigger.dev + +Build production-ready AI agents using Trigger.dev's durable execution. + +## Pattern Selection + +``` +Need to... → Use +───────────────────────────────────────────────────── +Process items in parallel → Parallelization +Route to different models/handlers → Routing +Chain steps with validation gates → Prompt Chaining +Coordinate multiple specialized tasks → Orchestrator-Workers +Self-improve until quality threshold → Evaluator-Optimizer +Pause for human approval → Human-in-the-Loop (waitpoints.md) +Stream progress to frontend → Realtime Streams (streaming.md) +Let LLM call your tasks as tools → ai.tool (ai-tool.md) +``` + +--- + +## Core Patterns + +### 1. Prompt Chaining (Sequential with Gates) + +Chain LLM calls with validation between steps. Fail early if intermediate output is bad. + +```typescript +import { task } from "@trigger.dev/sdk"; +import { generateText } from "ai"; +import { openai } from "@ai-sdk/openai"; + +export const translateCopy = task({ + id: "translate-copy", + run: async ({ text, targetLanguage, maxWords }) => { + // Step 1: Generate + const draft = await generateText({ + model: openai("gpt-4o"), + prompt: `Write marketing copy about: ${text}`, + }); + + // Gate: Validate before continuing + const wordCount = draft.text.split(/\s+/).length; + if (wordCount > maxWords) { + throw new Error(`Draft too long: ${wordCount} > ${maxWords}`); + } + + // Step 2: Translate (only if gate passed) + const translated = await generateText({ + model: openai("gpt-4o"), + prompt: `Translate to ${targetLanguage}: ${draft.text}`, + }); + + return { draft: draft.text, translated: translated.text }; + }, +}); +``` + +--- + +### 2. Routing (Classify → Dispatch) + +Use a cheap model to classify, then route to appropriate handler. + +```typescript +import { task } from "@trigger.dev/sdk"; +import { generateText } from "ai"; +import { openai } from "@ai-sdk/openai"; +import { z } from "zod"; + +const routingSchema = z.object({ + model: z.enum(["gpt-4o", "o1-mini"]), + reason: z.string(), +}); + +export const routeQuestion = task({ + id: "route-question", + run: async ({ question }) => { + // Cheap classification call + const routing = await generateText({ + model: openai("gpt-4o-mini"), + messages: [ + { + role: "system", + content: `Classify question complexity. Return JSON: {"model": "gpt-4o" | "o1-mini", "reason": "..."} + - gpt-4o: simple factual questions + - o1-mini: complex reasoning, math, code`, + }, + { role: "user", content: question }, + ], + }); + + const { model } = routingSchema.parse(JSON.parse(routing.text)); + + // Route to selected model + const answer = await generateText({ + model: openai(model), + prompt: question, + }); + + return { answer: answer.text, routedTo: model }; + }, +}); +``` + +--- + +### 3. Parallelization + +Run independent LLM calls simultaneously with `batch.triggerByTaskAndWait`. + +```typescript +import { batch, task } from "@trigger.dev/sdk"; + +export const analyzeContent = task({ + id: "analyze-content", + run: async ({ text }) => { + // All three run in parallel + const { runs: [sentiment, summary, moderation] } = await batch.triggerByTaskAndWait([ + { task: analyzeSentiment, payload: { text } }, + { task: summarizeText, payload: { text } }, + { task: moderateContent, payload: { text } }, + ]); + + // Check moderation first + if (moderation.ok && moderation.output.flagged) { + return { error: "Content flagged", reason: moderation.output.reason }; + } + + return { + sentiment: sentiment.ok ? sentiment.output : null, + summary: summary.ok ? summary.output : null, + }; + }, +}); +``` + +**See:** `references/orchestration.md` for advanced patterns + +--- + +### 4. Orchestrator-Workers (Fan-out/Fan-in) + +Orchestrator extracts work items, fans out to workers, aggregates results. + +```typescript +import { batch, task } from "@trigger.dev/sdk"; + +export const factChecker = task({ + id: "fact-checker", + run: async ({ article }) => { + // Step 1: Extract claims (sequential - need output first) + const { runs: [extractResult] } = await batch.triggerByTaskAndWait([ + { task: extractClaims, payload: { article } }, + ]); + + if (!extractResult.ok) throw new Error("Failed to extract claims"); + const claims = extractResult.output; + + // Step 2: Fan-out - verify all claims in parallel + const { runs } = await batch.triggerByTaskAndWait( + claims.map(claim => ({ task: verifyClaim, payload: claim })) + ); + + // Step 3: Fan-in - aggregate results + const verified = runs + .filter((r): r is typeof r & { ok: true } => r.ok) + .map(r => r.output); + + return { claims, verifications: verified }; + }, +}); +``` + +--- + +### 5. Evaluator-Optimizer (Self-Refining Loop) + +Generate → Evaluate → Retry with feedback until approved. + +```typescript +import { task } from "@trigger.dev/sdk"; + +export const refineTranslation = task({ + id: "refine-translation", + run: async ({ text, targetLanguage, feedback, attempt = 0 }) => { + // Bail condition + if (attempt >= 5) { + return { text, status: "MAX_ATTEMPTS", attempts: attempt }; + } + + // Generate (with feedback if retrying) + const prompt = feedback + ? `Improve this translation based on feedback:\n${feedback}\n\nOriginal: ${text}` + : `Translate to ${targetLanguage}: ${text}`; + + const translation = await generateText({ + model: openai("gpt-4o"), + prompt, + }); + + // Evaluate + const evaluation = await generateText({ + model: openai("gpt-4o"), + prompt: `Evaluate translation quality. Reply APPROVED or provide specific feedback:\n${translation.text}`, + }); + + if (evaluation.text.includes("APPROVED")) { + return { text: translation.text, status: "APPROVED", attempts: attempt + 1 }; + } + + // Recursive self-call with feedback + return refineTranslation.triggerAndWait({ + text, + targetLanguage, + feedback: evaluation.text, + attempt: attempt + 1, + }).unwrap(); + }, +}); +``` + +--- + +## Trigger-Specific Features + +| Feature | What it enables | Reference | +|---------|-----------------|-----------| +| **Waitpoints** | Human approval gates, external callbacks | `references/waitpoints.md` | +| **Streams** | Real-time progress to frontend | `references/streaming.md` | +| **ai.tool** | Let LLMs call your tasks as tools | `references/ai-tool.md` | +| **batch.triggerByTaskAndWait** | Typed parallel execution | `references/orchestration.md` | + +--- + +## Error Handling + +```typescript +const { runs } = await batch.triggerByTaskAndWait([...]); + +// Check individual results +for (const run of runs) { + if (run.ok) { + console.log(run.output); // Typed output + } else { + console.error(run.error); // Error details + console.log(run.taskIdentifier); // Which task failed + } +} + +// Or filter by task type +const verifications = runs + .filter((r): r is typeof r & { ok: true } => + r.ok && r.taskIdentifier === "verify-claim" + ) + .map(r => r.output); +``` + +--- + +## Quick Reference + +```typescript +// Trigger and wait for result +const result = await myTask.triggerAndWait(payload); +if (result.ok) console.log(result.output); + +// Batch trigger same task +const results = await myTask.batchTriggerAndWait([ + { payload: item1 }, + { payload: item2 }, +]); + +// Batch trigger different tasks (typed) +const { runs } = await batch.triggerByTaskAndWait([ + { task: taskA, payload: { foo: 1 } }, + { task: taskB, payload: { bar: "x" } }, +]); + +// Self-recursion with unwrap +return myTask.triggerAndWait(newPayload).unwrap(); +``` diff --git a/.agents/skills/trigger-agents/references/ai-tool.md b/.agents/skills/trigger-agents/references/ai-tool.md new file mode 100644 index 0000000..8258afc --- /dev/null +++ b/.agents/skills/trigger-agents/references/ai-tool.md @@ -0,0 +1,237 @@ +# ai.tool Integration + +Convert Trigger.dev tasks to Vercel AI SDK tools. Let LLMs call your tasks autonomously. + +## Basic Usage + +```typescript +import { schemaTask, ai } from "@trigger.dev/sdk"; +import { generateText } from "ai"; +import { openai } from "@ai-sdk/openai"; +import { z } from "zod"; + +// 1. Define task with schema +const lookupWeather = schemaTask({ + id: "lookup-weather", + schema: z.object({ + location: z.string().describe("City name"), + units: z.enum(["celsius", "fahrenheit"]).default("celsius"), + }), + run: async ({ location, units }) => { + const weather = await fetchWeather(location, units); + return { temperature: weather.temp, conditions: weather.conditions }; + }, +}); + +// 2. Convert to AI tool +const weatherTool = ai.tool(lookupWeather); + +// 3. Use with AI SDK +export const weatherAgent = schemaTask({ + id: "weather-agent", + schema: z.object({ question: z.string() }), + run: async ({ question }) => { + const result = await generateText({ + model: openai("gpt-4o"), + prompt: question, + tools: { + lookupWeather: weatherTool, + }, + }); + + return { answer: result.text }; + }, +}); +``` + +--- + +## Schema Requirements + +The task **must** use `schemaTask` with a Zod schema: + +```typescript +// ✅ Works - has schema +const myTask = schemaTask({ + id: "my-task", + schema: z.object({ + query: z.string(), + }), + run: async (payload) => { ... }, +}); + +// ❌ Won't work - no schema +const myTask = task({ + id: "my-task", + run: async (payload: { query: string }) => { ... }, +}); +``` + +**Supported schema libraries:** +- Zod +- ArkType +- Any schema with `.toJsonSchema()` method + +--- + +## Tool Result Customization + +Customize how results are sent back to the LLM: + +```typescript +const searchTool = ai.tool(searchDatabase, { + experimental_toToolResultContent: (result) => { + // Return structured content for the LLM + return [ + { + type: "text", + text: `Found ${result.count} results:\n${result.items.map(i => i.title).join("\n")}`, + }, + ]; + }, +}); +``` + +--- + +## Accessing Tool Options + +Get execution context inside the task: + +```typescript +const myToolTask = schemaTask({ + id: "my-tool-task", + schema: z.object({ input: z.string() }), + run: async (payload) => { + // Access AI SDK tool execution options + const toolOptions = ai.currentToolOptions(); + + console.log(toolOptions); + // { toolCallId: "...", messages: [...], ... } + + return processInput(payload.input); + }, +}); +``` + +--- + +## Multiple Tools + +```typescript +const searchTool = ai.tool(searchDatabase); +const calculateTool = ai.tool(calculate); +const summarizeTool = ai.tool(summarize); + +export const agentTask = schemaTask({ + id: "agent", + schema: z.object({ task: z.string() }), + run: async ({ task }) => { + const result = await generateText({ + model: openai("gpt-4o"), + prompt: task, + tools: { + search: searchTool, + calculate: calculateTool, + summarize: summarizeTool, + }, + maxSteps: 10, // Allow multiple tool calls + }); + + return { result: result.text }; + }, +}); +``` + +--- + +## With Tool Choice + +```typescript +const result = await generateText({ + model: openai("gpt-4o"), + prompt: "What's the weather in Tokyo?", + tools: { + weather: weatherTool, + news: newsTool, + }, + toolChoice: "required", // Force tool use + // or: toolChoice: { type: "tool", toolName: "weather" } +}); +``` + +--- + +## Description from Schema + +Add descriptions for better LLM understanding: + +```typescript +const searchTask = schemaTask({ + id: "search-database", + description: "Search the product database for items matching a query", + schema: z.object({ + query: z.string().describe("Search terms"), + limit: z.number().min(1).max(100).describe("Max results to return"), + category: z.enum(["electronics", "clothing", "books"]).optional() + .describe("Filter by product category"), + }), + run: async (payload) => { ... }, +}); +``` + +--- + +## Common Pattern: Research Agent + +```typescript +const webSearch = schemaTask({ + id: "web-search", + schema: z.object({ + query: z.string(), + maxResults: z.number().default(5), + }), + run: async ({ query, maxResults }) => { + return await searchWeb(query, maxResults); + }, +}); + +const readUrl = schemaTask({ + id: "read-url", + schema: z.object({ + url: z.string().url(), + }), + run: async ({ url }) => { + return await fetchAndParse(url); + }, +}); + +export const researchAgent = schemaTask({ + id: "research-agent", + schema: z.object({ topic: z.string() }), + run: async ({ topic }) => { + const result = await generateText({ + model: openai("gpt-4o"), + system: "Research the topic thoroughly using available tools.", + prompt: topic, + tools: { + search: ai.tool(webSearch), + read: ai.tool(readUrl), + }, + maxSteps: 20, + }); + + return { research: result.text }; + }, +}); +``` + +--- + +## Tips + +1. **Always use schemaTask** - regular `task` won't work +2. **Add descriptions** - helps LLM understand when to use the tool +3. **Use `.describe()`** - on schema fields for parameter hints +4. **Set maxSteps** - allow multiple tool calls for complex tasks +5. **Customize results** - use `experimental_toToolResultContent` for better LLM context diff --git a/.agents/skills/trigger-agents/references/orchestration.md b/.agents/skills/trigger-agents/references/orchestration.md new file mode 100644 index 0000000..c2524d2 --- /dev/null +++ b/.agents/skills/trigger-agents/references/orchestration.md @@ -0,0 +1,241 @@ +# Orchestration Patterns + +Advanced patterns for `batch.triggerByTaskAndWait` and task coordination. + +## Basic Usage + +```typescript +import { batch, task } from "@trigger.dev/sdk"; + +// Trigger different tasks, get typed results +const { runs } = await batch.triggerByTaskAndWait([ + { task: taskA, payload: { foo: "bar" } }, // payload typed to taskA + { task: taskB, payload: { num: 42 } }, // payload typed to taskB +]); + +// Results are typed based on position +if (runs[0].ok) { + console.log(runs[0].output); // typed as taskA output +} +``` + +## Destructured Results + +```typescript +const { + runs: [userRun, postsRun, settingsRun], +} = await batch.triggerByTaskAndWait([ + { task: fetchUser, payload: { id } }, + { task: fetchPosts, payload: { userId: id } }, + { task: fetchSettings, payload: { userId: id } }, +]); + +// Each run is individually typed +const user = userRun.ok ? userRun.output : null; +const posts = postsRun.ok ? postsRun.output : []; +``` + +--- + +## Error Handling Per-Task + +```typescript +const { runs } = await batch.triggerByTaskAndWait([ + { task: riskyTask, payload: item1 }, + { task: riskyTask, payload: item2 }, + { task: riskyTask, payload: item3 }, +]); + +// Individual error handling +const results = runs.map(run => { + if (run.ok) { + return { success: true, data: run.output }; + } + return { + success: false, + error: run.error, + taskId: run.taskIdentifier, + runId: run.id, + }; +}); + +// Or throw if any failed +const failed = runs.filter(r => !r.ok); +if (failed.length > 0) { + throw new Error(`${failed.length} tasks failed`); +} +``` + +--- + +## Filtering by Task Identifier + +When running mixed task types, filter results by `taskIdentifier`: + +```typescript +const { runs } = await batch.triggerByTaskAndWait([ + ...claims.map(c => ({ task: verifySource, payload: c })), + ...claims.map(c => ({ task: analyzeHistory, payload: c })), +]); + +// Filter to specific task results +const verifications = runs + .filter((r): r is typeof r & { ok: true } => + r.ok && r.taskIdentifier === "verify-source" + ) + .map(r => r.output as SourceVerification); + +const analyses = runs + .filter((r): r is typeof r & { ok: true } => + r.ok && r.taskIdentifier === "analyze-history" + ) + .map(r => r.output as HistoricalAnalysis); +``` + +--- + +## Fan-out/Fan-in Pattern + +```typescript +export const processItems = task({ + id: "process-items", + run: async ({ items }) => { + // Fan-out: process all items in parallel + const { runs } = await batch.triggerByTaskAndWait( + items.map(item => ({ task: processItem, payload: item })) + ); + + // Fan-in: aggregate results + const successful = runs.filter(r => r.ok).map(r => r.output); + const failed = runs.filter(r => !r.ok); + + return { + processed: successful.length, + failed: failed.length, + results: successful, + errors: failed.map(f => ({ id: f.id, error: f.error })), + }; + }, +}); +``` + +--- + +## Sequential Then Parallel + +```typescript +export const orchestrator = task({ + id: "orchestrator", + run: async ({ input }) => { + // Step 1: Sequential preprocessing + const { runs: [prepResult] } = await batch.triggerByTaskAndWait([ + { task: preprocess, payload: { input } }, + ]); + + if (!prepResult.ok) { + throw new Error(`Preprocessing failed: ${prepResult.error}`); + } + + const items = prepResult.output; + + // Step 2: Parallel processing + const { runs } = await batch.triggerByTaskAndWait( + items.map(item => ({ task: processItem, payload: item })) + ); + + // Step 3: Sequential aggregation + const { runs: [aggResult] } = await batch.triggerByTaskAndWait([ + { task: aggregate, payload: { results: runs.filter(r => r.ok).map(r => r.output) } }, + ]); + + return aggResult.ok ? aggResult.output : null; + }, +}); +``` + +--- + +## Same Task, Multiple Items + +For batch processing the same task: + +```typescript +// Using batchTriggerAndWait (single task type) +const results = await processItem.batchTriggerAndWait([ + { payload: item1 }, + { payload: item2 }, + { payload: item3 }, +]); + +// Equivalent using batch.triggerByTaskAndWait +const { runs } = await batch.triggerByTaskAndWait([ + { task: processItem, payload: item1 }, + { task: processItem, payload: item2 }, + { task: processItem, payload: item3 }, +]); +``` + +--- + +## Concurrency Control + +Control parallelism via queue settings on child tasks: + +```typescript +import { queue, task } from "@trigger.dev/sdk"; + +const rateLimitedQueue = queue({ + name: "api-calls", + concurrencyLimit: 5, // Max 5 concurrent +}); + +export const callExternalApi = task({ + id: "call-external-api", + queue: rateLimitedQueue, + run: async (payload) => { + // Rate limited to 5 concurrent executions + return fetch(payload.url); + }, +}); + +// Parent can batch trigger many - queue handles concurrency +export const batchProcess = task({ + id: "batch-process", + run: async ({ urls }) => { + // Will queue up, respecting concurrencyLimit: 5 + return callExternalApi.batchTriggerAndWait( + urls.map(url => ({ payload: { url } })) + ); + }, +}); +``` + +--- + +## Streaming Batch Items + +For large batches, stream items instead of loading all at once: + +```typescript +import { batch } from "@trigger.dev/sdk"; + +// Generator function for items +async function* generateItems() { + for await (const record of database.cursor()) { + yield { task: processRecord, payload: record }; + } +} + +// Stream to batch trigger +const { runs } = await batch.triggerByTaskAndWait(generateItems()); +``` + +--- + +## Tips + +1. **Use destructuring** for known task counts - cleaner code +2. **Filter by taskIdentifier** when mixing task types +3. **Check `.ok`** before accessing `.output` +4. **Control concurrency** on child task queues, not in orchestrator +5. **Avoid parallel waits** - use batch methods, not Promise.all with triggerAndWait diff --git a/.agents/skills/trigger-agents/references/streaming.md b/.agents/skills/trigger-agents/references/streaming.md new file mode 100644 index 0000000..15394e1 --- /dev/null +++ b/.agents/skills/trigger-agents/references/streaming.md @@ -0,0 +1,252 @@ +# Realtime Streams + +Stream data from tasks to your frontend in real-time. Perfect for AI completions, progress updates, and live status. + +## Define Streams + +Create typed stream definitions in a shared file: + +```typescript +// trigger/streams.ts +import { streams } from "@trigger.dev/sdk"; + +// Define with type and unique ID +export const progressStream = streams.define({ + id: "progress", +}); + +export const aiOutputStream = streams.define({ + id: "ai-output", +}); + +// Export type for frontend +export type STREAMS = typeof progressStream | typeof aiOutputStream; +``` + +--- + +## Emit from Tasks + +### Basic emit + +```typescript +import { task } from "@trigger.dev/sdk"; +import { progressStream } from "./streams"; + +export const processItems = task({ + id: "process-items", + run: async ({ items }) => { + for (const [i, item] of items.entries()) { + await processItem(item); + + // Emit progress + progressStream.append( + JSON.stringify({ + current: i + 1, + total: items.length, + status: `Processing ${item.name}`, + }) + ); + } + + return { processed: items.length }; + }, +}); +``` + +### Stream AI completion + +```typescript +import { task } from "@trigger.dev/sdk"; +import { streamText } from "ai"; +import { aiOutputStream } from "./streams"; + +export const generateText = task({ + id: "generate-text", + run: async ({ prompt }) => { + const result = streamText({ + model: openai("gpt-4o"), + prompt, + }); + + // Pipe AI stream to Trigger stream + for await (const chunk of result.textStream) { + aiOutputStream.append(chunk); + } + + return { text: await result.text }; + }, +}); +``` + +--- + +## Child → Parent Streaming + +When child tasks need to emit to the parent's stream: + +```typescript +// Child task +export const workerTask = task({ + id: "worker", + run: async ({ item }) => { + const result = await processItem(item); + + // Emit to PARENT's stream, not this task's + progressStream.append( + JSON.stringify({ item: item.id, status: "done" }), + { target: "parent" } + ); + + return result; + }, +}); + +// Parent task - frontend subscribes to this run +export const orchestrator = task({ + id: "orchestrator", + run: async ({ items }) => { + // Child emits bubble up to this task's stream + return workerTask.batchTriggerAndWait( + items.map(item => ({ payload: { item } })) + ); + }, +}); +``` + +--- + +## Frontend Subscription + +### Using useRealtimeStream (Recommended) + +```tsx +import { useRealtimeStream } from "@trigger.dev/react-hooks"; +import type { progressStream } from "@/trigger/streams"; + +function Progress({ runId, accessToken }: { runId: string; accessToken: string }) { + const { data } = useRealtimeStream(runId, { + accessToken, + stream: "progress", + }); + + if (!data) return
Waiting...
; + + // data is array of emitted values + const latest = data[data.length - 1]; + const progress = JSON.parse(latest); + + return ( +
+ {progress.current} / {progress.total}: {progress.status} +
+ ); +} +``` + +### Using useRealtimeRunWithStreams + +```tsx +import { useRealtimeRunWithStreams } from "@trigger.dev/react-hooks"; +import type { processItems, STREAMS } from "@/trigger/tasks"; + +function TaskProgress({ runId, accessToken }: Props) { + const { run, streams } = useRealtimeRunWithStreams( + runId, + { accessToken } + ); + + const progressUpdates = streams.progress ?? []; + const latest = progressUpdates[progressUpdates.length - 1]; + + return ( +
+

Status: {run?.status}

+ {latest &&

Progress: {latest}

} +
+ ); +} +``` + +--- + +## Backend Consumption + +Read streams from your backend: + +```typescript +import { aiOutputStream } from "./trigger/streams"; + +async function consumeStream(runId: string) { + const stream = await aiOutputStream.read(runId, { + timeoutInSeconds: 120, + }); + + let fullText = ""; + for await (const chunk of stream) { + fullText += chunk; + console.log("Received:", chunk); + } + + return fullText; +} +``` + +--- + +## JSON Serialization Pattern + +Streams serialize as strings. For objects, use JSON: + +```typescript +// Define helper functions +export function emitProgress(update: ProgressUpdate, options?: { target: "parent" }) { + progressStream.append(JSON.stringify(update), options); +} + +// Parse on frontend +const updates = streams.progress?.map(s => JSON.parse(s) as ProgressUpdate) ?? []; +``` + +--- + +## Throttling Frontend Updates + +Prevent excessive re-renders: + +```tsx +const { data } = useRealtimeStream(runId, { + accessToken, + stream: "progress", + throttleInMs: 100, // Max 10 updates/second +}); +``` + +--- + +## AI SDK Tool Calls + +Stream tool calls and results: + +```tsx +const { streams } = useRealtimeRunWithStreams(runId, { + accessToken, +}); + +// streams.openai is TextStreamPart[] +const toolCalls = streams.openai?.filter(s => s.type === "tool-call") ?? []; +const toolResults = streams.openai?.filter(s => s.type === "tool-result") ?? []; +const textDeltas = streams.openai?.filter(s => s.type === "text-delta") ?? []; + +const fullText = textDeltas.map(d => d.textDelta).join(""); +``` + +--- + +## Tips + +1. **Use streams.define()** - always define in shared file for type safety +2. **JSON stringify objects** - streams are strings internally +3. **Use `{ target: "parent" }`** - for child-to-parent bubbling +4. **Throttle on frontend** - prevent excessive re-renders +5. **Set appropriate timeouts** - AI completions may need longer waits diff --git a/.agents/skills/trigger-agents/references/waitpoints.md b/.agents/skills/trigger-agents/references/waitpoints.md new file mode 100644 index 0000000..c3fb2ef --- /dev/null +++ b/.agents/skills/trigger-agents/references/waitpoints.md @@ -0,0 +1,247 @@ +# Human-in-the-Loop with Waitpoints + +Pause task execution for human approval, external callbacks, or async events. + +## Core API + +```typescript +import { wait } from "@trigger.dev/sdk"; + +// Create a token (pauses execution point) +const token = await wait.createToken({ + timeout: "10m", // "1h", "1d", etc. +}); + +// Wait for completion (blocks until resolved) +const result = await wait.forToken(token.id); + +if (result.ok) { + console.log(result.output); // Typed as ApprovalPayload +} else { + console.log("Timed out:", result.error); +} +``` + +--- + +## Complete Pattern: Slack Approval + +```typescript +import { task, wait } from "@trigger.dev/sdk"; + +type ApprovalToken = { + approved: boolean; + selectedOption: "optionA" | "optionB"; + approvedBy: string; +}; + +export const generateWithApproval = task({ + id: "generate-with-approval", + maxDuration: 600, // 10 min to account for human delay + run: async ({ prompt }) => { + // 1. Generate options + const options = await generateOptions(prompt); + + // 2. Create approval token + const token = await wait.createToken({ + timeout: "1h", + }); + + // 3. Send to Slack/email/webhook + await sendSlackMessage({ + text: "Please approve one option:", + options, + approvalUrl: `${process.env.APP_URL}/approve?token=${token.id}`, + // Or use: token.url for direct callback + }); + + // 4. Wait for human (task suspends here) + const result = await wait.forToken(token.id); + + if (!result.ok) { + throw new Error("Approval timed out"); + } + + // 5. Continue with approved option + return { + selected: result.output.selectedOption, + approvedBy: result.output.approvedBy, + options, + }; + }, +}); +``` + +--- + +## Completing Tokens + +### From your backend + +```typescript +import { wait } from "@trigger.dev/sdk"; + +// In your approval endpoint +export async function POST(request: Request) { + const { tokenId, approved, option, userId } = await request.json(); + + await wait.completeToken(tokenId, { + approved, + selectedOption: option, + approvedBy: userId, + }); + + return Response.json({ success: true }); +} +``` + +### Via HTTP callback (webhooks) + +```typescript +const token = await wait.createToken({ timeout: "10m" }); + +// token.url is a webhook URL that completes the token +// POST to token.url with JSON body → becomes the output +await externalService.startJob({ + callbackUrl: token.url, // Service POSTs result here +}); + +const result = await wait.forToken(token.id); +``` + +### From React (useWaitToken) + +```typescript +import { useWaitToken } from "@trigger.dev/react-hooks"; + +function ApprovalButton({ tokenId, publicToken }) { + const { complete, isCompleting } = useWaitToken(tokenId, { + accessToken: publicToken, + }); + + return ( + + ); +} +``` + +--- + +## Timeout Handling + +```typescript +const result = await wait.forToken(token.id); + +if (result.ok) { + // Human responded in time + return processApproval(result.output); +} else { + // Timed out - handle gracefully + await notifyTimeout(); + return { status: "timeout", defaultAction: "rejected" }; +} +``` + +### Using .unwrap() for cleaner code + +```typescript +try { + const approval = await wait.forToken(token.id).unwrap(); + // approval is directly typed, throws on timeout + return processApproval(approval); +} catch (error) { + // Timeout throws here + return handleTimeout(); +} +``` + +--- + +## Idempotency + +Prevent duplicate tokens for the same workflow: + +```typescript +const token = await wait.createToken({ + timeout: "1h", + idempotencyKey: `review-${workflowId}`, +}); +``` + +--- + +## Tags for Tracking + +```typescript +const token = await wait.createToken({ + timeout: "1h", + tags: [`workflow:${workflowId}`, `user:${userId}`], +}); +``` + +--- + +## Public Access Token + +For frontend completion without server round-trip: + +```typescript +const token = await wait.createToken({ timeout: "10m" }); + +// Pass to frontend +return { + tokenId: token.id, + publicToken: token.publicAccessToken, // Auto-generated, expires in 1h +}; +``` + +--- + +## Example: Multi-step Review + +```typescript +export const contentPipeline = task({ + id: "content-pipeline", + run: async ({ content }) => { + // Step 1: AI generation + const draft = await generateDraft(content); + + // Step 2: Human review + const reviewToken = await wait.createToken({ timeout: "24h" }); + await sendForReview(draft, reviewToken.id); + const review = await wait.forToken(reviewToken.id); + + if (!review.ok || !review.output.approved) { + return { status: "rejected", feedback: review.output?.feedback }; + } + + // Step 3: Final approval + const publishToken = await wait.createToken({ timeout: "1h" }); + await sendForPublishApproval(draft, publishToken.id); + const publish = await wait.forToken(publishToken.id); + + if (!publish.ok || !publish.output.approved) { + return { status: "not_published" }; + } + + // Step 4: Publish + await publishContent(draft); + return { status: "published" }; + }, +}); +``` + +--- + +## Tips + +1. **Set realistic timeouts** - account for human response time +2. **Handle timeouts gracefully** - don't throw, provide default behavior +3. **Use idempotencyKey** - prevent duplicate tokens on retries +4. **Increase maxDuration** - task needs enough time for human + processing +5. **Use publicAccessToken** - for direct frontend completion diff --git a/.agents/skills/trigger-config/SKILL.md b/.agents/skills/trigger-config/SKILL.md new file mode 100644 index 0000000..f3764cc --- /dev/null +++ b/.agents/skills/trigger-config/SKILL.md @@ -0,0 +1,240 @@ +--- +name: trigger-config +description: Configure Trigger.dev projects with trigger.config.ts. Use when setting up build extensions for Prisma, Playwright, FFmpeg, Python, or customizing deployment settings. +--- + +# Trigger.dev Configuration + +Configure your Trigger.dev project with `trigger.config.ts` and build extensions. + +## When to Use + +- Setting up a new Trigger.dev project +- Adding database support (Prisma, TypeORM) +- Configuring browser automation (Playwright, Puppeteer) +- Adding media processing (FFmpeg) +- Running Python scripts from tasks +- Syncing environment variables +- Installing system packages + +## Basic Configuration + +```ts +// trigger.config.ts +import { defineConfig } from "@trigger.dev/sdk"; + +export default defineConfig({ + project: "", + dirs: ["./trigger"], + runtime: "node", // "node", "node-22", or "bun" + logLevel: "info", + + retries: { + enabledInDev: false, + default: { + maxAttempts: 3, + minTimeoutInMs: 1000, + maxTimeoutInMs: 10000, + factor: 2, + }, + }, + + build: { + extensions: [], // Add extensions here + }, +}); +``` + +## Common Build Extensions + +### Prisma + +```ts +import { prismaExtension } from "@trigger.dev/build/extensions/prisma"; + +export default defineConfig({ + // ... + build: { + extensions: [ + prismaExtension({ + schema: "prisma/schema.prisma", + migrate: true, + directUrlEnvVarName: "DIRECT_DATABASE_URL", + }), + ], + }, +}); +``` + +### Playwright (Browser Automation) + +```ts +import { playwright } from "@trigger.dev/build/extensions/playwright"; + +extensions: [ + playwright({ + browsers: ["chromium"], // or ["chromium", "firefox", "webkit"] + }), +] +``` + +### Puppeteer + +```ts +import { puppeteer } from "@trigger.dev/build/extensions/puppeteer"; + +extensions: [puppeteer()] + +// Set env var: PUPPETEER_EXECUTABLE_PATH="/usr/bin/google-chrome-stable" +``` + +### FFmpeg (Media Processing) + +```ts +import { ffmpeg } from "@trigger.dev/build/extensions/core"; + +extensions: [ + ffmpeg({ version: "7" }), +] +// Automatically sets FFMPEG_PATH and FFPROBE_PATH +``` + +### Python + +```ts +import { pythonExtension } from "@trigger.dev/build/extensions/python"; + +extensions: [ + pythonExtension({ + scripts: ["./python/**/*.py"], + requirementsFile: "./requirements.txt", + devPythonBinaryPath: ".venv/bin/python", + }), +] + +// Usage in tasks: +const result = await python.runScript("./python/process.py", ["arg1"]); +``` + +### System Packages (apt-get) + +```ts +import { aptGet } from "@trigger.dev/build/extensions/core"; + +extensions: [ + aptGet({ + packages: ["imagemagick", "curl"], + }), +] +``` + +### Additional Files + +```ts +import { additionalFiles } from "@trigger.dev/build/extensions/core"; + +extensions: [ + additionalFiles({ + files: ["./assets/**", "./templates/**"], + }), +] +``` + +### Environment Variable Sync + +```ts +import { syncEnvVars } from "@trigger.dev/build/extensions/core"; + +extensions: [ + syncEnvVars(async (ctx) => { + return [ + { name: "API_KEY", value: await getSecret(ctx.environment) }, + { name: "ENV", value: ctx.environment }, + ]; + }), +] +``` + +## Common Extension Combinations + +### Full-Stack Web App + +```ts +extensions: [ + prismaExtension({ schema: "prisma/schema.prisma", migrate: true }), + additionalFiles({ files: ["./assets/**"] }), + syncEnvVars(async (ctx) => [...envVars]), +] +``` + +### AI/ML Processing + +```ts +extensions: [ + pythonExtension({ + scripts: ["./ai/**/*.py"], + requirementsFile: "./requirements.txt", + }), + ffmpeg({ version: "7" }), +] +``` + +### Web Scraping + +```ts +extensions: [ + playwright({ browsers: ["chromium"] }), + additionalFiles({ files: ["./selectors.json"] }), +] +``` + +## Global Lifecycle Hooks + +```ts +export default defineConfig({ + // ... + onStartAttempt: async ({ payload, ctx }) => { + console.log("Task starting:", ctx.task.id); + }, + onSuccess: async ({ payload, output, ctx }) => { + console.log("Task succeeded"); + }, + onFailure: async ({ payload, error, ctx }) => { + console.error("Task failed:", error); + }, +}); +``` + +## Machine Defaults + +```ts +export default defineConfig({ + // ... + defaultMachine: "medium-1x", + maxDuration: 300, // seconds +}); +``` + +## Telemetry Integration + +```ts +import { PrismaInstrumentation } from "@prisma/instrumentation"; + +export default defineConfig({ + // ... + telemetry: { + instrumentations: [new PrismaInstrumentation()], + }, +}); +``` + +## Best Practices + +1. **Pin versions** for reproducible builds +2. **Use `syncEnvVars`** for dynamic secrets +3. **Add native modules** to `build.external` array +4. **Debug with** `--log-level debug --dry-run` + +Extensions only affect deployment, not local development. + +See `references/config.md` for complete documentation. diff --git a/.agents/skills/trigger-config/references/config.md b/.agents/skills/trigger-config/references/config.md new file mode 100644 index 0000000..f6a4db1 --- /dev/null +++ b/.agents/skills/trigger-config/references/config.md @@ -0,0 +1,346 @@ +# Trigger.dev Configuration + +**Complete guide to configuring `trigger.config.ts` with build extensions** + +## Basic Configuration + +```ts +import { defineConfig } from "@trigger.dev/sdk"; + +export default defineConfig({ + project: "", // Required: Your project reference + dirs: ["./trigger"], // Task directories + runtime: "node", // "node", "node-22", or "bun" + logLevel: "info", // "debug", "info", "warn", "error" + + // Default retry settings + retries: { + enabledInDev: false, + default: { + maxAttempts: 3, + minTimeoutInMs: 1000, + maxTimeoutInMs: 10000, + factor: 2, + randomize: true, + }, + }, + + // Build configuration + build: { + autoDetectExternal: true, + keepNames: true, + minify: false, + extensions: [], // Build extensions go here + }, + + // Global lifecycle hooks + onStartAttempt: async ({ payload, ctx }) => { + console.log("Global task start"); + }, + onSuccess: async ({ payload, output, ctx }) => { + console.log("Global task success"); + }, + onFailure: async ({ payload, error, ctx }) => { + console.log("Global task failure"); + }, +}); +``` + +## Build Extensions + +### Database & ORM + +#### Prisma + +```ts +import { prismaExtension } from "@trigger.dev/build/extensions/prisma"; + +extensions: [ + prismaExtension({ + schema: "prisma/schema.prisma", + version: "5.19.0", // Optional: specify version + migrate: true, // Run migrations during build + directUrlEnvVarName: "DIRECT_DATABASE_URL", + typedSql: true, // Enable TypedSQL support + }), +]; +``` + +#### TypeScript Decorators (for TypeORM) + +```ts +import { emitDecoratorMetadata } from "@trigger.dev/build/extensions/typescript"; + +extensions: [ + emitDecoratorMetadata(), // Enables decorator metadata +]; +``` + +### Scripting Languages + +#### Python + +```ts +import { pythonExtension } from "@trigger.dev/build/extensions/python"; + +extensions: [ + pythonExtension({ + scripts: ["./python/**/*.py"], // Copy Python files + requirementsFile: "./requirements.txt", // Install packages + devPythonBinaryPath: ".venv/bin/python", // Dev mode binary + }), +]; + +// Usage in tasks +const result = await python.runInline(`print("Hello, world!")`); +const output = await python.runScript("./python/script.py", ["arg1"]); +``` + +### Browser Automation + +#### Playwright + +```ts +import { playwright } from "@trigger.dev/build/extensions/playwright"; + +extensions: [ + playwright({ + browsers: ["chromium", "firefox", "webkit"], // Default: ["chromium"] + headless: true, // Default: true + }), +]; +``` + +#### Puppeteer + +```ts +import { puppeteer } from "@trigger.dev/build/extensions/puppeteer"; + +extensions: [puppeteer()]; + +// Environment variable needed: +// PUPPETEER_EXECUTABLE_PATH: "/usr/bin/google-chrome-stable" +``` + +#### Lightpanda + +```ts +import { lightpanda } from "@trigger.dev/build/extensions/lightpanda"; + +extensions: [ + lightpanda({ + version: "latest", // or "nightly" + disableTelemetry: false, + }), +]; +``` + +### Media Processing + +#### FFmpeg + +```ts +import { ffmpeg } from "@trigger.dev/build/extensions/core"; + +extensions: [ + ffmpeg({ version: "7" }), // Static build, or omit for Debian version +]; + +// Automatically sets FFMPEG_PATH and FFPROBE_PATH +// Add fluent-ffmpeg to external packages if using +``` + +#### Audio Waveform + +```ts +import { audioWaveform } from "@trigger.dev/build/extensions/audioWaveform"; + +extensions: [ + audioWaveform(), // Installs Audio Waveform 1.1.0 +]; +``` + +### System & Package Management + +#### System Packages (apt-get) + +```ts +import { aptGet } from "@trigger.dev/build/extensions/core"; + +extensions: [ + aptGet({ + packages: ["ffmpeg", "imagemagick", "curl=7.68.0-1"], // Can specify versions + }), +]; +``` + +#### Additional NPM Packages + +Only use this for installing CLI tools, NOT packages you import in your code. + +```ts +import { additionalPackages } from "@trigger.dev/build/extensions/core"; + +extensions: [ + additionalPackages({ + packages: ["wrangler"], // CLI tools and specific versions + }), +]; +``` + +#### Additional Files + +```ts +import { additionalFiles } from "@trigger.dev/build/extensions/core"; + +extensions: [ + additionalFiles({ + files: ["wrangler.toml", "./assets/**", "./fonts/**"], // Glob patterns supported + }), +]; +``` + +### Environment & Build Tools + +#### Environment Variable Sync + +```ts +import { syncEnvVars } from "@trigger.dev/build/extensions/core"; + +extensions: [ + syncEnvVars(async (ctx) => { + // ctx contains: environment, projectRef, env + return [ + { name: "SECRET_KEY", value: await getSecret(ctx.environment) }, + { name: "API_URL", value: ctx.environment === "prod" ? "api.prod.com" : "api.dev.com" }, + ]; + }), +]; +``` + +#### ESBuild Plugins + +```ts +import { esbuildPlugin } from "@trigger.dev/build/extensions"; +import { sentryEsbuildPlugin } from "@sentry/esbuild-plugin"; + +extensions: [ + esbuildPlugin( + sentryEsbuildPlugin({ + org: process.env.SENTRY_ORG, + project: process.env.SENTRY_PROJECT, + authToken: process.env.SENTRY_AUTH_TOKEN, + }), + { placement: "last", target: "deploy" } // Optional config + ), +]; +``` + +## Custom Build Extensions + +```ts +import { defineConfig } from "@trigger.dev/sdk"; + +const customExtension = { + name: "my-custom-extension", + + externalsForTarget: (target) => { + return ["some-native-module"]; // Add external dependencies + }, + + onBuildStart: async (context) => { + console.log(`Build starting for ${context.target}`); + // Register esbuild plugins, modify build context + }, + + onBuildComplete: async (context, manifest) => { + console.log("Build complete, adding layers"); + // Add build layers, modify deployment + context.addLayer({ + id: "my-layer", + files: [{ source: "./custom-file", destination: "/app/custom" }], + commands: ["chmod +x /app/custom"], + }); + }, +}; + +export default defineConfig({ + project: "my-project", + build: { + extensions: [customExtension], + }, +}); +``` + +## Advanced Configuration + +### Telemetry + +```ts +import { PrismaInstrumentation } from "@prisma/instrumentation"; +import { OpenAIInstrumentation } from "@langfuse/openai"; + +export default defineConfig({ + // ... other config + telemetry: { + instrumentations: [new PrismaInstrumentation(), new OpenAIInstrumentation()], + exporters: [customExporter], // Optional custom exporters + }, +}); +``` + +### Machine & Performance + +```ts +export default defineConfig({ + // ... other config + defaultMachine: "large-1x", // Default machine for all tasks + maxDuration: 300, // Default max duration (seconds) + enableConsoleLogging: true, // Console logging in development +}); +``` + +## Common Extension Combinations + +### Full-Stack Web App + +```ts +extensions: [ + prismaExtension({ schema: "prisma/schema.prisma", migrate: true }), + additionalFiles({ files: ["./public/**", "./assets/**"] }), + syncEnvVars(async (ctx) => [...envVars]), +]; +``` + +### AI/ML Processing + +```ts +extensions: [ + pythonExtension({ + scripts: ["./ai/**/*.py"], + requirementsFile: "./requirements.txt", + }), + ffmpeg({ version: "7" }), + additionalPackages({ packages: ["wrangler"] }), +]; +``` + +### Web Scraping + +```ts +extensions: [ + playwright({ browsers: ["chromium"] }), + puppeteer(), + additionalFiles({ files: ["./selectors.json", "./proxies.txt"] }), +]; +``` + +## Best Practices + +- **Use specific versions**: Pin extension versions for reproducible builds +- **External packages**: Add modules with native addons to the `build.external` array +- **Environment sync**: Use `syncEnvVars` for dynamic secrets +- **File paths**: Use glob patterns for flexible file inclusion +- **Debug builds**: Use `--log-level debug --dry-run` for troubleshooting + +Extensions only affect deployment, not local development. Use `external` array for packages that shouldn't be bundled. diff --git a/.agents/skills/trigger-cost-savings/SKILL.md b/.agents/skills/trigger-cost-savings/SKILL.md new file mode 100644 index 0000000..74efb32 --- /dev/null +++ b/.agents/skills/trigger-cost-savings/SKILL.md @@ -0,0 +1,158 @@ +--- +name: trigger-cost-savings +description: Analyze Trigger.dev tasks, schedules, and runs for cost optimization opportunities. Use when asked to reduce spend, optimize costs, audit usage, right-size machines, or review task efficiency. Requires Trigger.dev MCP tools for run analysis. +--- + +# Trigger.dev Cost Savings Analysis + +Analyze task runs and configurations to find cost reduction opportunities. + +## Prerequisites: MCP Tools + +This skill requires the **Trigger.dev MCP server** to analyze live run data. + +### Check MCP availability + +Before analysis, verify these MCP tools are available: +- `list_runs` — list runs with filters (status, task, time period, machine size) +- `get_run_details` — get run logs, duration, and status +- `get_current_worker` — get registered tasks and their configurations + +If these tools are **not available**, instruct the user: + +``` +To analyze your runs, you need the Trigger.dev MCP server installed. + +Run this command to install it: + + npx trigger.dev@latest install-mcp + +This launches an interactive wizard that configures the MCP server for your AI client. +``` + +Do NOT proceed with run analysis without MCP tools. You can still review source code for static issues (see Static Analysis below). + +### Load latest cost reduction documentation + +Before giving recommendations, fetch the latest guidance: + +``` +WebFetch: https://trigger.dev/docs/how-to-reduce-your-spend +``` + +Use the fetched content to ensure recommendations are current. If the fetch fails, fall back to the reference documentation in `references/cost-reduction.md`. + +## Analysis Workflow + +### Step 1: Static Analysis (source code) + +Scan task files in the project for these issues: + +1. **Oversized machines** — tasks using `large-1x` or `large-2x` without clear need +2. **Missing `maxDuration`** — tasks without execution time limits (runaway cost risk) +3. **Excessive retries** — `maxAttempts` > 5 without `AbortTaskRunError` for known failures +4. **Missing debounce** — high-frequency triggers without debounce configuration +5. **Missing idempotency** — payment/critical tasks without idempotency keys +6. **Polling instead of waits** — `setTimeout`/`setInterval`/sleep loops instead of `wait.for()` +7. **Short waits** — `wait.for()` with < 5 seconds (not checkpointed, wastes compute) +8. **Sequential instead of batch** — multiple `triggerAndWait()` calls that could use `batchTriggerAndWait()` +9. **Over-scheduled crons** — schedules running more frequently than necessary + +### Step 2: Run Analysis (requires MCP tools) + +Use MCP tools to analyze actual usage patterns: + +#### 2a. Identify expensive tasks + +``` +list_runs with filters: +- period: "30d" or "7d" +- Sort by duration or cost +- Check across different task IDs +``` + +Look for: +- Tasks with high total compute time (duration x run count) +- Tasks with high failure rates (wasted retries) +- Tasks running on large machines with short durations (over-provisioned) + +#### 2b. Analyze failure patterns + +``` +list_runs with status: "FAILED" or "CRASHED" +``` + +For high-failure tasks: +- Check if failures are retryable (transient) vs permanent +- Suggest `AbortTaskRunError` for known non-retryable errors +- Calculate wasted compute from failed retries + +#### 2c. Check machine utilization + +``` +get_run_details for sample runs of each task +``` + +Compare actual resource usage against machine preset: +- If a task on `large-2x` consistently runs in < 1 second, it's over-provisioned +- If tasks are I/O-bound (API calls, DB queries), they likely don't need large machines + +#### 2d. Review schedule frequency + +``` +get_current_worker to list scheduled tasks and their cron patterns +``` + +Flag schedules that may be too frequent for their purpose. + +### Step 3: Generate Recommendations + +Present findings as a prioritized list with estimated impact: + +```markdown +## Cost Optimization Report + +### High Impact +1. **Right-size `process-images` machine** — Currently `large-2x`, average run 2s. + Switching to `small-2x` could reduce this task's cost by ~16x. + ```ts + machine: { preset: "small-2x" } // was "large-2x" + ``` + +### Medium Impact +2. **Add debounce to `sync-user-data`** — 847 runs/day, often triggered in bursts. + ```ts + debounce: { key: `user-${userId}`, delay: "5s" } + ``` + +### Low Impact / Best Practices +3. **Add `maxDuration` to `generate-report`** — No timeout configured. + ```ts + maxDuration: 300 // 5 minutes + ``` +``` + +## Machine Preset Costs (relative) + +Larger machines cost proportionally more per second of compute: + +| Preset | vCPU | RAM | Relative Cost | +|--------|------|-----|---------------| +| micro | 0.25 | 0.25 GB | 0.25x | +| small-1x | 0.5 | 0.5 GB | 1x (baseline) | +| small-2x | 1 | 1 GB | 2x | +| medium-1x | 1 | 2 GB | 2x | +| medium-2x | 2 | 4 GB | 4x | +| large-1x | 4 | 8 GB | 8x | +| large-2x | 8 | 16 GB | 16x | + +## Key Principles + +- **Waits > 5 seconds are free** — checkpointed, no compute charge +- **Start small, scale up** — default `small-1x` is right for most tasks +- **I/O-bound tasks don't need big machines** — API calls, DB queries wait on network +- **Debounce saves the most on high-frequency tasks** — consolidates bursts into single runs +- **Idempotency prevents duplicate work** — especially important for expensive operations +- **`AbortTaskRunError` stops wasteful retries** — don't retry permanent failures + +See `references/cost-reduction.md` for detailed strategies with code examples. diff --git a/.agents/skills/trigger-cost-savings/references/cost-reduction.md b/.agents/skills/trigger-cost-savings/references/cost-reduction.md new file mode 100644 index 0000000..012f7ec --- /dev/null +++ b/.agents/skills/trigger-cost-savings/references/cost-reduction.md @@ -0,0 +1,206 @@ +# Cost Reduction Strategies + +Detailed strategies for reducing Trigger.dev spend. For the latest version, fetch: +https://trigger.dev/docs/how-to-reduce-your-spend + +## 1. Monitor Usage + +Review your usage dashboard regularly to identify: +- Most expensive tasks (by total compute time) +- Run counts and daily spikes +- Failure rates and wasted retries + +## 2. Configure Billing Alerts + +Set up alerts in the Trigger.dev dashboard: +- **Standard alerts**: Notifications at 75%, 90%, 100%, 200%, 500% of budget +- **Spike alerts**: Protection at 10x, 20x, 50x, 100x of monthly budget + +Keep spike alerts enabled as a safety net against runaway costs. + +## 3. Right-Size Machines + +Start with the smallest machine and scale only when necessary: + +```ts +// Default (small-1x) is right for most tasks +export const apiTask = task({ + id: "call-api", + // No machine preset needed — defaults to small-1x + run: async (payload) => { + const response = await fetch("https://api.example.com/data"); + return response.json(); + }, +}); + +// Only use larger machines for CPU/memory-intensive work +export const imageProcessor = task({ + id: "process-image", + machine: { preset: "medium-1x" }, // Only if actually needed + run: async (payload) => { + // Heavy image processing that needs more RAM + }, +}); + +// Override machine at trigger time for variable workloads +await imageProcessor.trigger(largePayload, { + machine: { preset: "large-1x" }, // Larger only for this specific run +}); +``` + +## 4. Use Idempotency Keys + +Prevent duplicate execution of expensive operations: + +```ts +import { task, idempotencyKeys } from "@trigger.dev/sdk"; + +export const expensiveTask = task({ + id: "expensive-operation", + run: async (payload: { orderId: string }) => { + const key = await idempotencyKeys.create(`order-${payload.orderId}`); + + // This won't re-execute if triggered again with same key + await costlyChildTask.trigger(payload, { + idempotencyKey: key, + idempotencyKeyTTL: "24h", + }); + }, +}); +``` + +## 5. Parallelize Within Tasks + +Consolidate multiple async operations into single tasks instead of spawning many: + +```ts +// Expensive: 3 separate task runs +await taskA.triggerAndWait(data); +await taskB.triggerAndWait(data); +await taskC.triggerAndWait(data); + +// Cheaper: single task with parallel I/O (when work is I/O-bound) +export const combinedTask = task({ + id: "combined-api-calls", + run: async (payload) => { + const [a, b, c] = await Promise.all([ + fetch("https://api-a.com"), + fetch("https://api-b.com"), + fetch("https://api-c.com"), + ]); + return { a: await a.json(), b: await b.json(), c: await c.json() }; + }, +}); +``` + +Note: Only use `Promise.all` for regular async operations (fetch, DB queries), NOT for `triggerAndWait()` or `wait.*` calls. + +## 6. Optimize Retries + +Reduce wasted compute from retries: + +```ts +import { task, AbortTaskRunError } from "@trigger.dev/sdk"; + +export const smartRetryTask = task({ + id: "smart-retry", + retry: { + maxAttempts: 3, // Not 10 — be realistic + }, + catchError: async ({ error }) => { + // Don't retry known permanent failures + if (error.message?.includes("NOT_FOUND")) { + throw new AbortTaskRunError("Resource not found — won't retry"); + } + if (error.message?.includes("UNAUTHORIZED")) { + throw new AbortTaskRunError("Auth failed — won't retry"); + } + // Only retry transient errors + }, + run: async (payload) => { + // task logic + }, +}); +``` + +## 7. Set maxDuration + +Prevent runaway tasks from consuming unlimited compute: + +```ts +export const boundedTask = task({ + id: "bounded-task", + maxDuration: 300, // 5 minutes max + run: async (payload) => { + // If this takes longer than 5 minutes, it's killed + }, +}); +``` + +## 8. Use Waitpoints Instead of Polling + +Waits > 5 seconds are checkpointed and free: + +```ts +// Expensive: polling loop burns compute +export const pollingTask = task({ + id: "polling-bad", + run: async (payload) => { + while (true) { + const status = await checkStatus(payload.id); + if (status === "ready") break; + await new Promise((r) => setTimeout(r, 5000)); // WASTES compute + } + }, +}); + +// Free: checkpointed wait +import { wait } from "@trigger.dev/sdk"; + +export const waitTask = task({ + id: "wait-good", + run: async (payload) => { + await wait.for({ minutes: 5 }); // FREE — checkpointed + const status = await checkStatus(payload.id); + if (status !== "ready") { + await wait.for({ minutes: 5 }); // Still free + } + }, +}); +``` + +## 9. Debounce High-Frequency Triggers + +Consolidate bursts into single executions: + +```ts +// Without debounce: 100 webhook events = 100 task runs +await syncTask.trigger({ userId: "123" }); + +// With debounce: 100 events in 5s = 1 task run +await syncTask.trigger( + { userId: "123" }, + { + debounce: { + key: "sync-user-123", + delay: "5s", + mode: "trailing", // Use latest payload + }, + } +); +``` + +## Cost Checklist + +Use this checklist when reviewing tasks: + +- [ ] Machine preset matches actual resource needs (start with `small-1x`) +- [ ] `maxDuration` is set to a reasonable limit +- [ ] Retry `maxAttempts` is appropriate (not excessive) +- [ ] `AbortTaskRunError` used for known permanent failures +- [ ] Idempotency keys used for expensive/critical operations +- [ ] `wait.for()` used instead of polling loops (with delays > 5s) +- [ ] Debounce configured for high-frequency trigger sources +- [ ] Batch triggering used instead of sequential `triggerAndWait()` loops +- [ ] Scheduled task frequency matches actual business needs +- [ ] Billing alerts configured in dashboard diff --git a/.agents/skills/trigger-realtime/SKILL.md b/.agents/skills/trigger-realtime/SKILL.md new file mode 100644 index 0000000..5368360 --- /dev/null +++ b/.agents/skills/trigger-realtime/SKILL.md @@ -0,0 +1,288 @@ +--- +name: trigger-realtime +description: Subscribe to Trigger.dev task runs in real-time from frontend and backend. Use when building progress indicators, live dashboards, streaming AI/LLM responses, or React components that display task status. +--- + +# Trigger.dev Realtime + +Subscribe to task runs and stream data in real-time from frontend and backend. + +## When to Use + +- Building progress indicators for long-running tasks +- Creating live dashboards showing task status +- Streaming AI/LLM responses to the UI +- React components that trigger and monitor tasks +- Waiting for user approval in tasks + +## Authentication + +### Create Public Access Token (Backend) + +```ts +import { auth } from "@trigger.dev/sdk"; + +// Read-only token for specific runs +const publicToken = await auth.createPublicToken({ + scopes: { + read: { + runs: ["run_123"], + tasks: ["my-task"], + }, + }, + expirationTime: "1h", +}); + +// Pass this token to your frontend +``` + +### Create Trigger Token (for frontend triggering) + +```ts +const triggerToken = await auth.createTriggerPublicToken("my-task", { + expirationTime: "30m", +}); +``` + +## Backend Subscriptions + +```ts +import { runs, tasks } from "@trigger.dev/sdk"; + +// Trigger and subscribe +const handle = await tasks.trigger("my-task", { data: "value" }); + +for await (const run of runs.subscribeToRun(handle.id)) { + console.log(`Status: ${run.status}`); + console.log(`Progress: ${run.metadata?.progress}`); + + if (run.status === "COMPLETED") { + console.log("Output:", run.output); + break; + } +} + +// Subscribe to tagged runs +for await (const run of runs.subscribeToRunsWithTag("user-123")) { + console.log(`Run ${run.id}: ${run.status}`); +} + +// Subscribe to batch +for await (const run of runs.subscribeToBatch(batchId)) { + console.log(`Batch run ${run.id}: ${run.status}`); +} +``` + +## React Hooks + +### Installation + +```bash +npm add @trigger.dev/react-hooks +``` + +### Trigger Task from React + +```tsx +"use client"; +import { useRealtimeTaskTrigger } from "@trigger.dev/react-hooks"; +import type { myTask } from "../trigger/tasks"; + +function TaskTrigger({ accessToken }: { accessToken: string }) { + const { submit, run, isLoading } = useRealtimeTaskTrigger( + "my-task", + { accessToken } + ); + + return ( +
+ + + {run && ( +
+

Status: {run.status}

+

Progress: {run.metadata?.progress}%

+ {run.output &&

Result: {JSON.stringify(run.output)}

} +
+ )} +
+ ); +} +``` + +### Subscribe to Existing Run + +```tsx +"use client"; +import { useRealtimeRun } from "@trigger.dev/react-hooks"; +import type { myTask } from "../trigger/tasks"; + +function RunStatus({ runId, accessToken }: { runId: string; accessToken: string }) { + const { run, error } = useRealtimeRun(runId, { + accessToken, + onComplete: (run) => { + console.log("Completed:", run.output); + }, + }); + + if (error) return
Error: {error.message}
; + if (!run) return
Loading...
; + + return ( +
+

Status: {run.status}

+

Progress: {run.metadata?.progress || 0}%

+
+ ); +} +``` + +### Subscribe to Tagged Runs + +```tsx +"use client"; +import { useRealtimeRunsWithTag } from "@trigger.dev/react-hooks"; + +function UserTasks({ userId, accessToken }: { userId: string; accessToken: string }) { + const { runs } = useRealtimeRunsWithTag(`user-${userId}`, { accessToken }); + + return ( +
    + {runs.map((run) => ( +
  • {run.id}: {run.status}
  • + ))} +
+ ); +} +``` + +## Realtime Streams (AI/LLM) + +### Define Stream (shared location) + +```ts +// trigger/streams.ts +import { streams } from "@trigger.dev/sdk"; + +export const aiStream = streams.define({ + id: "ai-output", +}); +``` + +### Pipe Stream in Task + +```ts +import { task } from "@trigger.dev/sdk"; +import { aiStream } from "./streams"; + +export const streamingTask = task({ + id: "streaming-task", + run: async (payload: { prompt: string }) => { + const completion = await openai.chat.completions.create({ + model: "gpt-4", + messages: [{ role: "user", content: payload.prompt }], + stream: true, + }); + + const { waitUntilComplete } = aiStream.pipe(completion); + await waitUntilComplete(); + }, +}); +``` + +### Read Stream in React + +```tsx +"use client"; +import { useRealtimeStream } from "@trigger.dev/react-hooks"; +import { aiStream } from "../trigger/streams"; + +function AIResponse({ runId, accessToken }: { runId: string; accessToken: string }) { + const { parts, error } = useRealtimeStream(aiStream, runId, { + accessToken, + throttleInMs: 50, + }); + + if (error) return
Error: {error.message}
; + if (!parts) return
Waiting for response...
; + + return
{parts.join("")}
; +} +``` + +## Wait Tokens (Human-in-the-loop) + +### In Task + +```ts +import { task, wait } from "@trigger.dev/sdk"; + +export const approvalTask = task({ + id: "approval-task", + run: async (payload) => { + // Process initial data + const processed = await processData(payload); + + // Wait for human approval + const approval = await wait.forToken<{ approved: boolean }>({ + token: `approval-${payload.id}`, + timeoutInSeconds: 86400, // 24 hours + }); + + if (approval.approved) { + return await finalizeData(processed); + } + + throw new Error("Not approved"); + }, +}); +``` + +### Complete Token from React + +```tsx +"use client"; +import { useWaitToken } from "@trigger.dev/react-hooks"; + +function ApprovalButton({ tokenId, accessToken }: { tokenId: string; accessToken: string }) { + const { complete } = useWaitToken(tokenId, { accessToken }); + + return ( +
+ + +
+ ); +} +``` + +## Run Object Properties + +| Property | Description | +|----------|-------------| +| `id` | Unique run identifier | +| `status` | `QUEUED`, `EXECUTING`, `COMPLETED`, `FAILED`, `CANCELED` | +| `payload` | Task input (typed) | +| `output` | Task result (typed, when completed) | +| `metadata` | Real-time updatable data | +| `createdAt` | Start timestamp | +| `costInCents` | Execution cost | + +## Best Practices + +1. **Scope tokens narrowly** — only grant necessary permissions +2. **Set expiration times** — don't use long-lived tokens +3. **Use typed hooks** — pass task types for proper inference +4. **Handle errors** — always check for errors in hooks +5. **Throttle streams** — use `throttleInMs` to control re-renders + +See `references/realtime.md` for complete documentation. diff --git a/.agents/skills/trigger-realtime/references/realtime.md b/.agents/skills/trigger-realtime/references/realtime.md new file mode 100644 index 0000000..c1c4c58 --- /dev/null +++ b/.agents/skills/trigger-realtime/references/realtime.md @@ -0,0 +1,244 @@ +# Trigger.dev Realtime + +**Real-time monitoring and updates for runs** + +## Core Concepts + +Realtime allows you to: + +- Subscribe to run status changes, metadata updates, and streams +- Build real-time dashboards and UI updates +- Monitor task progress from frontend and backend + +## Authentication + +### Public Access Tokens + +```ts +import { auth } from "@trigger.dev/sdk"; + +// Read-only token for specific runs +const publicToken = await auth.createPublicToken({ + scopes: { + read: { + runs: ["run_123", "run_456"], + tasks: ["my-task-1", "my-task-2"], + }, + }, + expirationTime: "1h", // Default: 15 minutes +}); +``` + +### Trigger Tokens (Frontend only) + +```ts +// Single-use token for triggering tasks +const triggerToken = await auth.createTriggerPublicToken("my-task", { + expirationTime: "30m", +}); +``` + +## Backend Usage + +### Subscribe to Runs + +```ts +import { runs, tasks } from "@trigger.dev/sdk"; + +// Trigger and subscribe +const handle = await tasks.trigger("my-task", { data: "value" }); + +// Subscribe to specific run +for await (const run of runs.subscribeToRun(handle.id)) { + console.log(`Status: ${run.status}, Progress: ${run.metadata?.progress}`); + if (run.status === "COMPLETED") break; +} + +// Subscribe to runs with tag +for await (const run of runs.subscribeToRunsWithTag("user-123")) { + console.log(`Tagged run ${run.id}: ${run.status}`); +} + +// Subscribe to batch +for await (const run of runs.subscribeToBatch(batchId)) { + console.log(`Batch run ${run.id}: ${run.status}`); +} +``` + +### Realtime Streams v2 + +```ts +import { streams, InferStreamType } from "@trigger.dev/sdk"; + +// 1. Define streams (shared location) +export const aiStream = streams.define({ + id: "ai-output", +}); + +export type AIStreamPart = InferStreamType; + +// 2. Pipe from task +export const streamingTask = task({ + id: "streaming-task", + run: async (payload) => { + const completion = await openai.chat.completions.create({ + model: "gpt-4", + messages: [{ role: "user", content: payload.prompt }], + stream: true, + }); + + const { waitUntilComplete } = aiStream.pipe(completion); + await waitUntilComplete(); + }, +}); + +// 3. Read from backend +const stream = await aiStream.read(runId, { + timeoutInSeconds: 300, + startIndex: 0, // Resume from specific chunk +}); + +for await (const chunk of stream) { + console.log("Chunk:", chunk); // Fully typed +} +``` + +## React Frontend Usage + +### Installation + +```bash +npm add @trigger.dev/react-hooks +``` + +### Triggering Tasks + +```tsx +"use client"; +import { useTaskTrigger, useRealtimeTaskTrigger } from "@trigger.dev/react-hooks"; +import type { myTask } from "../trigger/tasks"; + +function TriggerComponent({ accessToken }: { accessToken: string }) { + // Basic trigger + const { submit, handle, isLoading } = useTaskTrigger("my-task", { + accessToken, + }); + + // Trigger with realtime updates + const { + submit: realtimeSubmit, + run, + isLoading: isRealtimeLoading, + } = useRealtimeTaskTrigger("my-task", { accessToken }); + + return ( +
+ + + + + {run &&
Status: {run.status}
} +
+ ); +} +``` + +### Subscribing to Runs + +```tsx +"use client"; +import { useRealtimeRun, useRealtimeRunsWithTag } from "@trigger.dev/react-hooks"; +import type { myTask } from "../trigger/tasks"; + +function SubscribeComponent({ runId, accessToken }: { runId: string; accessToken: string }) { + // Subscribe to specific run + const { run, error } = useRealtimeRun(runId, { + accessToken, + onComplete: (run) => { + console.log("Task completed:", run.output); + }, + }); + + // Subscribe to tagged runs + const { runs } = useRealtimeRunsWithTag("user-123", { accessToken }); + + if (error) return
Error: {error.message}
; + if (!run) return
Loading...
; + + return ( +
+
Status: {run.status}
+
Progress: {run.metadata?.progress || 0}%
+ {run.output &&
Result: {JSON.stringify(run.output)}
} + +

Tagged Runs:

+ {runs.map((r) => ( +
+ {r.id}: {r.status} +
+ ))} +
+ ); +} +``` + +### Realtime Streams with React + +```tsx +"use client"; +import { useRealtimeStream } from "@trigger.dev/react-hooks"; +import { aiStream } from "../trigger/streams"; + +function StreamComponent({ runId, accessToken }: { runId: string; accessToken: string }) { + // Pass defined stream directly for type safety + const { parts, error } = useRealtimeStream(aiStream, runId, { + accessToken, + timeoutInSeconds: 300, + throttleInMs: 50, // Control re-render frequency + }); + + if (error) return
Error: {error.message}
; + if (!parts) return
Loading...
; + + const text = parts.join(""); // parts is typed as AIStreamPart[] + + return
Streamed Text: {text}
; +} +``` + +### Wait Tokens + +```tsx +"use client"; +import { useWaitToken } from "@trigger.dev/react-hooks"; + +function WaitTokenComponent({ tokenId, accessToken }: { tokenId: string; accessToken: string }) { + const { complete } = useWaitToken(tokenId, { accessToken }); + + return ; +} +``` + +## Run Object Properties + +Key properties available in run subscriptions: + +- `id`: Unique run identifier +- `status`: `QUEUED`, `EXECUTING`, `COMPLETED`, `FAILED`, `CANCELED`, etc. +- `payload`: Task input data (typed) +- `output`: Task result (typed, when completed) +- `metadata`: Real-time updatable data +- `createdAt`, `updatedAt`: Timestamps +- `costInCents`: Execution cost + +## Best Practices + +- **Use Realtime over SWR**: Recommended for most use cases due to rate limits +- **Scope tokens properly**: Only grant necessary read/trigger permissions +- **Handle errors**: Always check for errors in hooks and subscriptions +- **Type safety**: Use task types for proper payload/output typing +- **Cleanup subscriptions**: Backend subscriptions auto-complete, frontend hooks auto-cleanup diff --git a/.agents/skills/trigger-setup/SKILL.md b/.agents/skills/trigger-setup/SKILL.md new file mode 100644 index 0000000..9e30b4e --- /dev/null +++ b/.agents/skills/trigger-setup/SKILL.md @@ -0,0 +1,127 @@ +--- +name: trigger-setup +description: Set up Trigger.dev in your project. Use when adding Trigger.dev for the first time, creating trigger.config.ts, or initializing the trigger directory. +--- + +# Trigger.dev Setup + +Get Trigger.dev running in your project in minutes. + +## When to Use + +- Adding Trigger.dev to an existing project +- Creating your first task +- Setting up trigger.config.ts +- Connecting to Trigger.dev cloud + +## Prerequisites + +- Node.js 18+ or Bun +- A Trigger.dev account (https://cloud.trigger.dev) + +## Quick Start + +### 1. Install the SDK + +```bash +npm install @trigger.dev/sdk +``` + +### 2. Initialize Your Project + +```bash +npx trigger init +``` + +This creates: +- `trigger.config.ts` - project configuration +- `trigger/` directory - where your tasks live +- `trigger/example.ts` - a sample task + +### 3. Configure trigger.config.ts + +```ts +import { defineConfig } from "@trigger.dev/sdk"; + +export default defineConfig({ + project: "proj_xxxxx", // From dashboard + dirs: ["./trigger"], +}); +``` + +### 4. Create Your First Task + +```ts +// trigger/my-task.ts +import { task } from "@trigger.dev/sdk"; + +export const myFirstTask = task({ + id: "my-first-task", + run: async (payload: { name: string }) => { + console.log(`Hello, ${payload.name}!`); + return { message: `Processed ${payload.name}` }; + }, +}); +``` + +### 5. Start Development Server + +```bash +npx trigger dev +``` + +### 6. Trigger Your Task + +From your app code: + +```ts +import { tasks } from "@trigger.dev/sdk"; +import type { myFirstTask } from "./trigger/my-task"; + +await tasks.trigger("my-first-task", { + name: "World", +}); +``` + +Or from the Trigger.dev dashboard "Test" tab. + +## Project Structure + +``` +your-project/ +├── trigger.config.ts # Required - project config +├── trigger/ # Required - task files +│ ├── my-task.ts +│ └── another-task.ts +├── package.json +└── ... +``` + +## Environment Variables + +Create `.env` or set in your environment: + +```bash +TRIGGER_SECRET_KEY=tr_dev_xxxxx # From dashboard > API Keys +``` + +## Common Issues + +### "No tasks found" +- Ensure tasks are **exported** from files in `dirs` folders +- Check `trigger.config.ts` points to correct directories + +### "Project not found" +- Verify `project` in config matches dashboard +- Check `TRIGGER_SECRET_KEY` is set + +### "Task not registered" +- Restart `npx trigger dev` after adding new tasks +- Tasks must use `task()` or `schemaTask()` from `@trigger.dev/sdk` + +## Next Steps + +- Add retry logic → see **trigger-tasks** skill +- Configure build extensions → see **trigger-config** skill +- Build AI workflows → see **trigger-agents** skill +- Add real-time UI → see **trigger-realtime** skill diff --git a/.agents/skills/trigger-setup/references/environment-setup.md b/.agents/skills/trigger-setup/references/environment-setup.md new file mode 100644 index 0000000..bf1293c --- /dev/null +++ b/.agents/skills/trigger-setup/references/environment-setup.md @@ -0,0 +1,104 @@ +# Environment Setup + +## Required Variables + +| Variable | Description | Where to find | +|----------|-------------|---------------| +| `TRIGGER_SECRET_KEY` | API key for authentication | Dashboard > API Keys | + +## Development vs Production Keys + +```bash +# Development (starts with tr_dev_) +TRIGGER_SECRET_KEY=tr_dev_xxxxx + +# Production (starts with tr_prod_) +TRIGGER_SECRET_KEY=tr_prod_xxxxx +``` + +## Local Development + +### Option 1: .env File + +```bash +# .env +TRIGGER_SECRET_KEY=tr_dev_xxxxx +``` + +Add to `.gitignore`: + +``` +.env +.env.local +``` + +### Option 2: Shell Export + +```bash +export TRIGGER_SECRET_KEY=tr_dev_xxxxx +npx trigger dev +``` + +## CI/CD Deployment + +### GitHub Actions + +```yaml +- name: Deploy Trigger.dev + env: + TRIGGER_SECRET_KEY: ${{ secrets.TRIGGER_SECRET_KEY }} + run: npx trigger deploy +``` + +### Vercel + +Add `TRIGGER_SECRET_KEY` in Project Settings > Environment Variables. + +### Other Platforms + +Set `TRIGGER_SECRET_KEY` in your platform's secret management. + +## Multi-Environment Setup + +Use different keys per environment: + +| Environment | Key Prefix | Dashboard Section | +|-------------|------------|-------------------| +| Development | `tr_dev_` | Dev environment | +| Staging | `tr_stg_` | Staging environment | +| Production | `tr_prod_` | Prod environment | + +## Task Environment Variables + +Tasks run in Trigger.dev's infrastructure. To use env vars in tasks: + +1. **Sync from local** (using `syncEnvVars` extension): + +```ts +// trigger.config.ts +import { defineConfig } from "@trigger.dev/sdk"; +import { syncEnvVars } from "@trigger.dev/build/extensions/core"; + +export default defineConfig({ + project: "proj_xxxxx", + build: { + extensions: [ + syncEnvVars(), + ], + }, +}); +``` + +2. **Set in dashboard**: Project Settings > Environment Variables + +3. **Access in tasks**: + +```ts +export const myTask = task({ + id: "my-task", + run: async () => { + const apiKey = process.env.EXTERNAL_API_KEY; + // ... + }, +}); +``` diff --git a/.agents/skills/trigger-setup/references/project-structure.md b/.agents/skills/trigger-setup/references/project-structure.md new file mode 100644 index 0000000..2c73984 --- /dev/null +++ b/.agents/skills/trigger-setup/references/project-structure.md @@ -0,0 +1,94 @@ +# Project Structure + +## Default Layout + +``` +your-project/ +├── trigger.config.ts # Required - project configuration +├── trigger/ # Default task directory +│ ├── example.ts # Created by `npx trigger init` +│ └── ... +├── package.json +└── src/ # Your app code +``` + +## Monorepo Layout + +For monorepos, place `trigger.config.ts` in the package that contains your tasks: + +``` +monorepo/ +├── packages/ +│ ├── api/ +│ │ ├── trigger.config.ts # Config here +│ │ ├── trigger/ # Tasks here +│ │ └── src/ +│ └── web/ +└── package.json +``` + +## Multiple Task Directories + +Configure multiple directories in `trigger.config.ts`: + +```ts +import { defineConfig } from "@trigger.dev/sdk"; + +export default defineConfig({ + project: "proj_xxxxx", + dirs: [ + "./trigger", // Default + "./src/jobs", // Additional + "./src/scheduled", // Another + ], +}); +``` + +## Collocated Tasks + +Keep tasks next to related code: + +``` +src/ +├── users/ +│ ├── routes.ts +│ └── tasks/ +│ └── send-welcome-email.ts +├── orders/ +│ ├── routes.ts +│ └── tasks/ +│ └── process-order.ts +``` + +```ts +// trigger.config.ts +export default defineConfig({ + project: "proj_xxxxx", + dirs: ["./src/**/tasks"], // Glob pattern +}); +``` + +## Task File Requirements + +Each task file must: + +1. **Export** tasks (named exports) +2. Use `task()` or `schemaTask()` from `@trigger.dev/sdk` +3. Have unique task IDs across all files + +```ts +// ✅ Correct - exported task +export const myTask = task({ + id: "my-task", // Unique ID + run: async (payload) => { ... }, +}); + +// ❌ Wrong - not exported +const privateTask = task({ ... }); + +// ❌ Wrong - duplicate ID will error +export const anotherTask = task({ + id: "my-task", // Conflicts! + run: async (payload) => { ... }, +}); +``` diff --git a/.agents/skills/trigger-tasks/SKILL.md b/.agents/skills/trigger-tasks/SKILL.md new file mode 100644 index 0000000..7610c73 --- /dev/null +++ b/.agents/skills/trigger-tasks/SKILL.md @@ -0,0 +1,347 @@ +--- +name: trigger-tasks +description: Build AI agents, workflows and durable background tasks with Trigger.dev. Use when creating tasks, triggering jobs, handling retries, scheduling cron jobs, or implementing queues and concurrency control. +--- + +# Trigger.dev Tasks + +Build durable background tasks that run reliably with automatic retries, queuing, and observability. + +## When to Use + +- Creating background jobs or async workflows +- Building AI agents that need long-running execution +- Processing webhooks, emails, or file uploads +- Scheduling recurring tasks (cron) +- Any work that shouldn't block your main application + +## Critical Rules + +1. **Always use `@trigger.dev/sdk`** — never use deprecated `client.defineJob` +2. **Check `result.ok`** before accessing `result.output` from `triggerAndWait()` +3. **Never use `Promise.all`** with `triggerAndWait()` or `wait.*` calls +4. **Export tasks** from files in your `trigger/` directory + +## Basic Task + +```ts +import { task } from "@trigger.dev/sdk"; + +export const processData = task({ + id: "process-data", + retry: { + maxAttempts: 10, + factor: 1.8, + minTimeoutInMs: 500, + maxTimeoutInMs: 30_000, + }, + run: async (payload: { userId: string; data: any[] }) => { + console.log(`Processing ${payload.data.length} items`); + return { processed: payload.data.length }; + }, +}); +``` + +## Schema Task (Validated Input) + +```ts +import { schemaTask } from "@trigger.dev/sdk"; +import { z } from "zod"; + +export const validatedTask = schemaTask({ + id: "validated-task", + schema: z.object({ + name: z.string(), + email: z.string().email(), + }), + run: async (payload) => { + // payload is typed and validated + return { message: `Hello ${payload.name}` }; + }, +}); +``` + +## Triggering Tasks + +### From Backend Code + +```ts +import { tasks } from "@trigger.dev/sdk"; +import type { processData } from "./trigger/tasks"; + +// Single trigger (fire and forget) +const handle = await tasks.trigger("process-data", { + userId: "123", + data: [{ id: 1 }], +}); + +// Batch trigger (up to 1,000 items, 3MB per payload) +const batchHandle = await tasks.batchTrigger("process-data", [ + { payload: { userId: "123", data: [] } }, + { payload: { userId: "456", data: [] } }, +]); +``` + +### From Inside Tasks + +```ts +export const parentTask = task({ + id: "parent-task", + run: async (payload) => { + // Fire and forget + const handle = await childTask.trigger({ data: "value" }); + + // Wait for result - returns Result object, NOT direct output + const result = await childTask.triggerAndWait({ data: "value" }); + if (result.ok) { + console.log("Output:", result.output); + } else { + console.error("Failed:", result.error); + } + + // Quick unwrap (throws on error) + const output = await childTask.triggerAndWait({ data: "value" }).unwrap(); + + // Batch with wait + const results = await childTask.batchTriggerAndWait([ + { payload: { data: "item1" } }, + { payload: { data: "item2" } }, + ]); + }, +}); +``` + +## Waits + +```ts +import { task, wait } from "@trigger.dev/sdk"; + +export const taskWithWaits = task({ + id: "task-with-waits", + run: async (payload) => { + await wait.for({ seconds: 30 }); + await wait.for({ minutes: 5 }); + await wait.until({ date: new Date("2024-12-25") }); + + // Wait for external approval + await wait.forToken({ + token: "user-approval-token", + timeoutInSeconds: 3600, + }); + }, +}); +``` + +> Waits > 5 seconds are checkpointed and don't count toward compute. + +## Concurrency & Queues + +```ts +import { task, queue } from "@trigger.dev/sdk"; + +// Shared queue +const emailQueue = queue({ + name: "email-processing", + concurrencyLimit: 5, +}); + +// Task-level concurrency +export const oneAtATime = task({ + id: "sequential-task", + queue: { concurrencyLimit: 1 }, + run: async (payload) => { + // Only one instance runs at a time + }, +}); + +// Use shared queue +export const emailTask = task({ + id: "send-email", + queue: emailQueue, + run: async (payload) => {}, +}); + +// Per-tenant concurrency (at trigger time) +await childTask.trigger(payload, { + queue: { + name: `user-${userId}`, + concurrencyLimit: 2, + }, +}); +``` + +## Debouncing + +Consolidate rapid triggers into a single execution: + +```ts +await myTask.trigger( + { userId: "123" }, + { + debounce: { + key: "user-123-update", + delay: "5s", + mode: "trailing", // Use latest payload (default: "leading") + }, + } +); +``` + +## Idempotency + +```ts +import { task, idempotencyKeys } from "@trigger.dev/sdk"; + +export const paymentTask = task({ + id: "process-payment", + run: async (payload: { orderId: string }) => { + const key = await idempotencyKeys.create(`payment-${payload.orderId}`); + + await chargeCustomer.trigger(payload, { + idempotencyKey: key, + idempotencyKeyTTL: "24h", + }); + }, +}); +``` + +## Error Handling & Retries + +```ts +import { task, retry, AbortTaskRunError } from "@trigger.dev/sdk"; + +export const resilientTask = task({ + id: "resilient-task", + retry: { + maxAttempts: 10, + factor: 1.8, + minTimeoutInMs: 500, + maxTimeoutInMs: 30_000, + }, + catchError: async ({ error, ctx }) => { + if (error.code === "FATAL_ERROR") { + throw new AbortTaskRunError("Cannot retry"); + } + return { retryAt: new Date(Date.now() + 60000) }; + }, + run: async (payload) => { + // Retry specific operations + const result = await retry.onThrow( + async () => unstableApiCall(payload), + { maxAttempts: 3 } + ); + + // HTTP retries with conditions + const response = await retry.fetch("https://api.example.com", { + retry: { + maxAttempts: 5, + condition: (res, err) => res?.status === 429 || res?.status >= 500, + }, + }); + }, +}); +``` + +## Scheduled Tasks (Cron) + +```ts +import { schedules } from "@trigger.dev/sdk"; + +// Declarative schedule +export const dailyTask = schedules.task({ + id: "daily-cleanup", + cron: "0 0 * * *", // Midnight UTC + run: async (payload) => { + // payload.timestamp - scheduled time + // payload.timezone - IANA timezone + // payload.scheduleId - schedule identifier + }, +}); + +// With timezone +export const tokyoTask = schedules.task({ + id: "tokyo-morning", + cron: { pattern: "0 9 * * *", timezone: "Asia/Tokyo" }, + run: async () => {}, +}); + +// Dynamic/multi-tenant schedules +await schedules.create({ + task: "reminder-task", + cron: "0 8 * * *", + timezone: "America/New_York", + externalId: userId, + deduplicationKey: `${userId}-daily`, +}); +``` + +## Metadata & Progress + +```ts +import { task, metadata } from "@trigger.dev/sdk"; + +export const batchProcessor = task({ + id: "batch-processor", + run: async (payload: { items: any[] }) => { + metadata.set("progress", 0).set("total", payload.items.length); + + for (let i = 0; i < payload.items.length; i++) { + await processItem(payload.items[i]); + metadata.set("progress", ((i + 1) / payload.items.length) * 100); + } + + metadata.set("status", "completed"); + }, +}); +``` + +## Tags + +```ts +import { task, tags } from "@trigger.dev/sdk"; + +export const processUser = task({ + id: "process-user", + run: async (payload: { userId: string }) => { + await tags.add(`user_${payload.userId}`); + }, +}); + +// Trigger with tags +await processUser.trigger( + { userId: "123" }, + { tags: ["priority", "user_123"] } +); +``` + +## Machine Presets + +```ts +export const heavyTask = task({ + id: "heavy-computation", + machine: { preset: "large-2x" }, // 8 vCPU, 16 GB RAM + maxDuration: 1800, // 30 minutes + run: async (payload) => {}, +}); +``` + +| Preset | vCPU | RAM | +|--------|------|-----| +| micro | 0.25 | 0.25 GB | +| small-1x | 0.5 | 0.5 GB (default) | +| small-2x | 1 | 1 GB | +| medium-1x | 1 | 2 GB | +| medium-2x | 2 | 4 GB | +| large-1x | 4 | 8 GB | +| large-2x | 8 | 16 GB | + +## Best Practices + +1. **Make tasks idempotent** — safe to retry without side effects +2. **Use queues** to prevent overwhelming external services +3. **Configure appropriate retries** with exponential backoff +4. **Track progress with metadata** for long-running tasks +5. **Use debouncing** for user activity and webhook bursts +6. **Match machine size** to computational requirements + +See `references/` for detailed documentation on each feature. diff --git a/.agents/skills/trigger-tasks/references/advanced-tasks.md b/.agents/skills/trigger-tasks/references/advanced-tasks.md new file mode 100644 index 0000000..32a0033 --- /dev/null +++ b/.agents/skills/trigger-tasks/references/advanced-tasks.md @@ -0,0 +1,485 @@ +# Trigger.dev Advanced Tasks (v4) + +**Advanced patterns and features for writing tasks** + +## Tags & Organization + +```ts +import { task, tags } from "@trigger.dev/sdk"; + +export const processUser = task({ + id: "process-user", + run: async (payload: { userId: string; orgId: string }, { ctx }) => { + // Add tags during execution + await tags.add(`user_${payload.userId}`); + await tags.add(`org_${payload.orgId}`); + + return { processed: true }; + }, +}); + +// Trigger with tags +await processUser.trigger( + { userId: "123", orgId: "abc" }, + { tags: ["priority", "user_123", "org_abc"] } // Max 10 tags per run +); + +// Subscribe to tagged runs +for await (const run of runs.subscribeToRunsWithTag("user_123")) { + console.log(`User task ${run.id}: ${run.status}`); +} +``` + +**Tag Best Practices:** + +- Use prefixes: `user_123`, `org_abc`, `video:456` +- Max 10 tags per run, 1-64 characters each +- Tags don't propagate to child tasks automatically + +## Batch Triggering v2 + +Enhanced batch triggering with larger payloads and streaming ingestion. + +### Limits + +- **Maximum batch size**: 1,000 items (increased from 500) +- **Payload per item**: 3MB each (increased from 1MB combined) +- Payloads > 512KB automatically offload to object storage + +### Rate Limiting (per environment) + +| Tier | Bucket Size | Refill Rate | +|------|-------------|-------------| +| Free | 1,200 runs | 100 runs/10 sec | +| Hobby | 5,000 runs | 500 runs/5 sec | +| Pro | 5,000 runs | 500 runs/5 sec | + +### Concurrent Batch Processing + +| Tier | Concurrent Batches | +|------|-------------------| +| Free | 1 | +| Hobby | 10 | +| Pro | 10 | + +### Usage + +```ts +import { myTask } from "./trigger/myTask"; + +// Basic batch trigger (up to 1,000 items) +const runs = await myTask.batchTrigger([ + { payload: { userId: "user-1" } }, + { payload: { userId: "user-2" } }, + { payload: { userId: "user-3" } }, +]); + +// Batch trigger with wait +const results = await myTask.batchTriggerAndWait([ + { payload: { userId: "user-1" } }, + { payload: { userId: "user-2" } }, +]); + +for (const result of results) { + if (result.ok) { + console.log("Result:", result.output); + } +} + +// With per-item options +const batchHandle = await myTask.batchTrigger([ + { + payload: { userId: "123" }, + options: { + idempotencyKey: "user-123-batch", + tags: ["priority"], + }, + }, + { + payload: { userId: "456" }, + options: { + idempotencyKey: "user-456-batch", + }, + }, +]); +``` + +## Debouncing + +Consolidate multiple triggers into a single execution by debouncing task runs with a unique key and delay window. + +### Use Cases + +- **User activity updates**: Batch rapid user actions into a single run +- **Webhook deduplication**: Handle webhook bursts without redundant processing +- **Search indexing**: Combine document updates instead of processing individually +- **Notification batching**: Group notifications to prevent user spam + +### Basic Usage + +```ts +await myTask.trigger( + { userId: "123" }, + { + debounce: { + key: "user-123-update", // Unique identifier for debounce group + delay: "5s", // Wait duration ("5s", "1m", or milliseconds) + }, + } +); +``` + +### Execution Modes + +**Leading Mode** (default): Uses payload/options from the first trigger; subsequent triggers only reschedule execution time. + +```ts +// First trigger sets the payload +await myTask.trigger({ action: "first" }, { + debounce: { key: "my-key", delay: "10s" } +}); + +// Second trigger only reschedules - payload remains "first" +await myTask.trigger({ action: "second" }, { + debounce: { key: "my-key", delay: "10s" } +}); +// Task executes with { action: "first" } +``` + +**Trailing Mode**: Uses payload/options from the most recent trigger. + +```ts +await myTask.trigger( + { data: "latest-value" }, + { + debounce: { + key: "trailing-example", + delay: "10s", + mode: "trailing", + }, + } +); +``` + +In trailing mode, these options update with each trigger: +- `payload` — task input data +- `metadata` — run metadata +- `tags` — run tags (replaces existing) +- `maxAttempts` — retry attempts +- `maxDuration` — maximum compute time +- `machine` — machine preset + +### Important Notes + +- Idempotency keys take precedence over debounce settings +- Compatible with `triggerAndWait()` — parent runs block correctly on debounced execution +- Debounce key is scoped to the task + +## Concurrency & Queues + +```ts +import { task, queue } from "@trigger.dev/sdk"; + +// Shared queue for related tasks +const emailQueue = queue({ + name: "email-processing", + concurrencyLimit: 5, // Max 5 emails processing simultaneously +}); + +// Task-level concurrency +export const oneAtATime = task({ + id: "sequential-task", + queue: { concurrencyLimit: 1 }, // Process one at a time + run: async (payload) => { + // Critical section - only one instance runs + }, +}); + +// Per-user concurrency +export const processUserData = task({ + id: "process-user-data", + run: async (payload: { userId: string }) => { + // Override queue with user-specific concurrency + await childTask.trigger(payload, { + queue: { + name: `user-${payload.userId}`, + concurrencyLimit: 2, + }, + }); + }, +}); + +export const emailTask = task({ + id: "send-email", + queue: emailQueue, // Use shared queue + run: async (payload: { to: string }) => { + // Send email logic + }, +}); +``` + +## Error Handling & Retries + +```ts +import { task, retry, AbortTaskRunError } from "@trigger.dev/sdk"; + +export const resilientTask = task({ + id: "resilient-task", + retry: { + maxAttempts: 10, + factor: 1.8, // Exponential backoff multiplier + minTimeoutInMs: 500, + maxTimeoutInMs: 30_000, + randomize: false, + }, + catchError: async ({ error, ctx }) => { + // Custom error handling + if (error.code === "FATAL_ERROR") { + throw new AbortTaskRunError("Cannot retry this error"); + } + + // Log error details + console.error(`Task ${ctx.task.id} failed:`, error); + + // Allow retry by returning nothing + return { retryAt: new Date(Date.now() + 60000) }; // Retry in 1 minute + }, + run: async (payload) => { + // Retry specific operations + const result = await retry.onThrow( + async () => { + return await unstableApiCall(payload); + }, + { maxAttempts: 3 } + ); + + // Conditional HTTP retries + const response = await retry.fetch("https://api.example.com", { + retry: { + maxAttempts: 5, + condition: (response, error) => { + return response?.status === 429 || response?.status >= 500; + }, + }, + }); + + return result; + }, +}); +``` + +## Machines & Performance + +```ts +export const heavyTask = task({ + id: "heavy-computation", + machine: { preset: "large-2x" }, // 8 vCPU, 16 GB RAM + maxDuration: 1800, // 30 minutes timeout + run: async (payload, { ctx }) => { + // Resource-intensive computation + if (ctx.machine.preset === "large-2x") { + // Use all available cores + return await parallelProcessing(payload); + } + + return await standardProcessing(payload); + }, +}); + +// Override machine when triggering +await heavyTask.trigger(payload, { + machine: { preset: "medium-1x" }, // Override for this run +}); +``` + +**Machine Presets:** + +- `micro`: 0.25 vCPU, 0.25 GB RAM +- `small-1x`: 0.5 vCPU, 0.5 GB RAM (default) +- `small-2x`: 1 vCPU, 1 GB RAM +- `medium-1x`: 1 vCPU, 2 GB RAM +- `medium-2x`: 2 vCPU, 4 GB RAM +- `large-1x`: 4 vCPU, 8 GB RAM +- `large-2x`: 8 vCPU, 16 GB RAM + +## Idempotency + +```ts +import { task, idempotencyKeys } from "@trigger.dev/sdk"; + +export const paymentTask = task({ + id: "process-payment", + retry: { + maxAttempts: 3, + }, + run: async (payload: { orderId: string; amount: number }) => { + // Automatically scoped to this task run, so if the task is retried, the idempotency key will be the same + const idempotencyKey = await idempotencyKeys.create(`payment-${payload.orderId}`); + + // Ensure payment is processed only once + await chargeCustomer.trigger(payload, { + idempotencyKey, + idempotencyKeyTTL: "24h", // Key expires in 24 hours + }); + }, +}); + +// Payload-based idempotency +import { createHash } from "node:crypto"; + +function createPayloadHash(payload: any): string { + const hash = createHash("sha256"); + hash.update(JSON.stringify(payload)); + return hash.digest("hex"); +} + +export const deduplicatedTask = task({ + id: "deduplicated-task", + run: async (payload) => { + const payloadHash = createPayloadHash(payload); + const idempotencyKey = await idempotencyKeys.create(payloadHash); + + await processData.trigger(payload, { idempotencyKey }); + }, +}); +``` + +## Metadata & Progress Tracking + +```ts +import { task, metadata } from "@trigger.dev/sdk"; + +export const batchProcessor = task({ + id: "batch-processor", + run: async (payload: { items: any[] }, { ctx }) => { + const totalItems = payload.items.length; + + // Initialize progress metadata + metadata + .set("progress", 0) + .set("totalItems", totalItems) + .set("processedItems", 0) + .set("status", "starting"); + + const results = []; + + for (let i = 0; i < payload.items.length; i++) { + const item = payload.items[i]; + + // Process item + const result = await processItem(item); + results.push(result); + + // Update progress + const progress = ((i + 1) / totalItems) * 100; + metadata + .set("progress", progress) + .increment("processedItems", 1) + .append("logs", `Processed item ${i + 1}/${totalItems}`) + .set("currentItem", item.id); + } + + // Final status + metadata.set("status", "completed"); + + return { results, totalProcessed: results.length }; + }, +}); + +// Update parent metadata from child task +export const childTask = task({ + id: "child-task", + run: async (payload, { ctx }) => { + // Update parent task metadata + metadata.parent.set("childStatus", "processing"); + metadata.root.increment("childrenCompleted", 1); + + return { processed: true }; + }, +}); +``` + +## Logging & Tracing + +```ts +import { task, logger } from "@trigger.dev/sdk"; + +export const tracedTask = task({ + id: "traced-task", + run: async (payload, { ctx }) => { + logger.info("Task started", { userId: payload.userId }); + + // Custom trace with attributes + const user = await logger.trace( + "fetch-user", + async (span) => { + span.setAttribute("user.id", payload.userId); + span.setAttribute("operation", "database-fetch"); + + const userData = await database.findUser(payload.userId); + span.setAttribute("user.found", !!userData); + + return userData; + }, + { userId: payload.userId } + ); + + logger.debug("User fetched", { user: user.id }); + + try { + const result = await processUser(user); + logger.info("Processing completed", { result }); + return result; + } catch (error) { + logger.error("Processing failed", { + error: error.message, + userId: payload.userId, + }); + throw error; + } + }, +}); +``` + +## Hidden Tasks + +```ts +// Hidden task - not exported, only used internally +const internalProcessor = task({ + id: "internal-processor", + run: async (payload: { data: string }) => { + return { processed: payload.data.toUpperCase() }; + }, +}); + +// Public task that uses hidden task +export const publicWorkflow = task({ + id: "public-workflow", + run: async (payload: { input: string }) => { + // Use hidden task internally + const result = await internalProcessor.triggerAndWait({ + data: payload.input, + }); + + if (result.ok) { + return { output: result.output.processed }; + } + + throw new Error("Internal processing failed"); + }, +}); +``` + +## Best Practices + +- **Concurrency**: Use queues to prevent overwhelming external services +- **Retries**: Configure exponential backoff for transient failures +- **Idempotency**: Always use for payment/critical operations +- **Metadata**: Track progress for long-running tasks +- **Machines**: Match machine size to computational requirements +- **Tags**: Use consistent naming patterns for filtering +- **Debouncing**: Use for user activity, webhooks, and notification batching +- **Batch triggering**: Use for bulk operations up to 1,000 items +- **Error Handling**: Distinguish between retryable and fatal errors + +Design tasks to be stateless, idempotent, and resilient to failures. Use metadata for state tracking and queues for resource management. diff --git a/.agents/skills/trigger-tasks/references/basic-tasks.md b/.agents/skills/trigger-tasks/references/basic-tasks.md new file mode 100644 index 0000000..56bff34 --- /dev/null +++ b/.agents/skills/trigger-tasks/references/basic-tasks.md @@ -0,0 +1,199 @@ +# Trigger.dev Basic Tasks (v4) + +**MUST use `@trigger.dev/sdk`, NEVER `client.defineJob`** + +## Basic Task + +```ts +import { task } from "@trigger.dev/sdk"; + +export const processData = task({ + id: "process-data", + retry: { + maxAttempts: 10, + factor: 1.8, + minTimeoutInMs: 500, + maxTimeoutInMs: 30_000, + randomize: false, + }, + run: async (payload: { userId: string; data: any[] }) => { + // Task logic - runs for long time, no timeouts + console.log(`Processing ${payload.data.length} items for user ${payload.userId}`); + return { processed: payload.data.length }; + }, +}); +``` + +## Schema Task (with validation) + +```ts +import { schemaTask } from "@trigger.dev/sdk"; +import { z } from "zod"; + +export const validatedTask = schemaTask({ + id: "validated-task", + schema: z.object({ + name: z.string(), + age: z.number(), + email: z.string().email(), + }), + run: async (payload) => { + // Payload is automatically validated and typed + return { message: `Hello ${payload.name}, age ${payload.age}` }; + }, +}); +``` + +## Triggering Tasks + +### From Backend Code + +```ts +import { tasks } from "@trigger.dev/sdk"; +import type { processData } from "./trigger/tasks"; + +// Single trigger +const handle = await tasks.trigger("process-data", { + userId: "123", + data: [{ id: 1 }, { id: 2 }], +}); + +// Batch trigger (up to 1,000 items, 3MB per payload) +const batchHandle = await tasks.batchTrigger("process-data", [ + { payload: { userId: "123", data: [{ id: 1 }] } }, + { payload: { userId: "456", data: [{ id: 2 }] } }, +]); +``` + +### Debounced Triggering + +Consolidate multiple triggers into a single execution: + +```ts +// Multiple rapid triggers with same key = single execution +await myTask.trigger( + { userId: "123" }, + { + debounce: { + key: "user-123-update", // Unique key for debounce group + delay: "5s", // Wait before executing + }, + } +); + +// Trailing mode: use payload from LAST trigger +await myTask.trigger( + { data: "latest-value" }, + { + debounce: { + key: "trailing-example", + delay: "10s", + mode: "trailing", // Default is "leading" (first payload) + }, + } +); +``` + +**Debounce modes:** +- `leading` (default): Uses payload from first trigger, subsequent triggers only reschedule +- `trailing`: Uses payload from most recent trigger + +### From Inside Tasks (with Result handling) + +```ts +export const parentTask = task({ + id: "parent-task", + run: async (payload) => { + // Trigger and continue + const handle = await childTask.trigger({ data: "value" }); + + // Trigger and wait - returns Result object, NOT task output + const result = await childTask.triggerAndWait({ data: "value" }); + if (result.ok) { + console.log("Task output:", result.output); // Actual task return value + } else { + console.error("Task failed:", result.error); + } + + // Quick unwrap (throws on error) + const output = await childTask.triggerAndWait({ data: "value" }).unwrap(); + + // Batch trigger and wait + const results = await childTask.batchTriggerAndWait([ + { payload: { data: "item1" } }, + { payload: { data: "item2" } }, + ]); + + for (const run of results) { + if (run.ok) { + console.log("Success:", run.output); + } else { + console.log("Failed:", run.error); + } + } + }, +}); + +export const childTask = task({ + id: "child-task", + run: async (payload: { data: string }) => { + return { processed: payload.data }; + }, +}); +``` + +> Never wrap triggerAndWait or batchTriggerAndWait calls in a Promise.all or Promise.allSettled as this is not supported in Trigger.dev tasks. + +## Waits + +```ts +import { task, wait } from "@trigger.dev/sdk"; + +export const taskWithWaits = task({ + id: "task-with-waits", + run: async (payload) => { + console.log("Starting task"); + + // Wait for specific duration + await wait.for({ seconds: 30 }); + await wait.for({ minutes: 5 }); + await wait.for({ hours: 1 }); + await wait.for({ days: 1 }); + + // Wait until specific date + await wait.until({ date: new Date("2024-12-25") }); + + // Wait for token (from external system) + await wait.forToken({ + token: "user-approval-token", + timeoutInSeconds: 3600, // 1 hour timeout + }); + + console.log("All waits completed"); + return { status: "completed" }; + }, +}); +``` + +> Never wrap wait calls in a Promise.all or Promise.allSettled as this is not supported in Trigger.dev tasks. + +## Key Points + +- **Result vs Output**: `triggerAndWait()` returns a `Result` object with `ok`, `output`, `error` properties - NOT the direct task output +- **Type safety**: Use `import type` for task references when triggering from backend +- **Waits > 5 seconds**: Automatically checkpointed, don't count toward compute usage +- **Debounce + idempotency**: Idempotency keys take precedence over debounce settings + +## NEVER Use (v2 deprecated) + +```ts +// BREAKS APPLICATION +client.defineJob({ + id: "job-id", + run: async (payload, io) => { + /* ... */ + }, +}); +``` + +Use SDK (`@trigger.dev/sdk`), check `result.ok` before accessing `result.output` diff --git a/.agents/skills/trigger-tasks/references/scheduled-tasks.md b/.agents/skills/trigger-tasks/references/scheduled-tasks.md new file mode 100644 index 0000000..b314753 --- /dev/null +++ b/.agents/skills/trigger-tasks/references/scheduled-tasks.md @@ -0,0 +1,113 @@ +# Scheduled Tasks (Cron) + +Recurring tasks using cron. For one-off future runs, use the **delay** option. + +## Define a Scheduled Task + +```ts +import { schedules } from "@trigger.dev/sdk"; + +export const task = schedules.task({ + id: "first-scheduled-task", + run: async (payload) => { + payload.timestamp; // Date (scheduled time, UTC) + payload.lastTimestamp; // Date | undefined + payload.timezone; // IANA, e.g. "America/New_York" (default "UTC") + payload.scheduleId; // string + payload.externalId; // string | undefined + payload.upcoming; // Date[] + + payload.timestamp.toLocaleString("en-US", { timeZone: payload.timezone }); + }, +}); +``` + +> Scheduled tasks need at least one schedule attached to run. + +## Attach Schedules + +**Declarative (sync on dev/deploy):** + +```ts +schedules.task({ + id: "every-2h", + cron: "0 */2 * * *", // UTC + run: async () => {}, +}); + +schedules.task({ + id: "tokyo-5am", + cron: { pattern: "0 5 * * *", timezone: "Asia/Tokyo", environments: ["PRODUCTION", "STAGING"] }, + run: async () => {}, +}); +``` + +**Imperative (SDK or dashboard):** + +```ts +await schedules.create({ + task: task.id, + cron: "0 0 * * *", + timezone: "America/New_York", // DST-aware + externalId: "user_123", + deduplicationKey: "user_123-daily", // updates if reused +}); +``` + +### Dynamic / Multi-tenant Example + +```ts +// /trigger/reminder.ts +export const reminderTask = schedules.task({ + id: "todo-reminder", + run: async (p) => { + if (!p.externalId) throw new Error("externalId is required"); + const user = await db.getUser(p.externalId); + await sendReminderEmail(user); + }, +}); +``` + +```ts +// app/reminders/route.ts +export async function POST(req: Request) { + const data = await req.json(); + return Response.json( + await schedules.create({ + task: reminderTask.id, + cron: "0 8 * * *", + timezone: data.timezone, + externalId: data.userId, + deduplicationKey: `${data.userId}-reminder`, + }) + ); +} +``` + +## Cron Syntax (no seconds) + +``` +* * * * * +| | | | └ day of week (0–7 or 1L–7L; 0/7=Sun; L=last) +| | | └── month (1–12) +| | └──── day of month (1–31 or L) +| └────── hour (0–23) +└──────── minute (0–59) +``` + +## When Schedules Won't Trigger + +- **Dev:** only when the dev CLI is running. +- **Staging/Production:** only for tasks in the **latest deployment**. + +## SDK Management + +```ts +await schedules.retrieve(id); +await schedules.list(); +await schedules.update(id, { cron: "0 0 1 * *", externalId: "ext", deduplicationKey: "key" }); +await schedules.deactivate(id); +await schedules.activate(id); +await schedules.del(id); +await schedules.timezones(); // list of IANA timezones +``` diff --git a/.gitignore b/.gitignore index 10dcd9e..b51644e 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,8 @@ yarn-error.log* next-env.d.ts /app/generated/prisma + +# trigger.dev +.trigger + +.trigger \ No newline at end of file diff --git a/app/api/ai/design/route.ts b/app/api/ai/design/route.ts new file mode 100644 index 0000000..0658435 --- /dev/null +++ b/app/api/ai/design/route.ts @@ -0,0 +1,67 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { tasks } from "@trigger.dev/sdk/v3"; +import { prisma } from "@/lib/prisma"; +import { getProjectAccess } from "@/lib/project-access"; +import type { designAgent } from "@/trigger/design-agent"; + +export async function POST(request: Request) { + const { userId } = await auth(); + if (!userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + let prompt: unknown; + let roomId: unknown; + let projectId: unknown; + + try { + const body = await request.json(); + prompt = body.prompt; + roomId = body.roomId; + projectId = body.projectId; + } catch { + return NextResponse.json({ error: "Invalid request body" }, { status: 400 }); + } + + if (typeof prompt !== "string" || !prompt.trim()) { + return NextResponse.json({ error: "Prompt is required" }, { status: 400 }); + } + + if (typeof roomId !== "string" || !roomId) { + return NextResponse.json({ error: "Room ID is required" }, { status: 400 }); + } + + if (typeof projectId !== "string" || !projectId) { + return NextResponse.json({ error: "Project ID is required" }, { status: 400 }); + } + + const access = await getProjectAccess(projectId); + if (!access.hasAccess) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + try { + const handle = await tasks.trigger("design-agent", { + prompt: prompt.trim(), + roomId, + projectId, + userId, + }); + + await prisma.taskRun.create({ + data: { + runId: handle.id, + projectId, + userId, + }, + }); + + return NextResponse.json({ runId: handle.id }, { status: 201 }); + } catch { + return NextResponse.json( + { error: "Failed to trigger design task" }, + { status: 500 } + ); + } +} diff --git a/app/api/ai/design/token/route.ts b/app/api/ai/design/token/route.ts new file mode 100644 index 0000000..32622f5 --- /dev/null +++ b/app/api/ai/design/token/route.ts @@ -0,0 +1,54 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { auth as triggerAuth } from "@trigger.dev/sdk/v3"; +import { prisma } from "@/lib/prisma"; + +export async function POST(request: Request) { + const { userId } = await auth(); + if (!userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + let runId: unknown; + + try { + const body = await request.json(); + runId = body.runId; + } catch { + return NextResponse.json({ error: "Invalid request body" }, { status: 400 }); + } + + if (typeof runId !== "string" || !runId) { + return NextResponse.json({ error: "Run ID is required" }, { status: 400 }); + } + + try { + const taskRun = await prisma.taskRun.findUnique({ + where: { runId }, + }); + + if (!taskRun) { + return NextResponse.json({ error: "Run not found" }, { status: 404 }); + } + + if (taskRun.userId !== userId) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const publicToken = await triggerAuth.createPublicToken({ + scopes: { + read: { + runs: [runId], + }, + }, + expirationTime: "15m", + }); + + return NextResponse.json({ token: publicToken }); + } catch { + return NextResponse.json( + { error: "Failed to generate token" }, + { status: 500 } + ); + } +} diff --git a/app/api/ai/spec/route.ts b/app/api/ai/spec/route.ts new file mode 100644 index 0000000..de7f96d --- /dev/null +++ b/app/api/ai/spec/route.ts @@ -0,0 +1,64 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { tasks } from "@trigger.dev/sdk/v3"; +import { prisma } from "@/lib/prisma"; +import { getProjectAccess } from "@/lib/project-access"; +import type { generateSpec } from "@/trigger/generate-spec"; + +export async function POST(request: Request) { + const { userId } = await auth(); + if (!userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + let roomId: unknown; + let chatHistory: unknown; + let nodes: unknown; + let edges: unknown; + + try { + const body = await request.json(); + roomId = body.roomId; + chatHistory = body.chatHistory; + nodes = body.nodes; + edges = body.edges; + } catch { + return NextResponse.json({ error: "Invalid request body" }, { status: 400 }); + } + + if (typeof roomId !== "string" || !roomId) { + return NextResponse.json({ error: "Room ID is required" }, { status: 400 }); + } + + const access = await getProjectAccess(roomId); + if (!access.hasAccess || !access.project) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const projectId = access.project.id; + + try { + const handle = await tasks.trigger("generate-spec", { + projectId, + roomId, + chatHistory: Array.isArray(chatHistory) ? chatHistory : [], + nodes: Array.isArray(nodes) ? nodes : [], + edges: Array.isArray(edges) ? edges : [], + }); + + await prisma.taskRun.create({ + data: { + runId: handle.id, + projectId, + userId, + }, + }); + + return NextResponse.json({ runId: handle.id }, { status: 201 }); + } catch { + return NextResponse.json( + { error: "Failed to trigger spec generation task" }, + { status: 500 }, + ); + } +} diff --git a/app/api/ai/spec/token/route.ts b/app/api/ai/spec/token/route.ts new file mode 100644 index 0000000..ec14dd2 --- /dev/null +++ b/app/api/ai/spec/token/route.ts @@ -0,0 +1,54 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { auth as triggerAuth } from "@trigger.dev/sdk/v3"; +import { prisma } from "@/lib/prisma"; + +export async function POST(request: Request) { + const { userId } = await auth(); + if (!userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + let runId: unknown; + + try { + const body = await request.json(); + runId = body.runId; + } catch { + return NextResponse.json({ error: "Invalid request body" }, { status: 400 }); + } + + if (typeof runId !== "string" || !runId) { + return NextResponse.json({ error: "Run ID is required" }, { status: 400 }); + } + + try { + const taskRun = await prisma.taskRun.findUnique({ + where: { runId }, + }); + + if (!taskRun) { + return NextResponse.json({ error: "Run not found" }, { status: 404 }); + } + + if (taskRun.userId !== userId) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const publicToken = await triggerAuth.createPublicToken({ + scopes: { + read: { + runs: [runId], + }, + }, + expirationTime: "1h", + }); + + return NextResponse.json({ token: publicToken }); + } catch { + return NextResponse.json( + { error: "Failed to generate token" }, + { status: 500 }, + ); + } +} diff --git a/app/api/projects/[id]/specs/[specId]/content/route.ts b/app/api/projects/[id]/specs/[specId]/content/route.ts new file mode 100644 index 0000000..65c2313 --- /dev/null +++ b/app/api/projects/[id]/specs/[specId]/content/route.ts @@ -0,0 +1,58 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { prisma } from "@/lib/prisma"; +import { getProjectAccess } from "@/lib/project-access"; +import { get } from "@vercel/blob"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string; specId: string }> } +) { + try { + const session = await auth(); + if (!session.userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const { id: projectId, specId } = await params; + + const access = await getProjectAccess(projectId); + if (!access.hasAccess) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const spec = await prisma.projectSpec.findUnique({ + where: { id: specId }, + }); + + if (!spec || spec.projectId !== projectId) { + return NextResponse.json({ error: "Spec not found" }, { status: 404 }); + } + + if (!spec.filePath) { + return NextResponse.json({ error: "Spec content not yet uploaded" }, { status: 503 }); + } + + const blobData = await get(spec.filePath, { access: "private" }); + if (!blobData || blobData.statusCode === 304 || !blobData.stream) { + return NextResponse.json({ error: "Spec content not found" }, { status: 404 }); + } + + const reader = blobData.stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + text += decoder.decode(); + + return NextResponse.json({ content: text }); + } catch (e) { + return NextResponse.json( + { error: e instanceof Error ? e.message : "Failed to get spec content" }, + { status: 500 } + ); + } +} diff --git a/app/api/projects/[id]/specs/[specId]/download/route.ts b/app/api/projects/[id]/specs/[specId]/download/route.ts new file mode 100644 index 0000000..cc19014 --- /dev/null +++ b/app/api/projects/[id]/specs/[specId]/download/route.ts @@ -0,0 +1,64 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { prisma } from "@/lib/prisma"; +import { getProjectAccess } from "@/lib/project-access"; +import { get } from "@vercel/blob"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string; specId: string }> } +) { + try { + const session = await auth(); + if (!session.userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const { id: projectId, specId } = await params; + + const access = await getProjectAccess(projectId); + if (!access.hasAccess) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const spec = await prisma.projectSpec.findUnique({ + where: { id: specId }, + }); + + if (!spec || spec.projectId !== projectId) { + return NextResponse.json({ error: "Spec not found" }, { status: 404 }); + } + + if (!spec.filePath) { + return NextResponse.json({ error: "Spec content not yet uploaded" }, { status: 503 }); + } + + const blobData = await get(spec.filePath, { access: "private" }); + if (!blobData || blobData.statusCode === 304 || !blobData.stream) { + return NextResponse.json({ error: "Spec content not found" }, { status: 404 }); + } + + const reader = blobData.stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + text += decoder.decode(); + + return new NextResponse(text, { + status: 200, + headers: { + "Content-Type": "text/markdown", + "Content-Disposition": `attachment; filename="${specId}.md"`, + }, + }); + } catch (e) { + return NextResponse.json( + { error: e instanceof Error ? e.message : "Failed to download spec" }, + { status: 500 } + ); + } +} diff --git a/app/api/projects/[id]/specs/route.ts b/app/api/projects/[id]/specs/route.ts new file mode 100644 index 0000000..83b5193 --- /dev/null +++ b/app/api/projects/[id]/specs/route.ts @@ -0,0 +1,41 @@ +import { NextResponse } from "next/server"; +import { auth } from "@clerk/nextjs/server"; +import { prisma } from "@/lib/prisma"; +import { getProjectAccess } from "@/lib/project-access"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const session = await auth(); + if (!session.userId) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const projectId = (await params).id; + + const access = await getProjectAccess(projectId); + if (!access.hasAccess) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const specs = await prisma.projectSpec.findMany({ + where: { projectId }, + select: { + id: true, + projectId: true, + filePath: true, + createdAt: true, + }, + orderBy: { createdAt: "desc" }, + }); + + return NextResponse.json(specs); + } catch (e) { + return NextResponse.json( + { error: e instanceof Error ? e.message : "Failed to list specs" }, + { status: 500 } + ); + } +} diff --git a/components/editor/ai-sidebar.tsx b/components/editor/ai-sidebar.tsx index baf7cf5..6534a67 100644 --- a/components/editor/ai-sidebar.tsx +++ b/components/editor/ai-sidebar.tsx @@ -1,20 +1,40 @@ "use client" -import { useRef, useState } from "react" -import { X, Bot, Sparkles, FileText, Download, Send } from "lucide-react" +import { useRef, useState, useEffect, useCallback, useContext } from "react" +import { X, Bot, Sparkles, FileText, Download, Send, Loader2 } from "lucide-react" import { Button } from "@/components/ui/button" import { Textarea } from "@/components/ui/textarea" import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs" import { ScrollArea } from "@/components/ui/scroll-area" +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog" +import { useFeeds, useCreateFeed, useFeedMessages, useCreateFeedMessage } from "@liveblocks/react" +import { useUser } from "@clerk/nextjs" +import { useRealtimeRun } from "@trigger.dev/react-hooks" +import ReactMarkdown from "react-markdown" +import type { AiStatusPayload } from "@/types/canvas" +import { AiChatMessageSchema } from "@/types/task" +import type { designAgent } from "@/trigger/design-agent" +import type { generateSpec } from "@/trigger/generate-spec" +import type { RealtimeRun } from "@trigger.dev/core/v3" +import { CanvasStateContext } from "./canvas" + +interface ProjectSpecItem { + id: string + projectId: string + filePath: string | null + createdAt: string +} interface AiSidebarProps { isOpen: boolean onClose: () => void -} - -interface ChatMessage { - role: "user" | "assistant" - content: string + aiStatus?: AiStatusPayload | null + projectId: string } const STARTER_CHIPS = [ @@ -23,19 +43,214 @@ const STARTER_CHIPS = [ "Build a CI/CD pipeline", ] -export function AiSidebar({ isOpen, onClose }: AiSidebarProps) { - const [messages, setMessages] = useState([]) +export function AiSidebar({ isOpen, onClose, aiStatus, projectId }: AiSidebarProps) { const [input, setInput] = useState("") + const [sendError, setSendError] = useState(null) + const [isSubmitting, setIsSubmitting] = useState(false) + const [runId, setRunId] = useState(null) + const [publicToken, setPublicToken] = useState(null) const textareaRef = useRef(null) + const { user } = useUser() + const { feeds } = useFeeds() + const createFeed = useCreateFeed() + const { messages: feedMessages } = useFeedMessages("ai-chat") + const createFeedMessage = useCreateFeedMessage() + + const [specs, setSpecs] = useState([]) + const [specsLoading, setSpecsLoading] = useState(false) + const [selectedSpec, setSelectedSpec] = useState(null) + const [previewOpen, setPreviewOpen] = useState(false) + const [previewContent, setPreviewContent] = useState(null) + const [previewLoading, setPreviewLoading] = useState(false) + const [specRunId, setSpecRunId] = useState(null) + const [specRunToken, setSpecRunToken] = useState(null) + + const { getCanvasState } = useContext(CanvasStateContext) + + const fetchSpecs = useCallback(async () => { + setSpecsLoading(true) + try { + const res = await fetch(`/api/projects/${projectId}/specs`) + if (res.ok) { + const data = await res.json() + setSpecs(data) + } + } catch { + // silent + } + setSpecsLoading(false) + }, [projectId]) + + useEffect(() => { + fetchSpecs() + }, [fetchSpecs]) + + const openPreview = useCallback(async (spec: ProjectSpecItem) => { + setSelectedSpec(spec) + setPreviewOpen(true) + setPreviewLoading(true) + setPreviewContent(null) + try { + const res = await fetch(`/api/projects/${projectId}/specs/${spec.id}/content`) + if (res.ok) { + const data = await res.json() + setPreviewContent(data.content) + } else { + setPreviewContent("*Failed to load spec content*") + } + } catch { + setPreviewContent("*Failed to load spec content*") + } + setPreviewLoading(false) + }, [projectId]) + + const handleDownload = useCallback(async (spec: ProjectSpecItem) => { + const a = document.createElement("a") + a.href = `/api/projects/${projectId}/specs/${spec.id}/download` + a.download = `${spec.id}.md` + document.body.appendChild(a) + a.click() + document.body.removeChild(a) + }, [projectId]) + + const senderName = user?.fullName || user?.username || user?.primaryEmailAddress?.emailAddress || "Anonymous" + const isRunActive = isSubmitting || !!runId + + const { run } = useRealtimeRun(runId ?? undefined, { + accessToken: publicToken ?? undefined, + enabled: !!runId && !!publicToken, + onComplete: useCallback( + async (runResult: RealtimeRun) => { + const output = runResult.output as + | { status?: string; nodesGenerated?: number; edgesGenerated?: number } + | undefined + const content = + output?.status === "complete" + ? `Design generated with ${output.nodesGenerated} nodes and ${output.edgesGenerated} edges` + : "Could not generate a design from that prompt" + try { + await createFeedMessage("ai-chat", { + sender: "AI Architect", + role: "assistant", + content, + timestamp: Date.now(), + }) + } catch { + // silent + } + setRunId(null) + setPublicToken(null) + }, + [createFeedMessage], + ), + }) + + useRealtimeRun(specRunId ?? undefined, { + accessToken: specRunToken ?? undefined, + enabled: !!specRunId && !!specRunToken, + onComplete: useCallback(async () => { + setSpecRunId(null) + setSpecRunToken(null) + await fetchSpecs() + }, [fetchSpecs]), + }) + + useEffect(() => { + if (!feeds) return + const exists = feeds.some((f) => f.feedId === "ai-chat") + if (!exists) { + createFeed("ai-chat").catch(() => {}) + } + }, [feeds, createFeed]) + + const validatedMessages = (feedMessages ?? []) + .map((msg) => { + const parsed = AiChatMessageSchema.safeParse(msg.data) + if (parsed.success) { + return { ...parsed.data, id: msg.id } + } + return null + }) + .filter((m): m is NonNullable => m !== null) + + const submitPrompt = useCallback( + async (text: string) => { + if (isRunActive) return + setSendError(null) + setIsSubmitting(true) + try { + await createFeedMessage("ai-chat", { + sender: senderName, + role: "user", + content: text, + timestamp: Date.now(), + }) + } catch { + setSendError("Failed to send message") + setIsSubmitting(false) + return + } + + try { + const res = await fetch("/api/ai/design", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ prompt: text, roomId: projectId, projectId }), + }) + if (!res.ok) { + const err = await res.json().catch(() => ({})) + await createFeedMessage("ai-chat", { + sender: "System", + role: "assistant", + content: err.error || "Failed to start design task", + timestamp: Date.now(), + }).catch(() => {}) + setIsSubmitting(false) + return + } + const { runId: newRunId } = await res.json() + + const tokenRes = await fetch("/api/ai/design/token", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ runId: newRunId }), + }) + if (!tokenRes.ok) { + await createFeedMessage("ai-chat", { + sender: "System", + role: "assistant", + content: "Failed to get access token", + timestamp: Date.now(), + }).catch(() => {}) + setIsSubmitting(false) + return + } + const { token } = await tokenRes.json() + + setRunId(newRunId) + setPublicToken(token) + } catch { + setSendError("Network error") + await createFeedMessage("ai-chat", { + sender: "System", + role: "assistant", + content: "Network error: failed to send message", + timestamp: Date.now(), + }).catch(() => {}) + } + setIsSubmitting(false) + }, + [senderName, projectId, createFeedMessage, isRunActive], + ) - const handleSend = () => { + const handleSend = async () => { const text = input.trim() if (!text) return - setMessages((prev) => [...prev, { role: "user", content: text }]) setInput("") if (textareaRef.current) { textareaRef.current.style.height = "auto" } + await submitPrompt(text) } const handleInputChange = (e: React.ChangeEvent) => { @@ -52,13 +267,45 @@ export function AiSidebar({ isOpen, onClose }: AiSidebarProps) { } } - const handleStarterClick = (chip: string) => { - setMessages((prev) => [...prev, { role: "user", content: chip }]) - setInput("") + const handleStarterClick = async (chip: string) => { + if (isRunActive) return + await submitPrompt(chip) } + const handleGenerateSpec = useCallback(async () => { + if (specRunId) return + const { nodes, edges } = getCanvasState() + const chatHistory = validatedMessages.map(m => ({ role: m.role, content: m.content })) + try { + const res = await fetch("/api/ai/spec", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ roomId: projectId, chatHistory, nodes, edges }), + }) + if (!res.ok) { + return + } + const { runId: newRunId } = await res.json() + const tokenRes = await fetch("/api/ai/spec/token", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ runId: newRunId }), + }) + if (!tokenRes.ok) { + return + } + const { token } = await tokenRes.json() + setSpecRunId(newRunId) + setSpecRunToken(token) + } catch { + // silent + } + }, [projectId, getCanvasState, validatedMessages, specRunId]) + return (