simple<T>(producer: (() => T | Promise<T>) | T): Source<T>

// Static value
.id('hello-route')
.from(simple('Hello, World!'))

// Array of values (each becomes a separate exchange)
.id('items-route')
.from(simple(['item1', 'item2', 'item3']))

// Dynamic function
.id('api-route')
.from(simple(async () => {
  const response = await fetch('https://api.example.com/data')
  return response.json()
}))

// With custom ID
.id('data-loader')
.from(simple(() => loadData()))

log<T>(formatter?: (exchange: Exchange<T>) => unknown, options?: LogOptions): Destination<T, void>

// Log final result (default: logs exchange ID, body, and headers at info level)
.to(log())

// Log intermediate data without changing flow
.tap(log())

// Log with custom formatter function
.tap(log((ex) => `Exchange with id: ${ex.id}`))
.tap(log((ex) => `Body: ${JSON.stringify(ex.body)}`))
.tap(log((ex) => `Exchange with uuid: ${ex.headers.uuid}`))

// Log at different levels
.tap(log(undefined, { level: 'debug' }))
.tap(log((ex) => ex.body, { level: 'warn' }))
.tap(log((ex) => ex.body, { level: 'error' }))

// For debug logging, use the convenience helper
.tap(debug())
.tap(debug((ex) => ex.body))

debug<T>(formatter?: (exchange: Exchange<T>) => unknown): Destination<T, void>

// Log at debug level (default format)
.tap(debug())

// Log with custom formatter at debug level
.tap(debug((ex) => `Debug: ${JSON.stringify(ex.body)}`))
.tap(debug((ex) => ({ id: ex.id, bodySize: JSON.stringify(ex.body).length })))

// Use throughout development workflow
craft().from(source).tap(debug((ex) => `Input: ${ex.body}`)).transform(processData).tap(debug((ex) => `Processed: ${ex.body}`)).to(destination)

timer(options?: TimerOptions): Source<undefined>

// Simple interval (every second)
.id('ticker')
.from(timer({ intervalMs: 1000 }))

// Limited runs (10 times, then stop)
.id('batch-job')
.from(timer({ intervalMs: 5000, repeatCount: 10 }))

// Start with delay
.id('delayed-start')
.from(timer({ intervalMs: 1000, delayMs: 5000 }))

// Daily at specific time
.id('daily-report')
.from(timer({ exactTime: '09:30:00' }))

// Fixed rate (ignore execution time)
.id('heartbeat')
.from(timer({ intervalMs: 1000, fixedRate: true }))

// Add random jitter to prevent synchronized execution
.id('distributed-task')
.from(timer({ intervalMs: 1000, jitterMs: 200 }))

cron(expression: string, options?: CronOptions): Source<undefined>

// Every 5 minutes
.id('poller')
.from(cron('*/5 * * * *'))

// Weekdays at 9am Eastern
.id('morning-report')
.from(cron('0 9 * * 1-5', { timezone: 'America/New_York' }))

// Daily at midnight (nickname)
.id('nightly-cleanup')
.from(cron('@daily'))

// Every 30 seconds (6-field)
.id('health-check')
.from(cron('*/30 * * * * *'))

// First day of month, limited to 12 fires
.id('monthly-report')
.from(cron('@monthly', { maxFires: 12, name: 'monthly-report' }))

// With jitter to prevent thundering herd
.id('distributed-poll')
.from(cron('*/5 * * * *', { jitterMs: 5000 }))

// Run only during Q1 2026
.id('q1-campaign')
.from(cron('@daily', { startAt: '2026-01-01', stopAt: '2026-04-01' }))

import { event } from '@routecraft/routecraft'

// Single event
craft().from(event('route:started')).to(log())

// Multiple events
craft().from(event(['route:started', 'route:stopped'])).to(log())

// Source (endpoint = route id). Body types are unknown at the adapter
// layer; schemas live on the route builder via `.input()` / `.output()`.
direct(options?: Partial<DirectServerOptions>): Source<unknown>

// Destination (registry-aware: body type resolves from DirectEndpointRegistry when populated)
direct<K extends RegisteredDirectEndpoint>(endpoint: K): Destination<ResolveBody<DirectEndpointRegistry, K>, unknown>

// Destination (names a target route)
direct<T>(endpoint: string | ((exchange: Exchange<T>) => string)): Destination<T, T>

// Destination with explicit input != output (e.g. in-process agent call)
direct<TIn, TOut>(
  endpoint: RegisteredDirectEndpoint | ((exchange: Exchange<TIn>) => string),
): Destination<TIn, TOut>

