> Part of the walkerOS documentation. Project overview and full index: <https://www.walkeros.io/llms.txt>

# Understanding walkerOS Sources

## Overview[​](#overview "Direct link to Overview")

Sources capture events from the external world (browser DOM, dataLayer, HTTP requests, cloud functions) and feed them to the collector.

**Core principle:** Sources capture. They don't process or deliver—that's collector and destinations.

## Source Interface[​](#source-interface "Direct link to Source Interface")

See [packages/core/src/types/source.ts](https://github.com/elbwalker/walkerOS/blob/main/packages/core/src/types/source.ts) for canonical interface.

### Init Function (Context Pattern)[​](#init-function-context-pattern "Direct link to Init Function (Context Pattern)")

Sources use a context-based initialization pattern:

```
import type { Source } from '@walkeros/core';



export const sourceMySource: Source.Init<Types> = async (context) => {

  const { config = {}, env, logger, id } = context;

  // ...

};
```

**Context contains:**

| Property    | Type                                   | Purpose                                                                                     |
| ----------- | -------------------------------------- | ------------------------------------------------------------------------------------------- |
| `config`    | `Source.Config<T>`                     | Settings, mapping, options                                                                  |
| `env`       | `Types['env']`                         | Environment (push, logger)                                                                  |
| `logger`    | `Logger`                               | Logging functions                                                                           |
| `id`        | `string`                               | Source identifier                                                                           |
| `collector` | `Collector.Instance`                   | Reference to collector                                                                      |
| `withScope` | `(scope, respond, body) => Promise<R>` | Bind ingest + respond to a single scope (server sources). Takes a normalized `Source.Scope` |

### Push Method[​](#push-method "Direct link to Push Method")

| Method        | Purpose                             |
| ------------- | ----------------------------------- |
| `push(input)` | Receive external input, emit events |

### Init Method[​](#init-method "Direct link to Init Method")

`init?: () => void | Promise<void>` — Optional eager-startup hook on the returned `Source.Instance`. The factory must be **side-effect-free**: build the instance and return it. The collector calls `init()` on every source eagerly after all factories register, regardless of `config.require`. Use `init` for work that previously sat in the factory body: adopting a pre-init window queue (e.g., `window.elbLayer`), attaching DOM listeners, opening sockets, intercepting `window.dataLayer`. After `init` runs the collector flips `Source.Config.init` to `true`.

### queueOn Buffer[​](#queueon-buffer "Direct link to queueOn Buffer")

`queueOn?: Array<{ type: On.Types; data: unknown }>` — Optional buffer on the `Source.Instance` for lifecycle events delivered before the source is **started** (started ≡ `config.init === true && !config.require?.length`). The collector pushes `{ type, data }` here when it would otherwise call `source.on(type, data)`. Once the source becomes started, the collector replays each entry via `source.on(...)` and clears the queue.

### Destroy Method[​](#destroy-method "Direct link to Destroy Method")

`destroy?: DestroyFn` — Optional cleanup method. Called during `command('shutdown')`. Use to close HTTP servers, timers, or connections. Receives `{ id, config, env, logger }`.

## Source Lifecycle[​](#source-lifecycle "Direct link to Source Lifecycle")

The collector and every source agree on three lifecycle markers, all on `Source.Config`:

| Field                | Purpose                                                                                                                                                                                                                                                                                                                                                                                                                                      |
| -------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `init?: boolean`     | Set by the collector to `true` after `Instance.init()` resolves. Authors do not write to it. Reflects "init has run", not "is started".                                                                                                                                                                                                                                                                                                      |
| `require?: string[]` | Author-supplied **timing hint**. Lists collector state a source needs before its first `on()` delivery. Satisfied by the collector's **current recorded state** (level), not only by a future event — so order does not matter: the gate clears whether the required state was recorded before or after this source registered. It is not a correctness dependency: a source reacts to state correctly whether or not it declares `require`. |
| `disabled?: boolean` | Hard skip — no factory invocation, no init, no event capture.                                                                                                                                                                                                                                                                                                                                                                                |

The flow is:

1. **Register** — collector invokes the factory. Factory returns a fresh `Source.Instance` with no side effects.
2. **Init pass** — collector calls `Instance.init()` on every registered source, then sets `config.init = true`.
3. **Lifecycle delivery** — for each collector event (`consent`, `user`, `session`, `run`, …) the collector decrements every source's `require` in place AND reconciles every still-pending source/destination against current state, so a gate also clears from state that was already recorded. If a source is started, it calls `source.on(type, data)` directly. Otherwise it pushes `{ type, data }` into `Instance.queueOn`.
4. **Replay** — when a source becomes started (require empties), the collector replays its `queueOn` via `source.on(...)` and clears the queue.

`require` therefore gates the **timing** of `on()` delivery, not code execution and not correctness. `Instance.init()` always runs eagerly. There is no `collector.pending.sources` map: per-source state lives entirely on `Source.Instance` and `Source.Config`.

### Exactly-once state delivery is a collector guarantee[​](#exactly-once-state-delivery-is-a-collector-guarantee "Direct link to Exactly-once state delivery is a collector guarantee")

State commands (`consent`, `user`, `globals`, `custom`) are recorded by the collector immediately, even before `run`, and delivered to each source's `on` handler exactly once per change. The collector tracks what each subscriber has already received, so re-running or re-registering never double-fires a state reaction. Sources should **not** hand-roll their own deduplication for state deliveries: the collector enforces exactly-once, and delivery is order-independent (it does not depend on source init order or on whether the state arrived before or after `run`).

This mirrors the destination model: `Destination.Instance.init` handles one-time bootstrap, `Destination.Config.init` is the collector-managed "init has run" flag, and `Destination.Config.require` gates event delivery the same way. See [walkeros-understanding-destinations](https://www.walkeros.io/skills/walkeros-understanding-destinations.md).

### Pre-run events are held and replayed by the collector[​](#pre-run-events-are-held-and-replayed-by-the-collector "Direct link to Pre-run events are held and replayed by the collector")

Events a source emits before the collector runs are held in a bounded FIFO buffer (`collector.preRunQueue`, capped by `config.queueMax`) and replayed at `run`, with the pipeline running once at replay time. A source therefore does **not** need bespoke `on('run')` deferral to avoid losing its startup events. Deferring emission to `on('run')` is still fine and often clearer for **ordering** (the session source does this so `session start` lands in the post-run pipeline), but it is no longer a loss-protection requirement.

The one case that still needs an explicit run gate is a **queue consumer** (`pubsub-pull`, `sqs`): starting to consume means acknowledging messages to a broker, and an ack must never precede delivery. These sources start their subscription or poll loop in `on('run')` and never ack a message the pipeline did not accept (`result.ok === false` routes to the source's `onPushError` disposition, nack by default).

## Push Signatures by Type[​](#push-signatures-by-type "Direct link to Push Signatures by Type")

| Source Type    | Signature                           | Example      |
| -------------- | ----------------------------------- | ------------ |
| Cloud Function | `push(req, res) → Promise<void>`    | HTTP handler |
| Browser        | `push(event, data) → Promise<void>` | DOM events   |
| DataLayer      | `push(event, data) → Promise<void>` | GTM-style    |

**Key insight:** Source `push` IS the handler. No wrappers needed.

```
// Direct deployment

http('handler', source.push);
```

**Do not confuse `Instance.push` with `env.push`.** They point opposite ways:

* `Instance.push` is **inward**: the handler callers hand input to (an HTTP request, a `window.elb(...)` call, a synthetic test message). `startFlow` exports the primary source's `Instance.push` as its `elb`.
* `env.push` is **outward**: where the source sends the events it produced.

A source can legitimately have both, and they need not be related: the browser source's `Instance.push` is a flexible `BrowserPush` that accepts commands and argument forms, while its events leave via `env.push`. The dataLayer and session sources keep `Instance.push: env.elb`, because callers do not feed them events at all — they capture from `dataLayer.push` and from session detection respectively.

When a test or integration code needs to invoke a source's `push` through the collector bag, `collector.sources` erases the per-source generic on read. Use `Source.getSource<T>(collector, id)` to recover the narrow signature without a cast. See the testing-strategy skill for the full pattern (symmetric helpers exist for destinations, transformers, stores).

## Source Paths[​](#source-paths "Direct link to Source Paths")

| Type   | Path                       | Examples           |
| ------ | -------------------------- | ------------------ |
| Web    | `packages/web/sources/`    | browser, dataLayer |
| Server | `packages/server/sources/` | gcp                |

## Browser Source[​](#browser-source "Direct link to Browser Source")

The browser source captures events from DOM using data attributes.

```
<button data-elb="product" data-elb-product="id:P123;name:Laptop">

  <span data-elbaction="click">Add to Cart</span>

</button>
```

See [packages/web/sources/browser/](https://github.com/elbwalker/walkerOS/blob/main/packages/web/sources/browser) for implementation.

### Parallel sources[​](#parallel-sources "Direct link to Parallel sources")

Several browser sources can run on one page. Trigger, element, and visibility state is per source, and the DOM is shared: any number of sources may scan the same node, each emitting into its own pipeline. Only two resources are single slots, `window[elb]` and the named `elbLayer` array, and each belongs to the first source that takes it. A source refused one of them logs an error and keeps capturing without it. Give each parallel flow its own `elb` and `elbLayer` name, or `elb: false` and `elbLayer: false` for an embedded source with no page-window footprint.

Full contract, recipes, and known limits: [Multiple sources on one page](https://www.walkeros.io/docs/sources/web/browser.md).

## DataLayer Source[​](#datalayer-source "Direct link to DataLayer Source")

Captures events from a GTM-style dataLayer array.

```
window.dataLayer.push({

  event: 'product view',

  product: { id: 'P123', name: 'Laptop' },

});
```

See [packages/web/sources/dataLayer/](https://github.com/elbwalker/walkerOS/blob/main/packages/web/sources/dataLayer) for implementation.

## Server Sources[​](#server-sources "Direct link to Server Sources")

Handle HTTP requests in cloud functions. Server sources use the context pattern:

```
import type { Source } from '@walkeros/core';



export const sourceCloudFunction: Source.Init<Types> = async (context) => {

  const { config = {}, env } = context;

  const { push: envPush } = env;



  // Apply defaults inline — flow.json is developer-controlled, so no

  // runtime validation. Shape checks live in ./schemas and are used by

  // `walkeros validate` and dev tooling, never at runtime.

  const userSettings = config.settings || {};

  const settings = {

    ...userSettings,

    // example default: port: userSettings.port ?? 3000,

  };



  const push = async (req: Request, res: Response): Promise<void> => {

    // Transform HTTP request → walkerOS event

    const event = transformRequest(req);

    await envPush(event);

    res.json({ success: true });

  };



  return { type: 'cloudfunction', config: { ...config, settings }, push };

};



// Direct deployment

export const handler = source.push;
```

See [packages/server/sources/gcp/](https://github.com/elbwalker/walkerOS/blob/main/packages/server/sources/gcp) for implementation.

**Forward the whole body.** A server source hands the collector the full `DeepPartialEvent` it received, renaming only its wire-level event-name key. It must not project the body through a field allowlist, or it strips `source` provenance (`release`, `trace`) and other legitimate fields. `express` and `fetch` are the reference pass-through implementations.

## Env Pattern (Dependency Injection)[​](#env-pattern-dependency-injection "Direct link to Env Pattern (Dependency Injection)")

Platform dependencies go through `env` with fallback to globals or direct imports. This enables testing and simulation without touching globals.

```
// Express source: env.express ?? express (import fallback)

const expressLib = env.express ?? express;

const app = expressLib();



// Web sources: env.window ?? window (global fallback)

const win = env.window ?? window;

const doc = env.document ?? document;
```

Every source's `Env` interface extends `Source.BaseEnv` with optional platform deps:

```
export interface Env extends Source.BaseEnv {

  window?: Window & typeof globalThis; // web sources

  document?: Document; // web sources

  express?: typeof express; // express source

  cors?: typeof cors; // express source

}
```

Tests inject mocks via `env` instead of mocking globals. See [testing-strategy](https://www.walkeros.io/skills/walkeros-testing-strategy.md).

### `env.push` for events, `env.elb` for commands[​](#envpush-for-events-envelb-for-commands "Direct link to envpush-for-events-envelb-for-commands")

`BaseEnv` gives every source two outward exits, and they are not interchangeable:

| Exit       | Use for                         | What it is                                          |
| ---------- | ------------------------------- | --------------------------------------------------- |
| `env.push` | **Events** (event objects only) | The collector's per-source pipeline (`wrappedPush`) |
| `env.elb`  | **`walker *` commands**         | The raw collector adapter                           |

Emit events through `env.push`. That is what makes the source's own `next`/`before` chains, `mapping` (including `ignore` and `consent`), `cache`, `state`, per-source `Ingest`, and `collector.status.sources.<id>` apply. A source that emits events through `env.elb` bypasses all of it, so those config fields become silently dead: accepted, never executed.

Commands must keep using `env.elb`. `env.push` takes `WalkerOS.DeepPartialEvent` only, so a command string has no meaning there.

Both exits, plus `env.command`, `env.logger` and `env.sources`, are runtime-owned: the collector writes these five capabilities into the env LAST, so an author value for any of them is ignored, uniformly. `env` stays the author's bag for platform and vendor dependencies only. To end the pipeline somewhere else, declare `terminus` on the source registration (a sibling of `code`); it is explicit and TOTAL, and stored flow configs cannot carry it (validation rejects unknown source keys).

```
// Event → pipeline

await env.push({ name: 'page view', data: { title } });



// Command → raw collector

await env.elb('walker consent', { marketing: true });
```

Call `env.push` with the event alone. The collector supplies `id`, `ingest`, `mapping` and `preChain` itself; passing your own `options` is for server sources threading a per-request scope via `context.withScope`.

Two consequences worth knowing when testing a source:

* The pipeline's end is captured **when the source is constructed**. Replacing `collector.push` after `startFlow` resolves is invisible to pipeline pushes, so assert on a spy **destination** instead of a `collector.push` mock. For a deterministic raw capture at the source→collector boundary, declare `terminus` on the registration: it receives the event exactly as the source emitted it, and the ENTIRE pipeline is skipped (no `before`/`next` chains, no `cache`, no `state`, no `mapping`, no minted span id, no ingest, no `respond`, no `status.sources` counting, no Observe records).
* A `next`/`before` chain adds `await` hops between the emit and delivery. For a fire-and-forget emit, settle microtasks before asserting.

## Transformer Wiring[​](#transformer-wiring "Direct link to Transformer Wiring")

Sources can wire to pre-collector transformer chains via the `next` property:

```
sources: {

  browser: {

    code: sourceBrowser,

    next: 'validate'  // First transformer to run after this source

  }

}
```

The transformer chain runs before events reach the collector. See [understanding-transformers](https://www.walkeros.io/skills/walkeros-understanding-transformers.md) for chain details.

## Per-Scope Context (server sources)[​](#per-scope-context-server-sources "Direct link to Per-Scope Context (server sources)")

A single source factory instance handles many concurrent invocations: Express processes overlapping requests, Lambda reuses one handler across calls, queue consumers loop over messages. Each logical unit of work is a **scope**. Server sources MUST wrap each invocation with `context.withScope(rawScope, respond, body)`:

A server source's job is to adapt its platform's request into the shared `Source.Scope` shape, so everything downstream is source-independent. Build the scope first, then pass it:

```
const push = async (req, res) => {

  const respond = createRespond((options) => {

    /* wire options into res */

  });



  const scope = buildScope(req); // normalize: method, url, path, query,

  // headers, body, ip, raw



  await context.withScope(scope, respond, async (env) => {

    for (const event of toEventList(scope.body)) {

      await env.push(event);

    }

  });

};
```

`normalizeHeaders`, `normalizeQuery` and `normalizeBody` from `@walkeros/core` do the per-field work; `toEventList` and `isBatchBody` decide how many events the body carries. Never pass the platform object straight to `withScope`: the parameter is typed `Source.Scope | undefined`, and a `Headers` instance or a WHATWG `Request` cannot be descended by a mapping path at all.

Inside `body`, `env.push` carries that scope's `ingest` (extracted from the scope via `config.ingest` mapping) and `respond` end to end through the pipeline. Concurrent scopes never share ingest or respond.

**`ingest` is request scoped, not event scoped.** With N events in one scope, a transformer writing per-event annotations into `ingest` records only the last.

**Browser sources skip `withScope`.** A browser tab is a single logical scope for its lifetime; calling `env.push` directly is correct.

## Response Delegation (env.respond)[​](#response-delegation-envrespond "Direct link to Response Delegation (env.respond)")

When a server source passes a `respond` to `withScope`, every transformer and destination in the pipeline can call `env.respond?.({ body, status?, headers? })` to customize the HTTP response. First call wins (`createRespond` is idempotent), so the source's default response is a no-op if a step already responded.

See `@walkeros/server-source-express` for the reference implementation.

## createTrigger Pattern[​](#createtrigger-pattern "Direct link to createTrigger Pattern")

Every source exports a `createTrigger` factory from its examples (`dev` entry) that follows the unified `Trigger.CreateFn` interface:

```
type CreateFn<TContent, TResult> = (

  config: Collector.InitConfig,

  options?: unknown,

) => Promise<Trigger.Instance<TContent, TResult>>;
```

`createTrigger` simulates real-world invocations from the outside — full blackbox, no source instance access. Each package implements it differently:

| Source  | Content        | Trigger type       | Mechanism               |
| ------- | -------------- | ------------------ | ----------------------- |
| Browser | HTML string    | `click`, `load`... | DOM injection + events  |
| Express | HTTP req shape | `POST`, `GET`      | Real `fetch()` requests |

The trigger lazily calls `startFlow(config)` on first invocation. Tests capture events via spy destinations. See [using-step-examples](https://www.walkeros.io/skills/walkeros-using-step-examples.md) for testing patterns.

## Setup (optional)[​](#setup-optional "Direct link to Setup (optional)")

Sources can implement an optional `setup()` lifecycle to provision external resources, for example registering a webhook with a third-party provider, creating a Pub/Sub subscription, or pre-allocating queue resources. Setup is **never** invoked by the runtime, push, init, or deploy. It runs only when an operator explicitly types `walkeros setup source.<name>`.

The signature is `(ctx: LifecycleContext<Config<T>, Env<T>>) => Promise<unknown>`, where `LifecycleContext` carries `{ id, config, env, logger }`. Idempotency is the package's responsibility: the framework adds no opinion. Use `resolveSetup(ctx.config.setup, DEFAULTS)` from `@walkeros/core` to normalize the `boolean | object` shape into a concrete options object.

See [walkeros-create-source](https://www.walkeros.io/skills/walkeros-create-source.md), [walkeros-understanding-destinations](https://www.walkeros.io/skills/walkeros-understanding-destinations.md), [walkeros-understanding-stores](https://www.walkeros.io/skills/walkeros-understanding-stores.md), and the `walkeros setup` CLI documentation for the authoring template and operator workflow.

## Related Skills[​](#related-skills "Direct link to Related Skills")

* [walkeros-understanding-flow](https://www.walkeros.io/skills/walkeros-understanding-flow.md) - How sources fit in architecture
* [walkeros-understanding-events](https://www.walkeros.io/skills/walkeros-understanding-events.md) - Events that sources emit
* [walkeros-create-source](https://www.walkeros.io/skills/walkeros-create-source.md) - Create new source

**Source Files:**

* [packages/core/src/types/source.ts](https://github.com/elbwalker/walkerOS/blob/main/packages/core/src/types/source.ts) - Interface

**Package READMEs:**

* [packages/web/sources/browser/README.md](https://github.com/elbwalker/walkerOS/blob/main/packages/web/sources/browser/README.md) - Browser source
* [packages/web/sources/dataLayer/README.md](https://github.com/elbwalker/walkerOS/blob/main/packages/web/sources/dataLayer/README.md) - DataLayer source

**Documentation:**

* [Website: Sources](https://www.walkeros.io/docs/sources.md) - Overview
* [Website: Browser Source](https://www.walkeros.io/docs/sources/web/browser.md) - Browser docs
* [Website: Create Your Own](https://www.walkeros.io/docs/sources/create-your-own.md) - Guide