Cel Programs

cel-programs

When to use

Use this skill when tasks include:

• creating or editing cel.yml.hbs agent stream templates
• configuring data stream manifests for the cel input type
• writing CEL programs with pagination, cursor management, or authentication
• testing or debugging a CEL program locally with mito
• setting up system tests with mock APIs for CEL-based data streams
• prototyping a new CEL-based data stream's collection logic
• any CEL or mito question, regardless of context

When not to use

Do not use this skill as the primary guide for:

• ingest pipeline processor design (ingest-pipelines)
• ECS field mapping (ecs-field-mappings)
• package scaffolding (create-integration)
• system test execution with the Elastic stack (integration-testing → references/system-testing.md)

Mandatory workflow — mock → mito → template

This is not a suggestion. Every CEL program MUST be developed in this order. The subagent must not write cel.yml.hbs until the CEL program has been validated with mito against a running mock. Skipping steps or reordering causes failures that are hard to debug.

Do NOT write more than ~10–15 new lines of CEL before running mito. Build the program incrementally in phases (skeleton → error handling → event mapping → pagination → cursor guard), validating with mito after each phase. Writing a large program in one shot leads to cascading compilation errors that are extremely hard to debug. Follow the phased approach in references/cel-incremental-build.md.

Step 3 detail — translating template vars to mito state: When the future cel.yml.hbs will have a state: block like api_key: {{api_key}} and batch_size: {{batch_size}}, the state.json for mito testing uses the same key names with literal test values:

{
  "url": "http://localhost:8090",
  "api_key": "test-key",
  "batch_size": 50,
  "initial_interval": "24h"
}

This mirrors the runtime state the CEL input would provide. Add cursor to test subsequent-run behavior.

For the full mock-first workflow details, CLI flags, execution model, and quality standards: load references/mito-reference.md.

---

cel.yml.hbs template anatomy

The cel.yml.hbs file at data_stream/<stream>/agent/stream/cel.yml.hbs is a Handlebars template that renders the final CEL input configuration. It has these sections in order:

