npm package: @kognitivedev/workflows
Description: Graph-based workflow engine with suspend/resume and persistence
Latest version: 0.2.22

# @kognitivedev/workflows

> Graph-based workflow engine with suspend/resume, persistence, streaming, and visualization.

## Table of Contents

- [Installation](#installation)
- [Quick Start](#quick-start)
- [Steps](#steps)
- [Control Flow](#control-flow)
- [Suspend and Resume](#suspend-and-resume)
- [Streaming](#streaming)
- [Retry Policies](#retry-policies)
- [Error Handling](#error-handling)
- [Agent Steps](#agent-steps)
- [Persistence](#persistence)
- [Visualization](#visualization)
- [State Management](#state-management)
- [API Reference](#api-reference)

---

## Installation

```bash
bun add @kognitivedev/workflows zod
```

---

## Quick Start

```typescript
import { createWorkflow, createStep } from "@kognitivedev/workflows";

const greet = createStep({
  id: "greet",
  execute: async (input: { name: string }) => ({ message: `Hello, ${input.name}!` }),
});

const workflow = createWorkflow({ name: "greeting" }).then(greet).build();

const result = await workflow.execute({ name: "Alice" });
console.log(result.output);  // { message: "Hello, Alice!" }
console.log(result.status);  // "completed"
console.log(result.runId);   // unique run identifier
```

---

## Steps

A step is the smallest unit of work. Create one with `createStep()`:

```typescript
import { createStep } from "@kognitivedev/workflows";
import { z } from "zod";

const processOrder = createStep({
  id: "process-order",
  description: "Validate and process an incoming order",
  inputSchema: z.object({ orderId: z.string(), amount: z.number() }),
  outputSchema: z.object({ confirmed: z.boolean(), total: z.number() }),
  execute: async (input, ctx) => {
    return { confirmed: true, total: input.amount * 1.1 };
  },
  retryPolicy: { maxRetries: 3, baseDelayMs: 500 },
});
```

The `execute` function receives the input and a **`StepContext`**:

- **`ctx.state`** — `Map<string, unknown>` with outputs from all previous steps
- **`ctx.suspend(data?)`** — pauses execution, returns resume data when continued
- **`ctx.emit(event, data)`** — emits custom events for streaming
- **`ctx.abortSignal`** — `AbortSignal` for cancellation
- **`ctx.resourceId`** — resource/project identifier
- **`ctx.traceId`** — distributed tracing ID

---

## Control Flow

### Sequential

Chain steps with `.then()`. Each step receives the previous step's output:

```typescript
const workflow = createWorkflow({ name: "transform" })
  .then(addPrefix)
  .then(addSuffix)
  .build();
```

### Branching

Route execution based on a condition. Returns a key to select a branch:

```typescript
const workflow = createWorkflow({ name: "router" })
  .then(classifyStep)
  .branch(
    (input) => input.category,                         // returns "billing" or "support"
    { billing: billingStep, support: supportStep }
  )
  .build();
```

If the key doesn't match any branch, the step is skipped and execution continues.

### Parallel

Execute steps concurrently. All receive the same input, output is an array:

```typescript
const workflow = createWorkflow({ name: "analyze" })
  .parallel([sentimentStep, toxicityStep, languageStep])
  .build();

const result = await workflow.execute({ text: "Hello" });
// result.output = [sentimentResult, toxicityResult, languageResult]
```

### ForEach

Apply a step to each item in an array:

```typescript
const workflow = createWorkflow({ name: "batch" })
  .foreach(
    (input: number[]) => input,  // selector extracts the array
    doubleStep                    // applied to each item
  )
  .build();

const result = await workflow.execute([1, 2, 3]);
// result.output = [2, 4, 6]
```

### Nested Workflows

Embed a child workflow. Suspend/resume propagates to the parent:

```typescript
const childWorkflow = createWorkflow({ name: "doubler" }).then(doubleStep).build();

const parentWorkflow = createWorkflow({ name: "parent" })
  .then(addOneStep)
  .workflow(childWorkflow)           // optionally: .workflow(child, { id: "custom-id" })
  .build();

const result = await parentWorkflow.execute({ value: 5 });
// 5 + 1 = 6, then 6 * 2 = 12
```

### Combining

All control flow operators can be freely combined:

```typescript
const workflow = createWorkflow({ name: "pipeline" })
  .then(validateOrder)
  .parallel([checkInventory, calculateTax])
  .then(mergeResults)
  .branch((s) => s.status, { in_stock: fulfillStep, backordered: backorderStep })
  .then(sendConfirmation)
  .build();
```

---

## Suspend and Resume

Any step can pause execution by calling `ctx.suspend()`. The workflow returns with `status: "suspended"` and a snapshot. Resume later with new data:

```typescript
const reviewStep = createStep({
  id: "human-review",
  execute: async (input, ctx) => {
    const feedback = await ctx.suspend({ draft: input.draft });
    return { feedback, draft: input.draft };
  },
});

const workflow = createWorkflow({ name: "review" })
  .then(draftStep)
  .then(reviewStep)
  .then(reviseStep)
  .build();

// Start — suspends at reviewStep
const result = await workflow.execute({ topic: "AI agents" });
console.log(result.status);   // "suspended"
console.log(result.snapshot);  // WorkflowSnapshot with state + suspend data

// Resume with human feedback
const resumed = await workflow.resume(result.snapshot!, "Looks great, approved");
console.log(resumed.status);  // "completed"
console.log(resumed.output);  // revised result
```

When a nested workflow suspends, the parent also suspends. Resume on the parent and it routes to the child automatically.

---

## Streaming

Use `workflow.stream()` to receive real-time events as steps complete:

```typescript
const { result, stream } = await workflow.stream(input, {
  streamModes: ["values", "debug", "custom"],
});

const reader = stream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(value.event, value.data);
}

const finalResult = await result;
```

**Stream modes:**

| Mode | Event | Description |
|------|-------|-------------|
| `"values"` | `values` | Full state snapshot after each step |
| `"updates"` | `updates` | Only the new step's output delta |
| `"debug"` | `debug` | Step entry/exit events with timing |
| `"custom"` | `custom` | Application events from `ctx.emit()` |

A `result` event is always emitted at the end.

**Emitting custom events:**

```typescript
const step = createStep({
  id: "process",
  execute: async (input, ctx) => {
    ctx.emit("progress", { percent: 50 });
    // ...
    ctx.emit("progress", { percent: 100 });
    return result;
  },
});
```

**Resume with streaming:** `workflow.resumeStream(snapshot, resumeData, options)` works identically.

---

## Retry Policies

Steps can automatically retry on failure with exponential backoff:

```typescript
const step = createStep({
  id: "flaky-api",
  execute: async (input) => {
    const res = await fetch(input.url);
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return res.json();
  },
  retryPolicy: { maxRetries: 3, baseDelayMs: 500 },
});
```

When retries occur, `result.steps[i].retryMetadata` contains `{ totalAttempts, attempts[] }` with error details per attempt.

---

## Error Handling

Step errors are caught and recorded — the workflow continues to the next node:

```typescript
const result = await workflow.execute(input);
const step = result.steps[0];

if (step.status === "error") {
  console.log(step.error!.message);  // error message
  console.log(step.error!.type);     // "TypeError", "Error", etc.
  console.log(step.error!.code);     // custom error code if set
  console.log(step.error!.stack);    // full stack trace
}
```

---

## Agent Steps

`agentStep()` creates a step that invokes an AI agent (must have a `generate()` method):

```typescript
import { agentStep } from "@kognitivedev/workflows";

// Text
const classifyStep = agentStep({
  id: "classify",
  agent: classifierAgent,
  buildMessages: (input: { text: string }) => [
    { role: "user", content: `Classify: ${input.text}` },
  ],
});

// Vision / multimodal
const visionStep = agentStep({
  id: "describe-image",
  agent: visionAgent,
  buildMessages: (input: { imageUrl: string }) => [{
    role: "user",
    content: [
      { type: "text", text: "Describe this image." },
      { type: "image", image: input.imageUrl },
    ],
  }],
});

// Custom output extraction
const extractStep = agentStep({
  id: "extract",
  agent: extractorAgent,
  buildMessages: (input) => [{ role: "user", content: input.text }],
  extractOutput: (result) => JSON.parse(result.text),
});
```

Default output is `{ text, toolCalls, usage?, steps?, runId? }`.

---

## OCR Steps

`createOcrStep()` creates a workflow step that extracts text from files using either:
- `mode: "manual"` (default): local `@kognitivedev/documents` processing with `DocumentProcessor`.
- `mode: "cloud"`: Kognitive cloud endpoint `/api/cognitive/documents/extract`.

```typescript
import { createOcrStep } from "@kognitivedev/workflows";

const ocrManual = createOcrStep<{ data: Uint8Array; filename: string; mimeType?: string }, { text: string }>({
  id: "ocr-manual",
  mode: "manual",
  buildSource: (input) => ({
    data: input.data,
    filename: input.filename,
    mimeType: input.mimeType,
  }),
  processorConfig: {
    ocrInstructions: "Preserve headings, tables, and values.",
  },
});

const ocrCloud = createOcrStep({
  id: "ocr-cloud",
  mode: "cloud",
  apiKey: process.env.KOGNITIVE_API_KEY,
  baseUrl: "http://localhost:3001",
  cloud: {
    pdfStrategy: "screenshot-first",
    ocrInstructions: "Preserve headings, tables, and values.",
  },
  buildSource: (input) => ({
    data: input.data,
    filename: input.filename,
    mimeType: input.mimeType,
  }),
});
```

Cloud mode requires `apiKey` and sends the file as multipart form data to the OCR endpoint.

---

## Persistence

### WorkflowStore Interface

```typescript
interface WorkflowStore {
  saveSnapshot(snapshot: WorkflowSnapshot): Promise<void>;
  getSnapshot(runId: string): Promise<WorkflowSnapshot | null>;
  listSnapshots(workflowName: string): Promise<WorkflowSnapshot[]>;
  deleteSnapshot(runId: string): Promise<void>;
  saveEvent?(event: WorkflowTimelineEvent): Promise<void>;
}
```

### InMemoryWorkflowStore

For development and testing:

```typescript
import { InMemoryWorkflowStore } from "@kognitivedev/workflows";

const store = new InMemoryWorkflowStore();
const result = await workflow.execute(input, { store });

await store.getSnapshot(result.runId);
await store.listSnapshots("my-workflow");
await store.deleteSnapshot(result.runId);
store.clear();        // remove all
console.log(store.size); // count
```

### PgWorkflowStore (PostgreSQL)

Production-ready persistence with Drizzle ORM. Auto-checkpoints at every step:

```typescript
import { PgWorkflowStore } from "@kognitivedev/workflows";

const store = new PgWorkflowStore({ db, schema, projectId: "my-project" });

const result = await workflow.execute(input, { store });

// Browse checkpoints
const checkpoints = await store.getCheckpoints(result.runId);
const checkpoint = await store.getCheckpointAt(result.runId, 2);

// Rollback to step index 1 (deletes later checkpoints)
const snapshot = await store.rollbackToStep(result.runId, 1);
const resumed = await workflow.resume(snapshot, newData, { store });

// Run tracking
const run = await store.getRun(result.runId);
const runs = await store.listRuns("my-workflow", { status: "completed", limit: 10 });

// Timeline events (auto-emitted: "node_enter", "node_exit")
const events = await store.getEvents(result.runId);
```

---

## Visualization

`toGraph()` returns React Flow-compatible node/edge data:

```typescript
const graph = workflow.toGraph();

console.log(graph.nodes);
// [{ id: "__start", type: "step", label: "Start" },
//  { id: "step1", type: "step", label: "step1" },
//  { id: "__end", type: "step", label: "End" }]

console.log(graph.edges);
// [{ source: "__start", target: "step1" },
//  { source: "step1", target: "__end" }]
```

Node types: `step`, `branch`, `parallel`, `foreach`, `workflow`. Branch edges include `label` with the branch key. Nested workflows include `data.graph` with the child's graph.

Lower-level: `toGraphJSON(name, nodes)` converts raw `WorkflowNode[]` directly.

---

## State Management

Workflow state is a `Map<string, unknown>` that accumulates step outputs:

- **`__input`** — the original workflow input
- **`stepId`** — each step's output is stored under its ID

```typescript
const step = createStep({
  id: "send-welcome",
  execute: async (input, ctx) => {
    const user = ctx.state.get("fetch-user") as { name: string };
    return { sent: true, to: user.name };
  },
});
```

State values must be JSON-serializable when using persistence.

---

## API Reference

### Exports

| Export | Type | Description |
|--------|------|-------------|
| `createWorkflow(config)` | function | Create a workflow builder |
| `createStep(definition)` | function | Create a step definition |
| `agentStep(config)` | function | Create an agent-powered step |
| `createOcrStep(config)` | function | Create an OCR step (manual processor or cloud extraction) |
| `executeWorkflowWithStream()` | function | Low-level streaming execution |
| `toGraphJSON(name, nodes)` | function | Convert nodes to graph |
| `InMemoryWorkflowStore` | class | In-memory persistence |
| `PgWorkflowStore` | class | PostgreSQL persistence |

### Workflow Methods

| Method | Returns | Description |
|--------|---------|-------------|
| `execute(input, options?)` | `Promise<WorkflowRunResult>` | Run workflow |
| `stream(input, options?)` | `Promise<{ result, stream }>` | Run with streaming |
| `resume(snapshot, data?, options?)` | `Promise<WorkflowRunResult>` | Resume suspended workflow |
| `resumeStream(snapshot, data?, options?)` | `Promise<{ result, stream }>` | Resume with streaming |
| `toGraph()` | `WorkflowGraph` | Get visualization graph |

### Key Types

| Type | Description |
|------|-------------|
| `StepDefinition<TInput, TOutput>` | Step config: `id`, `description?`, `inputSchema?`, `outputSchema?`, `execute`, `retryPolicy?` |
| `StepContext` | Step runtime: `state`, `suspend()`, `emit()`, `abortSignal`, `resourceId`, `traceId?` |
| `WorkflowRunResult<T>` | Result: `output`, `steps[]`, `runId`, `status`, `snapshot?` |
| `StepResult` | Per-step: `stepId`, `output`, `durationMs`, `status`, `error?`, `retryMetadata?` |
| `WorkflowSnapshot` | Suspended state: `workflowName`, `runId`, `currentStepId`, `state`, `suspendedAt`, `resumeData?` |
| `WorkflowExecuteOptions` | Options: `runId?`, `store?`, `abortSignal?`, `resourceId?`, `traceId?` |
| `RetryPolicy` | Retry config: `maxRetries`, `baseDelayMs` |
| `CheckpointData` | Checkpoint: `stepId`, `stepIndex`, `stateSnapshot`, `stepStatus`, `error?`, `retryMetadata?` |
| `WorkflowRunData` | Run record: `workflowName`, `runId`, `status`, `input?`, `output?`, `totalSteps`, `totalDurationMs?` |
| `WorkflowGraph` | Graph: `nodes: GraphNode[]`, `edges: GraphEdge[]` |
| `StreamMode` | `"values" \| "updates" \| "debug" \| "custom"` |

---

## License

MIT