// Producer route that sends to a direct endpoint
craft()
  .id('data-producer')
  .from(source)
  .transform(processData)
  .to(direct('processed-data'))

// Consumer route that receives from the endpoint (route id = endpoint)
craft()
  .id('processed-data')
  .from(direct())
  .process(businessLogic)
  .to(destination)

// Consumer with framework-enforced validation
craft()
  .id('order-processing')
  .description('Validate and persist an incoming order')
  .input({ body: z.object({ orderId: z.string() }) })
  .output({ body: z.object({ status: z.literal('created'), orderId: z.string() }) })
  .from(direct())
  .process(validateOrder)
  .process(saveOrder)
  .transform(() => ({ status: 'created', orderId: '12345' }))

// Dynamic endpoint based on message content (destination side)
craft()
  .id('dynamic-router')
  .from(source)
  .to(direct((ex) => `handler-${ex.body.type}`))

// Route messages to different handlers based on priority
craft()
  .id('priority-router')
  .from(source)
  .to(direct((ex) => {
    const priority = ex.headers['priority'] || 'normal';
    return `processing-${priority}`;
  }))

// Consumer routes -- their ids match the dynamic target names
craft()
  .id('processing-high')
  .from(direct())
  .to(urgentProcessor)

craft()
  .id('processing-normal')
  .from(direct())
  .to(standardProcessor)

// Agent-only capability -- no .id() means a UUID endpoint,
// discoverable by agents but not callable from code
craft()
  .description('Internal knowledge base lookup')
  .input({ body: z.object({ query: z.string() }) })
  .from(direct())
  .process(fetchSnippets)

// Destination where the callee returns a different body shape than the caller sends.
// Supply two type arguments to express the response shape (e.g. an in-process agent).
craft()
  .id('agent-caller')
  .from(httpSource)
  .transform((body) => ({ name: body.agent, query: body.text }))
  .enrich(direct<{ name: string; query: string }, AgentResult>('agent'))

craft()
  .id('user-processor')
  .from(direct())  // No schema -- all data passes through
  .process(processUser)

import { z } from 'zod'

// z.object() - strips extra fields (default behavior)
const strictSchema = z.object({
  userId: z.string().uuid(),
  action: z.enum(['create', 'update', 'delete'])
})

craft()
  .id('user-processor')
  .input({ body: strictSchema })
  .from(direct())
  .process(processUser)

// Passes: { userId: '...', action: 'create' }
// Passes: { userId: '...', action: 'create', extra: 'field' }
//    Extra fields silently removed from result
// RC5002: { userId: '...', missing: 'action' }

// z.looseObject() - preserves extra fields
const looseSchema = z.looseObject({
  userId: z.string().uuid(),
  action: z.enum(['create', 'update'])
})

craft()
  .id('user-processor')
  .input({ body: looseSchema })
  .from(direct())
  .process(processUser)

// Passes: { userId: '...', action: 'create', extra: 'field' }
//    All fields preserved including extra

// z.strictObject() - rejects extra fields with error
const veryStrictSchema = z.strictObject({
  userId: z.string().uuid(),
  action: z.enum(['create', 'update'])
})

craft()
  .id('user-processor')
  .input({ body: veryStrictSchema })
  .from(direct())
  .process(processUser)

// Passes: { userId: '...', action: 'create' }
// RC5002: { userId: '...', action: 'create', extra: 'field' }

// No header schema - all headers pass through unchanged
craft()
  .id('api-handler')
  .input({ body: z.object({ id: z.string() }) })
  // input.headers not specified - all headers preserved
  .from(direct())
  .process(handleRequest)

// z.looseObject() - validate required headers, keep extras
craft()
  .id('api-handler')
  .input({
    headers: z.looseObject({
      'x-tenant-id': z.string().uuid(),
      'x-trace-id': z.string().optional(),
    }),
  })
  .from(direct())
  .process(handleRequest)

// Passes: { 'x-tenant-id': '...', 'x-other': '...' } (validates x-tenant-id, keeps x-other)

// z.object() - validate declared headers; merge preserves pass-through keys
craft()
  .id('api-handler')
  .input({
    headers: z.object({
      'x-tenant-id': z.string().uuid(),
    }),
  })
  .from(direct())
  .process(handleRequest)

// Passes: { 'x-tenant-id': '...', 'x-other': '...' } (x-other preserved via merge)

const schema = z.object({
  userId: z.string(),
  createdAt: z.coerce.date()  // Transforms string to Date
})

craft()
  .id('processor')
  .input({ body: schema })
  .from(direct())
  .process((data) => {
    // data.createdAt is Date, not string
    console.log(data.createdAt.getFullYear())
  })