interval: {{interval}}
resource.tracer:
  enabled: {{enable_request_tracer}}
  filename: "../../logs/cel/http-request-trace-*.ndjson"
  maxbackups: 5
{{#if proxy_url}}
resource.proxy_url: {{proxy_url}}
{{/if}}
{{#if ssl}}
resource.ssl: {{ssl}}
{{/if}}
{{#if http_client_timeout}}
resource.timeout: {{http_client_timeout}}
{{/if}}
resource.url: <constructed from vars>
state:
  <credentials and pagination config from vars>
redact:
  fields:
    - <sensitive state keys>
max_executions: <number, for heavy pagination>
program: |
  <CEL expression>
tags:
{{#if preserve_original_event}}
  - preserve_original_event
{{/if}}
{{#each tags as |tag|}}
  - {{tag}}
{{/each}}
{{#contains "forwarded" tags}}
publisher_pipeline.disable_host: true
{{/contains}}
{{#if processors}}
processors:
{{processors}}
{{/if}}

Handlebars patterns

Key template fields

• resource.url — base URL, often constructed from multiple vars (e.g., {{url}}/api/v1/endpoint)
• resource.headers (ga 8.18.1) — static headers the same for every request (Content-Type, Accept, API version headers). Set here rather than in-program when headers never vary. Applied before auth headers.
• state: — block where manifest vars are injected as CEL state; credentials and pagination settings go here
• redact.fields — list state keys containing secrets to redact from debug logs
• max_executions — override default 1000 for integrations with heavy pagination (e.g., 5000)
• program: | — the CEL expression; must be a YAML literal block scalar

Do NOT set data_stream.dataset in integration packages

Integration packages (type: integration) must never include data_stream.dataset in cel.yml.hbs or define a data_stream.dataset manifest var. The framework automatically routes documents to the correct data stream. Setting data_stream.dataset overrides this routing and causes documents to land in the wrong index — typically resulting in "0 hits" during system tests.

Only input-type packages (type: input) use data_stream.dataset because they have no predefined data streams.

Data stream manifest configuration

The data stream manifest.yml defines the CEL input stream and its variables.

Standard vars every CEL stream should include

Auth-specific vars depend on the API (API key, OAuth client_id/secret/token_url, bearer token, etc.).

Declare enable_request_tracer in the data stream manifest, not at the input level. Input-level tracing enables logging for all data streams in the policy.

Package-level vs data-stream-level vars

• Package-level vars in the root manifest.yml under policy_templates[].inputs[].vars: shared across streams (e.g., url, auth credentials)
• Data-stream-level vars in data_stream/<stream>/manifest.yml under streams[].vars: stream-specific (e.g., interval, batch_size, initial_interval)

Scope of the CEL program

The CEL program's responsibility is data collection only:

1. Fetch data from the API endpoint(s) 2. Handle pagination — walk through all pages within a single polling cycle 3. Manage cursor state — store timestamps or page tokens in cursor so the next polling interval resumes where the last one left off, avoiding re-collection of already-fetched events 4. Emit raw events — output {"message": e.encode_json()} for each record

The CEL program does not handle:

• Elasticsearch-level deduplication — if overlapping time windows cause a few duplicate events to be collected, that is acceptable. The ingest pipeline or Elasticsearch _id routing handles dedup at index time, not the CEL program.
• Field mapping or transformation — the ingest pipeline handles parsing, ECS mapping, and enrichment.
• Filtering by content — unless the API supports server-side filtering parameters, do not filter events in the CEL program. Emit everything and let the pipeline decide.

Do not search the codebase for _id, document_id, or deduplication patterns. These are not CEL concerns.

CEL program structure patterns

Pagination strategy selection

Cursor timestamp selection — use the last record's timestamp when the API sorts ascending; first when descending; max() with a regression guard when sort order is not guaranteed.

Full code, package references, and YAML snippets for each pattern: references/cel-pagination-patterns.md.

Authentication patterns

Three strategies: header (credentials in state:, passed via Header map), query parameter (credentials appended to URL via .format_query()), signed query (HMAC signature computed in CEL). Config-level auth.oauth2/auth.digest/auth.aws applies to all requests including .do_request(); auth.basic/auth.token applies only to direct calls (get(), post()). Prefer input-level auth over in-program token fetching.

For full code examples, optional-header syntax, and config-level auth scope details: load references/cel-auth-patterns.md.

State management rules

1. `state.url` is populated from resource.url config; must be preserved in output or hardcoded 2. `cursor` is the only state persisted across input restarts; store pagination positions and timestamps here 3. `events` array is removed after each evaluation; never rely on it in subsequent runs 4. `want_more: true` triggers immediate re-evaluation, but only if events is non-empty. Pagination continuation guardrail: when a next-page cursor/token exists, always set want_more: true regardless of how many events were collected on the current page. Tying want_more to size(events) > 0 stalls pagination silently — the next cursor is valid, and an empty events array is safe to emit. The correct pattern is "want_more": next_cursor != "". 5. All other state keys are retained within a session but lost on restart — use state.with() to propagate them automatically 6. Numbers are serialized as floats in state JSON; cast with int() when using as integers 7. Optional access with state.?cursor.last_timestamp.orValue(default) prevents errors when cursor is absent 8. Secrets — every sensitive field in state must have a corresponding redact entry. state.secret is always redacted automatically. When secret_state (ga 9.4.0) is available, prefer it. 9. Cursor updates require a published event — the input only persists cursor updates when at least one event is published. If a program updates the cursor but returns zero events, the cursor change is lost. 10. Do not duplicate request/response handling across branches — when an initialization branch (cursor creation, subscription, token exchange) and a steady-state branch both need the same fetch logic, consolidate it. Two approaches: split the init into a separate evaluation via want_more: true (Technique 6 Variant A), or use an intermediate result map to unify the branches within one evaluation (Variant B). Both are valid — see references/cel-code-style.md Technique 6 and the init-then-steady-state pattern in references/cel-pagination-patterns.md. 11. Nesting depth — .as() chain depth must not exceed 5 levels on any execution path. HTTP programs must target 2 levels inside state.with() (resp + body). Cursor defaults, window bounds, and page tokens must be extracted as pre-bindings before state.with(). Single-use values such as int(state.batch_size) must be inlined at the call site, not wrapped in .as(). Load references/cel-code-style.md for flattening techniques and before/after examples.

Map merge and field removal

with(), with_replace(), with_update(), and drop() are general-purpose map operations — they work on any map, not just state or cursor. with() does a shallow merge: nested objects are replaced entirely. This makes it a tool for cursor state transitions (omitting a sub-object removes it via clobber) as well as for building request headers, transforming response data, and constructing intermediate maps. Full semantics and examples: references/cel-code-style.md.

Event output format

Events must contain ONLY `"message"` — {"message": e.encode_json()}. Do not set @timestamp or any other field; the framework adds @timestamp, and duplicates cause silent document rejection in ES 9.x. See references/cel-idioms.md for correct/incorrect examples.

Response handling

`resp.Body.decode_json()` — the bytes(resp.Body) wrapper was required in older runtime versions but is no longer needed. Use resp.Body.decode_json() directly.

Error handling

Every CEL program must handle HTTP errors. Two forms:

• Single-object error (retry): "events": {"error": {...}} — logs at ERROR, sets degraded status, deletes the cursor so the next evaluation retries. Use when data was not collected.
• Array error (advance): "events": [{"error": {...}}] — cursor is updated. Use when the program should advance past the error. Requires a terminate processor in the ingest pipeline (ES 8.16.0+).

Error message format: "METHOD path: body-or-status". Code examples: references/cel-idioms.md.

Placeholder events

When advancing the cursor with no real events, emit a placeholder ([{"retry": true}]) and add a - drop_event.when.equals.retry: true entry in the processors: section so it is discarded before indexing. Full pattern and alternatives: references/cel-idioms.md.

Rate limiting and retry

Do NOT implement rate limiting or retry logic in the CEL program. No rate_limit() calls, no "rate_limit" state propagation, no 429-specific branches, no retry loops. These add excessive nesting and complexity for marginal benefit.

When an API has a documented rate limit, use config-only YAML settings in cel.yml.hbs:

resource.rate_limit.limit: 10   # max requests per second
resource.rate_limit.burst: 5    # max burst above sustained rate

When custom retry behavior is needed, use config-only YAML settings:

resource.retry.max_attempts: 5    # default: 5
resource.retry.wait_min: 1s       # default: 1s
resource.retry.wait_max: 60s      # default: 60s

The input framework enforces both transparently. See references/cel-rate-limiting.md for guidance on when to add these settings.

Type safety

Avoid `dyn()` — defeats type checking. Rarely needed in practice.

All numbers are float64 — the CEL input transmits all numbers as float64. Numbers >=1e7 render in scientific notation in Elasticsearch. Convert intended-integer fields to strings in the CEL program or via ingest pipeline. Safe integer range: [-(2^53 - 1), 2^53 - 1].

Debugging aids

• `debug(tag, value)` — logs to cel_debug at DEBUG level.
• `try(expr)` / `is_error(value)` — structured error handling without crashing the program.
• `failure_dump` (ga 8.18.0) — full evaluation state dump on failure. Note: dumps may contain secrets.
• `remaining_executions` (ga 9.2) — how many evaluations remain in the max_executions budget.

---

Mito CLI

Mito (github.com/elastic/mito) is the local CEL evaluation CLI. A CEL program that has not been tested with mito is not acceptable. Follow the mandatory workflow at the top of this skill: mock → mito → template. Do NOT write cel.yml.hbs until the program passes mito validation.

For installation, CLI flags, input state structure, execution model, the full mock-first workflow steps, mito→integration mapping, and quality standards: load references/mito-reference.md.

---

Data anonymization

All data committed to the repository must be fully anonymized. This applies to default values in manifest vars (use https://api.example.com), example values in CEL state, mock API responses, pipeline test fixtures from CEL output, and sample payloads captured during mito prototyping.

Refer to the anonymize-logs skill for the full anonymization policy and placeholder conventions.

Handoff to other skills

• integration-testing → references/system-testing.md to run system tests (the mock API is already in place from CEL development)
• integration-testing → references/pipeline-testing.md to validate ingest pipeline behavior on CEL-produced events
• create-integration skill for overall package layout

Reference files

IMPORTANT: These reference files contain the actual working code examples and patterns. The summaries above are not sufficient to write correct CEL programs — you MUST load the relevant references before writing code.

Always load these five when building a CEL program — in this order (mock/mito before templates):

Load these based on the task:

See also: CEL input docs · Mito lib docs · Mito repo · CEL language spec

CEL program builder subagent guidance

Operating manual for a subagent building or modifying CEL programs on behalf of the create-integration or maintain-integration orchestrator.

This subagent is an orchestrator for CEL integration development. Its primary responsibilities are mock design, mock validation, template wrapping, manifest configuration, and system test setup. For the CEL expression itself, it delegates to the cel-expression-builder subagent (see references/expression-builder-subagent-guidance.md), which translates test-api.py into a validated .cel file.

The orchestrator dispatches you with a brief task prompt that points you at this file by path. Read this entire file end-to-end before doing any other work, then read the skills and reference files listed in the "First steps" section below — they are mandatory. The orchestrator does not paste this file's content into your task prompt (to avoid burning context twice); you load it here in your own fresh context.

The orchestrator's task prompt tells you what to build or fix, which package and data stream to work on, the API details and sample data, and any constraints. This file tells you how to operate as a CEL program builder subagent. Follow both.

Scope

Your responsibility is strictly limited to:

• Building the cel.yml.hbs template (the CEL program and its Handlebars

wrapper) at data_stream/<stream>/agent/stream/cel.yml.hbs

• Configuring the data stream manifest vars for CEL input (data_stream/<stream>/manifest.yml)

and any package-level vars in the root manifest.yml that the template consumes

• Setting up system test mock API files (_dev/deploy/docker/docker-compose.yml,

_dev/deploy/docker/files/config-<stream>.yml) and the test config (data_stream/<stream>/_dev/test/system/test-default-config.yml)

• Defining initial field mappings (fields/fields.yml) for the raw API

response fields the CEL program emits

You do NOT:

• Create or modify pipeline test fixtures (_dev/test/pipeline/) — the

pipeline builder owns these (see ingest-pipelines/references/builder-subagent-guidance.md)

• Create or modify the ingest pipeline (elasticsearch/ingest_pipeline/) or

fields/ecs.yml — the pipeline builder handles this

• Create or modify sample_event.json — generated only by

elastic-package test system --generate, run by the system-test subagent (see integration-testing/references/builder-system-test-subagent-guidance.md)

• Run system tests (elastic-package test system) — the system-test subagent

runs them after the pipeline builder completes pipeline work for the data stream

• Implement document deduplication logic — overlapping windows producing a

small number of duplicates is acceptable; dedup is not a CEL concern

If the orchestrator's prompt asks for pipeline work, sample event handling, or system test execution rather than CEL program development, stop and report that the wrong subagent or guidance file was invoked.

Skill authority

The rules and patterns in the cel-programs skill and its reference files are the authoritative source of truth. When examining reference integrations in the official elastic/integrations repository for patterns (authentication, pagination structures, mock configs), many existing integrations contain legacy patterns that predate current standards — always follow the skills over patterns observed in other integrations.

First steps — read the skills and their references

Before doing any work, read these skill files and the specific reference files listed to load the rules, patterns, and working code examples you must follow. Reading only the SKILL.md files is not sufficient — the reference files contain the working code examples and patterns you need.

1. `cel-programs` skill (SKILL.md) — start with the Mandatory workflow table at the top, then template anatomy, manifest configuration, state management rules, error handling, and event output format. Then read these references in this order (mock/mito references before template examples — order matters):

• `references/cel-system-tests.md` — MUST READ FIRST: mock API setup

with elastic/stream, docker-compose config, rule format, variable- capture patterns, hit_count calculation, debugging 0-hits failures

• `references/cel-incremental-build.md` — MUST READ SECOND: the

phased build ladder you MUST follow (skeleton → error handling → events → pagination → cursor), syntax anti-patterns that cause compilation failures (bytes(), parse_time(), tuples, unbalanced parens), and debugging guidance. You will build the CEL program in these phases — do NOT write the full program before validating a skeleton.

• `references/mito-reference.md` — MUST READ THIRD: mito CLI flags,

input state structure, mock-first workflow steps, translating template vars to state.json, extension library quick-reference, syntax pitfalls

• `references/cel-template-examples.md` — MUST READ FOURTH (these are

FINAL output — do not write templates until mito validation passes): complete working cel.yml.hbs examples with manifest configs

• `references/cel-code-style.md` — MUST READ FIFTH: nesting discipline

rules, the 2-level HTTP core structure, flattening techniques with before/after examples — read this before writing any multi-line CEL so structure is correct from the start

• `references/cel-pagination-patterns.md` — read when writing

pagination logic: all 6 pattern types with code

• `references/cel-auth-patterns.md` — read when implementing

authentication

• `references/cel-rate-limiting.md` — config-only rate limiting and

retry policy (do not use rate_limit() in CEL programs)

• `references/cel-idioms.md` — quick-reference for common idioms, HTTP

patterns, structure conventions (anti-patterns table at the top)

• `references/cel-polymorphic-patterns.md` — read when choosing between

config auth, lib functions, and manual CEL: version-tagged tables

2. `integration-testing` skill (SKILL.md) — then read `references/system-testing.md` fully, focusing on the CEL system test section, mock API setup, and required test config fields (wait_for_data_timeout: 1m, service, vars, assert.hit_count).

3. `elastic-package-cli` skill — elastic-package format / lint / check and the build commands you will run.

4. `anonymize-logs` skill — placeholder conventions for any data committed (mock responses, default manifest values, sample state values).

Read all skills and their MUST READ references before writing any CEL code. Do not rely on the SKILL.md summaries — the reference files contain the working code examples and patterns you must follow.

Mandatory workflow — do NOT skip or reorder steps

The core principle: build the system test mock first, then develop the CEL program against it with mito, then — and only then — write the template. This eliminates custom Python mock servers and keeps the mock, the CEL program, and the system test infrastructure consistent from the start.

CRITICAL — do NOT write `cel.yml.hbs` first. Even if you have template examples, API documentation, and a clear picture of the final template, you MUST follow this order:

1. Investigate the API via test-api.py (the ground truth) 2. Derive the system test mock from test-api.py's request/response flow (with the full two-round flow described in step 3c) 3. Start the mock locally 4. Validate the mock with test-api.py (hard gate — do not proceed until it passes) 5. Verify mock completeness gate (2+ pages, terminal page, second-round cursor resume, optional regression guard) 6. Build the CEL expression (preferred: delegate to cel-expression-builder with test-api.py as the translation source) 7. ONLY THEN write cel.yml.hbs by wrapping the validated program in Handlebars 8. Run celfmt -s -agent, configure manifest vars, define field mappings, validate

Writing the template first and trying to extract a .cel file from Handlebars for mito does not work — Handlebars syntax is not valid CEL. Always develop the raw CEL program first, validate it with mito, then wrap it in the template.

If the orchestrator provides real API credentials, you may additionally test against the live API for highest-confidence validation, but mock-first development is always the primary path — it is faster, reproducible, and does not depend on API availability or rate limits.

Step 1 — investigate the API via test-api.py

Start with the research `test-api.py` script. This is always present for CEL integrations and is the ground truth — it has been tested against the real API. Read its collection function (run_collection() or equivalent) to identify:

• Base URL and endpoint paths
• Authentication method (header auth, query parameter auth, signed query

params, OAuth2 client credentials, OAuth2 token refresh)

• Pagination pattern — which loop structure and termination condition does

the Python script use? Map it to the patterns in the cel-programs skill / references/cel-pagination-patterns.md

• Response JSON structure (where the event array lives, total count fields,

next-page indicators) — visible in the script's response navigation

• Time-range filtering parameters
• Batch size constraints
• Error handling branches (status codes, missing fields, malformed responses)

The research brief provides supplementary context (field meanings, edge cases not exercised by the script). If the API documentation is incomplete and credentials are available, use curl or python to make exploratory requests against the real API, but the Python script remains the primary specification.

Step 2 — derive the system test mock from test-api.py

Before writing any CEL code, build the mock HTTP API the system test will use. The mock is derived from test-api.py's request/response flow, not independently designed from the research brief. The Python script defines which endpoints are called, what request shapes are sent, and what response structures are expected — the mock replays this interaction with anonymised data. If a trace.json exists from a real API run, use it as a reference for response structure and field presence.

Use the elastic/stream http-server — do NOT write a custom Python mock.

1. Define the mock service in _dev/deploy/docker/docker-compose.yml using docker.elastic.co/observability/stream:v0.20.0. 2. Create a rule-based config file at _dev/deploy/docker/files/config-<stream>.yml that models the complete API request flow the CEL program will make. Follow the patterns and rule examples in references/cel-system-tests.md — mock API flow design, variable capture ({varName:.*}), OAuth token endpoints, GraphQL request-body matching. A poorly designed mock is the single largest cause of system test failures. 3. Write the test config at data_stream/<stream>/_dev/test/system/test-default-config.yml with:

• wait_for_data_timeout: 1m (required — caps how long

elastic-package test system waits for data)

• input: cel
• service: <mock-service-name>
• Vars pointing to http://{{Hostname}}:{{Port}} — never a real URL
• `interval: 2s` — short enough that the agent completes pagination,

persists the cursor, and fires a second evaluation cycle that verifies cursor persistence within the test window

• assert.hit_count summing events from both evaluation rounds (the

first pagination round + the second cursor-based round) — see step 3c

Step 3 — start the mock server locally

Start the elastic/stream mock so mito can make requests against it.

Option A — `stream` CLI (preferred, lightweight):

stream http-server --addr=:8090 --config=_dev/deploy/docker/files/config-<stream>.yml &
MOCK_PID=$!

The mock listens on http://localhost:8090. Use this URL in your mito input state.

Option B — docker-compose: start only the mock service from the system test docker-compose:

cd _dev/deploy/docker
docker-compose up -d <service-name>

Find the mapped port with docker-compose port <service-name> 8090 and use http://localhost:<mapped-port> in your mito input state.

Either approach gives mito the exact same mock the system test will use — no divergence between what mito tests and what elastic-package test system tests.

Step 3b — validate the mock with test-api.py (hard gate)

This is a hard gate, not optional. The research test-api.py script is always present for CEL integrations. Run it against the running mock before writing any CEL:

python3 test-api.py --base-url http://localhost:8090 --mock

If the script fails against the mock, assume the mock is wrong unless there is an obvious flaw in the script (wrong endpoint path, missing optional header). Fix the mock rules to match what the script expects (missing query params, wrong response shape, missing auth endpoint, wrong status codes).

Do NOT proceed to step 3c until test-api.py passes against the mock. This gate ensures the mock faithfully models the API interaction that the CEL program will be translated from.

If a trace.json exists from a real API run, compare the mock's responses against it as an additional fidelity check — field presence, response shape, and pagination state transitions should match.

Step 3c — verify mock completeness gate (hard gate)

Before writing any CEL, verify the mock implements the full two-round pagination flow. This is a hard gate, not a suggestion:

• Round 1 — initial fetch with full pagination:
• Page 1: returns events + a non-terminal pagination signal

(hasNextPage: true, non-empty next cursor, offset < total)

• Page 2: returns events + a non-terminal signal (at minimum 2 pages

of results)

• Terminal page: returns events (or empty) + the terminal pagination

signal (hasNextPage: false, empty next cursor, offset >= total) that stops want_more.

• Round 2 — cursor-persisted resume (interval-driven):

After the short interval: 2s, the agent fires a new cycle using the persisted cursor bookmark (e.g. last_from). The mock must have a rule matching this second-round request pattern; this round returns at least one additional event so cursor persistence is observable in hit_count.

• Optional auth route: include an auth endpoint rule when authentication

is not a static API key (OAuth, token refresh).

• Regression guard (recommended): add a catch-all rule that fires if the

cursor was incorrectly cleared (e.g. the program re-requests from now - initial_interval instead of using the persisted bookmark). The rule returns extra synthetic events that cause hit_count to exceed the expected value, failing the test if the cursor regresses.

Sum events from all pages across both rounds — that sum is the assert.hit_count value in test-default-config.yml.

Do NOT proceed to step 4 until the mock implements all of the above.

Step 4 — build the CEL expression

Every CEL program must be developed and validated through mito. This is not optional. The mock from step 3 is already running.

Preferred path — delegate to cel-expression-builder

When the orchestrator provided a test-api.py script (which is always the case for new CEL integrations), launch a cel-expression-builder subagent with:

• The test-api.py file content
• A state.json with literal test values (url pointing at the running mock,

credentials, batch_size, initial_interval)

• The mock URL
• The research brief (if available)

Point the subagent at references/expression-builder-subagent-guidance.md (relative to the cel-programs skill) as its operating manual and instruct it to read that file (plus the skill SKILL.md it lists) end-to-end before doing any other work. Do NOT read the guidance file yourself or paste its contents into the task prompt — pass only the path plus the task-specific context. The expression builder returns a validated .cel file and a taxonomy classification.

Step 4b — structured review (when complexity warrants it)

After receiving the .cel file and classification from the expression builder, run ceplx -diag -json program.cel to get complexity metrics.

Skip the review if cognitive complexity is below the class p50 (from references/cel-complexity-baselines.md) and below 40.

Otherwise, launch a cel-expression-reviewer subagent with:

• The generated .cel file
• The taxonomy classification
• The ceplx -diag -json output
• The test-api.py file content
• The research brief

Point the subagent at references/reviewer-subagent-guidance.md (relative to the cel-programs skill) as its operating manual and instruct it to read that file end-to-end before doing any other work. Do NOT paste the guidance file's contents into the task prompt — pass only the path plus the task-specific context.

If the reviewer returns revise, pass the challenges back to the expression builder (resume the same subagent) and ask it to address each challenge — either justify the current approach or produce a revised .cel file.

If a revision is produced, re-run ceplx and compare. Select the version with lower complexity unless the revision drops fidelity.

Proceed to step 5 with the final .cel file.

Direct path — build expression yourself

If the orchestrator did not provide test-api.py, or if you are fixing an existing CEL program rather than building one from scratch, build the expression directly following the incremental approach below.

Build incrementally — never write the full program at once. Even if you "know" what the final program looks like, follow the phased ladder in references/cel-incremental-build.md. Each phase is a hard gate; do not add the next phase until the current one runs cleanly under mito.

If mito reports a compilation error, do NOT rewrite the program from scratch. Revert to the last working phase and re-add changes incrementally. Do NOT use Python, sed, or bash scripts to modify .cel files — use the editor's StrReplace/Write tools only.

Nesting discipline applies at every phase. Before writing any multi-line CEL, read references/cel-code-style.md. Target the 2-level HTTP core inside state.with() (resp + body); extract cursor defaults, window math, page tokens, and URL construction as pre-bindings before state.with(); inline single-use values such as int(state.batch_size). The hard cap is 5 .as() levels on any execution path. If at any phase nesting exceeds this, STOP and refactor using the flattening techniques before continuing — do NOT defer.

You can also validate and simplify the standalone .cel syntax with celfmt during this step — it works on plain .cel files too:

celfmt -s -i program.cel -o /dev/null

If celfmt hangs on a standalone .cel file (a known issue in some environments), skip it during prototyping and rely on celfmt -s -agent against cel.yml.hbs once the template exists in step 6.

For complex multi-phase programs (e.g. subscribe-then-fetch), test each phase separately by adjusting state.json or the mock rules so mito exercises one branch at a time before integrating them.

If the orchestrator provided real API credentials, you may additionally point a copy of the program at the live API for highest-confidence validation. Mock-first development remains the primary path.

Step 5 — write cel.yml.hbs (only after mito validation passes)

Do not reach this step until step 4 is complete and the program works in mito. Embed the mito-validated CEL program into data_stream/<stream>/agent/stream/cel.yml.hbs following the template anatomy in the cel-programs skill (and the worked examples in references/cel-template-examples.md):

• interval: {{interval}}, resource block (tracer, proxy, ssl,

timeout) using standard Handlebars conditionals

• resource.url constructed from template vars
• resource.headers (ga 8.18.1) for any static headers identical across

every request (Accept, Content-Type, API version headers) — prefer this over per-request header construction in CEL

• state: block injecting credentials and pagination config from manifest

vars

• redact.fields listing all secret state keys
• max_executions if heavy pagination is expected (override the default

1000)

• program: | containing the validated CEL expression as a YAML literal

block scalar — every line indented exactly 2 spaces relative to program:

• Tags, publisher_pipeline, and processors blocks using the standard

patterns

Write the full program with StrReplace/Write rather than hand-pasting and re-indenting — transcription errors (extra parentheses, wrong indentation levels) are easy to introduce in deep .as() nesting. Then use celfmt -s -agent (step 6) to normalize indentation and verify the result.

Step 6 — format and simplify with celfmt -s -agent

After writing the template, run celfmt -s -agent to validate, simplify, and format the embedded CEL. The -s flag inlines single-use .as() bindings, eliminates boolean comparisons (x == true → x, x == false → !x), and rewrites has(x.f) ? x.f : d to x.?f.orValue(d). The -agent flag tells celfmt the input is an agent template (YAML with embedded CEL in the program: block).

WARNING: `-o` overwrites the output file unconditionally — even on syntax errors. Always validate first before writing back:

cd data_stream/<stream>/agent/stream

# Step 1: validate only — /dev/null protects the source if there are errors.
celfmt -s -agent -i cel.yml.hbs -o /dev/null

# Step 2: only if step 1 succeeds (exit code 0), apply formatting in place.
celfmt -s -agent -i cel.yml.hbs -o cel.yml.hbs

Or as a single safe one-liner:

celfmt -s -agent -i cel.yml.hbs -o /dev/null && celfmt -s -agent -i cel.yml.hbs -o cel.yml.hbs

If validation fails (non-zero exit, diagnostics on stderr), fix the source and re-run validation. Do not proceed until validation passes cleanly.

Step 7 — configure the data stream manifest

Edit data_stream/<stream>/manifest.yml:

• Set input: cel and template_path: cel.yml.hbs
• Define stream-level vars (interval, initial_interval, batch_size,

and any stream-specific settings)

• Ensure package-level vars in the root manifest.yml cover shared settings

(url, auth credentials)

• Include the standard CEL stream vars from the cel-programs skill

(enable_request_tracer, http_client_timeout, proxy_url, ssl, tags, preserve_original_event, processors)

• **Strip every var the scaffold added that is not consumed by

cel.yml.hbs** — do not leave the verbose generic scaffold as-is. The CEL scaffold generates 300+ lines of generic vars; most of them are dead weight in a real integration.

• Audit package-level and stream-level manifests together so shared vars and

stream vars match actual template usage

Step 8 — define initial field mappings

• Create fields/fields.yml with custom integration-specific fields based

on the API response structure discovered during mito prototyping. These are the raw fields before pipeline processing.

• Do not create fields/ecs.yml — that is the pipeline builder's

responsibility, and ECS root fields are provided automatically by ecs@mappings (8.13+).

• Verify fields/base-fields.yml exists with the standard data stream

fields (all six entries use external: ecs; override type/value only for event.module and event.dataset).

Step 9 — validate (do NOT run system tests)

Run format, lint, and check from the package directory:

elastic-package format
elastic-package lint
elastic-package check

Fix any issues. Do not run `elastic-package test system` — the system-test subagent runs system tests after the pipeline builder completes pipeline work for this data stream. Do not create or modify sample_event.json — it is generated exclusively by elastic-package test system --generate in that later pass.

If validation surfaces issues unrelated to your CEL work (e.g. pre-existing pipeline errors), note them in your report and move on rather than fixing them yourself.

Note on script tests: the same elastic/stream mock you set up here can be reused (or extended) for txtar-based script tests that cover failure paths and partial failures. Setting them up is out of scope for your role; in your report, note which CEL error paths would be good script-test candidates (which HTTP error codes the program routes to error events, any partial-failure branches). See integration-testing/references/script-testing.md for details.

Stop the local mock when done

# Option A (stream CLI):
kill $MOCK_PID

# Option B (docker-compose):
cd _dev/deploy/docker && docker-compose down

Critical: integration packages vs input-type packages

Never set `data_stream.dataset` in a `cel.yml.hbs` template for an integration package (type: integration in the root manifest.yml). Integration packages have named data streams (e.g., data_stream/event/) and the framework automatically routes documents to the correct index (e.g., logs-<package>.<stream>-default). Overriding data_stream.dataset breaks this routing and causes documents to land in the wrong index — system tests report "0 hits" because they look in the expected data stream.

Only input-type packages (type: input, such as the generic cel, httpjson, and log packages) use a data_stream.dataset var because they have no predefined data streams. Do not copy this pattern from input-type package references into integration packages.

If you encounter a reference integration that includes data_stream:\n dataset: {{data_stream.dataset}}, check whether it is an input-type package (type: input in manifest) before copying.

Data anonymization

All data committed to the repository must be fully anonymized. When prototyping with mito against a real API using real credentials, the raw responses contain real data — none of it may be committed as-is. Replace every identifying value with a synthetic example of the same format before using it in:

• Mock API responses in system test config files
• Default values in manifest vars (use https://api.example.com, never a

real vendor URL)

• CEL state example values and template comments

Use RFC 5737 documentation IP ranges (198.51.100.x, 203.0.113.x), example.com domains, realistic placeholder names (Alice Johnson, Example Corp), and synthetic IDs. Refer to the anonymize-logs skill for the full placeholder convention list.

What to return

When you finish, report:

• Files created or modified (with paths)
• Field files: list each field file created or modified

(fields/fields.yml, fields/base-fields.yml) and what was added or changed, so the pipeline builder knows which definitions are already in place

• API interaction summary: endpoints used, auth method, pagination

pattern selected from the patterns table

• Python script validation: if test-api.py was available, whether it

passed against the mock and any mock fixes you applied

• System test setup: mock config file path, rules added, expected

hit_count, mock response data structure (so the pipeline builder knows what shape to expect). Confirm explicitly that the mock implements: 2+ pages in round 1, terminal page, round 2 cursor resume, regression guard.

• Mito validation: list each phase you ran (skeleton, error handling,

events, pagination, cursor) and confirm each one passed individually against the running mock — not just that the final program works

• CEL program structure: single-request vs paginated, cursor fields used

(and whether you separated page tokens from time bookmarks), re-evaluation logic

• Template configuration: vars defined, redacted fields, any

resource.rate_limit.* or resource.retry.* settings added

• Manifest/template usage audit: list package-level and stream-level

vars kept, and list vars removed as unused because they were not referenced by cel.yml.hbs

• celfmt result: confirm celfmt -s -agent validation passed
• Validation results: from elastic-package format / lint / check
• Any open issues or decisions that need user input

Authentication patterns

Header, query-parameter, and signed-query authentication in CEL programs. Credentials live in the state: block and are passed into requests as shown below.

Header auth

Most APIs authenticate via HTTP headers. Pass credentials in the Header map of the request:

request("GET", state.url).with({
  "Header": {
    "Authorization": ["Bearer " + state.api_token],
  }
}).do_request()

For optional headers (credentials may or may not be configured), use the optional syntax:

request("GET", state.url).with({
  "Header": {
    ?"Authorization": has(state.api_token) ?
      optional.of(["Bearer " + state.api_token]) : optional.none(),
  }
}).do_request()

Query parameter auth

Some APIs authenticate via query parameters in the URL rather than headers. Build the full URL using .format_query() with credentials as query parameters:

(state.url + "/current?" + {
    "access_key": [state.api_key],
    "query": [state.location],
}.format_query()).as(target_url,
  request("GET", target_url).do_request().as(resp, ...)
)

Multiple query params with different credential keys:

request(
  "GET",
  state.url.trim_right("/") + "/feed/nod/?" + {
    "api_username": [state.api_username],
    "api_key": [state.api_key],
    "sessionID": [state.session_id],
  }.format_query()
).with({
  "Header": {"Accept": ["application/x-ndjson"]}
}).do_request()

When using query param auth, redact.fields in the template must still list the secret state keys (e.g., api_key) to prevent them from appearing in debug logs.

Signed query parameter auth

Some APIs require HMAC-signed parameters in the URL. Compute the signature in CEL and include it as a query parameter:

state.url.trim_right("/") + "/ingestion/rules/save_result_set/?" + {
  "AccessID": [state.access_id],
  "Expires": [string(state.expires)],
  "Signature": [(
    [state.access_id, string(state.expires)].join("\n")
    .hmac("sha1", bytes(state.secret_key))
    .base64()
  )],
}.format_query()

Choosing auth strategy

Auth scope at config level

Auth mechanisms at the config level differ in scope — this is critical for choosing request style:

• **auth.digest**, **auth.oauth2**, **auth.file**, **auth.aws** — applied to all requests including .do_request(). No CEL-side auth logic needed.
• **auth.basic, `auth.token` — applied only to direct calls (`get()`, `post()`, `head()`), not** .do_request(). Prefer direct calls with these. If .do_request() is needed (e.g. for custom headers beyond auth), use basic_authentication(user, pass) on the request map instead of manual header construction.

Prefer input-level auth over in-program token fetching. Use in-program auth only when the API requires non-standard authentication (session cookies, HMAC signing, custom token endpoints not supported by auth.oauth2).

CEL code style and nesting discipline

The core structure rule

Every HTTP-based CEL program has the same fundamental shape. Keep all computation outside state.with() or inline at the call site. Inside state.with(), the depth target is 2 .as() levels: resp and body.

[pre-bindings: cursor defaults, window math, URL construction]
state.with(
  request(...).do_request().as(resp,        ← level 1
    resp.StatusCode == 200 ?
      resp.Body.decode_json().as(body, {    ← level 2
        ...result map...
      })
    : { ...error... }
  )
)

Hard cap: `.as()` depth must never exceed 5 levels on any execution path. If you count more than 5, stop and refactor using the techniques below. HTTP programs should target 2 levels inside state.with().

---

Automated simplification with celfmt -s

celfmt -s applies three rewrites that enforce several of the style rules below automatically:

1. Inline single-use `.as()` bindings — removes .as(name, ...) when name is used zero or one times in the body, replacing the binding with a direct substitution. This enforces the "do not bind single-use values" rule. 2. Eliminate boolean comparisons — rewrites x == true to x and x == false to !x. This enforces the "do not compare booleans" rule. 3. Rewrite `has()` ternaries — rewrites has(x.f) ? x.f : d (and the negated !has(x.f) ? d : x.f) to x.?f.orValue(d).

Always run `celfmt -s` on CEL programs. Pass -s in every celfmt invocation — both during development (on standalone .cel files) and when formatting the final cel.yml.hbs template. The has-ternary rewrite skips cases where it would lose comments. The x != true and x != false forms are not handled because they would change a value-producing expression into a runtime error under dyn.

The == true and == false rewrites are not strictly semantics-preserving under dyn either — if x is not a bool, x == true evaluates to false via heterogeneous equality, while the simplified x evaluates to whatever x is. In practice, nobody writes x == true unless x is boolean, and code that wraps a bool in dyn to exploit the heterogeneous equality behaviour is already wrong. If the rewrite breaks something, the original code should be fixed rather than the simplification reverted.

celfmt -s -agent -i cel.yml.hbs -o /dev/null && celfmt -s -agent -i cel.yml.hbs -o cel.yml.hbs

The simplifier runs before formatting, so its output is always correctly formatted. During development, write code in whatever way is clearest — celfmt -s will clean up redundant bindings and boolean comparisons at the end.

---

General style rules

Do not compare booleans to `true` or `false`. JSON deserialization always produces a CEL bool. Use the value directly or negate it. (celfmt -s fixes == true and == false automatically, but not != true or != false — rewrite those by hand.)

// WRONG
result.ok == false
body.?more_to_read.orValue(false) == true

// CORRECT
!result.ok
body.?more_to_read.orValue(false)

Do not bind single-use values with `.as()`. Every .as() adds a nesting level. If a value is used exactly once, inline it. See Technique 2 below for examples. This is not optional — unnecessary bindings are a common source of excessive nesting in generated programs. (celfmt -s removes these automatically.)

---

Flattening techniques

Technique 1 — Extract cursor and window defaults before state.with()

The most common cause of deep nesting is binding cursor fields inside the HTTP chain. Move them out.

Before (deep — 5 levels inside `state.with`):

state.with(
  state.?cursor.last_timestamp.orValue(
    string(now - duration(state.initial_interval))
  ).as(since,
    int(state.batch_size).as(limit,
      request("GET", state.url.trim_right("/") + "?" + {
        "since": [since],
        "limit": [string(limit)],
      }.format_query()).with({...}).do_request().as(resp,
        resp.StatusCode == 200 ?
          resp.Body.decode_json().as(body, {...})
        : {...}
      )
    )
  )
)

After (flat — 2 levels inside `state.with`):

state.?cursor.last_timestamp.orValue(
  string(now - duration(state.initial_interval))
).as(since,
  state.with(
    request("GET", state.url.trim_right("/") + "?" + {
      "since": [since],
      "limit": [string(int(state.batch_size))],
    }.format_query()).with({...}).do_request().as(resp,
      resp.StatusCode == 200 ?
        resp.Body.decode_json().as(body, {...})
      : {...}
    )
  )
)

The same technique applies to page tokens, offsets, window start/end times, and any other value derived purely from state without making an HTTP call.

---

Technique 2 — Inline single-use bindings

Every .as() costs a nesting level. If a value is used exactly once, inline it. Do not bind it.

Wrong — unnecessary `.as()` adds a nesting level for no benefit:

int(state.batch_size).as(limit,
  request("GET", url + "?" + {
    "limit": [string(limit)],
  }.format_query())...
)

// Also wrong — binding body fields used once each:
string(body.?next_cursor.orValue("")).as(next_c,
  body.?more_to_read.orValue(false).as(more,
    {"cursor": {"next_cursor": next_c}, "want_more": more}
  )
)

Correct — inline at call site:

request("GET", url + "?" + {
  "limit": [string(int(state.batch_size))],
}.format_query())...

// Inline body fields directly:
{
  "cursor": {"next_cursor": string(body.?next_cursor.orValue(""))},
  "want_more": body.?more_to_read.orValue(false),
}

Bind with .as() only when a value is referenced more than once, or when the expression is complex enough that a name genuinely aids comprehension. "I might need it later" is not a reason to bind.

---

Technique 3 — Sequential .as(state, ...) pipeline for multi-step flows

When a program must initialize state, normalize cursor fields, or perform multi-phase work (e.g., subscribe → list → fetch), chain top-level state transforms sequentially instead of nesting deeper.

Wrong — deeply nested state init inside HTTP chain:

state.with(
  state.?initial_start_time.orValue("").as(ist,
    (ist != "" ? state : state.with({"initial_start_time": string(int(timestamp(now - duration(state.initial_interval))))})).as(st,
      request(...).do_request().as(resp, ...)
    )
  )
)

Correct — sequential `.as(state, ...)` pipeline:

(
  !has(state.initial_start_time) ?
    state.with({"initial_start_time": string(int(timestamp(now - duration(state.initial_interval))))})
  :
    state
).as(state,
  (
    state.?want_more.orValue(false) ?
      state
    :
      state.with({"page": null})
  ).as(state,
    state.with(
      request(...).do_request().as(resp,
        resp.StatusCode == 200 ?
          resp.Body.decode_json().as(body, {...})
        : {...}
      )
    )
  )
)

Each .as(state, ...) block is at the top level — they do not nest inside each other. This pattern scales to arbitrarily complex multi-phase programs without increasing depth.

---

Technique 4 — Compute URL outside the response chain

URL construction belongs before the HTTP call, not as a nested .as() inside the response chain.

Wrong:

state.with(
  (state.url.trim_right("/") + "/api/v1/events?" + params.format_query()).as(reqUrl,
    request("GET", reqUrl).do_request().as(resp, ...)
  )
)

Correct — inline short URLs, or bind before `state.with()` for complex ones:

// Short URL: inline directly
state.with(
  request("GET", state.url.trim_right("/") + "/api/v1/events?" + params.format_query())
  .do_request().as(resp, ...)
)

// Complex URL: bind before state.with
(state.url.trim_right("/") + "/admin/v1/orgs/" + state.org_id + "/events").as(endpoint,
  state.with(
    request("GET", endpoint + "?" + params.format_query()).do_request().as(resp, ...)
  )
)

---

Technique 5 — Consolidate multiple cursor fields into a single map binding

When a program needs two or more cursor fields (e.g. a page token and a timestamp bookmark), nested .as() bindings add a level of indent per field. Constructing a map and binding it once removes those levels.

Optional values can be bound and carried through intermediate computation — they only need to be concretised (via orValue) when assigned to fields in an object that will be serialised to JSON, since JSON has no concept of optional values. Do not concretise at binding time; carry the optionals through and resolve them at the point of use.

Before (nested — one level per field, early concretisation):

state.?cursor.next_token.orValue("").as(next_token,
  state.?cursor.last_from.orValue(
    string(now - duration(state.initial_interval))
  ).as(since,
    state.with(
      request(...)...
    )
  )
)

After (single map binding, optionals carried through):

{
  "next_token": state.?cursor.next_token,
  "since": state.?cursor.last_from,
}.as(cursor,
  state.with(
    request("GET", state.url.trim_right("/") + "/v1/events?" + (
      cursor.next_token.orValue("") != "" ?
        {"limit": [...], "cursor": [cursor.next_token.orValue("")]}
      :
        {"limit": [...], "since": [cursor.since.orValue(
          string(now - duration(state.initial_interval))
        )]}
    ).format_query())...
  )
)

The map values are optional-typed — cursor.next_token and cursor.since remain optional until orValue is called at the point where a concrete value is needed (query parameter construction, cursor output). This preserves the semantic distinction between "field absent" and "field present but empty".

Optional keys (?"next_token": state.?cursor.next_token) are also valid — absent optionals omit the key entirely, and access uses cursor.?next_token. Either form works; the choice depends on whether downstream code benefits from the key always being present (with an optional value) or conditionally absent.

Technique 6 — Eliminate duplicated request/response handling

When an initialization branch (cursor creation, subscription, token exchange) and a steady-state branch both contain the same fetch logic, the duplication must be removed. Two approaches work equally well — choose based on the situation.

Variant A — Split initialization into a separate evaluation

When the init request's output is a cursor or token that the steady-state path already uses, split the work across two evaluations.

Problem — duplicated fetch block:

state.?cursor.realtime.as(persisted,
  persisted.hasValue() ?
    // Phase A: fetch events with persisted cursor (30 lines)
    state.with(
      request("GET", fetch_url + "?cursor=" + persisted.orValue(""))
        .do_request().as(resp, ...)
    )
  :
    // Phase B: create cursor, then immediately fetch events
    state.with(
      request("GET", create_url).do_request().as(r1,
        r1.StatusCode == 200 ?
          // fetch events — identical 30 lines copy-pasted from Phase A
          request("GET", fetch_url + "?cursor=" + r1.Body.decode_json().next_cursor)
            .do_request().as(r2, ...)
        : // create error
      )
    )
)

Phase B inlines the same fetch-and-process logic as Phase A. Two copies diverge over time.

Solution — Phase B stores the cursor and defers fetching:

state.?cursor.realtime.as(persisted,
  persisted.hasValue() ?
    // Phase A: fetch events with persisted cursor (30 lines — single copy)
    state.with(
      request("GET", fetch_url + "?cursor=" + persisted.orValue(""))
        .do_request().as(resp, ...)
    )
  :
    // Phase B: create cursor, persist it, let Phase A handle fetching
    state.with(
      request("GET", create_url).do_request().as(r1,
        r1.StatusCode == 200 ?
          {
            "events": [{"retry": true}],
            "cursor": {
              ?"realtime": r1.Body.decode_json().?next_cursor.optMap(v, string(v)),
            },
            "want_more": true,
          }
        : // create error
      )
    )
)

Phase B now makes one request, stores the cursor via a placeholder event, and sets want_more: true. The immediate re-evaluation enters Phase A with the cursor present — the fetch logic exists in one place. Each evaluation has a single HTTP concern.

Trade-offs: each step's success is durable independently — if the init succeeds and the subsequent fetch fails, the cursor is already persisted and the retry is a normal steady-state attempt. The init work is not repeated. However, it introduces a transient state (cursor stored, no events fetched yet) and requires a placeholder event with a matching drop processor.

---

Variant B — Intermediate result map within a single evaluation

CEL has no functions. When multiple branches need the same downstream processing, unify the branches into an intermediate result map, then bind it once before the shared code.

Trade-offs: the init and first fetch are atomic — both happen in one evaluation. But if the fetch fails after a successful init, the init's result (e.g. the created cursor) is lost because no event was published to persist it. Handling this requires careful use of error event forms (array to preserve cursor vs single-object to reset), adding complexity to the error paths.

Problem — duplicated fetch block (anti-pattern):

(stored != "") ?
  // fetch events with stored cursor + error handling (50 lines)
:
  create_cursor_request(...).as(rc,
    rc.StatusCode == 200 ?
      // identical fetch events + error handling (50 lines, copy-pasted)
    :
      // create error
  )

Solution — intermediate result, shared downstream:

(
  (stored != "") ?
    {"ok": true, "cursor": stored}
  :
    request("GET", create_url).do_request().as(rc,
      rc.StatusCode == 200 ?
        {"ok": true, "cursor": string(rc.Body.decode_json().?next_cursor.orValue(""))}
      :
        {
          "ok": false,
          "code": string(rc.StatusCode),
          "status": string(rc.Status),
          "body": string(rc.Body),
        }
    )
).as(result,
  !result.ok ?
    {
      "events": {
        "error": {
          "code": result.code,
          "id": result.status,
          "message": "GET /create: " + (size(result.body) != 0 ? result.body : result.status),
        },
      },
      "want_more": false,
    }
  :
    // Single copy of fetch + error handling using result.cursor
    request("GET", fetch_url + "?cursor=" + result.cursor)
      .do_request().as(resp, ...)
)

The intermediate map acts as a result type: ok distinguishes success from failure, and the remaining fields carry the payload for each case. This pattern keeps the fetch logic in one place and avoids the nesting that comes from inlining both paths.

---

Map merge, update, and field removal

These are general-purpose map operations. They apply to any map — request headers, query parameter maps, response objects, cursor state, or intermediate values. All examples in this section can be verified with mito. See tests/map_merge.txt for testscript cases.

Merge strategies

{"a": 1, "b": 2}.with({"a": 10, "c": 3})         // {"a": 10, "b": 2, "c": 3}
{"a": 1, "b": 2}.with_replace({"a": 10, "c": 3})  // {"a": 10, "b": 2}
{"a": 1, "b": 2}.with_update({"a": 10, "c": 3})   // {"a": 1, "b": 2, "c": 3}

with() is the most common — it's what state.with() uses. All three do a shallow merge: top-level keys are merged according to the strategy, but nested objects are replaced entirely, not merged recursively.

Field removal with drop()

drop() removes fields by name, supporting dot-path navigation into nested structures:

m.drop("key")                              // remove a single key
m.drop(["a", "b"])                         // remove multiple keys
{"a": [{"b": 1, "c": 2}]}.drop("a.b")    // {"a": [{"c": 2}]} — dot-path into arrays

Cursor key naming — avoid stuttering the parent

Keys inside cursor should not repeat the word "cursor". The parent path already provides context:

// WRONG — stutters the parent key
"cursor": {"realtime_cursor": ...}   // cursor.realtime_cursor
"cursor": {"next_cursor": ...}       // cursor.next_cursor

// CORRECT — concise, no redundancy
"cursor": {"realtime": ...}          // cursor.realtime
"cursor": {"next": ...}              // cursor.next

The same principle applies to sub-objects: cursor.page.token not cursor.page.page_token.

Application: cursor state transitions via clobber

Since with() replaces nested objects entirely, omitting a field from the cursor output removes it. This is useful for managing state transitions — namespace transient pagination state in a sub-object so the phase end cleans it up:

// During backfill — transient pagination state in cursor.next
"cursor": {
  "bookmark": string(body.?first_id.orValue("")),
  "next": {"after_id": string(body.?last_id.orValue(""))},
}

// Backfill complete — omitting "next" removes it via clobber
"cursor": {
  "bookmark": cursor.bookmark,
}

// Selective update — keep all cursor fields, update one
"cursor": cursor.with({"bookmark": new_bookmark}),

Namespacing transient state (e.g. cursor.next) separately from persistent state (e.g. cursor.bookmark) means the entire sub-object is cleaned up by clobber when the phase ends. No need to remove individual fields or rely on sentinel values.

---

Before/after: complete example

A typical deeply-nested generated program with cursor fields, pagination metadata, and URL construction stacked in one chain:

Before (12 levels — anti-pattern):

state.with(
  int(state.batch_size).as(limit,
    state.?cursor.next_token.orValue("").as(nt,
      state.?cursor.last_from.orValue(
        string(now - duration(state.initial_interval))
      ).as(since,
        (nt != "" ?
          {"limit": [string(limit)], "cursor": [nt]}
        :
          {"limit": [string(limit)], "since": [since]}
        ).format_query().as(qs,
          (state.url.trim_right("/") + "/v1/events?" + qs).as(reqUrl,
            request("GET", reqUrl).with({
              "Header": {"Authorization": ["Bearer " + state.api_key]},
            }).do_request().as(resp,
              resp.StatusCode == 200 ?
                resp.Body.decode_json().as(body,
                  (has(body.data) ? body.data : []).as(items,
                    body.?meta.next_cursor.orValue("").as(next,
                      {
                        "events": items.map(e, {"message": e.encode_json()}),
                        "cursor": {
                          "next_token": next,
                          "last_from": since,
                        },
                        "want_more": next != "",
                      }
                    )
                  )
                )
              : { "events": {"error": {...}}, "want_more": false }
            )
          )
        )
      )
    )
  )
)

After (4 levels — correct):

state.?cursor.next_token.orValue("").as(next_token,
  state.?cursor.last_from.orValue(
    string(now - duration(state.initial_interval))
  ).as(since,
    state.with(
      request("GET", state.url.trim_right("/") + "/v1/events?" + (
        next_token != "" ?
          {"limit": [string(int(state.batch_size))], "cursor": [next_token]}
        :
          {"limit": [string(int(state.batch_size))], "since": [since]}
      ).format_query()).with({
        "Header": {"Authorization": ["Bearer " + state.api_key]},
      }).do_request().as(resp,
        resp.StatusCode == 200 ?
          resp.Body.decode_json().as(body, {
            "events": body.?data.orValue([]).map(e, {"message": e.encode_json()}),
            "cursor": {
              "next_token": body.?meta.next_cursor.orValue(""),
              "last_from": since,
            },
            "want_more": body.?meta.next_cursor.orValue("") != "",
          })
        :
          {
            "events": {
              "error": {
                "code": string(resp.StatusCode),
                "id": string(resp.Status),
                "message": "GET /v1/events: " + (
                  size(resp.Body) != 0 ? string(resp.Body) : string(resp.Status)
                ),
              },
            },
            "want_more": false,
          }
      )
    )
  )
)

Changes made:

• int(state.batch_size) inlined (used once)
• nt, since extracted before state.with()
• qs and reqUrl eliminated — query and URL built inline
• items and next inlined into the result map
• Net result: 4 levels instead of 12, same logic, same correctness

---

Well-structured reference integrations

These programs from the public elastic/integrations repository demonstrate clean structure. Read them for real-world examples of the techniques above:

• vectra_rux audit — cursor bound before state.with(), 3 levels total
• openai completions — sequential .as(state, ...) pipeline for state normalization, then shallow HTTP
• o365 audit — long complex program kept readable with sequential state phases and comments
• authentik event — minimal 2-level HTTP pattern, clean pagination
• airlock_digital execution_histories — checkpoint cursor, 2 levels inside state.with()

CEL complexity baselines

Per-pattern-class complexity baselines from a survey of 316 CEL programs in elastic/integrations, measured with ceplx. The reviewer uses these to challenge generated programs that exceed expected complexity for their pattern class.

Source: ceplx survey of elastic/integrations (May 2026), joined with celir taxonomy classifications. Raw data in ~/thinking/cel_complexity/ceplx-joined.csv.

How to use

1. Classify the program using references/cel-taxonomy.md. 2. Look up the class in the tables below. 3. If the class has n >= 10, use the class-specific baselines. 4. If n < 10, use the global percentiles as a fallback. 5. A program above p90 for its class needs justification — the API's requirements may warrant it, but the burden is on the generator to explain why.

Skip threshold

Skip the review when both conditions hold:

• The program's cognitive complexity is below the class p50
• Total cognitive complexity is below 40

These programs are simple enough that review overhead is not justified.

By pagination pattern

Only classes with n >= 10 are reliable baselines. Classes with smaller samples are included for reference but should be used cautiously.

Composite patterns (e.g., offset + time_window) have very small samples (n <= 4) and are not listed as reliable baselines.

By state management pattern

Global percentiles (fallback)

When the pattern class has n < 10, use these global baselines derived from all 316 programs:

Interpreting ceplx diagnostic output

Run ceplx -diag -json program.cel to get per-node complexity contributions. The reviewer should focus on:

• High-cost comprehensions: .map() and .filter() inside nested

.as() chains multiply cognitive cost

• Deep ternary nesting: each level of ? ... : ... inside .as()

adds both cyclomatic and cognitive complexity

• Logical chains: && / || chains inside conditions add

cyclomatic complexity

• `.as()` depth: each level amplifies the cognitive cost of

everything inside it

The diagnostic output identifies which nodes contribute most, guiding specific refactoring suggestions (extract pre-bindings, inline single-use bindings, split branches into separate evaluations).

Reviewer challenge examples

Given a cursor_token program with cognitive complexity 85:

"This is a cursor_token program. The p90 for cursor_token is 49

(n=16). Your program's cognitive complexity is 85, which is well

above the baseline. The diagnostic shows 40 points from the nested

ternary at line 25 inside a 3-level .as() chain. Can this be

flattened with pre-bindings?"

Given a graphql_relay program with cognitive complexity 35:

"This is a graphql_relay program. The p50 is 31, p90 is 64 (n=6,

treat as approximate). Your program is within expected range. No

complexity challenge."

CEL expression reference

This reference covers the CEL expression itself — the program that runs inside program: | in a cel.yml.hbs template. It does not cover Handlebars, YAML template anatomy, manifest configuration, or system test setup.

The expression builder's job: given state.json (literal test values) and a running mock URL, produce a validated .cel file via incremental mito development.

---

Interface contract

Inputs:

• test-api.py — the Python implementation of the API interaction

(the specification; the ground truth)

• state.json — keys matching the future state: block, with literal

test values (url pointing at the mock, credentials, batch_size, etc.)

• Mock URL — a running elastic/stream mock derived from test-api.py
• Research brief — supplementary context (field meanings, edge cases

not exercised by the script)

Outputs:

• A validated .cel file that passes mito against the mock
• A taxonomy classification (pagination pattern + state management

pattern at the least complex class that satisfies requirements)

The expression builder never touches cel.yml.hbs, manifests, or system test files. It returns the working CEL expression and the classification. The orchestrator wraps it.

---

Translation framing

The task is translation from Python to CEL, not generation from prose. The test-api.py collection function (run_collection() or collect()) is the specification. Every construct has a direct CEL equivalent:

The builder translates the collection function, not the CLI scaffolding (argument parsing, logging, archiving).

---

Incremental build phases

Every expression must be built in phases, validating with mito after each. Do NOT write the full program before running mito.

Run mito after each phase (always use -fb for filebeat-compatible validation):

mito -fb -data state.json -log_requests program.cel

For pagination testing:

mito -fb -data state.json -log_requests -max_executions 5 program.cel

For cursor persistence testing:

mito -fb -data state_cursor.json -log_requests -max_executions 3 program.cel

---

Core structure

Every HTTP-based expression has this shape:

[pre-bindings: cursor defaults, window math, URL construction]
state.with(
  request(...).do_request().as(resp,        ← level 1
    resp.StatusCode == 200 ?
      resp.Body.decode_json().as(body, {    ← level 2
        ...result map...
      })
    : { ...error... }
  )
)

Hard cap: `.as()` depth must never exceed 5 levels on any execution path. HTTP programs target 2 levels inside state.with().

Pre-bindings

Extract these before state.with():

• Cursor defaults: state.?cursor.last_ts.orValue(...)
• Window bounds: now - duration(state.initial_interval)
• URL construction for complex paths
• Page tokens from previous state

Do NOT bind single-use values with .as() — inline them.

Result map

The map inside body.as(body, {...}) contains:

• "events": the event array
• "cursor": state to persist across restarts
• "want_more": pagination continuation signal
• "url": preserved from state (required)

Flat decode pattern (structural constraint)

After try(resp.Body.decode_json()).as(body, ...) or resp.Body.decode_json().as(body, ...), the success path must follow this shape:

body.as(body,
  error_check_1 ? error_result_1 :
  error_check_2 ? error_result_2 :
    {
      "items": body.?data.orValue([]),
      "next":  body.?pagination.next.orValue(""),
    }.as(page, {
      "events": page.items.map(e, {"message": e.encode_json()}),
      "want_more": size(page.items) > 0 && page.next != "",
      "cursor": { "next": page.next },
      "url": state.url,
    })
)

The key technique: extract response navigation into a flat intermediate map, then build the result from that map. Do NOT chain nested .as() calls to extract individual fields. Each .as() level multiplies cognitive complexity of everything inside it.

Wrong — nested extraction (high complexity):

body.?events.optMap(events, type(events)==type([]) ? dyn(events) : dyn([]))
  .orValue([]).as(events_list,
    body.?pagination.optMap(pg, pg.?next.optMap(n, n).orValue(""))
      .orValue("").as(next_str,
        { ... result using events_list and next_str ... }
      )
  )

Right — intermediate map (low complexity):

{
  "items": body.?events.orValue([]),
  "next":  body.?pagination.next.orValue(""),
}.as(page, {
  "events": page.items.map(e, {"message": e.encode_json()}),
  "want_more": size(page.items) > 0 && page.next != "",
  "cursor": { "next": page.next },
  "url": state.url,
})

Both extract the same two values from the response body. The intermediate map does it in one .as() level; the nested version uses two (or more). The complexity difference compounds with every additional field extracted.

Worked example: Airtable (cursor_token + time_window)

Python (run_collection):

events = body.get("events", [])
next_token = body.get("pagination", {}).get("next", "")
# stop if no events or no next token
if not events or not next_token:
    break

CEL (flat):

state.?cursor.next.orValue("").as(page_token,
  state.with(
    request("GET", url + "?" + query_params).with({...}).do_request().as(resp,
      resp.StatusCode == 200 ?
        try(resp.Body.decode_json()).as(parsed,
          is_error(parsed) ? { error result } :
          parsed.as(body,
            (has(body.error) && body.error != null) ? { error result } :
              {
                "items": body.?events.orValue([]),
                "next":  body.?pagination.next.orValue(""),
              }.as(page, {
                "events": page.items.map(e, {"message": e.encode_json()}),
                "want_more": size(page.items) > 0 && page.next != "",
                "cursor": { "next": page.next },
                "url": state.url,
              })
          )
        )
      : { error result }
    )
  )
)

.as() depth inside state.with(): resp (1) → parsed (2) → body (3) → page (4). The page level is the intermediate map — it doesn't add nesting depth to the result map contents because the result map is a flat literal.

Worked example: Buildkite (graphql_relay + time_window)

Python (run_collection):

edges = parsed["data"]["organization"]["auditEvents"]["edges"]
page_info = parsed["data"]["organization"]["auditEvents"]["pageInfo"]
has_next = page_info.get("hasNextPage", False)
end_cursor = page_info.get("endCursor", "")

CEL (flat, with relay condition pre-bound):

poll_from.as(poll_from,
  pages_before.as(pages_before,
    state.with(
      post_request(...).do_request().as(resp,
        resp.StatusCode == 200 ?
          try(resp.Body.decode_json()).as(parsed,
            is_error(parsed) ? { error } :
            size(parsed.?errors.orValue([])) > 0 ? { graphql error } :
            (!has(parsed.?data.organization) || ...) ? { org error } :
              parsed.data.organization.?auditEvents.orValue({"edges":[],"pageInfo":{}}).as(audit, {
                "has_next": audit.?pageInfo.?hasNextPage.orValue(false),
                "end_cursor": string(audit.?pageInfo.?endCursor.orValue("")),
                "edges": audit.?edges.orValue([]),
              }.as(page, {
                "events": page.edges.map(e, {"message": e.node.encode_json()}),
                "want_more": page.has_next && page.end_cursor != "" && (pages_before+1) < int(state.max_pages),
                "cursor": page.has_next && page.end_cursor != "" && (pages_before+1) < int(state.max_pages) ?
                  {"poll_occurred_at_from": poll_from, "after": page.end_cursor, "pages_in_cycle": pages_before+1}
                : {},
                "url": state.url,
              }))
          )
        : { error }
      )
    )
  )
)

The relay condition (has_next && end_cursor != "" && pages < max) is expressed once in page.has_next && page.end_cursor != "" rather than repeated with full optional-access chains. The intermediate map absorbs the navigation complexity; the result map stays flat.

---

Event output

Events contain ONLY "message":

body.items.map(e, {"message": e.encode_json()})

Do NOT set @timestamp or any other field. The framework handles metadata. Duplicating @timestamp causes silent document rejection in ES 9.x.

---

Error handling

Every HTTP request needs a status check. Two error forms:

Single-object error (retry — deletes cursor):

resp.StatusCode == 200 ?
  ...success...
: {
    "events": {
      "error": {"message": "GET /path: " + string(resp.StatusCode)}
    },
    "want_more": false,
  }

Array error (advance — preserves cursor):

"events": [{"error": {"message": "..."}}],
"cursor": state.cursor,
"want_more": false,

Use single-object (retry) when data was not collected. Use array (advance) when the program should skip past the error.

---

Pagination

The want_more field drives pagination. Set it based on the API's pagination signal, NOT based on whether events were returned:

"want_more": body.?meta.next.orValue("") != "",
"cursor": {
  "next": body.?meta.next.orValue(""),
  "last_ts": /* high-water mark */,
},

Cursor field separation: Store page tokens (transient, drive want_more) and time bookmarks (persistent, drive the starting point) as separate cursor fields. Never use one field for both.

---

State management rules

1. state.url — from resource.url config; preserve in output 2. cursor — only state persisted across restarts 3. events — removed after each evaluation; never rely on it 4. want_more: true — triggers immediate re-evaluation (only if events is non-empty) 5. Numbers are float64; cast with int() for integer operations 6. Optional access: state.?cursor.field.orValue(default) 7. After first ? in a chain, subsequent ? are automatic

---

Syntax rules

---

Quality checklist

Before returning the .cel file, verify:

• [ ] Passes mito against the mock at each phase
• [ ] All error paths from test-api.py are represented
• [ ] Pagination logic matches the Python loop's termination conditions
• [ ] Response fields navigated the same way as the Python script
• [ ] Cursor state captures the same info Python propagates between iterations
• [ ] No invented logic (branches that don't exist in the Python source)
• [ ] .as() depth <= 5 on every path
• [ ] No rate limiting or retry logic in the expression
• [ ] Events contain only "message"
• [ ] want_more driven by pagination signal, not event count
• [ ] celfmt -s -i program.cel -o program.cel run to simplify/format the final file

CEL function reference

Functions available in the CEL input, organized by mito extension. Each extension must be registered in beats for its functions to be available. Functions present since mito v1.0.0 are unmarked; later additions show the first mito version.

Extension availability

Regexp is conditional on regexp config being present. Globals provides variables (now, useragent, env, remaining_executions) not functions. The as macro is from Collections.

Full set as of v9.3.0: AWS, Collections, Crypto, Debug, File, Globals, HTTP, JSON, Limit, MIME, Printf, Regexp\*, Strings, Time, Try, XML.

Functions by extension

AWS

Registered from v8.19.0 / v9.1.0. All functions added in mito v1.21.0.

Collections

Registered from v8.6.0. Also registers the as macro.

Crypto

Registered from v8.6.0.

Debug

Registered from v8.11.0.

File

Registered from v8.6.0.

HTTP

Registered from v8.6.0. All functions added in v1.0.0.

JSON

Registered from v8.6.0.

Limit

Registered from v8.6.0.

Named policies: "okta", "draft".

Behavior change at v9.3.0: registered via LimitWithApply instead of Limit. The apply callback fires during evaluation, updating the HTTP client immediately. Before v9.3.0, the return map had to be placed in state.rate_limit and only took effect on the next evaluation cycle. From v9.3.0, rate limit changes take effect between requests within the same evaluation. Not back-ported to 8.19.

MIME

Registered from v8.6.0.

Printf

Registered from v8.18.0 / v9.0.0.

Regexp

Registered from v8.6.0. Conditional on regexp config being present. All functions use named precompiled patterns from the regexp config block.

Strings

Registered from v8.7.0. All functions added in v1.0.0 except where noted.

Time

Registered from v8.6.0.

`now` global vs `now()` function: The now global is a fixed value set once per evaluation by the beats input. The now() function calls time.Now() each invocation. Within a single evaluation the global is stable; the function is not. CEL programs should use the now global for consistency.

Try

Registered from v8.6.0. Both functions are non-strict.

XML

Registered from v8.9.0.

Determining minimum beats version

For quick lookups, check the extension availability table above and the per-function mito version annotations. For systematic version verification during formal reviews, use the review-integration skill which has the full beats-to-mito mapping table and the step-by-step verification procedure in its references.

Common CEL idioms and conventions

Quick reference for idioms, HTTP usage, pagination strategy bullets, events, structure, and YAML configuration notes used with CEL programs.

Syntax anti-patterns — NEVER use these

These are the most common mistakes that cause compilation failures. See references/cel-incremental-build.md for detailed explanations and correct alternatives.

Common CEL idioms

HTTP requests

Simple requests — use get(url) or post(url, ct, body). These direct calls automatically pick up auth.basic and auth.token config. Use request("METHOD", url).with({...}).do_request() when additional headers beyond auth are needed, or for methods other than GET/POST/HEAD.

`resource.headers` (ga 8.18.1) — static headers that are the same for every request (e.g. Content-Type, Accept, API version headers) can be set in the YAML config rather than in the CEL program. These are added before auth headers.

URL normalization — use trim_right("/") to handle trailing slashes.

Query strings — build with format_query() on a map with optional keys. Avoid constructing query strings via string concatenation.

(state.url.trim_right("/") + "/api/v1/alerts?" +
  {
    "limit": [string(state.batch_size)],
    "sort": ["updated_timestamp|asc"],
    ?"after": state.?cursor.token.optMap(v, [v]),
    ?"filter": state.?query.optMap(v, [v]),
  }.format_query()
)

optMap is safe here because the body always evaluates to a concrete value ([v]). Do not use optMap when the body evaluates to optional.of(...) or optional.none(). optMap is map, not flatMap — it wraps the result, so optional.none() from the body becomes optional.of(optional.none()), which serialises as null instead of omitting the key:

// WRONG — empty string produces "event_types": null, breaking format_query()
?"event_types": state.?event_types.optMap(et,
  (et != "") ? optional.of([et]) : optional.none()
),

// CORRECT — ternary at the top level; optional.none() omits the key
?"event_types": (state.?event_types.orValue("") != "") ?
  optional.of([state.event_types])
:
  optional.none(),

Pagination

Match the strategy to the API:

• Cursor/token — API returns a next-page token; pass it in the next request. Set "want_more": has(body.?meta.pagination.next).
• Offset — increment offset by page size each iteration.
• Next-link — API returns a full URL for the next page.
• Time-window — advance a timestamp cursor based on response data.
• Worklist — fetch a list of IDs, then iterate over each.

Cursor timestamp tracking — use the last record's timestamp when results are known to be sorted by the API (first record if reverse-sorted). Use max() with a regression guard when sort order is not guaranteed.

Event output

Standard structure — body.<path>.map(e, {"message": e.encode_json()}) where <path> matches the API's response (e.g. body.data, body.items, body.resources).

Events must contain ONLY `"message"`. Do not set @timestamp, event.original, or any other field. The Elastic Agent framework adds @timestamp; duplicates cause silent document rejection in ES 9.x (Duplicate field '@timestamp').

// CORRECT
items.map(e, {"message": e.encode_json()})

// WRONG — causes duplicate @timestamp, ES drops all events
items.map(e, {"message": e.encode_json(), "@timestamp": e.timestamp})

Placeholder events for cursor persistence

The input only persists cursor updates when at least one event is published. When a page returns no data but the cursor should advance, emit a placeholder and drop it before indexing:

"events": body.?data.orValue([]).map(e, {"message": e.encode_json()}).as(events,
  (size(events) > 0) ? dyn(events) : dyn([{"retry": true}])
),

Then add a drop_event processor in cel.yml.hbs to discard it before indexing:

processors:
- drop_event.when.equals.retry: true

The dyn() wrapping is required because the two branches have different compile-time types (list(map(string,string)) vs list(map(string,bool))). See cel-incremental-build.md for the full workflow: write without dyn() first, add it only when celfmt -s reports a type mismatch, then verify both paths with mito.

The alternative form uses [{"message": "retry"}] with - drop_event.when.equals.message: retry in processors:. The boolean form is preferred — retry is a dedicated control flag that no real event would have. The {"message": "retry"} form avoids the type mismatch (both branches are map(string,string)) so dyn() is not needed.

Error handling

Two error event forms with distinct semantics:

Single-object error (retry): "events": {"error": {...}} — the input logs at ERROR, sets degraded status, and deletes the cursor so the next evaluation retries. Use when data was not collected.

{
  "events": {
    "error": {
      "code": string(resp.StatusCode),
      "id": string(resp.Status),
      "message": "GET " + state.url.trim_right("/") + "/api/v1/items: " + (
        size(resp.Body) != 0 ? string(resp.Body) : string(resp.Status)
      ),
    },
  },
  "want_more": false,
}

Array error (advance): "events": [{"error": {...}}] — processed as a normal event array. The cursor is updated. Use when the program should advance past the error. Ensure the ingest pipeline has a terminate processor (ES 8.16.0+).

Error message format: include HTTP method and URL path without query params: "METHOD path: body-or-status".

Nested mapping with flatten() and drop_empty()

When an API response contains nested arrays or when mapping conditionally produces items, the result is a list of lists or a list with empty elements. Use flatten() to collapse nesting and drop_empty() to remove empties.

Expanding sub-arrays — each item contains a nested array that should become separate events:

// API returns: {"records": [{"id": "a", "events": [e1, e2]}, {"id": "b", "events": [e3]}]}
// Want: [e1, e2, e3] as separate events
body.records.map(r, r.events).flatten().map(e, {"message": e.encode_json()})

Without flatten(), the inner map produces [[e1, e2], [e3]] — a list of lists. flatten() collapses it to [e1, e2, e3].

Conditional mapping — some items produce events, others don't:

// Filter and transform: only include items with status "complete"
body.items.map(item,
  item.status == "complete" ?
    [{"message": item.encode_json()}]
  :
    []
).flatten()

Each item maps to either a single-element list or an empty list. flatten() merges them into a flat event list.

Cleaning up after `drop()` — removing fields from nested objects can leave empty maps:

body.items.drop("internal_id").drop_empty().map(e, {"message": e.encode_json()})

drop_empty() recursively removes all empty maps and lists, so items that contained only the dropped field disappear entirely rather than becoming {}.

Structure and readability

• 2-space indentation throughout, reflecting scope.
• Break long lines — only very simple ternary expressions should be one-liners.
• `.as(name, ...)` with meaningful names. For values extracted from a dotted path, use the final field name: state.?cursor.next.orValue("").as(next, ...), not .as(stored, ...) or .as(nc, ...).
• Comment non-obvious intent, not obvious code. Multi-phase state machines (subscribe-list-fetch, create-poll-download) must include a comment block explaining the phases and state variables.
• Describe each major branch — CEL has no functions, so a multi-branch ternary tree is the only way to express control flow. When a program has top-level branches (init vs steady-state, paginating vs backfilling, different API call types), add a short comment at each branch entry describing its purpose (e.g. // Steady state: fetch events with persisted cursor). These act as section headings in an expression that would otherwise require tracing every condition to navigate.
• Avoid stringly typed expressions when CEL extensions exist — e.g. use format_query() instead of string-concatenated query strings.

Nesting discipline

`.as()` depth must not exceed 5 levels on any execution path. HTTP programs must target 2 levels inside state.with(): resp and body.

See references/cel-code-style.md for before/after examples of all six techniques.

Configuration

Request tracer — use the always-present style (from v8.15):

resource.tracer:
  enabled: {{enable_request_tracer}}
  filename: "../../logs/cel/http-request-trace-*.ndjson"
  maxbackups: 5

rather than the conditional {{#if}} block. The newer form supports trace deletion when tracing is disabled. If the integration uses the old form, consider upgrading if the stack version allows.

Tracer at data stream level — declare enable_request_tracer in the data stream manifest, not at the input level. Input-level tracing enables logging for all data streams in the policy, increasing load and risk of secret leakage.

Polymorphic patterns

Many capabilities in CEL input programs can be implemented at multiple abstraction levels: raw CEL expressions, mito library functions, or YAML config options. This reference maps each capability to its available implementations and the minimum version required, so builders can choose the right approach for their target version.

Authentication

HTTP headers

Static headers can be declared in config via resource.headers (available from v8.18.1). Dynamic or conditional headers must still be set in CEL using .with({"Header": ...}).

Most existing integrations predate resource.headers and set everything in CEL. New integrations targeting v8.18.1+ should put static headers in config.

Rate limiting

Named policies for rate_limit(): "okta" (Okta rate limit headers) and "draft" (IETF rate limit draft headers).

Request construction

URL construction

Prefer parse_query() / format_query() when query parameter values may contain special characters.

Data encoding

When an API returns base64 without padding, use .trim_right("=").base64_raw_decode() instead of .base64_decode().

The decode_json_string_numbers config option (v8.19.0 / v9.1.0, mito v1.22.0) preserves numeric precision by decoding JSON numbers as strings rather than floats.

Prefer sprintf over string concatenation for readability when targeting v8.18.0+.

Environment and secrets

Before secret_state, secrets stored in state require explicit redact config entries for log redaction.

Debugging aids

Patterns worth noting

1. Auth header duplication. Some integrations set the same static header in both resource.headers and CEL .with(). This is harmless but redundant. New code should pick one.

2. HMAC signing cannot move to config. Vendor-specific HMAC schemes (Duo SHA1, ThreatConnect SHA256, Akamai EG1) are too varied. These remain pure CEL.

3. AWS SigV4 has three abstraction levels. Manual signing, library functions (sign_aws_from_env() / sign_aws_from_static()), and config (auth.aws). New code should use the highest abstraction available for the target version.

4. `request().with().do_request()` dominates. Convenience functions like get() and post() exist but most integrations need custom headers, making the full request builder the standard pattern.

5. `sprintf` adoption is slow. Available since v8.18.0 / v9.0.0 but many integrations still use string concatenation. Prefer sprintf in new code for readability.