The Exchange

What is an exchange?

Every piece of data that moves through a capability is wrapped in an exchange. When a source produces data, it becomes an exchange. Every operation receives that exchange and passes it along. The destination receives it last.

An exchange has two parts:

• body: the main payload. This is your data: an object, a string, a number, whatever your capability is working with.
• headers: metadata about the exchange. Timestamps, IDs, adapter-specific context, and anything you want to carry alongside the data without putting it in the body.

{
  "id": "a3f4e1b2-9c6d-4e8a-b1f3-2d7c0e5a9f12",
  "body": {
    "to": "alice@example.com",
    "subject": "Your order is confirmed"
  },
  "headers": {
    "routecraft.correlation_id": "req-abc-123",
    "routecraft.route": "send-confirmation"
  }
}

Body

The body is what your operations act on. .transform(), .filter(), and .process() all receive the current body (or the full exchange) and return something new.

craft()
  .id('greet')
  .from(simple({ name: 'Alice' }))
  .transform((body) => `Hello, ${body.name}!`)  // body is { name: 'Alice' }
  .to(log())                                      // body is now 'Hello, Alice!'

The body type flows through the DSL. TypeScript tracks what shape the body is at each step, giving you full type safety throughout the pipeline.

Headers

Headers travel alongside the body without being part of it. They are useful for metadata you want available throughout the pipeline but do not want polluting the body.

Set a header with .header():

craft()
  .id('process-order')
  .from(simple({ orderId: '123', amount: 49.99 }))
  .header('x-tenant', 'acme-corp')
  .header('x-priority', (exchange) => exchange.body.amount > 100 ? 'high' : 'normal')
  .process((exchange) => {
    const tenant = exchange.headers['x-tenant']     // 'acme-corp'
    const priority = exchange.headers['x-priority'] // 'normal'
    return exchange
  })
  .to(log())

Headers can be static values or derived from the exchange at runtime.

Built-in headers

Routecraft sets a number of routecraft.* headers automatically on every exchange:

These are useful for logging, debugging, and correlating exchanges across capability chains. Six of them are the engine's, and so are routecraft.trace.parent and routecraft.journey.joined, which mark an exchange that joined its journey rather than being called into it: .header() refuses all eight at build time. Adapter keys in the namespace, such as a mail message's UID, stay settable.

Journeys: correlation and causation

One request often runs many exchanges: the route it reached, the routes that one calls, the tools an agent picks, the routes an event() watcher starts in reaction. Two headers tie them together. The correlation id groups a journey; the causation id draws its tree, each exchange naming the one that caused it. The engine sets both at every hand-off it makes: a direct() call or forward(), an agent's tool call, a .tap(), .multicast() or .dispatch() clone, a .split() child, a .debounce() release, and an event() delivery, which continues the journey of the exchange that caused the event. The exchange's logger binds both, and every route:exchange:* event carries both, so a ledger rebuilds any journey from route:exchange:started together with route:exchange:correlated, which records a .correlate() re-stamp.

Across a door the two ids travel as W3C trace context. Outbound calls send traceparent and a tracestate member rc=c:<correlation>;x:<exchange>, which an OpenTelemetry service reads as a parent span and a Routecraft service reads as the exact ids, so never build a correlation id from personal data. Inbound, a door adopts a caller's context by default only from a caller it authenticated, because a caller who could attach its requests to any journey could pollute that journey's record; traceContext: 'always' adopts it from any caller, for a door whose callers are all trusted, and 'never' from none. Each door and client takes a traceContext option: see the http() client, httpPlugin, opsPlugin, remotesPlugin and mcpPlugin. The correlation id is an observability claim and never authority: nothing that authorizes, caches, rate-limits or resumes reads it, and an exchange that joined its journey rather than being called into it (a door adopted the id, .correlate() re-stamped it, an event() delivery continued it) does not reach what a called exchange reaches by correlation id, such as an editor turn's surface.

Where a channel carries no trace context (a webhook back from GitHub, a reply by mail), the application records where each artefact came from and joins the returning exchange to its journey with .correlate(). A source adapter that knows the journey supplies both ids on emit instead.

Adapter-specific headers

Chunked file-based adapters set additional headers on each emitted exchange:

Framework keys live on the exported HeadersKeys constant; each adapter exports its own key object (TimerHeaders, CronHeaders, FileHeaders, CsvHeaders, JsonlHeaders, MailHeaders, ...) for type-safe access:

import { JsonlHeaders } from '@routecraft/routecraft'

.process((exchange) => {
  const lineNum = exchange.headers[JsonlHeaders.LINE]
  const filePath = exchange.headers[JsonlHeaders.PATH]
  return exchange
})

Body vs full exchange access

Most operations give you a choice: work with just the body, or the full exchange.

Body only with .transform():

.transform((body) => body.toUpperCase())

Full exchange with .process():

.process((exchange) => {
  const tenantId = exchange.headers['x-tenant']
  return { ...exchange, body: { ...exchange.body, tenantId } }
})

Full exchange with .filter():

.filter((exchange) => exchange.headers['x-priority'] === 'high')

Use .transform() when you only need the data. Use .process() or .filter() when you need headers, correlation IDs, or the context.

Immutability

The exchange is immutable. DefaultExchange shallow-freezes the wrapper, its headers, and (when present) the principal at construction, and every field on Exchange<T> is readonly. The body is intentionally not deep-frozen so adapters can attach arbitrary payloads, but the framework never mutates it and your code should not either.

An exchange is often still needed after a step has used it: .retry() re-runs a step on the exchange that step was given, and an .error() handler receives the exchange as the failing step saw it. Because no step can replace an exchange's body or headers in place, a re-attempt or a handler starts from that input, and no step can quietly rewrite a header, such as the authenticated principal, that later steps rely on. The freeze is shallow: an object body can still be mutated, and a mutation made before a failure is what the retry and the handler then see. Return a new object rather than changing the one you were given.

Operations that change the exchange return a new one by copy-on-write (spread) rather than mutating in place. The framework re-wraps the returned plain object back into a proper instance, preserving the context binding, route binding, and id.

// Correct: copy-on-write
.process((exchange) => ({
  ...exchange,
  body: { ...exchange.body, stage: 'processed' },
  headers: { ...exchange.headers, 'x-stage': 'processed' },
}))

// Wrong: body is not deep-frozen, so this compiles and runs without throwing, but mutating
// in place bypasses copy-on-write, so the framework never re-wraps or tracks the change
.process((exchange) => {
  exchange.body.stage = 'processed'
  return exchange
})

Returning the same exchange unchanged is a valid no-op pass-through.

Exchange in taps

When you .tap(), the tap receives a deep copy of the exchange with a new ID. The correlation ID is preserved and the causation ID names the tapped exchange, so you can trace the tap back to its parent exchange. The main pipeline continues immediately without waiting for the tap.

craft()
  .id('order-pipeline')
  .from(source)
  .tap((exchange) => {
    // exchange.headers['routecraft.causation_id'] is the parent's id
    auditLog.write(exchange)
  })
  .to(destination)

Related