import { ADAPTER_DIRECT_REGISTRY } from '@routecraft/routecraft'

craft()
  .id('fetch-content')
  .title('Fetch content')
  .description('Fetch and summarize web content from URL')
  .input({ body: z.object({ url: z.string().url() }) })
  .output({ body: z.object({ summary: z.string() }) })
  .from(direct())
  .process(fetchAndSummarize)

// Later, query registered routes from context
const ctx = await new ContextBuilder().routes(...).build()
await ctx.start()

const registry = ctx.getStore(ADAPTER_DIRECT_REGISTRY)
const routes = registry ? Array.from(registry.values()) : []
// [{ endpoint, title?, description?, input?, output? }]

http<T, R>(options: HttpOptions<T>): Destination<T, HttpResult<R>>

// Static GET request - result merged into body
.enrich(http({ 
  method: 'GET',
  url: 'https://api.example.com/users'
}))

// Dynamic URL based on exchange data
.enrich(http({ 
  method: 'GET',
  url: (exchange) => `https://api.example.com/users/${exchange.body.userId}`
}))

// Custom aggregator to control merge behavior
.enrich(
  http({ url: 'https://api.example.com/profile' }),
  (original, result) => ({
    ...original,
    body: { ...original.body, profileData: result.body }
  })
)

// Fire-and-forget intent (code does not read the response); body is still replaced by HttpResult at runtime
.to(http({
  method: 'POST',
  url: 'https://api.example.com/webhook',
  body: (exchange) => exchange.body
}))

// http() returns HttpResult; .to() replaces exchange body with it
.to(http({ 
  method: 'GET',
  url: 'https://api.example.com/transform' 
}))
// Body is now the HttpResult (status, headers, body). Use .enrich() with an aggregator to merge or preserve the original body.

// With query parameters
.enrich(http({
  url: 'https://api.example.com/search',
  query: (exchange) => ({ q: exchange.body.searchTerm, limit: 10 })
}))

// Simple webhook endpoint
.id('webhook-receiver')
.from(http({ path: '/webhook', method: 'POST' }))

// Multiple methods on same path
.id('data-api')
.from(http({ path: '/api/data', method: ['GET', 'POST', 'PUT'] }))

noop<T>(): NoopAdapter<T>

// Conditional destination based on environment
.to(process.env.NODE_ENV === 'production' ? realDestination() : noop())

// Testing placeholder
.to(noop()) // Messages are discarded but logged

spy<T>(): SpyAdapter<T>

import { spy } from '@routecraft/routecraft'

const spyAdapter = spy()

const route = craft()
  .id('my-route')
  .from(simple('payload'))
  .to(spyAdapter)

const t = await testContext().routes(route).build()
await t.test()

expect(spyAdapter.received).toHaveLength(1)
expect(spyAdapter.received[0].body).toBe('payload')
expect(spyAdapter.calls.send).toBe(1)

pseudo<Opts>(name?: string, options?: PseudoOptions): PseudoFactory<Opts>
pseudo<Opts>(name: string, options: PseudoKeyedOptions): PseudoKeyedFactory<Opts>

import { craft, timer, log, pseudo } from "@routecraft/routecraft";

// Option types (move to real adapter package later)
interface McpCallOptions {
  server: string;
  tool: string;
  args?: Record<string, unknown>;
}

interface GmailListResult {
  messages: { id: string; subject?: string }[];
  nextPageToken?: string;
}

const mcp = pseudo<McpCallOptions>("mcp");

// Object-only call: mcp<Result>(options)
craft()
  .from(timer({ intervalMs: 60_000 }))
  .enrich(
    mcp<GmailListResult>({
      server: "gmail",
      tool: "messages.list",
      args: { query: "is:unread" },
    }),
  )
  .split((r) => r.messages)
  .tap(log());

const queue = pseudo<{ ttl?: number }>("queue", { args: "keyed" });

craft()
  .from(source)
  .to(queue<void>("outbound", { ttl: 5000 }));

// Before (pseudo)
import { pseudo } from "@routecraft/routecraft";
const mcp = pseudo<McpCallOptions>("mcp");

// After (real adapter)
import { mcp } from "@routecraft/mcp-adapter";
// mcp<GmailListResult>({ server, tool, args }) still works

// Default: route per-line parse errors through .error(), keep streaming.
craft()
  .from(jsonl({ path: './events.jsonl', chunked: true }))
  .error((err, exchange) => {
    log.warn({ err, line: exchange.headers['routecraft.jsonl.line'] }, 'bad line');
    return null;
  })
  .filter((e) => e.body != null)
  .to(db());

