From 5bc995ffb929cc4adc5015515cdb6b4df974bc4b Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 16:48:05 +0300 Subject: [PATCH 1/8] feat: add workflow-js skills --- skills/SKILL.md | 55 ++++ skills/agents.md | 240 ++++++++++++++++++ skills/basics/client.md | 180 +++++++++++++ skills/basics/context.md | 223 ++++++++++++++++ skills/basics/serve.md | 66 +++++ skills/features/flow-control.md | 135 ++++++++++ skills/features/invoke.md | 93 +++++++ .../features/retries-failures-reliability.md | 217 ++++++++++++++++ skills/features/wait-for-event.md | 94 +++++++ skills/features/webhooks.md | 79 ++++++ skills/how-to/local-dev.md | 122 +++++++++ skills/how-to/middleware.md | 155 +++++++++++ skills/how-to/migrations.md | 164 ++++++++++++ skills/how-to/realtime.md | 173 +++++++++++++ skills/rest-api.md | 156 ++++++++++++ skills/troubleshooting.md | 192 ++++++++++++++ 16 files changed, 2344 insertions(+) create mode 100644 skills/SKILL.md create mode 100644 skills/agents.md create mode 100644 skills/basics/client.md create mode 100644 skills/basics/context.md create mode 100644 skills/basics/serve.md create mode 100644 skills/features/flow-control.md create mode 100644 skills/features/invoke.md create mode 100644 skills/features/retries-failures-reliability.md create mode 100644 skills/features/wait-for-event.md create mode 100644 skills/features/webhooks.md create mode 100644 skills/how-to/local-dev.md create mode 100644 skills/how-to/middleware.md create mode 100644 skills/how-to/migrations.md create mode 100644 skills/how-to/realtime.md create mode 100644 skills/rest-api.md create mode 100644 skills/troubleshooting.md diff --git a/skills/SKILL.md b/skills/SKILL.md new file mode 100644 index 00000000..e3e5ead3 --- /dev/null +++ b/skills/SKILL.md @@ -0,0 +1,55 @@ +--- +name: upstash-workflow-sdk +description: Lightweight guidance for using the Upstash Workflow SDK to define, trigger, and manage workflows. Use this Skill whenever a user wants to create workflow endpoints, run steps, or interact with the Upstash Workflow client. +--- + +# Upstash Workflow SDK + +## Quick Start +The Upstash Workflow SDK lets you expose serverless workflow endpoints and run them reliably using QStash under the hood. + +Install: +```bash +npm install @upstash/workflow +``` + +Define a simple workflow endpoint: +```ts +import { serve } from "@upstash/workflow"; + +export const { POST } = serve(async (context) => { + await context.run("step-1", () => console.log("step 1")); + await context.run("step-2", () => console.log("step 2")); +}); +``` + +Trigger it from your backend: +```ts +import { Client } from "@upstash/workflow"; + +const client = new Client({ token: process.env.QSTASH_TOKEN! }); +await client.trigger({ url: "https://your-app.com/api/workflow" }); +``` + +## Other Skill Files +These files contain the full documentation. Use them for details, patterns, and advanced behavior. + +- basics: + - **basics/serve** – How to expose workflow endpoints. + - **basics/context** – Full API for workflow `context` (steps, waits, webhooks, events, invoke, etc.). + - **basics/client** – Using the Workflow client to trigger, cancel, inspect, and notify runs. +- features: + - **features/invoke** – Cross‑workflow invocation. + - **features/reliability** – Retries, failure callbacks, and DLQ. + - **features/flow-control** – Rate limits, concurrency, and parallelism. + - **features/wait-for-event** – Notify and wait-for-event patterns. + - **features/webhooks** – Webhook creation and consumption. +- how to: + - **how-to/local-dev** – Local QStash dev server and tunneling. + - **how-to/realtime** – Realtime and human‑in‑the‑loop workflows. + - **how-to/migrations** – Migrating workflows safely. + - **how-to/middleware** – Adding middleware to workflows. +- other files: + - **rest-api** – Low-level REST endpoints for interacting with QStash/Workflow. + - **troubleshooting** – Common debugging and environment issues. + - **agents** – Using Workflow with agents, orchestrators, and automation patterns. diff --git a/skills/agents.md b/skills/agents.md new file mode 100644 index 00000000..64ba28a8 --- /dev/null +++ b/skills/agents.md @@ -0,0 +1,240 @@ + --- +name: workflow-agents +description: Skill for building, configuring, and orchestrating Upstash Workflow Agents. Use when the user mentions agent workflows, multi-agent collaboration, tools, tasks, Upstash Workflow, or patterns like evaluator-optimizer, prompt chaining, parallelization, or orchestrator-workers. +--- + +# Workflow Agents + +This Skill provides guidance for defining and orchestrating agents, tools, models, and tasks inside **Upstash Workflow** using **TypeScript**. It explains core concepts, common patterns, workflow configuration, and pitfalls. + +## Quick Overview +Use this Skill when: +- Building agents with tools and backgrounds +- Creating multi-step or multi-agent tasks +- Implementing patterns like evaluator-optimizer or prompt chaining +- Integrating LangChain or AI SDK tools +- Running workflows reliably inside Upstash Workflow + +The Workflow Agents API centers around four elements: +- **Models** — LLM providers (OpenAI, Anthropic, or any AI SDK provider) +- **Tools** — Functions agents can call +- **Agents** — LLM instances with background, maxSteps, and tools +- **Tasks** — Executable prompts assigned to one or more agents + +--- + +# Defining Models +A model defines which provider the agent uses and includes optional configuration for retries, timeouts, or rate‑limits. + +```ts +const agents = agentWorkflow(context) + +// Basic OpenAI model +const model = agents.openai('gpt-3.5-turbo') + +// OpenAI-compatible provider +const deepseek = agents.openai('deepseek-chat', { + baseURL: 'https://api.deepseek.com', + apiKey: process.env.DEEPSEEK_API_KEY, +}) + +// AI SDK provider (ex: Anthropic) +const anthropic = agents.AISDKModel({ + context, + provider: createAnthropic, // imported from @ai-sdk/anthropic + providerParams: { apiKey: process.env.ANTHROPIC_KEY }, + agentCallParams: { + timeout: 1000, + retries: 0, + } +}) +``` + +**Key parameters:** +- **callSettings / agentCallParams** — timeout, retries, flow control +- **provider / providerParams** — when using AI SDK providers + +**Pitfall:** If you use an OpenAI-compatible provider, you **must** set `baseURL`. + +--- + +# Defining Tools +Tools extend what agents can do. Workflow supports: +- WorkflowTool (native) +- AI SDK tools +- LangChain tools +- Agentic toolkits + +```ts +import { z } from 'zod' +import { tool } from 'ai' +import { WorkflowTool } from '@upstash/workflow-agents' +import { WikipediaQueryRun } from '@langchain/community/tools/wikipedia_query_run' + +const mathTool = tool({ + description: 'Evaluate a math expression', + parameters: z.object({ expression: z.string() }), + execute: async ({ expression }) => mathjs.evaluate(expression), +}) + +const workflowMath = new WorkflowTool({ + description: 'Evaluate math (workflow step aware)', + schema: z.object({ expression: z.string() }), + invoke: async ({ expression }) => mathjs.evaluate(expression), + executeAsStep: false, // allows context.call, etc. +}) + +const wikiTool = new WikipediaQueryRun({ + topKResults: 1, + maxDocContentLength: 500, +}) +``` + +**Common mistakes:** +- Workflow wraps execute/invoke in `context.run` by default, so you **cannot use context.call** unless `executeAsStep: false` is set. +- LangChain tools must return **strings**, not objects. + +--- + +# Defining Agents +Agents wrap a model and add behavior via: +- `maxSteps` — how many LLM calls the agent is allowed to make +- `background` — system prompt +- `tools` — available actions + +```ts +const generator = agents.agent({ + model, + name: 'generator', + maxSteps: 1, + background: 'Generate text from prompts.', + tools: {}, +}) + +const evaluator = agents.agent({ + model, + name: 'evaluator', + maxSteps: 1, + background: 'Evaluate responses and give corrections.', + tools: {}, +}) +``` + +**Tips:** +- Pick maxSteps carefully; too low prevents tool use; too high increases cost. +- Names appear in Upstash Console logs; keep them descriptive. + +--- + +# Tasks (Single & Multi Agent) +A **task** is a single execution of an agent or a group of agents. + +```ts +// Single agent task +const single = agents.task({ + agent: generator, + prompt: 'Explain quantum mechanics.', +}) +const { text } = await single.run() + +// Multi-agent with manager agent +const multi = agents.task({ + model, // manager LLM + agents: [generator, evaluator], + maxSteps: 3, + prompt: 'Generate text and refine it until quality improves.', +}) +const result = await multi.run() +``` + +**Tip:** In multi-agent mode, the model becomes a "manager" system that decides which agent to call. + +--- + +# Common Agent Patterns +Below are the patterns supported by the source files in this skill. + +## Prompt Chaining +Sequential agent calls where each output becomes the next input. +Useful for: stepwise research, multi‑stage content generation, breaking down complex tasks. + +Pitfall: watch `maxSteps` for agents that need both tool calls and summarization. + +## Evaluator‑Optimizer +Loop until evaluator returns a PASS. Simple feedback‑refinement pattern. + +Pitfall: Always check evaluator output with `.includes("PASS")`, not strict equality. + +## Parallelization +Use multiple agents with `Promise.all` and then aggregate. + +Pitfall: Avoid extremely large aggregated prompts; summarizing before combining is recommended. + +## Orchestrator‑Workers +Manager delegates sub‑tasks to specialized workers. +Useful for structured Q&A, multi‑topic analysis, or complex synthesis. + +Pitfall: The manager must have enough `maxSteps` to orchestrate multiple workers. + +--- + +# Best Practices +- Give each agent a *clear* background; ambiguous roles cause incorrect tool use. +- Define tools with strict schemas so LLMs call them reliably. +- Use multi-agent tasks when the problem requires specialization. +- Inspect console logs to debug tool calls and agent decisions. +- Use local QStash dev server during development to avoid rate limits. + +--- + +# Example: Combined Setup (models + tools + agents + tasks) +This shows all core fields together in one concise example. +```ts +export const { POST } = serve(async (context) => { + const agents = agentWorkflow(context) + const model = agents.openai('gpt-4o') + + const mathTool = tool({ + description: 'Compute math', + parameters: z.object({ expression: z.string() }), + execute: async ({ expression }) => mathjs.evaluate(expression), + }) + + const researcher = agents.agent({ + model, + name: 'researcher', + maxSteps: 2, + background: 'Research topics using wiki.', + tools: { wikiTool }, + }) + + const mathematician = agents.agent({ + model, + name: 'math', + maxSteps: 2, + background: 'Solve numeric problems.', + tools: { mathTool }, + }) + + const task = agents.task({ + model, // manager + agents: [researcher, mathematician], + maxSteps: 3, + prompt: 'Tell me about 3 stars and compute the sum of their masses.', + }) + + return (await task.run()).text +}) +``` + +--- + +# When to Use This Skill +Use this Skill whenever the task involves: +- Designing agent workflows +- Multi-agent coordination or patterns +- Using LangChain/AI SDK tools inside Workflow +- Implementing refinement loops or multi-step processing +- Debugging or optimizing Upstash agent behavior + +This Skill ensures consistent, correct usage of the Workflow Agents API and provides reference patterns for building reliable TypeScript agent systems. diff --git a/skills/basics/client.md b/skills/basics/client.md new file mode 100644 index 00000000..d44b35b4 --- /dev/null +++ b/skills/basics/client.md @@ -0,0 +1,180 @@ +# Workflow Client Basics + +This skill provides guidance for using the Workflow Client to trigger, cancel, inspect, resume, and manage workflow runs. It focuses on TypeScript usage patterns, argument structures, and common pitfalls. + +--- + +## Core Operations + +### Triggering Workflow Runs (`client.trigger`) +Starts one or multiple workflow runs with optional payload, metadata, retry logic, and flow control. + +```ts +// Single and multiple triggers +const single = await client.trigger({ + url: "https://your-endpoint/workflow", // required + body: { msg: "hello" }, // optional payload + headers: { "x-user": "123" }, // optional headers + workflowRunId: "custom-id", // custom unique identifier + retries: 3, // retry count for steps + retryDelay: "1000 * (1 + retried)", // expression-based retry delay + delay: "10s", // start later + notBefore: Math.floor(Date.now() / 1000) + 60, // absolute schedule + label: "my-run", // dashboard filtering + disableTelemetry: true, // opt‑out telemetry + flowControl: { // concurrency & rate limiting + key: "user-key", + rate: 10, + parallelism: 5, + period: "10m", + }, +}); + +const multiple = await client.trigger([ + { url: "https://api/runA", body: { a: 1 } }, + { url: "https://api/runB", retries: 5 }, +]); +``` + +**Pitfalls**: +- `workflowRunId` must be unique; collisions cause errors. +- `delay` is ignored if `notBefore` is provided. +- Using `url` incorrectly (missing protocol or dynamic segments) leads to invalid workflow endpoints. + +--- + +### Canceling Workflow Runs (`client.cancel`) +Cancel runs by ID, by URL prefix, or all active/pending runs. + +```ts +await client.cancel({ ids: ["wfr_123", "wfr_456"] }); +await client.cancel({ urlStartingWith: "https://your-endpoint.com" }); +await client.cancel({ all: true }); +``` + +**Pitfalls**: +- `ids` accepts both string and array, but mixing with other filters is not supported. +- `urlStartingWith` cancels *all* descendant paths—use carefully. + +--- + +### Retrieving Logs (`client.logs`) +Fetch workflow execution logs with filtering and cursor-based pagination. + +```ts +const { runs, cursor } = await client.logs({ + workflowRunId: "wfr_123", // filter a specific run + workflowUrl: "https://endpoint", // fetch logs for a workflow + state: "RUN_FAILED", // filter by execution state + count: 50, // limit return size + workflowCreatedAt: 1700000000, // Unix timestamp + cursor: undefined, // pagination +}); +``` + +**Common mistakes**: +- Passing invalid state strings causes silent empty results. +- `workflowUrl` must match exactly; prefix filtering is not supported here. + +--- + +### Notifying Events (`client.notify`) +Notify workflows paused at `context.waitForEvent`. + +```ts +await client.notify({ + eventId: "order-paid", + eventData: { orderId: 1 }, // delivered to waiting workflow +}); +``` + +**Pitfall**: If no workflow is waiting, no error is thrown—ensure event IDs match exactly. + +--- + +### Fetching Waiting Workflows (`client.getWaiters`) +Retrieve active waiters waiting for a specific event. + +```ts +const waiters = await client.getWaiters({ eventId: "order-paid" }); +``` + +This is useful for debugging or coordinating external signals. + +--- + +## Dead Letter Queue (DLQ) Operations + +### Listing DLQ Messages (`client.dlq.list`) + +```ts +const { messages, cursor } = await client.dlq.list({ + cursor, + count: 20, + filter: { + fromDate: Date.now() - 86400000, + toDate: Date.now(), + url: "https://your-endpoint.com", + responseStatus: 500, + }, +}); +``` + +**Pitfalls**: +- Date fields use Unix **ms**, not seconds. +- `url` must match exactly. + +--- + +### Retry Failure Callback (`client.dlq.retryFailureFunction`) +If a workflow's `failureFunction`/`failureUrl` call failed: + +```ts +await client.dlq.retryFailureFunction({ dlqId: "dlq-123" }); +``` + +--- + +### Restarting DLQ Runs (`client.dlq.restart`) +Restart from the **beginning** of the workflow. + +```ts +await client.dlq.restart({ + dlqId: ["dlq-1", "dlq-2"], + retries: 5, + flowControl: { + key: "restart-group", + parallelism: 10, + }, +}); +``` + +--- + +### Resuming DLQ Runs (`client.dlq.resume`) +Continue from the **failed step**. + +```ts +await client.dlq.resume({ + dlqId: "dlq-123", + retries: 3, + flowControl: { + key: "resume-group", + rate: 5, + }, +}); +``` + +**Common confusion**: +- `restart` = new run from step 0. +- `resume` = same run continues at the failed step. + +--- + +## General Tips for TS Consumers +- Always validate URLs before triggering or canceling workflows. +- When dealing with arrays (`ids`, `dlqId`), ensure consistent type usage. +- Use flow control when bulk‑triggering or restarting large batches to avoid rate limits. +- Cursor values must be reused exactly as returned—do not modify. + +This skill equips AI agents with reliable patterns for orchestrating workflow runs, handling failures, and managing operational workflows end-to-end. \ No newline at end of file diff --git a/skills/basics/context.md b/skills/basics/context.md new file mode 100644 index 00000000..feafec85 --- /dev/null +++ b/skills/basics/context.md @@ -0,0 +1,223 @@ +# Workflow Context Basics + +This skill documents all workflow context functions available inside a TypeScript workflow. It focuses on how these APIs behave, how their parameters interact, and common pitfalls when composing them into reliable long‑running workflows. + +--- + +## context.api + +`context.api` provides typed integrations for OpenAI, Anthropic, and Resend. These calls behave like `context.call` but enforce the correct operation names and body types. + +### Key points +- Each provider exposes a `.call(stepName, config)` function. +- `token` is always required. +- The `operation` field selects the API method. +- `body` must match the provider's type definitions. +- Returns `{ status, body }`. + +### Example (combined) +```ts +// OpenAI + Anthropic + Resend usage +const openai = await context.api.openai.call("openai chat", { + token: process.env.OPENAI_API_KEY!, + operation: "chat.completions.create", + body: { + model: "gpt-4o", + messages: [ + { role: "system", content: "Hello" }, + { role: "user", content: "Hi" }, + ], + }, +}); + +const anthropic = await context.api.anthropic.call("anthropic message", { + token: process.env.ANTHROPIC_API_KEY!, + operation: "messages.create", + body: { + model: "claude-3-5-sonnet-20241022", + max_tokens: 1024, + messages: [{ role: "user", content: "Hello" }], + }, +}); + +// Resend email +const email = await context.api.resend.call("send email", { + token: process.env.RESEND_API_KEY!, + headers: { "content-type": "application/json" }, + body: { + from: "Acme ", + to: ["someone@example.com"], + subject: "Welcome", + html: "

Hi!

", + }, +}); +``` + +--- + +## context.call + +Performs an HTTP request with long execution windows (up to 12 hours). Always returns a response, even for non‑2xx status codes. + +### Important parameters +- `url`: required. +- `method`: defaults to `GET`. +- `headers`: object. +- `body`: body as string. +- `retries`: number of retry attempts. +- `retryDelay`: retry delay expression as string or number (in milliseconds). +- `flowControl`: throttling config. +- `timeout`: seconds for each attempt. +- `workflow`: used only when invoking a workflow inside `serveMany`. + +### Example (showing most fields) +```ts +const response = await context.call("sync user", { + url: "https://api.example.com/user", + method: "POST", + headers: { authorization: `Bearer ${process.env.TOKEN}` }, + body: JSON.stringify({ id: userId, name }), + retries: 3, + retryDelay: "pow(2, retried)", // exponential + timeout: 20, + flowControl: { + key: "user-sync", + rate: 5, + parallelism: 2, + }, +}); +``` + +### Pitfalls +- Localhost requests fail unless tunneled. + +--- + +## context.cancel + +Cancels the current workflow run. +- Workflow is marked **canceled**, not failed. +- `failureFunction` does not run. +- No DLQ entry. + +### Example +```ts +const result = await context.run("check something", () => ...); +if (result.cancel) await context.cancel(); +``` + +--- + +## context.createWebhook + +Creates a reusable webhook URL for external systems. + +### Example +```ts +const { webhookUrl, eventId } = await context.createWebhook("create webhook"); +// Use webhookUrl with an external service +``` + +--- + +## context.invoke + +Invokes another workflow (must be served under the same `serveMany`) and waits for its completion. + +### Key parameters +- `workflow`: the workflow definition. +- `body`: becomes `requestPayload` of invoked workflow. +- `headers`: forwarded headers. +- `workflowRunId`: optional override. +- `retries`, `retryDelay`, `flowControl`: control invocation retry strategy. + +### Example +```ts +const { body, isFailed, isCanceled } = await context.invoke("invoke child", { + workflow: childWorkflow, + body: { id: 123 }, + headers: { foo: "bar" }, + retries: 2, +}); +``` + +--- + +## context.notify + +Notifies waiting workflows created via `context.waitForEvent`. + +### Example +```ts +const { notifyResponse } = await context.notify( + "notify step", + "order-updated", + { status: "shipped" }, +); +``` + +`notifyResponse` lists all workflows resumed by this event. + +--- + +## context.run + +Runs custom synchronous or async logic as a step. Values must be JSON‑serializable. + +### Example +```ts +const user = await context.run("load user", () => getUser()); + +await context.run("process user", () => process(user)); + +// Running in parallel +const a = context.run("step A", () => foo()); +const b = context.run("step B", () => bar()); +await Promise.all([a, b]); +``` + +### Pitfalls +- Returned class instances lose methods due to serialization. Rehydrate via `Object.assign(new Class(), data)`. + +--- + +## context.sleep + +Pauses workflow execution for a duration. String or numeric seconds. + +### Example +```ts +await context.sleep("wait 1 day", "1d"); +``` + +--- + +## context.sleepUntil + +Pauses execution until a specific `Date`, timestamp, or parseable string. + +### Example +```ts +const ts = new Date(Date.now() + 1000 * 60 * 60); +await context.sleepUntil("wait until", ts); +``` + +--- + +## context.waitForEvent + +Waits for an event (triggered by `context.notify`) or until timeout. + +### Example +```ts +const { eventData, timeout } = await context.waitForEvent( + "wait for update", + "order-updated", + { timeout: "2h" } +); +``` + +--- + +## Summary +This skill centralizes all workflow context primitives. They enable long‑running, state‑persistent, event‑driven workflows with external API access, typed provider integrations, and fine‑grained control over retry strategies, sleeps, and cross‑workflow invocation. \ No newline at end of file diff --git a/skills/basics/serve.md b/skills/basics/serve.md new file mode 100644 index 00000000..c29a553a --- /dev/null +++ b/skills/basics/serve.md @@ -0,0 +1,66 @@ +This documentation explains how to use `serve()` to expose workflows as HTTP endpoints. + +## Core Concept +`serve()` turns an async route function into a workflow endpoint. When a request arrives, a workflow run is created, and the route function defines the execution logic using `context.run()`. + +### Full Example With Inline Comments +```typescript +import { serve } from "@upstash/workflow/nextjs"; +import { z } from "zod"; +import { Client, Receiver } from "@upstash/qstash"; +import { loggingMiddleware } from "@upstash/workflow"; + +type InitialPayload = { foo: string; bar: number }; + +const payloadSchema = z.object({ foo: z.string(), bar: z.number() }); + +export const { POST } = serve( + async (context) => { + const payload = context.requestPayload; + await context.run("step-1", async () => `hello ${payload.foo}`); + }, + { + // Runs when the workflow fails after all retries + failureFunction: async ({ context, failStatus, failResponse, failHeaders, failStack }) => console.error(failResponse), + + // Hooks for step lifecycle, run lifecycle, and debug events + middlewares: [loggingMiddleware], + + // Manually parse the initial request payload before workflow execution + initialPayloadParser: (raw) => JSON.parse(raw), + + // Automatically validate + parse the request payload using Zod + schema: payloadSchema, + + // Override the full URL used to schedule subsequent workflow steps + url: "https://myapp.com/api/workflow", + + // Override only the base portion of the inferred URL (keeps the route path) + baseUrl: "https://proxy-url.com", + + // Provide a custom QStash client + qstashClient: new Client({ token: process.env.QSTASH_TOKEN! }), + + // Provide a custom QStash Receiver for verifying that requests come from QStash + receiver: new Receiver({ + currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!, + nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!, + }), + + // Override environment variables available inside workflow execution + env: { + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: process.env.QSTASH_TOKEN!, + }, + + // Output detailed execution logs to stdout + verbose: true, + + // Disable anonymous telemetry for this workflow endpoint + disableTelemetry: true, + } +); +``` + +## Common Pitfalls +- **Incorrect URL inference** behind proxies or local tunnels—set `url` or `baseUrl`. \ No newline at end of file diff --git a/skills/features/flow-control.md b/skills/features/flow-control.md new file mode 100644 index 00000000..2d960c2d --- /dev/null +++ b/skills/features/flow-control.md @@ -0,0 +1,135 @@ +# Flow Control + +Flow Control defines how many workflow steps may run or start within a given time, ensuring predictable load and preventing overload of internal or third‑party systems. It applies per **flow control key**, allowing steps across different workflow runs to share the same limits. + +This skill documents: +- parallelism +- rate and period +- monitoring +- common behaviors and pitfalls + + +## Core Concepts + +### Flow Control Key +A string identifier grouping related workflow steps under shared limits. +All steps triggered with the same key share the same concurrency and rate budget. + +```ts +const client = new Client({ token: "" }) + +await client.trigger({ + url: "https://", + flowControl: { + key: "user-onboarding", // Groups steps under this key + parallelism: 5, // max concurrent steps + rate: 20, // max starts per period + period: "1m", // time window + }, +}) +``` + + +## Parallelism +Controls **how many steps may run concurrently**. +Works like a token bucket: each running step consumes one token; when it finishes, the token is released. + +- If tokens are available → step starts immediately. +- If no tokens → step enters the waitlist. +- Token handoff is **not FIFO**; later steps may start earlier. + +### Example Behavior +- `parallelism = 3` → at most 3 steps run at the same time. +- Step 4 waits until one of the first 3 completes. + +Parallelism affects **execution concurrency**, not rate‑of‑start per time window. + + +## Rate & Period +Controls **how many steps may START within a defined time window**. + +```ts +flowControl: { + key: "billing-jobs", + rate: 2, // allow 2 starts + period: "10s", // every 10 seconds + parallelism: 1, // can be combined +} +``` + +Behavior: +- First `rate` steps in each period start immediately. +- Additional steps wait until the next window. +- Execution duration **does not** count toward the period; only start time matters. + +This is rate limiting, not concurrency control. + + +## Combining Rate + Parallelism +Both limits apply **independently**. +A step must satisfy *both* to run. + +Example: +- `rate = 3 per minute` +- `parallelism = 7` + +Even if concurrency slots are free, steps may be delayed by the rate window. +Even if rate window is open, steps may be delayed by concurrency. + +This enables predictable load patterns. + + +## Monitoring +Flow control queues and waitlists are visible from: +- Console → **FlowControl** tab (waitlist size per key) +- REST API: + - List all keys + - Get details for one key + +Useful for diagnosing bottlenecks, long queues, or misconfigured limits. + + +## Important Behaviors & Pitfalls + +### Limits are stored on each step +If you deploy new limits, previously published steps keep their original configuration. + +### No per‑step config (mostly) +All steps inherit the workflow‑run’s flowControl block. +Exceptions: +- `context.call` – may define a separate key to isolate API throttling. +- `context.invoke` – spawns a new workflow run with its own flowControl. + +To isolate heavy or sensitive steps, move them into a separate workflow and invoke them. + +### Waitlist order is not guaranteed +A later step may receive a token before an earlier one. +Do not rely on ordering. + +### Queueing is automatic +Exceeding limits never rejects steps. Everything is eventually executed. + + +## Full Example +Demonstrates defining all parameters together and how they interact. + +```ts +import { Client } from "@upstash/workflow" + +const client = new Client({ token: "" }) + +await client.trigger({ + url: "https:///process", + flowControl: { + key: "media-processing", // Shared grouping key + parallelism: 4, // Max 4 at the same time + rate: 10, // Max 10 starts + period: 60_000, // in 60 seconds (ms or string supported) + }, +}) +``` + +This configuration: +- ensures no more than 4 processing tasks run concurrently +- ensures no more than 10 tasks start per minute +- queues everything automatically beyond those limits diff --git a/skills/features/invoke.md b/skills/features/invoke.md new file mode 100644 index 00000000..18c285bd --- /dev/null +++ b/skills/features/invoke.md @@ -0,0 +1,93 @@ +# Workflow Invocation + +This documentation covers how to invoke one workflow from another using `context.invoke`, along with how to define and expose multiple workflows using `createWorkflow` and `serveMany`. + +--- + +## Invoking Another Workflow + +A workflow can start another workflow run and wait for it to finish before continuing. This enables orchestration and dependency chains across workflows. + +Use `context.invoke` to trigger another workflow: + +```ts +const result = await context.invoke("analyze content", { + workflow: analyzeContent, // workflow object or function + body: "input payload", // request body passed to invoked workflow + headers: { "x-user": "123" }, // optional headers + retries: 5, // optional, default 3 + flowControl: { /* ... */ }, + workflowRunId: "optional-custom-id" +}); + +const { + body, // response returned by invoked workflow + isFailed, // true if invoked workflow failed + isCanceled // true if invoked workflow was canceled +} = result; +``` + +### Returning a Value +A workflow may return any serializable value, which becomes `body` for the caller. + +### Depth Limitation +Workflow invocation depth is limited to 100. Any recursive or cyclic chain exceeding this will fail. + +--- + +## Defining Workflows Without Exposing Endpoints + +Use `createWorkflow` to define workflows as objects. These are not automatically exposed as HTTP endpoints. + +```ts +const analyzeContent = createWorkflow(async (context) => { + await context.sleep("delay", 1); + return { message: "processed" }; +}); + +const processUser = createWorkflow(async (context) => { + const { body } = await context.invoke("invoke analyze", { + workflow: analyzeContent, // type‑safe invocation + body: "user-1" + }); + + return { result: body }; +}); +``` + +--- + +## Exposing Multiple Workflows With `serveMany` + +To allow workflows to invoke each other without requiring explicit URLs, define all related workflows under a shared parent path. + +```ts +// app/workflows/[...any]/route.ts +export const { POST } = serveMany({ + "analyze": analyzeContent, + "process": processUser +}); +``` + +- Keys become route names. +- All workflows that need to invoke each other must appear in the same `serveMany` call. +- Requests are routed based on the key names. + +### Triggering a Workflow + +```ts +import { Client } from "@upstash/workflow"; + +const client = new Client({ token: "" }); + +await client.trigger({ + url: "https://your-app/workflows/analyze" +}); +``` + +--- + +## Common Pitfalls + +- **Missing workflow in `serveMany`**: If a workflow invokes another that was not included in the same `serveMany` definition, the call will fail. +- **Infinite invocation loops**: Depth limit of 100 prevents runaway recursive chains. diff --git a/skills/features/retries-failures-reliability.md b/skills/features/retries-failures-reliability.md new file mode 100644 index 00000000..535f2a58 --- /dev/null +++ b/skills/features/retries-failures-reliability.md @@ -0,0 +1,217 @@ +# Workflow Reliability Features + +This documentation provides a consolidated reference for reliability mechanisms in Upstash Workflow, focusing on retries, failure handling, and the Dead Letter Queue (DLQ). It is optimized for AI‑driven execution but remains clear for developers. + +--- + +## Core Reliability Concepts + +Upstash Workflow ensures that workflow runs are durable, recoverable, and traceable. Reliability is achieved through: + +- Automatic retries for transient failures +- Failure handling via `failureFunction` or `failureUrl` +- DLQ for storing permanently failed runs +- Manual recovery actions: restart, resume, rerun failure function +- Mechanisms to prevent retries when failure is intentional + +These mechanisms work together to guarantee that no failed execution is lost and that failures can be observed and handled. + +--- + +## Automatic Retries + +Workflow steps automatically retry when they fail, using configurable retry count and delay. + +### Retry Parameters +- `retries`: How many times the step will retry +- `retryDelay`: A math expression (string) returning delay in milliseconds; can use `retried` variable + +### Common Pitfalls +- The delay expression must be a *string*; raw numbers disable dynamic calculation. +- A workflow only moves to DLQ after **all** retries fail. + +### Example: Configure Retry Count + Delay +```ts +import { Client, WorkflowNonRetryableError, serve } from "@upstash/workflow"; + +// Trigger with custom retry settings +const client = new Client({ token: process.env.WORKFLOW_TOKEN! }); +await client.trigger({ + url: "https://example.com/workflow", + retries: 5, + retryDelay: "(1 + retried) * 2000", // 2s, 4s, 6s... +}); + +// Workflow with steps that may intentionally skip retries +export const { POST } = serve(async (context) => { + const shouldFail = await context.run("check", () => true); + + if (shouldFail) { + // Skips retry cycle and sends directly to failure handling + throw new WorkflowNonRetryableError("Bad state"); + } + + await context.run("process", () => doWork()); +}); +``` + +--- + +## Preventing Retries + +### When to Stop Retries +Use these when failure is expected or should halt execution immediately: + +- `WorkflowNonRetryableError` → Fail fast, trigger failure function, enter DLQ +- `context.cancel()` → Stop workflow intentionally, no DLQ, no failure function +- Conditional returns → Graceful early exit without error + +### Pitfalls +- `context.cancel()` marks the run as **canceled**, not failed — no failure handling executes. +- Throwing any error *other than* `WorkflowNonRetryableError` will still trigger retries. + +--- + +## Failure Handling + +You can handle workflow failures using either: +- `failureFunction` (runs in your workflow environment) +- `failureUrl` (external callback endpoint) + +Only **one** can be configured — they are mutually exclusive. + +### Failure Function Behavior +- Runs after all retries fail +- Receives failure metadata and workflow context +- Cannot create new workflow steps; `context` is read‑only +- Is retried if it fails +- Can be manually retried from the DLQ + +### Failure Function Parameters +- `context.workflowRunId` +- `context.url` +- `context.requestPayload` +- `context.headers` +- `context.env` +- `failStatus` +- `failResponse` +- `failHeaders` + +### Example: Workflow + Failure Function + Prevent‑Retry Logic +```ts +export const { POST } = serve( + async (context) => { + const ok = await context.run("verify", () => verifySomething()); + + if (!ok) { + throw new WorkflowNonRetryableError("Verification failed"); + } + + const shouldAbort = await context.run("check", () => false); + if (shouldAbort) { + await context.cancel(); + return; + } + + await context.run("execute", () => doWork()); + }, + { + failureFunction: async ({ context, failStatus, failResponse }) => { + // Logging / cleanup / alerts + await logFailure(context.workflowRunId, failStatus, failResponse); + }, + } +); +``` + +--- + +## Dead Letter Queue (DLQ) + +A workflow enters the DLQ when: +- A step fails *after all retries* +- A `WorkflowNonRetryableError` is thrown +- `failureFunction` also fails (visible as failed in DLQ) + +### Retention +- Free: 3 days +- Pay‑as‑you‑go: 1 week +- Fixed: Up to 3 months + +After retention expires, items cannot be recovered. + +--- + +## DLQ Recovery Actions + +You may perform these actions from the dashboard or programmatically. + +### 1. Resume +Continue from the failed step while keeping previous step results. + +### 2. Restart +Start the entire workflow from scratch, discarding previous results. + +### 3. Rerun Failure Function +Retry the failure handler **only if it previously failed**. + +### Example: All DLQ Actions +```ts +import { Client } from "@upstash/workflow"; +const client = new Client({ token: process.env.WORKFLOW_TOKEN! }); + +await client.dlq.resume({ dlqId: "dlq-1", retries: 3 }); // Continue run +await client.dlq.restart({ dlqId: "dlq-2", retries: 3 }); // Restart +await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure handler +``` + +### Pitfalls +- `retryFailureFunction` is only allowed when the failure function failed. +- `resume` cannot be used if workflow code changed before the failure point. +- `restart` re-executes everything, including expensive steps. + +--- + +## Failure URL (Advanced Option) + +Use when failure must be handled by a separate service. + +### Notes +- Use **either** `failureFunction` **or** `failureUrl`. +- Recommended only when your primary app might be offline. +- Failure URL receives a structured JSON payload about the failed request. + +### Example +```ts +export const { POST } = serve(async (context) => { + await context.run("work", () => doWork()); +}, { + failureUrl: "https://example.com/failure-handler", +}); +``` + +--- + +## Debugging Failed Runs + +DLQ entries store: +- Request + response headers +- Payloads +- Failure metadata +- Failure function state + +Use the dashboard filters (workflow run ID, URL, failure function state) to identify and debug root causes. + +--- + +## Summary + +This skill unifies all Workflow reliability controls: + +- Retries (automatic, configurable) +- Failure handling (`failureFunction`, `failureUrl`) +- DLQ retention + recovery +- Fast‑fail mechanisms (`WorkflowNonRetryableError`, `cancel`, conditional exits) +- Manual recovery operations (resume, restart, retry failure function) + +Together these features ensure workflows are resilient, debuggable, and recoverable under all failure scenarios. \ No newline at end of file diff --git a/skills/features/wait-for-event.md b/skills/features/wait-for-event.md new file mode 100644 index 00000000..596f1504 --- /dev/null +++ b/skills/features/wait-for-event.md @@ -0,0 +1,94 @@ +# Wait for Event + +Use this feature to pause workflow execution until an external event arrives. It allows building asynchronous, event‑driven workflows without holding compute resources. + +--- + +## How It Works + +`context.waitForEvent()` suspends the workflow and registers a waiter under an event ID. When a matching `notify` call is sent, the workflow resumes with the provided event data. + +• Each waiter has a timeout. If the event does not arrive in time, the workflow continues with `{ timeout: true }`. +• Maximum timeout depends on your pricing tier. +• Multiple workflow runs may wait on the same event ID; a notify call resumes all of them. + +--- + +## Key Concepts & Pitfalls + +### Event IDs +Use unique event IDs to avoid heavy fan‑out notifications. + +Example patterns: +• `order-123-processing-complete` +• `user-42-email-verified` + +### Race Conditions +A notify call sent *before* the workflow begins waiting is lost. + +To avoid this: +• Always inspect the notify response. +• Retry if no waiters were found. + +--- + +## Combined Example +Below is a single TypeScript example showing waiting for an event, handling timeouts, and safely notifying: + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { Client } from "@upstash/workflow"; + +// Workflow that waits for an event +export const { POST } = serve(async (context) => { + const { orderId } = context.requestPayload; + const eventId = `order-${orderId}-processed`; + + // Wait for the event with timeout + const { eventData, timeout } = await context.waitForEvent( + "wait-for-order-processing", // step name + eventId, // event id + { + timeout: "1d", // optional timeout + } + ); + + if (timeout) { + // handle timeout here + await context.run("timeout-handler", async () => { + console.log("Order processing timed out"); + }); + return; + } + + // Continue workflow using eventData + await context.run("process-completed-order", async () => { + console.log("Order processed:", eventData); + }); +}); + +// External notifier (safe retry pattern) +const client = new Client({ token: "" }); + +async function notifyProcessingComplete(orderId: string, payload: any) { + const eventId = `order-${orderId}-processed`; + + // First attempt - returns array of NotifyResponse + const waiters = await client.notify({ eventId, eventData: payload }); + + if (waiters > 0) return first; + + // Retry if no workflows were waiting + await new Promise((r) => setTimeout(r, 3000)); + return await client.notify({ eventId, eventData: payload }); +} +``` + +--- + +## Best Practices + +• Generate event IDs that uniquely identify a specific workflow instance. +• Always handle the timeout case explicitly. +• On notification, check the returned array length to detect if any workflows were waiting. +• Prefer one event ID per workflow run to avoid notifying large groups. diff --git a/skills/features/webhooks.md b/skills/features/webhooks.md new file mode 100644 index 00000000..a9e39f99 --- /dev/null +++ b/skills/features/webhooks.md @@ -0,0 +1,79 @@ +## Webhooks + +This skill teaches agents how to use webhooks inside Upstash Workflow to pause execution, wait for external callbacks, and resume reliably. It covers creation, waiting, validation, and multi-step usage. + +--- + +### Key Concepts + +• **Webhook URLs are unique per context.createWebhook() call**. +• **Workflow execution pauses** when calling `context.waitForWebhook()` until the webhook receives a request or times out. +• **Lookback protection** ensures the workflow still receives early webhook calls. +• **Workflows remain dormant** during the waiting period, avoiding compute cost. +• **Webhooks are ideal for third‑party APIs** that support callback URLs. + +--- + +### Core Workflow: Create → Call External API → Wait → Resume + +Below is a consolidated example showing: +• Creating a webhook +• Passing it to an external API +• Handling the callback +• Waiting for multiple updates +• Error/timeout handling + +```ts +import { serve } from "@upstash/workflow/nextjs"; + +export const { POST } = serve(async (context) => { + // Create a unique webhook for this workflow + const webhook = await context.createWebhook("create webhook"); + + // Trigger an external service, sending it the callback URL + await context.call("trigger-external", { + url: "https://api.example.com/process", + method: "POST", + body: { webhookUrl: webhook.webhookUrl }, + }); + + // Loop: wait for progress updates until service signals completion + let updates = []; + while (true) { + const res = await context.waitForWebhook(`wait update`, webhook, "5m"); + + if (res.timeout) break; // no progress received + + const req = res.request; + const payload = await req.json(); + updates.push(payload); + + if (req.headers.get("x-task-finished") === "true") break; + } + + return updates; +}); +``` + +--- + +### Webhook Reliability: Lookback Protection + +**Common pitfall:** external systems may call the webhook *before* the workflow reaches `waitForWebhook()`. + +Upstash automatically stores early webhook calls. The next `waitForWebhook()` will immediately resolve with the stored request. + +This avoids race conditions and eliminates the need for manual buffering. + +--- + +### When to Use Webhooks vs. Wait for Event + +Use **webhooks** when: +• integrating with third‑party APIs +• the external system invokes a callback URL +• you need race‑condition‑safe behavior + +Use **Wait for Event** when: +• you control the notifying system and can use the Workflow Client +• you don't need lookback semantics diff --git a/skills/how-to/local-dev.md b/skills/how-to/local-dev.md new file mode 100644 index 00000000..01b1e8bc --- /dev/null +++ b/skills/how-to/local-dev.md @@ -0,0 +1,122 @@ +# Local Development for Upstash Workflow + +This documentation explains how to run Upstash Workflow locally using the QStash development server and how to expose your local app using a public tunnel such as ngrok. It focuses on practical usage and TypeScript integration. + +## Overview + +Upstash Workflow uses Upstash QStash under the hood. During development you can: + +- Run a **local QStash development server** to simulate workflow behavior entirely offline. +- Optionally expose your local server using **ngrok** if you want to test with the production QStash. + +Both approaches require updating environment variables and ensuring your workflow trigger URLs correctly reference your local or tunnel address. + +--- + +## 1. Start the QStash Local Development Server + +Use the QStash CLI: + +``` +npx @upstash/qstash-cli dev +``` + +The CLI prints: +- QSTASH_TOKEN +- QSTASH_CURRENT_SIGNING_KEY +- QSTASH_NEXT_SIGNING_KEY +- Local server URL (default: `http://127.0.0.1:8080`) + +Set these values in your `.env` file so your workflow client uses the local environment. + +--- + +## 2. Set Local Environment Variables + +Use the environment values printed by the CLI: + +``` +QSTASH_URL="http://127.0.0.1:8080" +QSTASH_TOKEN="" +QSTASH_CURRENT_SIGNING_KEY="" +QSTASH_NEXT_SIGNING_KEY="" +``` + +These ensure all workflow requests are routed locally. + +--- + +## 3. Trigger Workflows Using Local URLs + +A common pattern is determining the base URL dynamically based on environment variables. Below is an example that demonstrates: +- Local development (localhost) +- Production deployments (auto-detected env) + +```ts +import { Client } from "@upstash/workflow"; + +const client = Client(); + +// In production, Vercel sets VERCEL_URL. Otherwise use localhost. +const BASE_URL = process.env.VERCEL_URL + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:3000`; + +// Trigger a workflow with retries +const { workflowRunId } = await client.trigger({ + url: `${BASE_URL}/api/workflow`, // Local or production + retries: 3, // Optional retry logic +}); + +console.log("Workflow run:", workflowRunId); +``` + +**Common mistakes:** +- Forgetting to include the full URL including `http://` or `https://`. +- Using a production URL while the local QStash server is running. +- Missing environment variables. + +--- + +## 4. Using ngrok (Optional) + +If your workflow must be reachable from the managed Upstash servers (not local), expose your local server publicly. + +### Install & authenticate +``` +ngrok config add-authtoken +``` + +### Start a tunnel +``` +ngrok http 3000 +``` + +ngrok outputs a public URL: +``` +Forwarding https://1234abcd.ngrok.io -> http://localhost:3000 +``` + +Use this public URL instead of localhost: + +```ts +const BASE_URL = "https://1234abcd.ngrok.io"; // Public tunnel + +await client.trigger({ + url: `${BASE_URL}/api/workflow`, + retries: 3, +}); +``` + +**Pitfall:** Your ngrok port must match your dev server port, otherwise all workflow calls will fail. + +--- + +## Summary + +- Start QStash locally using `qstash-cli dev`. +- Copy generated environment variables into `.env`. +- Use local URLs in TypeScript clients while developing. +- Optionally expose your server with ngrok if you need remote access. + +This setup ensures fast workflow iteration without deploying your app. \ No newline at end of file diff --git a/skills/how-to/middleware.md b/skills/how-to/middleware.md new file mode 100644 index 00000000..65772db5 --- /dev/null +++ b/skills/how-to/middleware.md @@ -0,0 +1,155 @@ +# Middlewares + +This guide explains how to intercept workflow lifecycle and debug events using middleware. It covers how to attach middlewares, define event handlers, initialize resources, and combine functionality into efficient implementations. + +## Overview +Middlewares plug into the workflow engine and listen to: +- Lifecycle events such as run start, run completion, and step execution +- Debug messages such as errors, warnings, and informational logs + +They are useful for logging, monitoring, instrumentation, and integrating with external systems. + +## Using Built‑in Middleware +Add middleware inside the `middlewares` array when serving a workflow: + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { loggingMiddleware } from "@upstash/workflow"; + +export const { POST } = serve( + async (ctx) => { + await ctx.run("step-1", () => "Hello World"); + }, + { + middlewares: [loggingMiddleware], // emits run/step logs + } +); +``` + +## Creating Custom Middleware +You define middleware by instantiating `WorkflowMiddleware` with callbacks for the events you want to handle. + +### Direct Callback Definitions +All lifecycle and debug events can be defined in one object. + +```ts +import { WorkflowMiddleware } from "@upstash/workflow"; + +const customMiddleware = new WorkflowMiddleware({ + name: "custom-logger", + callbacks: { + // Lifecycle events + runStarted: async ({ context }) => { + console.log(`Run ${context.workflowRunId} started`); + }, + beforeExecution: async ({ context, stepName }) => { + console.log(`→ Executing step: ${stepName}`); + }, + afterExecution: async ({ context, stepName, result }) => { + console.log(`✓ Step ${stepName} completed`, result); + }, + runCompleted: async ({ context, result }) => { + console.log(`Run ${context.workflowRunId} finished`, result); + }, + + // Debug events + onError: async ({ workflowRunId, error }) => { + console.error(`Error in ${workflowRunId}`, error); + }, + onWarning: async ({ workflowRunId, warning }) => { + console.warn(`Warning from ${workflowRunId}`, warning); + }, + onInfo: async ({ workflowRunId, info }) => { + console.info(`Info from ${workflowRunId}`, info); + }, + }, +}); +``` + +### Initialization With `init` +Use `init` when you need setup logic (connecting to a DB, creating clients, loading configuration). `init` returns an object containing the callbacks. + +```ts +import { WorkflowMiddleware } from "@upstash/workflow"; + +const databaseMiddleware = new WorkflowMiddleware({ + name: "database-logger", + init: async () => { + const db = await connectToDatabase(); + + return { + runStarted: async ({ context }) => { + await db.insert({ workflowRunId: context.workflowRunId, status: "started" }); + }, + runCompleted: async ({ context, result }) => { + await db.update({ workflowRunId: context.workflowRunId, status: "completed", result }); + }, + onError: async ({ workflowRunId, error }) => { + await db.insert({ workflowRunId, level: "error", message: error.message }); + }, + }; + }, +}); +``` + +## Event Reference +Below is a concise list of supported event handlers and what they receive. + +### Lifecycle +- **runStarted**: `{ context }` +- **beforeExecution**: `{ context, stepName }` +- **afterExecution**: `{ context, stepName, result }` +- **runCompleted**: `{ context, result }` + +### Debug +- **onError**: `{ workflowRunId?, error }` +- **onWarning**: `{ workflowRunId?, warning }` +- **onInfo**: `{ workflowRunId?, info }` + +## Combined Example +Single snippet showing typical usage patterns. + +```ts +import { serve } from "@upstash/workflow/nextjs"; +import { WorkflowMiddleware } from "@upstash/workflow"; + +// Error tracking +const errorTrackingMiddleware = new WorkflowMiddleware({ + name: "errors", + callbacks: { + onError: async ({ workflowRunId, error }) => { + await fetch("https://monitoring/errors", { + method: "POST", + body: JSON.stringify({ workflowRunId, error: error.message }), + }); + }, + }, +}); + +// Performance measurement +const timings = new Map(); +const performanceMiddleware = new WorkflowMiddleware({ + name: "perf", + callbacks: { + beforeExecution: ({ stepName }) => timings.set(stepName, Date.now()), + afterExecution: ({ stepName }) => { + const t = timings.get(stepName); + if (t) console.log(`Step ${stepName} took ${Date.now() - t}ms`); + timings.delete(stepName); + }, + }, +}); + +export const { POST } = serve( + async (ctx) => { + await ctx.run("task", () => "done"); + }, + { + middlewares: [errorTrackingMiddleware, performanceMiddleware], + } +); +``` + +## Pitfalls and Tips +- Ensure `init` returns all callbacks; missing keys simply mean the event is ignored. +- Debug events may be called without a `workflowRunId` if errors occur before run id is identified. diff --git a/skills/how-to/migrations.md b/skills/how-to/migrations.md new file mode 100644 index 00000000..0cd5f834 --- /dev/null +++ b/skills/how-to/migrations.md @@ -0,0 +1,164 @@ +# Migration Guide for Upstash Workflow (TypeScript) + +This document provides a clear, task-focused overview of migrations between Workflow versions. It highlights breaking changes, how to update your code, and common pitfalls to avoid. Code samples combine multiple parameters in each example to reduce redundancy. + +## January 2026 – Major Release 1.0.0 + +### Agents API → moved to a separate package +The Agents API was removed from the core workflow package and placed into `@upstash/workflow-agents` so the base SDK no longer depends on large AI-related libraries. + +**Key changes**: +- Must install and import `agentWorkflow` from the new package. +- Agents access now happens through an `agents` helper created per workflow invocation. + +```ts +// Before +import { serve } from "@upstash/workflow/nextjs"; +export const { POST } = serve(async (context) => { + const model = context.agents.openai("gpt-3.5-turbo"); + const agent = context.agents.agent({ /* ... */ }); + const task = context.agents.task({ /* ... */ }); +}); + +// After +import { serve } from "@upstash/workflow/nextjs"; +import { agentWorkflow } from "@upstash/workflow-agents"; +export const { POST } = serve(async (context) => { + const agents = agentWorkflow(context); + const model = agents.openai("gpt-3.5-turbo"); + const agent = agents.agent({ /* ... */ }); + const task = agents.task({ /* ... */ }); +}); +``` + +--- +### `keepTriggerConfig` and `useFailureFunction` removed +These fields were redundant—both behaviors are now always enabled. + +```ts +// Before +await client.trigger({ + url: "...", + retries: 3, + keepTriggerConfig: true, + useFailureFunction: true +}); + +// After +await client.trigger({ url: "...", retries: 3 }); +``` + +--- +### Configuration moved from `serve()` → `client.trigger()` +Options such as `retries`, `flowControl`, `retryDelay`, and `failureUrl` no longer belong in `serve()`. + +```ts +// Before +export const { POST } = serve(async (context) => { /* ... */ }, { + retries: 3, + retryDelay: "1000 * (1 + retried)", + flowControl: { key: "my-key", rate: 10 } +}); +await client.trigger({ url: "..." }); + +// After +export const { POST } = serve(async (context) => { /* ... */ }); +await client.trigger({ + url: "...", + retries: 3, + retryDelay: "1000 * (1 + retried)", + flowControl: { key: "my-key", rate: 10 } +}); +``` + +--- +### `stringifyBody` removed from `context.call` and `context.invoke` +Bodies must now be strings. + +```ts +// Before +await context.call("step", { + url: "https://api.example.com", + method: "POST", + body: { key: "value" }, + stringifyBody: true +}); + +// After +await context.call("step", { + url: "https://api.example.com", + method: "POST", + body: JSON.stringify({ key: "value" }) +}); + +// Same change applies to invoke +await context.invoke("other", { + workflow: otherWorkflow, + body: JSON.stringify({ key: "value" }) +}); +``` + +--- +### Logger removed → middleware system added +Replace `verbose` logging with explicit middlewares. + +```ts +import { loggingMiddleware, WorkflowMiddleware } from "@upstash/workflow"; + +const stepFinish = new WorkflowMiddleware({ + name: "step-finish", + callbacks: { + afterExecution: async ({ stepName, result }) => + console.log(`Step ${stepName} finished`, result) + } +}); + +export const { POST } = serve(async (context) => { /* ... */ }, { + middlewares: [loggingMiddleware, stepFinish] +}); +``` + +--- +## October 2024 – Migration from `@upstash/qstash` + +### Install the new package +Replace all workflow usage from `@upstash/qstash` with `@upstash/workflow`. + +### Serve import and return shape changed +Most environments now require destructuring `{ POST }`. + +```ts +// Before +import { serve } from "@upstash/qstash/nextjs"; +export const POST = serve(...); + +// After +import { serve } from "@upstash/workflow/nextjs"; +export const { POST } = serve(...); +``` + +### `context.call` updated +Now returns `{ status, headers, body }` and **does not fail** the workflow if the HTTP call fails. + +```ts +const { status, headers, body } = await context.call("call-step", { + url: "https://example.com/api", + method: "POST" +}); +``` + +**Pitfall**: Old runs may not contain `status` or `headers` until fully migrated. + +### Errors renamed +- `QStashWorkflowError` → `WorkflowError` +- `QStashWorkflowAbort` → `WorkflowAbort` + +--- +## Summary of Common Mistakes +- Using `context.agents` instead of the new `agentWorkflow()` helper. +- Leaving old serve-config options (`retries`, `verbose`, etc.). +- Forgetting to `JSON.stringify` bodies. +- Assuming `context.call` failures stop workflow execution. +- Mix-and-matching `@upstash/qstash` and `@upstash/workflow` imports. + +This guide ensures smooth migration with minimal friction while highlighting where behavioral changes may break existing workflows. \ No newline at end of file diff --git a/skills/how-to/realtime.md b/skills/how-to/realtime.md new file mode 100644 index 00000000..9b6a8322 --- /dev/null +++ b/skills/how-to/realtime.md @@ -0,0 +1,173 @@ +# Realtime Workflow Integration (TypeScript) + +This Skill provides guidance for building real‑time and human‑in‑the‑loop workflows using Upstash Workflow + Upstash Realtime. It covers event schemas, workflow patterns, emitting/receiving updates, and handling interactive pauses. + +Key capabilities: +- Emit strongly typed workflow events. +- Subscribe to live updates in the frontend. +- Implement pause/resume workflows using `waitForEvent` and `notify`. + +--- + +## Core Concepts + +### Realtime schema design +Define all workflow event types in one schema. Keep event definitions minimal and stable. + +```ts +// lib/realtime.ts +const schema = { + workflow: { + // Completion events + runFinish: z.object({}), + stepFinish: z.object({ stepName: z.string(), result: z.unknown().optional() }), + + // Human‑in‑the‑loop events + waitingForInput: z.object({ eventId: z.string(), message: z.string() }), + inputResolved: z.object({ eventId: z.string() }), + } +}; + +export const realtime = new Realtime({ schema, redis }); +export type RealtimeEvents = InferRealtimeEvents; +``` + +Common mistakes: +- Forgetting to wrap events under a namespace (e.g. `workflow.*`). + +--- + +## Emitting Events Inside Workflows +Emit inside `context.run()` whenever possible to avoid duplicate emissions on retries. + +```ts +// app/api/workflow/route.ts +export const { POST } = serve(async (context) => { + const channel = realtime.channel(context.workflowRunId); + + // Emit step updates + await context.run("validate", async () => { + const result = { ok: true }; + await channel.emit("workflow.stepFinish", { stepName: "validate", result }); + return result; + }); + + // Final event + await context.run("finish", () => channel.emit("workflow.runFinish", {})); +}); +``` + +Pitfall: +- Emitting events *outside* of `context.run()` risks duplicate delivery if retries happen. + +--- + +## Human‑in‑the‑Loop Workflows +Use `waitForEvent()` with unique `eventId` values and emit a `waitingForInput` event immediately. + +```ts +// Step that pauses for human input +const eventId = `approval-${context.workflowRunId}`; + +const [{ eventData, timeout }] = await Promise.all([ + context.waitForEvent("wait-for-approval", eventId, { timeout: "5m" }), + context.run("notify-wait", () => + channel.emit("workflow.waitingForInput", { + eventId, + message: "Waiting for approval..." + }) + ) +]); + +if (timeout) return { success: false, reason: "timeout" }; + +await context.run("resolved", () => + channel.emit("workflow.inputResolved", { eventId }) +); +``` + +Important details: +- Always handle timeout: otherwise the workflow may appear stuck. +- Emit `inputResolved` so the UI can clear any pending state. +- Use deterministic `eventId` prefixes per action/approval. + +--- + +## Notify Endpoint (User Response) +The frontend sends input back to the workflow using `client.notify()`. + +```ts +// app/api/notify/route.ts +export async function POST(req: NextRequest) { + const { eventId, eventData } = await req.json(); + + await workflowClient.notify({ eventId, eventData }); // resumes workflow + return NextResponse.json({ success: true }); +} +``` + +Common mistakes: +- Sending the wrong `eventId` → workflow never resumes. +- Sending `eventData` that doesn’t match the awaited type. + +--- + +## Realtime Subscription in the Frontend +Use a single hook to consume all workflow‑related events and manage state. + +```ts +// lib/realtime-client.ts & useWorkflow hooks +useRealtime({ + enabled: !!workflowRunId, + channels: [workflowRunId], + events: [ + "workflow.stepFinish", + "workflow.runFinish", + "workflow.waitingForInput", + "workflow.inputResolved" + ], + onData({ event, data }) { + switch (event) { + case "workflow.stepFinish": + setSteps(prev => [...prev, data]); break; + case "workflow.waitingForInput": + setWaitingState(data); break; + case "workflow.inputResolved": + setWaitingState(prev => prev?.eventId === data.eventId ? null : prev); + break; + case "workflow.runFinish": + setIsRunFinished(true); + } + } +}); +``` + +Pitfalls: +- Forgetting to clear waiting state on `inputResolved`. +- Forgetting to reset state on re‑trigger. + +--- + +## Triggering the Workflow +One simple endpoint triggers both basic and human‑in‑the‑loop flows. + +```ts +// app/api/trigger/route.ts +export async function POST(req: NextRequest) { + const workflowUrl = `${req.nextUrl.origin}/api/workflow`; // or /human-in-loop + const { workflowRunId } = await workflowClient.trigger({ + url: workflowUrl, + body: { userId: "123", action: "process" } + }); + return NextResponse.json({ workflowRunId }); +} +``` + +--- + +## Recommended Patterns +- Use one channel per workflow run (`workflowRunId`). +- Always place event emissions inside `context.run()`. +- Use stable `eventId` formats for all interactive steps. +- Subscribe to only the events your UI needs. +- Reset UI state before every new trigger. diff --git a/skills/rest-api.md b/skills/rest-api.md new file mode 100644 index 00000000..28ed5fab --- /dev/null +++ b/skills/rest-api.md @@ -0,0 +1,156 @@ +# Workflow REST API Skill + +This Skill provides TypeScript-focused guidance for interacting with the Upstash QStash Workflow Dead Letter Queue (DLQ) REST API. It includes practical usage examples, consolidated patterns, and key considerations for safely restarting, resuming, listing, deleting, and inspecting failed workflow runs. + +The goal is to help agents construct correct TypeScript API requests, handle optional fields, and avoid common pitfalls when working with workflow recovery operations. + +--- + +## Core Concepts + +• **DLQ Entries** represent failed workflow runs containing metadata such as payload, headers, timestamps, failure callback state, and workflow configuration. +• **Restart** recreates the entire workflow run from the beginning (fresh execution, no preserved state). +• **Resume** creates a new run continuing from the failed step (preserves successful steps). +• **Bulk operations** process up to 50 DLQ entries per request and may return a cursor for pagination. +• **Flow-Control and Retry Overrides** are passed via headers and apply only to the newly created workflow run. + +Be careful: changing workflow code **before** the failed step may cause resume operations to break. + +--- + +# HTTP Endpoints Overview + +Below is a complete summary of the API endpoints supported by this skill. + +• `GET /v2/workflows/dlq` — List DLQ entries with filtering. +• `GET /v2/workflows/dlq/{dlqId}` — Retrieve a single DLQ entry. +• `DELETE /v2/workflows/dlq/{dlqId}` — Delete a DLQ entry. +• `POST /v2/workflows/dlq/restart/{dlqId}` — Restart a workflow from scratch. +• `POST /v2/workflows/dlq/resume/{dlqId}` — Resume a workflow from point of failure. +• `POST /v2/workflows/dlq/restart` — Bulk restart (up to 50). +• `POST /v2/workflows/dlq/resume` — Bulk resume (up to 50). +• `POST /v2/workflows/dlq/callback/{dlqId}` — Rerun a failed failure-callback. +• `GET /v2/flowControl` and `/v2/flowControl/{key}` — Flow-control inspection. + +--- + +# TypeScript Usage Patterns + +Below is a single integrated example illustrating several endpoints, optional query parameters, header overrides, and response field handling. + +```ts +import fetch from "node-fetch"; + +const token = process.env.QSTASH_TOKEN; +const base = "https://qstash.upstash.io/v2"; + +async function example() { + // 1. List DLQ items with filters + const list = await fetch(`${base}/workflows/dlq?workflowUrl=https://my.app/workflow&fromDate=1700000000000`, { + headers: { Authorization: `Bearer ${token}` } + }).then(r => r.json()); + + const firstDlqId = list.messages?.[0]?.dlqId; + + // 2. Restart or Resume a specific run (headers may override flow control or retries) + const restart = await fetch(`${base}/workflows/dlq/restart/${firstDlqId}`, { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Upstash-Flow-Control-Key": "my-key", // optional + "Upstash-Flow-Control-Value": "parallelism=1", // optional + "Upstash-Retries": "3" // optional + } + }).then(r => r.json()); + + // 3. Bulk resume using filters and cursor handling + const bulkResume = await fetch(`${base}/workflows/dlq/resume`, { + method: "POST", + headers: { Authorization: `Bearer ${token}` }, + body: JSON.stringify({ + fromDate: 1700000000000, + workflowUrl: "https://my.app/workflow", + dlqIds: [firstDlqId] // optional; filters also supported + }) + }).then(r => r.json()); + + // 4. Rerun failure callback + const callback = await fetch(`${base}/workflows/dlq/callback/${firstDlqId}`, { + method: "POST", + headers: { Authorization: `Bearer ${token}` } + }).then(r => r.json()); + + // 5. Delete a DLQ entry + await fetch(`${base}/workflows/dlq/${firstDlqId}`, { + method: "DELETE", + headers: { Authorization: `Bearer ${token}` } + }); + + // 6. Inspect Flow-Control + const fc = await fetch(`${base}/flowControl/my-key`, { + headers: { Authorization: `Bearer ${token}` } + }).then(r => r.json()); + + console.log({ list, restart, bulkResume, callback, fc }); +} +``` + +--- + +# Key Fields and Behaviors + +## Request Query Fields + +• **dlqIds: string[]** — Exact DLQ IDs for targeting specific failed runs (bulk restart/resume). +• **fromDate / toDate: number** — Millisecond timestamps; inclusive range filters. +• **workflowUrl: string** — Filters by workflow endpoint URL. +• **workflowRunId: string** — Can match full run ID or prefix. +• **workflowCreatedAt: number** — Timestamp filter for creation time. +• **responseStatus: number** — Filter by HTTP status of failed run. +• **callerIP: string** — Filter by origin IP. +• **failureCallbackState: string** — Filter runs whose failure callback succeeded or failed. +• **cursor: string** — Pagination cursor; returned by list & bulk ops. +• **count: number** — Limit for listing DLQ entries (default/max 100). + +## Header Overrides + +These apply only when creating a **new workflow run** (restart or resume). + +• **Upstash-Flow-Control-Key** — Override flow-control key. +• **Upstash-Flow-Control-Value** — Override flow-control config (e.g., "parallelism=1"). +• **Upstash-Retries** — Override step retry configuration. + +All are optional; original values are reused if omitted. + +## Response Fields + +• **cursor?: string** — If returned, additional entries exist. +• **workflowRuns: { workflowRunId: string; workflowCreatedAt: number }[]** — Returned by bulk restart/resume. +• **message objects** (DLQ list/get) include: + - **workflowRunId, workflowCreatedAt, workflowUrl** + - **responseStatus, responseHeader, responseBody** + - **failureCallbackInfo.state** — Identify failed callbacks. + - **dlqId** — Unique DLQ entry identifier. + +--- + +# Pitfalls & Best Practices + +• **Resume only works if workflow code before the failed step stays unchanged**. Any change may cause resume to fail. +• **Bulk operations process max 50 items**. Always check `cursor` to continue processing. +• **Deleting a DLQ entry is permanent**; once removed, it cannot be resumed or restarted. +• **Failure callback reruns are independent** and do not affect workflow execution. They only retry the failure-notification step. +• **Prefix matching on workflowRunId** can unintentionally match more runs than expected—use full ID for precision. + +--- + +# Recommended Patterns + +• Fetch → Check cursor → Continue looping for bulk operations. +• Always log `failureCallbackInfo.state` to determine whether a callback rerun is required. +• Apply header overrides sparingly; flow-control misconfiguration may stall large batches. +• Prefer filtering in bulk operations instead of manually passing large dlqIds arrays. + +--- + +This Skill equips agents to safely recover workflows, inspect DLQ state, and manage retry and flow-control semantics using TypeScript-based REST interactions. \ No newline at end of file diff --git a/skills/troubleshooting.md b/skills/troubleshooting.md new file mode 100644 index 00000000..d3634a7f --- /dev/null +++ b/skills/troubleshooting.md @@ -0,0 +1,192 @@ +# Workflow Troubleshooting (TypeScript) + +This guide provides clear, TypeScript‑focused patterns for diagnosing and fixing common issues in Upstash Workflow. It highlights correct usage of core workflow features and shows how to avoid pitfalls that commonly occur in real applications. + +--- + +## Steps Inside try/catch + +Workflow steps (`context.run`, `context.sleep`, `context.sleepUntil`, `context.call`) intentionally throw `WorkflowAbort` after completing. Catching this error prevents the workflow engine from progressing. + +**Correct pattern:** rethrow `WorkflowAbort` or avoid `try/catch` around steps. + +```ts +import { WorkflowAbort } from "@upstash/workflow"; + +try { + // Any step will throw WorkflowAbort + const result = await context.run("step", () => "ok"); + +} catch (err) { + if (err instanceof WorkflowAbort) throw err; // required + // handle real errors +} +``` + +If you *must* handle errors, move the `try/catch` into the function passed to `context.run`. + +--- + +## requestPayload Becoming Undefined + +During `context.call`, the workflow endpoint is invoked multiple times internally. Before the first step executes, `context.requestPayload` may appear `undefined`. + +**Fix:** capture the payload via a step: + +```ts +export const { POST } = serve(async (context) => { + const payload = await context.run("get-payload", () => context.requestPayload); + + // payload remains stable even during context.call + await context.call("my-call", { + url: "https://example.com", + body: payload, + }); +}); +``` + +**Header considerations:** When triggering workflows, ensure payload‑friendly headers like: +- `Content-Type: text/plain` +- `Content-Type: application/json` + +--- + +## Signature Verification Failures + +If QStash request verification fails: +- Ensure the workflow is triggered through QStash (`client.trigger` or publish) +- Verify `QSTASH_CURRENT_SIGNING_KEY` and `QSTASH_NEXT_SIGNING_KEY` +- Pass appropriate `Content-Type` headers when triggering + +Error looks like: +``` +Failed to verify that the Workflow request comes from QStash: ... +``` + +--- + +## Authorization Errors from Early Returns + +If your workflow returns **before running any step**, the SDK interprets this as failed authorization: +``` +HTTP 400 – Failed to authenticate Workflow request. +``` + +To safely perform non‑deterministic checks, wrap them in a step: + +```ts +export const { POST } = serve(async (context) => { + const shouldExit = await context.run("check-condition", () => Math.random() > 0.5); + if (shouldExit) return; + + // remaining workflow +}); +``` + +--- + +## Retry Configuration + +Retries come from two locations: +- Workflow start (`client.trigger`): default **3**, applies to workflow steps +- `context.call`: default **0** +- `context.invoke`: default **0** + +Use these in combination to tune behavior correctly. + +--- + +## Verbose Mode Diagnostics + +Enable verbose logging to debug rare edge cases: +```ts +serve(handler, { verbose: true }); +``` + +Useful warnings include: +- Localhost in workflow URL +- Duplicate step execution +- Network response anomalies +- Parallel `context.call` race warnings + +Each of these indicates environmental or routing issues rather than workflow logic problems. + +--- + +## HTTPS Protocol Issues (Proxy Environments) + +If deployed behind a proxy that downgrades HTTPS → HTTP (e.g., Railway), the SDK may infer the wrong protocol. + +**Fix:** explicitly set: +``` +UPSTASH_WORKFLOW_URL=https://your-deployment-url +``` + +--- + +## Workflow Stuck on First Step + +If the first step appears stuck: +- Check step logs in the dashboard +- Ensure the endpoint URL is correct and reachable +- Verify SDK versions +- If inference is failing, explicitly set `baseUrl`: + +```ts +serve(handler, { baseUrl: "https://your-url" }); +``` + +--- + +## Non‑workflow Destination Error + +Occurs when triggering a non‑workflow endpoint or using mismatched SDK versions. + +``` +detected a non-workflow destination for trigger/invoke +``` + +**Ensure:** +- The URL points to a `serve()` workflow endpoint +- Both caller and workflow endpoint use the latest SDK + +--- + +## Vercel Preview Deployment Protection + +Preview deployments block external requests unless bypassed. + +**Steps:** +1. Create a bypass secret in Vercel → Deployment Protection +2. Add it to the QStash client + pass it when triggering + +```ts +import { Client as QStash } from "@upstash/qstash"; +import { serve } from "@upstash/workflow/nextjs"; + +export const { POST } = serve(async (context) => { + // workflow logic +}, { + qstashClient: new QStash({ + token: process.env.QSTASH_TOKEN!, + headers: { + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET! + } + }) +}); + +// Triggering +await client.trigger({ + url: "https://preview.vercel.app/workflow", + headers: { + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET! + }, + body: "hello" +}); +``` + +Both trigger header and client header are required. + +--- + +This documentation consolidates the core troubleshooting patterns needed to debug and maintain healthy workflows in TypeScript applications. \ No newline at end of file From 0832ea3c10871d7716158247e1edec7eed7c8528 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 16:50:18 +0300 Subject: [PATCH 2/8] feat: add workflow-js skills --- skills/SKILL.md | 5 + skills/agents.md | 114 +++++++++++------- skills/basics/client.md | 53 +++++--- skills/basics/context.md | 35 ++++-- skills/basics/serve.md | 8 +- skills/features/flow-control.md | 50 +++++--- skills/features/invoke.md | 26 ++-- .../features/retries-failures-reliability.md | 46 +++++-- skills/features/wait-for-event.md | 9 +- skills/features/webhooks.md | 2 +- skills/how-to/local-dev.md | 10 +- skills/how-to/middleware.md | 11 ++ skills/how-to/migrations.md | 81 +++++++++---- skills/how-to/realtime.md | 41 ++++--- skills/rest-api.md | 46 +++---- skills/troubleshooting.md | 45 ++++--- 16 files changed, 383 insertions(+), 199 deletions(-) diff --git a/skills/SKILL.md b/skills/SKILL.md index e3e5ead3..d8df73e0 100644 --- a/skills/SKILL.md +++ b/skills/SKILL.md @@ -6,14 +6,17 @@ description: Lightweight guidance for using the Upstash Workflow SDK to define, # Upstash Workflow SDK ## Quick Start + The Upstash Workflow SDK lets you expose serverless workflow endpoints and run them reliably using QStash under the hood. Install: + ```bash npm install @upstash/workflow ``` Define a simple workflow endpoint: + ```ts import { serve } from "@upstash/workflow"; @@ -24,6 +27,7 @@ export const { POST } = serve(async (context) => { ``` Trigger it from your backend: + ```ts import { Client } from "@upstash/workflow"; @@ -32,6 +36,7 @@ await client.trigger({ url: "https://your-app.com/api/workflow" }); ``` ## Other Skill Files + These files contain the full documentation. Use them for details, patterns, and advanced behavior. - basics: diff --git a/skills/agents.md b/skills/agents.md index 64ba28a8..0fe89145 100644 --- a/skills/agents.md +++ b/skills/agents.md @@ -1,6 +1,8 @@ - --- +--- + name: workflow-agents description: Skill for building, configuring, and orchestrating Upstash Workflow Agents. Use when the user mentions agent workflows, multi-agent collaboration, tools, tasks, Upstash Workflow, or patterns like evaluator-optimizer, prompt chaining, parallelization, or orchestrator-workers. + --- # Workflow Agents @@ -8,7 +10,9 @@ description: Skill for building, configuring, and orchestrating Upstash Workflow This Skill provides guidance for defining and orchestrating agents, tools, models, and tasks inside **Upstash Workflow** using **TypeScript**. It explains core concepts, common patterns, workflow configuration, and pitfalls. ## Quick Overview + Use this Skill when: + - Building agents with tools and backgrounds - Creating multi-step or multi-agent tasks - Implementing patterns like evaluator-optimizer or prompt chaining @@ -16,6 +20,7 @@ Use this Skill when: - Running workflows reliably inside Upstash Workflow The Workflow Agents API centers around four elements: + - **Models** — LLM providers (OpenAI, Anthropic, or any AI SDK provider) - **Tools** — Functions agents can call - **Agents** — LLM instances with background, maxSteps, and tools @@ -24,19 +29,20 @@ The Workflow Agents API centers around four elements: --- # Defining Models + A model defines which provider the agent uses and includes optional configuration for retries, timeouts, or rate‑limits. ```ts -const agents = agentWorkflow(context) +const agents = agentWorkflow(context); // Basic OpenAI model -const model = agents.openai('gpt-3.5-turbo') +const model = agents.openai("gpt-3.5-turbo"); // OpenAI-compatible provider -const deepseek = agents.openai('deepseek-chat', { - baseURL: 'https://api.deepseek.com', +const deepseek = agents.openai("deepseek-chat", { + baseURL: "https://api.deepseek.com", apiKey: process.env.DEEPSEEK_API_KEY, -}) +}); // AI SDK provider (ex: Anthropic) const anthropic = agents.AISDKModel({ @@ -46,11 +52,12 @@ const anthropic = agents.AISDKModel({ agentCallParams: { timeout: 1000, retries: 0, - } -}) + }, +}); ``` **Key parameters:** + - **callSettings / agentCallParams** — timeout, retries, flow control - **provider / providerParams** — when using AI SDK providers @@ -59,45 +66,50 @@ const anthropic = agents.AISDKModel({ --- # Defining Tools + Tools extend what agents can do. Workflow supports: + - WorkflowTool (native) - AI SDK tools - LangChain tools - Agentic toolkits ```ts -import { z } from 'zod' -import { tool } from 'ai' -import { WorkflowTool } from '@upstash/workflow-agents' -import { WikipediaQueryRun } from '@langchain/community/tools/wikipedia_query_run' +import { z } from "zod"; +import { tool } from "ai"; +import { WorkflowTool } from "@upstash/workflow-agents"; +import { WikipediaQueryRun } from "@langchain/community/tools/wikipedia_query_run"; const mathTool = tool({ - description: 'Evaluate a math expression', + description: "Evaluate a math expression", parameters: z.object({ expression: z.string() }), execute: async ({ expression }) => mathjs.evaluate(expression), -}) +}); const workflowMath = new WorkflowTool({ - description: 'Evaluate math (workflow step aware)', + description: "Evaluate math (workflow step aware)", schema: z.object({ expression: z.string() }), invoke: async ({ expression }) => mathjs.evaluate(expression), executeAsStep: false, // allows context.call, etc. -}) +}); const wikiTool = new WikipediaQueryRun({ topKResults: 1, maxDocContentLength: 500, -}) +}); ``` **Common mistakes:** + - Workflow wraps execute/invoke in `context.run` by default, so you **cannot use context.call** unless `executeAsStep: false` is set. - LangChain tools must return **strings**, not objects. --- # Defining Agents + Agents wrap a model and add behavior via: + - `maxSteps` — how many LLM calls the agent is allowed to make - `background` — system prompt - `tools` — available actions @@ -105,46 +117,48 @@ Agents wrap a model and add behavior via: ```ts const generator = agents.agent({ model, - name: 'generator', + name: "generator", maxSteps: 1, - background: 'Generate text from prompts.', + background: "Generate text from prompts.", tools: {}, -}) +}); const evaluator = agents.agent({ model, - name: 'evaluator', + name: "evaluator", maxSteps: 1, - background: 'Evaluate responses and give corrections.', + background: "Evaluate responses and give corrections.", tools: {}, -}) +}); ``` **Tips:** + - Pick maxSteps carefully; too low prevents tool use; too high increases cost. - Names appear in Upstash Console logs; keep them descriptive. --- # Tasks (Single & Multi Agent) + A **task** is a single execution of an agent or a group of agents. ```ts // Single agent task const single = agents.task({ agent: generator, - prompt: 'Explain quantum mechanics.', -}) -const { text } = await single.run() + prompt: "Explain quantum mechanics.", +}); +const { text } = await single.run(); // Multi-agent with manager agent const multi = agents.task({ - model, // manager LLM + model, // manager LLM agents: [generator, evaluator], maxSteps: 3, - prompt: 'Generate text and refine it until quality improves.', -}) -const result = await multi.run() + prompt: "Generate text and refine it until quality improves.", +}); +const result = await multi.run(); ``` **Tip:** In multi-agent mode, the model becomes a "manager" system that decides which agent to call. @@ -152,25 +166,30 @@ const result = await multi.run() --- # Common Agent Patterns + Below are the patterns supported by the source files in this skill. ## Prompt Chaining + Sequential agent calls where each output becomes the next input. Useful for: stepwise research, multi‑stage content generation, breaking down complex tasks. Pitfall: watch `maxSteps` for agents that need both tool calls and summarization. ## Evaluator‑Optimizer + Loop until evaluator returns a PASS. Simple feedback‑refinement pattern. Pitfall: Always check evaluator output with `.includes("PASS")`, not strict equality. ## Parallelization + Use multiple agents with `Promise.all` and then aggregate. Pitfall: Avoid extremely large aggregated prompts; summarizing before combining is recommended. ## Orchestrator‑Workers + Manager delegates sub‑tasks to specialized workers. Useful for structured Q&A, multi‑topic analysis, or complex synthesis. @@ -179,7 +198,8 @@ Pitfall: The manager must have enough `maxSteps` to orchestrate multiple workers --- # Best Practices -- Give each agent a *clear* background; ambiguous roles cause incorrect tool use. + +- Give each agent a _clear_ background; ambiguous roles cause incorrect tool use. - Define tools with strict schemas so LLMs call them reliably. - Use multi-agent tasks when the problem requires specialization. - Inspect console logs to debug tool calls and agent decisions. @@ -188,49 +208,53 @@ Pitfall: The manager must have enough `maxSteps` to orchestrate multiple workers --- # Example: Combined Setup (models + tools + agents + tasks) + This shows all core fields together in one concise example. + ```ts export const { POST } = serve(async (context) => { - const agents = agentWorkflow(context) - const model = agents.openai('gpt-4o') + const agents = agentWorkflow(context); + const model = agents.openai("gpt-4o"); const mathTool = tool({ - description: 'Compute math', + description: "Compute math", parameters: z.object({ expression: z.string() }), execute: async ({ expression }) => mathjs.evaluate(expression), - }) + }); const researcher = agents.agent({ model, - name: 'researcher', + name: "researcher", maxSteps: 2, - background: 'Research topics using wiki.', + background: "Research topics using wiki.", tools: { wikiTool }, - }) + }); const mathematician = agents.agent({ model, - name: 'math', + name: "math", maxSteps: 2, - background: 'Solve numeric problems.', + background: "Solve numeric problems.", tools: { mathTool }, - }) + }); const task = agents.task({ model, // manager agents: [researcher, mathematician], maxSteps: 3, - prompt: 'Tell me about 3 stars and compute the sum of their masses.', - }) + prompt: "Tell me about 3 stars and compute the sum of their masses.", + }); - return (await task.run()).text -}) + return (await task.run()).text; +}); ``` --- # When to Use This Skill + Use this Skill whenever the task involves: + - Designing agent workflows - Multi-agent coordination or patterns - Using LangChain/AI SDK tools inside Workflow diff --git a/skills/basics/client.md b/skills/basics/client.md index d44b35b4..78d6422e 100644 --- a/skills/basics/client.md +++ b/skills/basics/client.md @@ -7,22 +7,24 @@ This skill provides guidance for using the Workflow Client to trigger, cancel, i ## Core Operations ### Triggering Workflow Runs (`client.trigger`) + Starts one or multiple workflow runs with optional payload, metadata, retry logic, and flow control. ```ts // Single and multiple triggers const single = await client.trigger({ - url: "https://your-endpoint/workflow", // required - body: { msg: "hello" }, // optional payload - headers: { "x-user": "123" }, // optional headers - workflowRunId: "custom-id", // custom unique identifier - retries: 3, // retry count for steps - retryDelay: "1000 * (1 + retried)", // expression-based retry delay - delay: "10s", // start later + url: "https://your-endpoint/workflow", // required + body: { msg: "hello" }, // optional payload + headers: { "x-user": "123" }, // optional headers + workflowRunId: "custom-id", // custom unique identifier + retries: 3, // retry count for steps + retryDelay: "1000 * (1 + retried)", // expression-based retry delay + delay: "10s", // start later notBefore: Math.floor(Date.now() / 1000) + 60, // absolute schedule - label: "my-run", // dashboard filtering - disableTelemetry: true, // opt‑out telemetry - flowControl: { // concurrency & rate limiting + label: "my-run", // dashboard filtering + disableTelemetry: true, // opt‑out telemetry + flowControl: { + // concurrency & rate limiting key: "user-key", rate: 10, parallelism: 5, @@ -37,6 +39,7 @@ const multiple = await client.trigger([ ``` **Pitfalls**: + - `workflowRunId` must be unique; collisions cause errors. - `delay` is ignored if `notBefore` is provided. - Using `url` incorrectly (missing protocol or dynamic segments) leads to invalid workflow endpoints. @@ -44,6 +47,7 @@ const multiple = await client.trigger([ --- ### Canceling Workflow Runs (`client.cancel`) + Cancel runs by ID, by URL prefix, or all active/pending runs. ```ts @@ -53,38 +57,42 @@ await client.cancel({ all: true }); ``` **Pitfalls**: + - `ids` accepts both string and array, but mixing with other filters is not supported. -- `urlStartingWith` cancels *all* descendant paths—use carefully. +- `urlStartingWith` cancels _all_ descendant paths—use carefully. --- ### Retrieving Logs (`client.logs`) + Fetch workflow execution logs with filtering and cursor-based pagination. ```ts const { runs, cursor } = await client.logs({ - workflowRunId: "wfr_123", // filter a specific run - workflowUrl: "https://endpoint", // fetch logs for a workflow - state: "RUN_FAILED", // filter by execution state - count: 50, // limit return size - workflowCreatedAt: 1700000000, // Unix timestamp - cursor: undefined, // pagination + workflowRunId: "wfr_123", // filter a specific run + workflowUrl: "https://endpoint", // fetch logs for a workflow + state: "RUN_FAILED", // filter by execution state + count: 50, // limit return size + workflowCreatedAt: 1700000000, // Unix timestamp + cursor: undefined, // pagination }); ``` **Common mistakes**: + - Passing invalid state strings causes silent empty results. - `workflowUrl` must match exactly; prefix filtering is not supported here. --- ### Notifying Events (`client.notify`) + Notify workflows paused at `context.waitForEvent`. ```ts await client.notify({ eventId: "order-paid", - eventData: { orderId: 1 }, // delivered to waiting workflow + eventData: { orderId: 1 }, // delivered to waiting workflow }); ``` @@ -93,6 +101,7 @@ await client.notify({ --- ### Fetching Waiting Workflows (`client.getWaiters`) + Retrieve active waiters waiting for a specific event. ```ts @@ -121,12 +130,14 @@ const { messages, cursor } = await client.dlq.list({ ``` **Pitfalls**: + - Date fields use Unix **ms**, not seconds. - `url` must match exactly. --- ### Retry Failure Callback (`client.dlq.retryFailureFunction`) + If a workflow's `failureFunction`/`failureUrl` call failed: ```ts @@ -136,6 +147,7 @@ await client.dlq.retryFailureFunction({ dlqId: "dlq-123" }); --- ### Restarting DLQ Runs (`client.dlq.restart`) + Restart from the **beginning** of the workflow. ```ts @@ -152,6 +164,7 @@ await client.dlq.restart({ --- ### Resuming DLQ Runs (`client.dlq.resume`) + Continue from the **failed step**. ```ts @@ -166,15 +179,17 @@ await client.dlq.resume({ ``` **Common confusion**: + - `restart` = new run from step 0. - `resume` = same run continues at the failed step. --- ## General Tips for TS Consumers + - Always validate URLs before triggering or canceling workflows. - When dealing with arrays (`ids`, `dlqId`), ensure consistent type usage. - Use flow control when bulk‑triggering or restarting large batches to avoid rate limits. - Cursor values must be reused exactly as returned—do not modify. -This skill equips AI agents with reliable patterns for orchestrating workflow runs, handling failures, and managing operational workflows end-to-end. \ No newline at end of file +This skill equips AI agents with reliable patterns for orchestrating workflow runs, handling failures, and managing operational workflows end-to-end. diff --git a/skills/basics/context.md b/skills/basics/context.md index feafec85..91f538f8 100644 --- a/skills/basics/context.md +++ b/skills/basics/context.md @@ -9,6 +9,7 @@ This skill documents all workflow context functions available inside a TypeScrip `context.api` provides typed integrations for OpenAI, Anthropic, and Resend. These calls behave like `context.call` but enforce the correct operation names and body types. ### Key points + - Each provider exposes a `.call(stepName, config)` function. - `token` is always required. - The `operation` field selects the API method. @@ -16,6 +17,7 @@ This skill documents all workflow context functions available inside a TypeScrip - Returns `{ status, body }`. ### Example (combined) + ```ts // OpenAI + Anthropic + Resend usage const openai = await context.api.openai.call("openai chat", { @@ -60,6 +62,7 @@ const email = await context.api.resend.call("send email", { Performs an HTTP request with long execution windows (up to 12 hours). Always returns a response, even for non‑2xx status codes. ### Important parameters + - `url`: required. - `method`: defaults to `GET`. - `headers`: object. @@ -71,6 +74,7 @@ Performs an HTTP request with long execution windows (up to 12 hours). Always re - `workflow`: used only when invoking a workflow inside `serveMany`. ### Example (showing most fields) + ```ts const response = await context.call("sync user", { url: "https://api.example.com/user", @@ -89,6 +93,7 @@ const response = await context.call("sync user", { ``` ### Pitfalls + - Localhost requests fail unless tunneled. --- @@ -96,11 +101,13 @@ const response = await context.call("sync user", { ## context.cancel Cancels the current workflow run. + - Workflow is marked **canceled**, not failed. - `failureFunction` does not run. - No DLQ entry. ### Example + ```ts const result = await context.run("check something", () => ...); if (result.cancel) await context.cancel(); @@ -113,6 +120,7 @@ if (result.cancel) await context.cancel(); Creates a reusable webhook URL for external systems. ### Example + ```ts const { webhookUrl, eventId } = await context.createWebhook("create webhook"); // Use webhookUrl with an external service @@ -125,6 +133,7 @@ const { webhookUrl, eventId } = await context.createWebhook("create webhook"); Invokes another workflow (must be served under the same `serveMany`) and waits for its completion. ### Key parameters + - `workflow`: the workflow definition. - `body`: becomes `requestPayload` of invoked workflow. - `headers`: forwarded headers. @@ -132,6 +141,7 @@ Invokes another workflow (must be served under the same `serveMany`) and waits f - `retries`, `retryDelay`, `flowControl`: control invocation retry strategy. ### Example + ```ts const { body, isFailed, isCanceled } = await context.invoke("invoke child", { workflow: childWorkflow, @@ -148,12 +158,11 @@ const { body, isFailed, isCanceled } = await context.invoke("invoke child", { Notifies waiting workflows created via `context.waitForEvent`. ### Example + ```ts -const { notifyResponse } = await context.notify( - "notify step", - "order-updated", - { status: "shipped" }, -); +const { notifyResponse } = await context.notify("notify step", "order-updated", { + status: "shipped", +}); ``` `notifyResponse` lists all workflows resumed by this event. @@ -165,6 +174,7 @@ const { notifyResponse } = await context.notify( Runs custom synchronous or async logic as a step. Values must be JSON‑serializable. ### Example + ```ts const user = await context.run("load user", () => getUser()); @@ -177,6 +187,7 @@ await Promise.all([a, b]); ``` ### Pitfalls + - Returned class instances lose methods due to serialization. Rehydrate via `Object.assign(new Class(), data)`. --- @@ -186,6 +197,7 @@ await Promise.all([a, b]); Pauses workflow execution for a duration. String or numeric seconds. ### Example + ```ts await context.sleep("wait 1 day", "1d"); ``` @@ -197,6 +209,7 @@ await context.sleep("wait 1 day", "1d"); Pauses execution until a specific `Date`, timestamp, or parseable string. ### Example + ```ts const ts = new Date(Date.now() + 1000 * 60 * 60); await context.sleepUntil("wait until", ts); @@ -209,15 +222,15 @@ await context.sleepUntil("wait until", ts); Waits for an event (triggered by `context.notify`) or until timeout. ### Example + ```ts -const { eventData, timeout } = await context.waitForEvent( - "wait for update", - "order-updated", - { timeout: "2h" } -); +const { eventData, timeout } = await context.waitForEvent("wait for update", "order-updated", { + timeout: "2h", +}); ``` --- ## Summary -This skill centralizes all workflow context primitives. They enable long‑running, state‑persistent, event‑driven workflows with external API access, typed provider integrations, and fine‑grained control over retry strategies, sleeps, and cross‑workflow invocation. \ No newline at end of file + +This skill centralizes all workflow context primitives. They enable long‑running, state‑persistent, event‑driven workflows with external API access, typed provider integrations, and fine‑grained control over retry strategies, sleeps, and cross‑workflow invocation. diff --git a/skills/basics/serve.md b/skills/basics/serve.md index c29a553a..0d5d1632 100644 --- a/skills/basics/serve.md +++ b/skills/basics/serve.md @@ -1,9 +1,11 @@ This documentation explains how to use `serve()` to expose workflows as HTTP endpoints. ## Core Concept + `serve()` turns an async route function into a workflow endpoint. When a request arrives, a workflow run is created, and the route function defines the execution logic using `context.run()`. ### Full Example With Inline Comments + ```typescript import { serve } from "@upstash/workflow/nextjs"; import { z } from "zod"; @@ -21,7 +23,8 @@ export const { POST } = serve( }, { // Runs when the workflow fails after all retries - failureFunction: async ({ context, failStatus, failResponse, failHeaders, failStack }) => console.error(failResponse), + failureFunction: async ({ context, failStatus, failResponse, failHeaders, failStack }) => + console.error(failResponse), // Hooks for step lifecycle, run lifecycle, and debug events middlewares: [loggingMiddleware], @@ -63,4 +66,5 @@ export const { POST } = serve( ``` ## Common Pitfalls -- **Incorrect URL inference** behind proxies or local tunnels—set `url` or `baseUrl`. \ No newline at end of file + +- **Incorrect URL inference** behind proxies or local tunnels—set `url` or `baseUrl`. diff --git a/skills/features/flow-control.md b/skills/features/flow-control.md index 2d960c2d..10bd1dbe 100644 --- a/skills/features/flow-control.md +++ b/skills/features/flow-control.md @@ -3,34 +3,35 @@ Flow Control defines how many workflow steps may run or start within a given time, ensuring predictable load and preventing overload of internal or third‑party systems. It applies per **flow control key**, allowing steps across different workflow runs to share the same limits. This skill documents: + - parallelism - rate and period - monitoring - common behaviors and pitfalls - ## Core Concepts ### Flow Control Key + A string identifier grouping related workflow steps under shared limits. All steps triggered with the same key share the same concurrency and rate budget. ```ts -const client = new Client({ token: "" }) +const client = new Client({ token: "" }); await client.trigger({ url: "https://", flowControl: { key: "user-onboarding", // Groups steps under this key - parallelism: 5, // max concurrent steps - rate: 20, // max starts per period - period: "1m", // time window + parallelism: 5, // max concurrent steps + rate: 20, // max starts per period + period: "1m", // time window }, -}) +}); ``` - ## Parallelism + Controls **how many steps may run concurrently**. Works like a token bucket: each running step consumes one token; when it finishes, the token is released. @@ -39,13 +40,14 @@ Works like a token bucket: each running step consumes one token; when it finishe - Token handoff is **not FIFO**; later steps may start earlier. ### Example Behavior + - `parallelism = 3` → at most 3 steps run at the same time. - Step 4 waits until one of the first 3 completes. Parallelism affects **execution concurrency**, not rate‑of‑start per time window. - ## Rate & Period + Controls **how many steps may START within a defined time window**. ```ts @@ -58,18 +60,20 @@ flowControl: { ``` Behavior: + - First `rate` steps in each period start immediately. - Additional steps wait until the next window. - Execution duration **does not** count toward the period; only start time matters. This is rate limiting, not concurrency control. - ## Combining Rate + Parallelism + Both limits apply **independently**. -A step must satisfy *both* to run. +A step must satisfy _both_ to run. Example: + - `rate = 3 per minute` - `parallelism = 7` @@ -78,9 +82,10 @@ Even if rate window is open, steps may be delayed by concurrency. This enables predictable load patterns. - ## Monitoring + Flow control queues and waitlists are visible from: + - Console → **FlowControl** tab (waitlist size per key) - REST API: - List all keys @@ -88,48 +93,53 @@ Flow control queues and waitlists are visible from: Useful for diagnosing bottlenecks, long queues, or misconfigured limits. - ## Important Behaviors & Pitfalls ### Limits are stored on each step + If you deploy new limits, previously published steps keep their original configuration. ### No per‑step config (mostly) + All steps inherit the workflow‑run’s flowControl block. Exceptions: + - `context.call` – may define a separate key to isolate API throttling. - `context.invoke` – spawns a new workflow run with its own flowControl. To isolate heavy or sensitive steps, move them into a separate workflow and invoke them. ### Waitlist order is not guaranteed + A later step may receive a token before an earlier one. Do not rely on ordering. ### Queueing is automatic -Exceeding limits never rejects steps. Everything is eventually executed. +Exceeding limits never rejects steps. Everything is eventually executed. ## Full Example + Demonstrates defining all parameters together and how they interact. ```ts -import { Client } from "@upstash/workflow" +import { Client } from "@upstash/workflow"; -const client = new Client({ token: "" }) +const client = new Client({ token: "" }); await client.trigger({ url: "https:///process", flowControl: { - key: "media-processing", // Shared grouping key - parallelism: 4, // Max 4 at the same time - rate: 10, // Max 10 starts - period: 60_000, // in 60 seconds (ms or string supported) + key: "media-processing", // Shared grouping key + parallelism: 4, // Max 4 at the same time + rate: 10, // Max 10 starts + period: 60_000, // in 60 seconds (ms or string supported) }, -}) +}); ``` This configuration: + - ensures no more than 4 processing tasks run concurrently - ensures no more than 10 tasks start per minute - queues everything automatically beyond those limits diff --git a/skills/features/invoke.md b/skills/features/invoke.md index 18c285bd..8e11a435 100644 --- a/skills/features/invoke.md +++ b/skills/features/invoke.md @@ -13,24 +13,28 @@ Use `context.invoke` to trigger another workflow: ```ts const result = await context.invoke("analyze content", { workflow: analyzeContent, // workflow object or function - body: "input payload", // request body passed to invoked workflow + body: "input payload", // request body passed to invoked workflow headers: { "x-user": "123" }, // optional headers - retries: 5, // optional, default 3 - flowControl: { /* ... */ }, - workflowRunId: "optional-custom-id" + retries: 5, // optional, default 3 + flowControl: { + /* ... */ + }, + workflowRunId: "optional-custom-id", }); const { - body, // response returned by invoked workflow - isFailed, // true if invoked workflow failed - isCanceled // true if invoked workflow was canceled + body, // response returned by invoked workflow + isFailed, // true if invoked workflow failed + isCanceled, // true if invoked workflow was canceled } = result; ``` ### Returning a Value + A workflow may return any serializable value, which becomes `body` for the caller. ### Depth Limitation + Workflow invocation depth is limited to 100. Any recursive or cyclic chain exceeding this will fail. --- @@ -48,7 +52,7 @@ const analyzeContent = createWorkflow(async (context) => { const processUser = createWorkflow(async (context) => { const { body } = await context.invoke("invoke analyze", { workflow: analyzeContent, // type‑safe invocation - body: "user-1" + body: "user-1", }); return { result: body }; @@ -64,8 +68,8 @@ To allow workflows to invoke each other without requiring explicit URLs, define ```ts // app/workflows/[...any]/route.ts export const { POST } = serveMany({ - "analyze": analyzeContent, - "process": processUser + analyze: analyzeContent, + process: processUser, }); ``` @@ -81,7 +85,7 @@ import { Client } from "@upstash/workflow"; const client = new Client({ token: "" }); await client.trigger({ - url: "https://your-app/workflows/analyze" + url: "https://your-app/workflows/analyze", }); ``` diff --git a/skills/features/retries-failures-reliability.md b/skills/features/retries-failures-reliability.md index 535f2a58..01adb594 100644 --- a/skills/features/retries-failures-reliability.md +++ b/skills/features/retries-failures-reliability.md @@ -23,14 +23,17 @@ These mechanisms work together to guarantee that no failed execution is lost and Workflow steps automatically retry when they fail, using configurable retry count and delay. ### Retry Parameters + - `retries`: How many times the step will retry - `retryDelay`: A math expression (string) returning delay in milliseconds; can use `retried` variable ### Common Pitfalls -- The delay expression must be a *string*; raw numbers disable dynamic calculation. + +- The delay expression must be a _string_; raw numbers disable dynamic calculation. - A workflow only moves to DLQ after **all** retries fail. ### Example: Configure Retry Count + Delay + ```ts import { Client, WorkflowNonRetryableError, serve } from "@upstash/workflow"; @@ -60,6 +63,7 @@ export const { POST } = serve(async (context) => { ## Preventing Retries ### When to Stop Retries + Use these when failure is expected or should halt execution immediately: - `WorkflowNonRetryableError` → Fail fast, trigger failure function, enter DLQ @@ -67,20 +71,23 @@ Use these when failure is expected or should halt execution immediately: - Conditional returns → Graceful early exit without error ### Pitfalls + - `context.cancel()` marks the run as **canceled**, not failed — no failure handling executes. -- Throwing any error *other than* `WorkflowNonRetryableError` will still trigger retries. +- Throwing any error _other than_ `WorkflowNonRetryableError` will still trigger retries. --- ## Failure Handling You can handle workflow failures using either: + - `failureFunction` (runs in your workflow environment) - `failureUrl` (external callback endpoint) Only **one** can be configured — they are mutually exclusive. ### Failure Function Behavior + - Runs after all retries fail - Receives failure metadata and workflow context - Cannot create new workflow steps; `context` is read‑only @@ -88,6 +95,7 @@ Only **one** can be configured — they are mutually exclusive. - Can be manually retried from the DLQ ### Failure Function Parameters + - `context.workflowRunId` - `context.url` - `context.requestPayload` @@ -98,6 +106,7 @@ Only **one** can be configured — they are mutually exclusive. - `failHeaders` ### Example: Workflow + Failure Function + Prevent‑Retry Logic + ```ts export const { POST } = serve( async (context) => { @@ -129,11 +138,13 @@ export const { POST } = serve( ## Dead Letter Queue (DLQ) A workflow enters the DLQ when: -- A step fails *after all retries* + +- A step fails _after all retries_ - A `WorkflowNonRetryableError` is thrown - `failureFunction` also fails (visible as failed in DLQ) ### Retention + - Free: 3 days - Pay‑as‑you‑go: 1 week - Fixed: Up to 3 months @@ -147,25 +158,30 @@ After retention expires, items cannot be recovered. You may perform these actions from the dashboard or programmatically. ### 1. Resume + Continue from the failed step while keeping previous step results. ### 2. Restart + Start the entire workflow from scratch, discarding previous results. ### 3. Rerun Failure Function + Retry the failure handler **only if it previously failed**. ### Example: All DLQ Actions + ```ts import { Client } from "@upstash/workflow"; const client = new Client({ token: process.env.WORKFLOW_TOKEN! }); -await client.dlq.resume({ dlqId: "dlq-1", retries: 3 }); // Continue run -await client.dlq.restart({ dlqId: "dlq-2", retries: 3 }); // Restart -await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure handler +await client.dlq.resume({ dlqId: "dlq-1", retries: 3 }); // Continue run +await client.dlq.restart({ dlqId: "dlq-2", retries: 3 }); // Restart +await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure handler ``` ### Pitfalls + - `retryFailureFunction` is only allowed when the failure function failed. - `resume` cannot be used if workflow code changed before the failure point. - `restart` re-executes everything, including expensive steps. @@ -177,17 +193,22 @@ await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure ha Use when failure must be handled by a separate service. ### Notes + - Use **either** `failureFunction` **or** `failureUrl`. - Recommended only when your primary app might be offline. - Failure URL receives a structured JSON payload about the failed request. ### Example + ```ts -export const { POST } = serve(async (context) => { - await context.run("work", () => doWork()); -}, { - failureUrl: "https://example.com/failure-handler", -}); +export const { POST } = serve( + async (context) => { + await context.run("work", () => doWork()); + }, + { + failureUrl: "https://example.com/failure-handler", + } +); ``` --- @@ -195,6 +216,7 @@ export const { POST } = serve(async (context) => { ## Debugging Failed Runs DLQ entries store: + - Request + response headers - Payloads - Failure metadata @@ -214,4 +236,4 @@ This skill unifies all Workflow reliability controls: - Fast‑fail mechanisms (`WorkflowNonRetryableError`, `cancel`, conditional exits) - Manual recovery operations (resume, restart, retry failure function) -Together these features ensure workflows are resilient, debuggable, and recoverable under all failure scenarios. \ No newline at end of file +Together these features ensure workflows are resilient, debuggable, and recoverable under all failure scenarios. diff --git a/skills/features/wait-for-event.md b/skills/features/wait-for-event.md index 596f1504..b5a7af64 100644 --- a/skills/features/wait-for-event.md +++ b/skills/features/wait-for-event.md @@ -17,6 +17,7 @@ Use this feature to pause workflow execution until an external event arrives. It ## Key Concepts & Pitfalls ### Event IDs + Use unique event IDs to avoid heavy fan‑out notifications. Example patterns: @@ -24,7 +25,8 @@ Example patterns: • `user-42-email-verified` ### Race Conditions -A notify call sent *before* the workflow begins waiting is lost. + +A notify call sent _before_ the workflow begins waiting is lost. To avoid this: • Always inspect the notify response. @@ -33,6 +35,7 @@ To avoid this: --- ## Combined Example + Below is a single TypeScript example showing waiting for an event, handling timeouts, and safely notifying: ```ts @@ -47,9 +50,9 @@ export const { POST } = serve(async (context) => { // Wait for the event with timeout const { eventData, timeout } = await context.waitForEvent( "wait-for-order-processing", // step name - eventId, // event id + eventId, // event id { - timeout: "1d", // optional timeout + timeout: "1d", // optional timeout } ); diff --git a/skills/features/webhooks.md b/skills/features/webhooks.md index a9e39f99..26c614e3 100644 --- a/skills/features/webhooks.md +++ b/skills/features/webhooks.md @@ -59,7 +59,7 @@ export const { POST } = serve(async (context) => { ### Webhook Reliability: Lookback Protection -**Common pitfall:** external systems may call the webhook *before* the workflow reaches `waitForWebhook()`. +**Common pitfall:** external systems may call the webhook _before_ the workflow reaches `waitForWebhook()`. Upstash automatically stores early webhook calls. The next `waitForWebhook()` will immediately resolve with the stored request. diff --git a/skills/how-to/local-dev.md b/skills/how-to/local-dev.md index 01b1e8bc..69c97a87 100644 --- a/skills/how-to/local-dev.md +++ b/skills/how-to/local-dev.md @@ -22,6 +22,7 @@ npx @upstash/qstash-cli dev ``` The CLI prints: + - QSTASH_TOKEN - QSTASH_CURRENT_SIGNING_KEY - QSTASH_NEXT_SIGNING_KEY @@ -49,6 +50,7 @@ These ensure all workflow requests are routed locally. ## 3. Trigger Workflows Using Local URLs A common pattern is determining the base URL dynamically based on environment variables. Below is an example that demonstrates: + - Local development (localhost) - Production deployments (auto-detected env) @@ -65,13 +67,14 @@ const BASE_URL = process.env.VERCEL_URL // Trigger a workflow with retries const { workflowRunId } = await client.trigger({ url: `${BASE_URL}/api/workflow`, // Local or production - retries: 3, // Optional retry logic + retries: 3, // Optional retry logic }); console.log("Workflow run:", workflowRunId); ``` **Common mistakes:** + - Forgetting to include the full URL including `http://` or `https://`. - Using a production URL while the local QStash server is running. - Missing environment variables. @@ -83,16 +86,19 @@ console.log("Workflow run:", workflowRunId); If your workflow must be reachable from the managed Upstash servers (not local), expose your local server publicly. ### Install & authenticate + ``` ngrok config add-authtoken ``` ### Start a tunnel + ``` ngrok http 3000 ``` ngrok outputs a public URL: + ``` Forwarding https://1234abcd.ngrok.io -> http://localhost:3000 ``` @@ -119,4 +125,4 @@ await client.trigger({ - Use local URLs in TypeScript clients while developing. - Optionally expose your server with ngrok if you need remote access. -This setup ensures fast workflow iteration without deploying your app. \ No newline at end of file +This setup ensures fast workflow iteration without deploying your app. diff --git a/skills/how-to/middleware.md b/skills/how-to/middleware.md index 65772db5..3be475ae 100644 --- a/skills/how-to/middleware.md +++ b/skills/how-to/middleware.md @@ -3,13 +3,16 @@ This guide explains how to intercept workflow lifecycle and debug events using middleware. It covers how to attach middlewares, define event handlers, initialize resources, and combine functionality into efficient implementations. ## Overview + Middlewares plug into the workflow engine and listen to: + - Lifecycle events such as run start, run completion, and step execution - Debug messages such as errors, warnings, and informational logs They are useful for logging, monitoring, instrumentation, and integrating with external systems. ## Using Built‑in Middleware + Add middleware inside the `middlewares` array when serving a workflow: ```ts @@ -27,9 +30,11 @@ export const { POST } = serve( ``` ## Creating Custom Middleware + You define middleware by instantiating `WorkflowMiddleware` with callbacks for the events you want to handle. ### Direct Callback Definitions + All lifecycle and debug events can be defined in one object. ```ts @@ -67,6 +72,7 @@ const customMiddleware = new WorkflowMiddleware({ ``` ### Initialization With `init` + Use `init` when you need setup logic (connecting to a DB, creating clients, loading configuration). `init` returns an object containing the callbacks. ```ts @@ -93,20 +99,24 @@ const databaseMiddleware = new WorkflowMiddleware({ ``` ## Event Reference + Below is a concise list of supported event handlers and what they receive. ### Lifecycle + - **runStarted**: `{ context }` - **beforeExecution**: `{ context, stepName }` - **afterExecution**: `{ context, stepName, result }` - **runCompleted**: `{ context, result }` ### Debug + - **onError**: `{ workflowRunId?, error }` - **onWarning**: `{ workflowRunId?, warning }` - **onInfo**: `{ workflowRunId?, info }` ## Combined Example + Single snippet showing typical usage patterns. ```ts @@ -151,5 +161,6 @@ export const { POST } = serve( ``` ## Pitfalls and Tips + - Ensure `init` returns all callbacks; missing keys simply mean the event is ignored. - Debug events may be called without a `workflowRunId` if errors occur before run id is identified. diff --git a/skills/how-to/migrations.md b/skills/how-to/migrations.md index 0cd5f834..cf7b7599 100644 --- a/skills/how-to/migrations.md +++ b/skills/how-to/migrations.md @@ -5,9 +5,11 @@ This document provides a clear, task-focused overview of migrations between Work ## January 2026 – Major Release 1.0.0 ### Agents API → moved to a separate package + The Agents API was removed from the core workflow package and placed into `@upstash/workflow-agents` so the base SDK no longer depends on large AI-related libraries. **Key changes**: + - Must install and import `agentWorkflow` from the new package. - Agents access now happens through an `agents` helper created per workflow invocation. @@ -16,8 +18,12 @@ The Agents API was removed from the core workflow package and placed into `@upst import { serve } from "@upstash/workflow/nextjs"; export const { POST } = serve(async (context) => { const model = context.agents.openai("gpt-3.5-turbo"); - const agent = context.agents.agent({ /* ... */ }); - const task = context.agents.task({ /* ... */ }); + const agent = context.agents.agent({ + /* ... */ + }); + const task = context.agents.task({ + /* ... */ + }); }); // After @@ -26,13 +32,19 @@ import { agentWorkflow } from "@upstash/workflow-agents"; export const { POST } = serve(async (context) => { const agents = agentWorkflow(context); const model = agents.openai("gpt-3.5-turbo"); - const agent = agents.agent({ /* ... */ }); - const task = agents.task({ /* ... */ }); + const agent = agents.agent({ + /* ... */ + }); + const task = agents.task({ + /* ... */ + }); }); ``` --- + ### `keepTriggerConfig` and `useFailureFunction` removed + These fields were redundant—both behaviors are now always enabled. ```ts @@ -41,7 +53,7 @@ await client.trigger({ url: "...", retries: 3, keepTriggerConfig: true, - useFailureFunction: true + useFailureFunction: true, }); // After @@ -49,30 +61,41 @@ await client.trigger({ url: "...", retries: 3 }); ``` --- + ### Configuration moved from `serve()` → `client.trigger()` + Options such as `retries`, `flowControl`, `retryDelay`, and `failureUrl` no longer belong in `serve()`. ```ts // Before -export const { POST } = serve(async (context) => { /* ... */ }, { - retries: 3, - retryDelay: "1000 * (1 + retried)", - flowControl: { key: "my-key", rate: 10 } -}); +export const { POST } = serve( + async (context) => { + /* ... */ + }, + { + retries: 3, + retryDelay: "1000 * (1 + retried)", + flowControl: { key: "my-key", rate: 10 }, + } +); await client.trigger({ url: "..." }); // After -export const { POST } = serve(async (context) => { /* ... */ }); +export const { POST } = serve(async (context) => { + /* ... */ +}); await client.trigger({ url: "...", retries: 3, retryDelay: "1000 * (1 + retried)", - flowControl: { key: "my-key", rate: 10 } + flowControl: { key: "my-key", rate: 10 }, }); ``` --- + ### `stringifyBody` removed from `context.call` and `context.invoke` + Bodies must now be strings. ```ts @@ -81,25 +104,27 @@ await context.call("step", { url: "https://api.example.com", method: "POST", body: { key: "value" }, - stringifyBody: true + stringifyBody: true, }); // After await context.call("step", { url: "https://api.example.com", method: "POST", - body: JSON.stringify({ key: "value" }) + body: JSON.stringify({ key: "value" }), }); // Same change applies to invoke await context.invoke("other", { workflow: otherWorkflow, - body: JSON.stringify({ key: "value" }) + body: JSON.stringify({ key: "value" }), }); ``` --- + ### Logger removed → middleware system added + Replace `verbose` logging with explicit middlewares. ```ts @@ -109,22 +134,30 @@ const stepFinish = new WorkflowMiddleware({ name: "step-finish", callbacks: { afterExecution: async ({ stepName, result }) => - console.log(`Step ${stepName} finished`, result) - } + console.log(`Step ${stepName} finished`, result), + }, }); -export const { POST } = serve(async (context) => { /* ... */ }, { - middlewares: [loggingMiddleware, stepFinish] -}); +export const { POST } = serve( + async (context) => { + /* ... */ + }, + { + middlewares: [loggingMiddleware, stepFinish], + } +); ``` --- + ## October 2024 – Migration from `@upstash/qstash` ### Install the new package + Replace all workflow usage from `@upstash/qstash` with `@upstash/workflow`. ### Serve import and return shape changed + Most environments now require destructuring `{ POST }`. ```ts @@ -138,27 +171,31 @@ export const { POST } = serve(...); ``` ### `context.call` updated + Now returns `{ status, headers, body }` and **does not fail** the workflow if the HTTP call fails. ```ts const { status, headers, body } = await context.call("call-step", { url: "https://example.com/api", - method: "POST" + method: "POST", }); ``` **Pitfall**: Old runs may not contain `status` or `headers` until fully migrated. ### Errors renamed + - `QStashWorkflowError` → `WorkflowError` - `QStashWorkflowAbort` → `WorkflowAbort` --- + ## Summary of Common Mistakes + - Using `context.agents` instead of the new `agentWorkflow()` helper. - Leaving old serve-config options (`retries`, `verbose`, etc.). - Forgetting to `JSON.stringify` bodies. - Assuming `context.call` failures stop workflow execution. - Mix-and-matching `@upstash/qstash` and `@upstash/workflow` imports. -This guide ensures smooth migration with minimal friction while highlighting where behavioral changes may break existing workflows. \ No newline at end of file +This guide ensures smooth migration with minimal friction while highlighting where behavioral changes may break existing workflows. diff --git a/skills/how-to/realtime.md b/skills/how-to/realtime.md index 9b6a8322..69f90787 100644 --- a/skills/how-to/realtime.md +++ b/skills/how-to/realtime.md @@ -3,6 +3,7 @@ This Skill provides guidance for building real‑time and human‑in‑the‑loop workflows using Upstash Workflow + Upstash Realtime. It covers event schemas, workflow patterns, emitting/receiving updates, and handling interactive pauses. Key capabilities: + - Emit strongly typed workflow events. - Subscribe to live updates in the frontend. - Implement pause/resume workflows using `waitForEvent` and `notify`. @@ -12,6 +13,7 @@ Key capabilities: ## Core Concepts ### Realtime schema design + Define all workflow event types in one schema. Keep event definitions minimal and stable. ```ts @@ -24,8 +26,8 @@ const schema = { // Human‑in‑the‑loop events waitingForInput: z.object({ eventId: z.string(), message: z.string() }), - inputResolved: z.object({ eventId: z.string() }), - } + inputResolved: z.object({ eventId: z.string() }), + }, }; export const realtime = new Realtime({ schema, redis }); @@ -33,11 +35,13 @@ export type RealtimeEvents = InferRealtimeEvents; ``` Common mistakes: + - Forgetting to wrap events under a namespace (e.g. `workflow.*`). --- ## Emitting Events Inside Workflows + Emit inside `context.run()` whenever possible to avoid duplicate emissions on retries. ```ts @@ -58,11 +62,13 @@ export const { POST } = serve(async (context) => { ``` Pitfall: -- Emitting events *outside* of `context.run()` risks duplicate delivery if retries happen. + +- Emitting events _outside_ of `context.run()` risks duplicate delivery if retries happen. --- ## Human‑in‑the‑Loop Workflows + Use `waitForEvent()` with unique `eventId` values and emit a `waitingForInput` event immediately. ```ts @@ -74,19 +80,18 @@ const [{ eventData, timeout }] = await Promise.all([ context.run("notify-wait", () => channel.emit("workflow.waitingForInput", { eventId, - message: "Waiting for approval..." + message: "Waiting for approval...", }) - ) + ), ]); if (timeout) return { success: false, reason: "timeout" }; -await context.run("resolved", () => - channel.emit("workflow.inputResolved", { eventId }) -); +await context.run("resolved", () => channel.emit("workflow.inputResolved", { eventId })); ``` Important details: + - Always handle timeout: otherwise the workflow may appear stuck. - Emit `inputResolved` so the UI can clear any pending state. - Use deterministic `eventId` prefixes per action/approval. @@ -94,6 +99,7 @@ Important details: --- ## Notify Endpoint (User Response) + The frontend sends input back to the workflow using `client.notify()`. ```ts @@ -107,12 +113,14 @@ export async function POST(req: NextRequest) { ``` Common mistakes: + - Sending the wrong `eventId` → workflow never resumes. - Sending `eventData` that doesn’t match the awaited type. --- ## Realtime Subscription in the Frontend + Use a single hook to consume all workflow‑related events and manage state. ```ts @@ -124,31 +132,35 @@ useRealtime({ "workflow.stepFinish", "workflow.runFinish", "workflow.waitingForInput", - "workflow.inputResolved" + "workflow.inputResolved", ], onData({ event, data }) { switch (event) { case "workflow.stepFinish": - setSteps(prev => [...prev, data]); break; + setSteps((prev) => [...prev, data]); + break; case "workflow.waitingForInput": - setWaitingState(data); break; + setWaitingState(data); + break; case "workflow.inputResolved": - setWaitingState(prev => prev?.eventId === data.eventId ? null : prev); + setWaitingState((prev) => (prev?.eventId === data.eventId ? null : prev)); break; case "workflow.runFinish": setIsRunFinished(true); } - } + }, }); ``` Pitfalls: + - Forgetting to clear waiting state on `inputResolved`. - Forgetting to reset state on re‑trigger. --- ## Triggering the Workflow + One simple endpoint triggers both basic and human‑in‑the‑loop flows. ```ts @@ -157,7 +169,7 @@ export async function POST(req: NextRequest) { const workflowUrl = `${req.nextUrl.origin}/api/workflow`; // or /human-in-loop const { workflowRunId } = await workflowClient.trigger({ url: workflowUrl, - body: { userId: "123", action: "process" } + body: { userId: "123", action: "process" }, }); return NextResponse.json({ workflowRunId }); } @@ -166,6 +178,7 @@ export async function POST(req: NextRequest) { --- ## Recommended Patterns + - Use one channel per workflow run (`workflowRunId`). - Always place event emissions inside `context.run()`. - Use stable `eventId` formats for all interactive steps. diff --git a/skills/rest-api.md b/skills/rest-api.md index 28ed5fab..aaf63683 100644 --- a/skills/rest-api.md +++ b/skills/rest-api.md @@ -46,9 +46,12 @@ const base = "https://qstash.upstash.io/v2"; async function example() { // 1. List DLQ items with filters - const list = await fetch(`${base}/workflows/dlq?workflowUrl=https://my.app/workflow&fromDate=1700000000000`, { - headers: { Authorization: `Bearer ${token}` } - }).then(r => r.json()); + const list = await fetch( + `${base}/workflows/dlq?workflowUrl=https://my.app/workflow&fromDate=1700000000000`, + { + headers: { Authorization: `Bearer ${token}` }, + } + ).then((r) => r.json()); const firstDlqId = list.messages?.[0]?.dlqId; @@ -57,11 +60,11 @@ async function example() { method: "POST", headers: { Authorization: `Bearer ${token}`, - "Upstash-Flow-Control-Key": "my-key", // optional - "Upstash-Flow-Control-Value": "parallelism=1", // optional - "Upstash-Retries": "3" // optional - } - }).then(r => r.json()); + "Upstash-Flow-Control-Key": "my-key", // optional + "Upstash-Flow-Control-Value": "parallelism=1", // optional + "Upstash-Retries": "3", // optional + }, + }).then((r) => r.json()); // 3. Bulk resume using filters and cursor handling const bulkResume = await fetch(`${base}/workflows/dlq/resume`, { @@ -70,26 +73,26 @@ async function example() { body: JSON.stringify({ fromDate: 1700000000000, workflowUrl: "https://my.app/workflow", - dlqIds: [firstDlqId] // optional; filters also supported - }) - }).then(r => r.json()); + dlqIds: [firstDlqId], // optional; filters also supported + }), + }).then((r) => r.json()); // 4. Rerun failure callback const callback = await fetch(`${base}/workflows/dlq/callback/${firstDlqId}`, { method: "POST", - headers: { Authorization: `Bearer ${token}` } - }).then(r => r.json()); + headers: { Authorization: `Bearer ${token}` }, + }).then((r) => r.json()); // 5. Delete a DLQ entry await fetch(`${base}/workflows/dlq/${firstDlqId}`, { method: "DELETE", - headers: { Authorization: `Bearer ${token}` } + headers: { Authorization: `Bearer ${token}` }, }); // 6. Inspect Flow-Control const fc = await fetch(`${base}/flowControl/my-key`, { - headers: { Authorization: `Bearer ${token}` } - }).then(r => r.json()); + headers: { Authorization: `Bearer ${token}` }, + }).then((r) => r.json()); console.log({ list, restart, bulkResume, callback, fc }); } @@ -127,10 +130,11 @@ All are optional; original values are reused if omitted. • **cursor?: string** — If returned, additional entries exist. • **workflowRuns: { workflowRunId: string; workflowCreatedAt: number }[]** — Returned by bulk restart/resume. • **message objects** (DLQ list/get) include: - - **workflowRunId, workflowCreatedAt, workflowUrl** - - **responseStatus, responseHeader, responseBody** - - **failureCallbackInfo.state** — Identify failed callbacks. - - **dlqId** — Unique DLQ entry identifier. + +- **workflowRunId, workflowCreatedAt, workflowUrl** +- **responseStatus, responseHeader, responseBody** +- **failureCallbackInfo.state** — Identify failed callbacks. +- **dlqId** — Unique DLQ entry identifier. --- @@ -153,4 +157,4 @@ All are optional; original values are reused if omitted. --- -This Skill equips agents to safely recover workflows, inspect DLQ state, and manage retry and flow-control semantics using TypeScript-based REST interactions. \ No newline at end of file +This Skill equips agents to safely recover workflows, inspect DLQ state, and manage retry and flow-control semantics using TypeScript-based REST interactions. diff --git a/skills/troubleshooting.md b/skills/troubleshooting.md index d3634a7f..938d628f 100644 --- a/skills/troubleshooting.md +++ b/skills/troubleshooting.md @@ -16,14 +16,13 @@ import { WorkflowAbort } from "@upstash/workflow"; try { // Any step will throw WorkflowAbort const result = await context.run("step", () => "ok"); - } catch (err) { - if (err instanceof WorkflowAbort) throw err; // required + if (err instanceof WorkflowAbort) throw err; // required // handle real errors } ``` -If you *must* handle errors, move the `try/catch` into the function passed to `context.run`. +If you _must_ handle errors, move the `try/catch` into the function passed to `context.run`. --- @@ -46,6 +45,7 @@ export const { POST } = serve(async (context) => { ``` **Header considerations:** When triggering workflows, ensure payload‑friendly headers like: + - `Content-Type: text/plain` - `Content-Type: application/json` @@ -54,11 +54,13 @@ export const { POST } = serve(async (context) => { ## Signature Verification Failures If QStash request verification fails: + - Ensure the workflow is triggered through QStash (`client.trigger` or publish) - Verify `QSTASH_CURRENT_SIGNING_KEY` and `QSTASH_NEXT_SIGNING_KEY` - Pass appropriate `Content-Type` headers when triggering Error looks like: + ``` Failed to verify that the Workflow request comes from QStash: ... ``` @@ -68,6 +70,7 @@ Failed to verify that the Workflow request comes from QStash: ... ## Authorization Errors from Early Returns If your workflow returns **before running any step**, the SDK interprets this as failed authorization: + ``` HTTP 400 – Failed to authenticate Workflow request. ``` @@ -88,6 +91,7 @@ export const { POST } = serve(async (context) => { ## Retry Configuration Retries come from two locations: + - Workflow start (`client.trigger`): default **3**, applies to workflow steps - `context.call`: default **0** - `context.invoke`: default **0** @@ -99,11 +103,13 @@ Use these in combination to tune behavior correctly. ## Verbose Mode Diagnostics Enable verbose logging to debug rare edge cases: + ```ts serve(handler, { verbose: true }); ``` Useful warnings include: + - Localhost in workflow URL - Duplicate step execution - Network response anomalies @@ -118,6 +124,7 @@ Each of these indicates environmental or routing issues rather than workflow log If deployed behind a proxy that downgrades HTTPS → HTTP (e.g., Railway), the SDK may infer the wrong protocol. **Fix:** explicitly set: + ``` UPSTASH_WORKFLOW_URL=https://your-deployment-url ``` @@ -127,6 +134,7 @@ UPSTASH_WORKFLOW_URL=https://your-deployment-url ## Workflow Stuck on First Step If the first step appears stuck: + - Check step logs in the dashboard - Ensure the endpoint URL is correct and reachable - Verify SDK versions @@ -147,6 +155,7 @@ detected a non-workflow destination for trigger/invoke ``` **Ensure:** + - The URL points to a `serve()` workflow endpoint - Both caller and workflow endpoint use the latest SDK @@ -157,6 +166,7 @@ detected a non-workflow destination for trigger/invoke Preview deployments block external requests unless bypassed. **Steps:** + 1. Create a bypass secret in Vercel → Deployment Protection 2. Add it to the QStash client + pass it when triggering @@ -164,24 +174,27 @@ Preview deployments block external requests unless bypassed. import { Client as QStash } from "@upstash/qstash"; import { serve } from "@upstash/workflow/nextjs"; -export const { POST } = serve(async (context) => { - // workflow logic -}, { - qstashClient: new QStash({ - token: process.env.QSTASH_TOKEN!, - headers: { - "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET! - } - }) -}); +export const { POST } = serve( + async (context) => { + // workflow logic + }, + { + qstashClient: new QStash({ + token: process.env.QSTASH_TOKEN!, + headers: { + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!, + }, + }), + } +); // Triggering await client.trigger({ url: "https://preview.vercel.app/workflow", headers: { - "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET! + "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!, }, - body: "hello" + body: "hello", }); ``` @@ -189,4 +202,4 @@ Both trigger header and client header are required. --- -This documentation consolidates the core troubleshooting patterns needed to debug and maintain healthy workflows in TypeScript applications. \ No newline at end of file +This documentation consolidates the core troubleshooting patterns needed to debug and maintain healthy workflows in TypeScript applications. From 736f201037b9782dadc71fabdfbc24eb850dd046 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 16:50:31 +0300 Subject: [PATCH 3/8] fix: fmt --- skills/agents.md | 2 -- 1 file changed, 2 deletions(-) diff --git a/skills/agents.md b/skills/agents.md index 0fe89145..707b0293 100644 --- a/skills/agents.md +++ b/skills/agents.md @@ -1,8 +1,6 @@ --- - name: workflow-agents description: Skill for building, configuring, and orchestrating Upstash Workflow Agents. Use when the user mentions agent workflows, multi-agent collaboration, tools, tasks, Upstash Workflow, or patterns like evaluator-optimizer, prompt chaining, parallelization, or orchestrator-workers. - --- # Workflow Agents From 1a84f1b77a9639fddef6802801b752fe3a11ab13 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 16:58:44 +0300 Subject: [PATCH 4/8] fix: simplify --- skills/basics/client.md | 19 ++----------------- skills/basics/context.md | 6 ------ skills/features/flow-control.md | 2 +- 3 files changed, 3 insertions(+), 24 deletions(-) diff --git a/skills/basics/client.md b/skills/basics/client.md index 78d6422e..4f3a039a 100644 --- a/skills/basics/client.md +++ b/skills/basics/client.md @@ -56,8 +56,6 @@ await client.cancel({ urlStartingWith: "https://your-endpoint.com" }); await client.cancel({ all: true }); ``` -**Pitfalls**: - - `ids` accepts both string and array, but mixing with other filters is not supported. - `urlStartingWith` cancels _all_ descendant paths—use carefully. @@ -70,7 +68,7 @@ Fetch workflow execution logs with filtering and cursor-based pagination. ```ts const { runs, cursor } = await client.logs({ workflowRunId: "wfr_123", // filter a specific run - workflowUrl: "https://endpoint", // fetch logs for a workflow + workflowUrl: "https://endpoint", // fetch logs for an endpoint state: "RUN_FAILED", // filter by execution state count: 50, // limit return size workflowCreatedAt: 1700000000, // Unix timestamp @@ -78,11 +76,6 @@ const { runs, cursor } = await client.logs({ }); ``` -**Common mistakes**: - -- Passing invalid state strings causes silent empty results. -- `workflowUrl` must match exactly; prefix filtering is not supported here. - --- ### Notifying Events (`client.notify`) @@ -96,7 +89,7 @@ await client.notify({ }); ``` -**Pitfall**: If no workflow is waiting, no error is thrown—ensure event IDs match exactly. +**Pitfall**: If no workflow is waiting, no error is thrown. --- @@ -129,10 +122,7 @@ const { messages, cursor } = await client.dlq.list({ }); ``` -**Pitfalls**: - - Date fields use Unix **ms**, not seconds. -- `url` must match exactly. --- @@ -187,9 +177,4 @@ await client.dlq.resume({ ## General Tips for TS Consumers -- Always validate URLs before triggering or canceling workflows. -- When dealing with arrays (`ids`, `dlqId`), ensure consistent type usage. - Use flow control when bulk‑triggering or restarting large batches to avoid rate limits. -- Cursor values must be reused exactly as returned—do not modify. - -This skill equips AI agents with reliable patterns for orchestrating workflow runs, handling failures, and managing operational workflows end-to-end. diff --git a/skills/basics/context.md b/skills/basics/context.md index 91f538f8..50f79a80 100644 --- a/skills/basics/context.md +++ b/skills/basics/context.md @@ -228,9 +228,3 @@ const { eventData, timeout } = await context.waitForEvent("wait for update", "or timeout: "2h", }); ``` - ---- - -## Summary - -This skill centralizes all workflow context primitives. They enable long‑running, state‑persistent, event‑driven workflows with external API access, typed provider integrations, and fine‑grained control over retry strategies, sleeps, and cross‑workflow invocation. diff --git a/skills/features/flow-control.md b/skills/features/flow-control.md index 10bd1dbe..0680b902 100644 --- a/skills/features/flow-control.md +++ b/skills/features/flow-control.md @@ -133,7 +133,7 @@ await client.trigger({ key: "media-processing", // Shared grouping key parallelism: 4, // Max 4 at the same time rate: 10, // Max 10 starts - period: 60_000, // in 60 seconds (ms or string supported) + period: "60s", // in 60 seconds }, }); ``` From abd08dd70de5005b8b2bd082b2931a6a3284edd6 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 16:59:32 +0300 Subject: [PATCH 5/8] fix: simplify --- skills/agents.md | 14 -------------- skills/rest-api.md | 4 ---- skills/troubleshooting.md | 4 ---- 3 files changed, 22 deletions(-) diff --git a/skills/agents.md b/skills/agents.md index 707b0293..3eb941c7 100644 --- a/skills/agents.md +++ b/skills/agents.md @@ -246,17 +246,3 @@ export const { POST } = serve(async (context) => { return (await task.run()).text; }); ``` - ---- - -# When to Use This Skill - -Use this Skill whenever the task involves: - -- Designing agent workflows -- Multi-agent coordination or patterns -- Using LangChain/AI SDK tools inside Workflow -- Implementing refinement loops or multi-step processing -- Debugging or optimizing Upstash agent behavior - -This Skill ensures consistent, correct usage of the Workflow Agents API and provides reference patterns for building reliable TypeScript agent systems. diff --git a/skills/rest-api.md b/skills/rest-api.md index aaf63683..6a1d6ef8 100644 --- a/skills/rest-api.md +++ b/skills/rest-api.md @@ -154,7 +154,3 @@ All are optional; original values are reused if omitted. • Always log `failureCallbackInfo.state` to determine whether a callback rerun is required. • Apply header overrides sparingly; flow-control misconfiguration may stall large batches. • Prefer filtering in bulk operations instead of manually passing large dlqIds arrays. - ---- - -This Skill equips agents to safely recover workflows, inspect DLQ state, and manage retry and flow-control semantics using TypeScript-based REST interactions. diff --git a/skills/troubleshooting.md b/skills/troubleshooting.md index 938d628f..1150741a 100644 --- a/skills/troubleshooting.md +++ b/skills/troubleshooting.md @@ -199,7 +199,3 @@ await client.trigger({ ``` Both trigger header and client header are required. - ---- - -This documentation consolidates the core troubleshooting patterns needed to debug and maintain healthy workflows in TypeScript applications. From 8603de114cfe7617b84a60e0bb968102adfa7c99 Mon Sep 17 00:00:00 2001 From: Cahid Arda Date: Wed, 28 Jan 2026 16:59:51 +0300 Subject: [PATCH 6/8] Update skills/features/wait-for-event.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- skills/features/wait-for-event.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/skills/features/wait-for-event.md b/skills/features/wait-for-event.md index b5a7af64..1cb03ee6 100644 --- a/skills/features/wait-for-event.md +++ b/skills/features/wait-for-event.md @@ -79,7 +79,7 @@ async function notifyProcessingComplete(orderId: string, payload: any) { // First attempt - returns array of NotifyResponse const waiters = await client.notify({ eventId, eventData: payload }); - if (waiters > 0) return first; + if (waiters > 0) return waiters; // Retry if no workflows were waiting await new Promise((r) => setTimeout(r, 3000)); From caecfe8a2037e125debfa06c09108d844813d0fc Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 17:03:00 +0300 Subject: [PATCH 7/8] fix: verbose info --- skills/how-to/migrations.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/skills/how-to/migrations.md b/skills/how-to/migrations.md index cf7b7599..b5c19f39 100644 --- a/skills/how-to/migrations.md +++ b/skills/how-to/migrations.md @@ -125,7 +125,7 @@ await context.invoke("other", { ### Logger removed → middleware system added -Replace `verbose` logging with explicit middlewares. +Removed `WorkflowLogger` class and added `WorkflowMiddleware`. Updated `verbose` param to only allow `true` (in this case, `loggingMiddleware` will be used which prints to console). ```ts import { loggingMiddleware, WorkflowMiddleware } from "@upstash/workflow"; @@ -193,7 +193,7 @@ const { status, headers, body } = await context.call("call-step", { ## Summary of Common Mistakes - Using `context.agents` instead of the new `agentWorkflow()` helper. -- Leaving old serve-config options (`retries`, `verbose`, etc.). +- Leaving old serve-config options (`retries`, `flowControl`, etc.). - Forgetting to `JSON.stringify` bodies. - Assuming `context.call` failures stop workflow execution. - Mix-and-matching `@upstash/qstash` and `@upstash/workflow` imports. From 1fe9eb3ad07a2a0c58c321e647ce9c0c82c06472 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Wed, 28 Jan 2026 17:28:03 +0300 Subject: [PATCH 8/8] fix: update skill name for clarity --- skills/SKILL.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/skills/SKILL.md b/skills/SKILL.md index d8df73e0..11c3401f 100644 --- a/skills/SKILL.md +++ b/skills/SKILL.md @@ -1,5 +1,5 @@ --- -name: upstash-workflow-sdk +name: @upstash/workflow TypeScript SDK Skill description: Lightweight guidance for using the Upstash Workflow SDK to define, trigger, and manage workflows. Use this Skill whenever a user wants to create workflow endpoints, run steps, or interact with the Upstash Workflow client. ---