// Stop the stream on the first malformed row (atomic-import semantics).
craft().from(csv({ path: './daily.csv', chunked: true, onParseError: 'abort' })).to(load());

// Drop unparseable mail with structured event observability.
craft().from(mail('INBOX', { onParseError: 'drop' })).to(process());

// Subscribe to parse drops across all routes:
ctx.on('route:*:exchange:dropped', ({ details }) => {
  if (details.reason === 'parse-failed') metrics.increment('source.parse.dropped');
});

file(options: FileOptions & { chunked: true }): Source<string>
file(options: FileOptions): FileAdapter   // Source<string> & Destination<unknown, void>

// Read file once
.from(file({ path: './input.txt' }))

// Custom encoding
.from(file({ path: './data.txt', encoding: 'latin1' }))

// Write to file (overwrite)
.to(file({ path: './output.txt', mode: 'write' }))

// Append to file
.to(file({ path: './log.txt', mode: 'append' }))

// Dynamic file paths with directory creation
.to(file({
  path: (exchange) => `./data/${exchange.body.date}.txt`,
  mode: 'write',
  createDirs: true
}))

// Per-line emission
.from(file({ path: './big.txt', chunked: true }))

json(options?: JsonOptions): JsonAdapter | JsonFileAdapter

// Parse JSON string from body
.transform(json())

// Extract nested data using dot notation
.transform(json({ path: 'data.items' }))

// Custom parsing with getValue
.transform(json({
  from: (b) => b.rawJson,
  getValue: (parsed) => parsed as User[]
}))

// Write to custom field
.transform(json({
  to: (body, result) => ({ ...body, parsed: result })
}))

// Read and parse JSON file
.from(json({ path: './data.json' }))

// With custom reviver
.from(json({
  path: './data.json',
  reviver: (key, value) => {
    if (key === 'date') return new Date(value);
    return value;
  }
}))

// Write with formatting
.to(json({
  path: './output.json',
  indent: 2
}))

// Dynamic paths with directory creation
.to(json({
  path: (exchange) => `./exports/${exchange.body.id}.json`,
  createDirs: true
}))

// With custom replacer
.to(json({
  path: './filtered.json',
  replacer: (key, value) => {
    if (key.startsWith('_')) return undefined;
    return value;
  }
}))

csv(options: CsvOptions & { chunked: true }): Source<CsvRow>
csv(options: CsvOptions): CsvAdapter   // Source<CsvData> & Destination<unknown, void>

npm install papaparse

// Read CSV with headers
.from(csv({ path: './data.csv', header: true }))
// Emits array of objects: [{ name: 'Alice', age: '30' }, ...]

// Read CSV without headers
.from(csv({ path: './data.csv', header: false }))
// Emits array of arrays: [['Alice', '30'], ['Bob', '25'], ...]

// Custom delimiter and encoding
.from(csv({
  path: './data.csv',
  delimiter: ';',
  encoding: 'latin1',
  header: true
}))

// Write array of objects to CSV
.to(csv({
  path: './output.csv',
  header: true
}))
// Automatically includes headers from object keys

// Write to tab-separated file
.to(csv({
  path: './data.tsv',
  delimiter: '\t',
  header: true
}))

// Dynamic paths with directory creation
.to(csv({
  path: (exchange) => `./reports/${exchange.body.reportDate}.csv`,
  createDirs: true,
  header: true
}))

// Append to existing CSV (skips header if file exists)
.to(csv({
  path: './log.csv',
  mode: 'append',
  header: true
}))

// Per-row emission
.from(csv({ path: './big.csv', chunked: true }))

jsonl<T>(options: JsonlSourceOptions & { chunked: true }): Source<T>
jsonl<T>(options: JsonlCombinedOptions): Source<T[]> & Destination<unknown, void>
jsonl(options: JsonlDestinationOptions): Destination<unknown, void>

// Read all lines as array
.from(jsonl({ path: './events.jsonl' }))
// Emits: [{ type: 'click', ts: 1 }, { type: 'view', ts: 2 }, ...]

// Per-line emission (chunked)
.from(jsonl({ path: './events.jsonl', chunked: true }))
// Emits one exchange per line with JSONL_LINE and JSONL_PATH headers

// Custom reviver
.from(jsonl({
  path: './data.jsonl',
  reviver: (key, value) => key === 'date' ? new Date(value) : value
}))

// Append to JSONL file (default)
.to(jsonl({ path: './output.jsonl' }))

// Overwrite file
.to(jsonl({ path: './output.jsonl', mode: 'write' }))

// Dynamic path with directory creation
.to(jsonl({
  path: (exchange) => `./logs/${exchange.body.date}.jsonl`,
  createDirs: true
}))

// Custom replacer (omit sensitive fields)
.to(jsonl({
  path: './output.jsonl',
  replacer: (key, value) => key === 'secret' ? undefined : value
}))

html(options: HtmlOptions): HtmlAdapter

// Extract text from title
.transform(html({ selector: 'title', extract: 'text' }))

// Extract multiple elements (returns array)
.transform(html({ selector: 'h2', extract: 'text' }))
// Result: ['First Heading', 'Second Heading', ...]

// Extract HTML content
.transform(html({ selector: '.content', extract: 'html' }))

// Extract attribute value
.transform(html({ selector: 'a', extract: 'attr', attr: 'href' }))

// Extract outer HTML (including element tag)
.transform(html({ selector: 'article', extract: 'outerHtml' }))

// Custom parsing from sub-field
.transform(html({
  selector: 'p',
  extract: 'text',
  from: (body) => body.htmlContent,
  to: (body, result) => ({ ...body, paragraphs: result })
}))

// Read HTML file and extract title
.from(html({
  path: './page.html',
  selector: 'title',
  extract: 'text'
}))

// Extract multiple links from file
.from(html({
  path: './page.html',
  selector: 'a',
  extract: 'attr',
  attr: 'href'
}))
// Emits array: ['https://example.com', '/about', ...]

// Write HTML string to file
.to(html({ path: './output.html' }))

// Dynamic paths with directory creation
.to(html({
  path: (exchange) => `./pages/${exchange.body.slug}.html`,
  createDirs: true
}))

// Append to HTML file
.to(html({
  path: './log.html',
  mode: 'append'
}))

mail(folder: string, options: Partial<MailServerOptions>): Source<MailMessage>
mail(folder: string): Destination<unknown, MailFetchResult>
mail(options: Partial<MailServerOptions>): Destination<unknown, MailFetchResult>
mail(action: MailAction): Destination<unknown, void>
mail(options?: Partial<MailClientOptions>): Destination<MailSendPayload, MailSendResult>

craft()
  .id('inbox-watcher')
  .from(mail('INBOX', { markSeen: true }))
  .to(log())

// Re-evaluate the inbox every minute; archive a message to mark it done.
// If you later extend `matchesCriteria`, previously-unmatched mail that is
// still in INBOX is picked up on the next cycle.
craft()
  .id('inbox-processor')
  .from(mail('INBOX', {
    pollIntervalMs: 60_000,
    markSeen: false,
    unseen: false,
  }))
  .filter(matchesCriteria)
  .process(processMessage)
  .to(mail({ action: 'move', folder: 'Archive' }))

craft()
  .id('check-inbox')
  .from(cron('0 */5 * * * *'))
  .enrich(mail('INBOX'))
  .to(log())

craft()
  .id('outbound')
  .from(direct())
  .to(mail())

// Forward unread mail to a different address
craft()
  .id('mail-forwarder')
  .from(mail('INBOX', { unseen: true, markSeen: true }))
  .transform((msg) => ({
    to: 'team@example.com',
    subject: `Fwd: ${msg.subject}`,
    text: msg.body.text ?? '',
  }))
  .to(mail())

// Archive after processing
craft()
  .id('archive-processed')
  .from(mail('INBOX', { unseen: true }))
  .tap(processMessage)
  .to(mail({ action: 'move', folder: 'Archive' }))

// Flag important messages
craft()
  .id('flag-important')
  .from(mail('INBOX', { subject: 'URGENT' }))
  .to(mail({ action: 'flag', flags: '\\Flagged' }))

// craft.config.ts
import type { CraftConfig } from '@routecraft/routecraft'

export const craftConfig: CraftConfig = {
  mail: {
    accounts: {
      default: {
        imap: {
          host: 'imap.gmail.com',
          auth: { user: process.env.MAIL_USER!, pass: process.env.MAIL_APP_PASSWORD! },
        },
        smtp: {
          host: 'smtp.gmail.com',
          auth: { user: process.env.MAIL_USER!, pass: process.env.MAIL_APP_PASSWORD! },
          from: process.env.MAIL_USER!,
        },
      },
    },
  },
}

// capabilities/inbox-watcher.ts
export { craftConfig } from '../craft.config'
import { craft, mail, log } from '@routecraft/routecraft'

export default craft()
  .id('inbox-watcher')
  .from(mail('INBOX', { markSeen: true }))
  .to(log())

.from(mail('INBOX', { account: 'support' }))
.to(mail({ account: 'notifications' }))