<!-- Sema documentation — Collectors & Taps
     Sema is a new AI-native, neurosymbolic programming language most models
     have not been trained on. Full corpus: https://sema.49.12.246.95.sslip.io/llms-full.txt
     Install language support: https://sema.49.12.246.95.sslip.io/install-skill/ -->

# Collectors & Taps

> Non-interfering instrumentation in Sema — collector blocks, the |> tap pipe that records a value without changing it, bounded retention, and async export.

A `collector` is a typed telemetry channel, and `|>` is the **tap pipe** that feeds
it. The defining property is *non-interference*: a tap records a value and evaluates
to that exact same value, so you can instrument any expression in a hot path without
changing control flow, dataflow, or the value's identity. Metrics never become the
bug.

Collectors are the **value/metric** telemetry channel. They are strictly observational
— unlike [events](/governance/events/), a collector can *never* drive control flow.

## Why instrumentation is a language construct

Bolted-on instrumentation is a classic source of bugs and drift: hand-written logging
calls around every scalar wrap real code in ceremony; plotting libraries that
monkey-patch values are not replayable or type-visible; unbounded in-memory metric
lists turn a long run into an OOM. Sema makes the telemetry channel typed, bounded,
and non-interfering by construction — you declare *what* you collect and *how much* is
retained, and the compiler enforces it.

## Syntax

A `collector` block declares typed aggregation channels; a `def` taps values into
them with `|>`:

```sema
collector TrainMetrics:
    loss: f32 mode series retention ring(100_000)
    activations: Tensor[f32] mode stack retention sample(rate=0.05)
    prediction: str mode set retention limit(10_000)
    batch: TrainingBatch mode bag retention ring(1_000)
    export wandb project="sema-train" run=TrainConfig.tenant

def train_step(batch: TrainingBatch) -> f32 !{model.invoke, observe.record}:
    logits = model.forward(batch.inputs) |> TrainMetrics.activations(layer="encoder.3")
    loss = cross_entropy(logits, batch.labels) |> TrainMetrics.loss(split="train")
    return loss
```

## The tap pipe `|>`

`left |> Collector.field(args...)` records `left` into the typed collector sink and
**evaluates to `left`** — same type, same value identity. The collector call may
attach labels, tags, run ids, source spans, or grouping keys, but it **must not
transform the value**. Because taps are expression-level, they need no special
assignment form; the verified `finops-ledger` reconciler taps a score inline:

```sema
score = candidate_score(bank, entry) |> ReconcileMetrics.score(bank_id=bank.id)
```

is equivalent, for dataflow, to assigning `candidate_score(...)` directly. The same
example taps whole decisions and candidates on their way through:

```sema
return decision |> ReconciliationMetrics.decision(bank_line_id=bank.id)
```

:::caution
`|>` is **reserved syntax, not an overloadable operator.** This is deliberate: a
redefinable pipe could be turned into control flow, defeating the non-interference
guarantee. The left expression's type must be assignable to the collector channel
type, the whole expression carries the left expression's type and trust label, and
its [effect row](/governance/effects/) adds `observe.record` (configured exporters add
`observe.export` at the export boundary).
:::

## Retention modes — bounded by construction

Each channel has a value type, a **mode**, a **retention** bound, an export policy,
and optional labels. Modes are compiler-known:

| Mode | For | Behavior |
| --- | --- | --- |
| `series` | scalars | append |
| `histogram` | numeric summaries | streaming summary |
| `stack` | fixed-shape arrays/tensors | stack fixed shapes |
| `set` | strings/enums | de-duplicate |
| `counts` | categorical values | count per category |
| `bag` | structured objects | collect objects |
| `last` | gauges | keep the latest |

If the mode is omitted the compiler infers the safest bounded mode from the type:
numeric scalars → `series`, arrays/tensors → `stack`, strings/enums → `set`/`counts`,
structs → `bag`. Retention bounds — `ring(N)` / `limit(N)` keep the last `N`,
`sample(rate=…)` subsamples — are **required**: an unbounded collector is a compile
error outside debug builds, so an experiment run can never quietly grow into the bug.
Heterogeneous data must use an explicit erased `Any`/`Dyn` collector so mixed bags are
visible in reviews.

## Non-interference and the hot path

Tap recording is a bounded, hot-path operation: append a typed sample or update a
sketch in the runtime journal / ring buffer, then return the original value. **The
runtime never plots, exports over the network, embeds, or calls a model in the tap
hot path.** Exporters run asynchronously.

Collector failures are non-interfering *by default*: if a sink is unavailable or a
retention bound is exceeded, the runtime records a `CollectorDropped` /
`CollectorBackpressure` event and returns the tapped value unchanged. A run does not
crash because telemetry hiccuped.

:::note[Opting into strict]
A channel may opt into `strict`, in which case failures become typed
`CollectorError` values and the enclosing function must declare and handle that
possibility. Use `strict` when a missing metric is itself a correctness bug; leave it
off (the default) when the telemetry is genuinely best-effort.
:::

## External export

The `export` clause names an asynchronous destination that ships the same payload the
local artifact holds. At run end the runtime writes each collector to
`.sema/collectors/<name>.json` — a real, inspectable artifact — and records the
configured `export` destination; **the local write is always produced, so data is
never lost** even if the networked sink is down. Supported exporters include local
Arrow/Parquet, OpenTelemetry, TensorBoard, and Weights & Biases:

```sema
collector ReconciliationMetrics:
    candidate_score: f32 mode series retention ring(100_000)
    candidate: MatchCandidate mode bag retention ring(25_000)
    decision: ReconciliationDecision mode bag retention ring(50_000)
    export local path="out/metrics/reconciliation.arrow"
```

Sending secret or `trusted` data to an *external* exporter without a
[policy](/governance/policy/) grant is a policy denial — telemetry export is
capability-gated like any other effect.

## Collectors vs monitors vs events

| Construct | Role | Drives control flow? |
| --- | --- | --- |
| `collector` | record values/metrics | never |
| [`monitor`](/governance/monitor/) | detect distribution drift | via `on drifted:` |
| [`event`](/governance/events/) | typed domain signals | yes, deliveries do work |

Collectors are the *value/metric* channel; the *narrative* channel — structured log
records, console output — shares this section's exporter machinery and
non-interference rules but is a separate construct.

## Failure modes

- **Type mismatch** between the tapped value and the channel → compile error.
- **Tensor/array shape mismatch for `stack`** → compile error when static,
  `CollectorShapeError` at dynamic boundaries.
- **Unbounded collector** → compile error outside debug builds.
- **Secret/trusted data to an external exporter without policy** → policy denial.
- **Exporter outage** → non-interfering drop/backpressure event, unless the channel
  is `strict`.

## How it's checked

- `sema check <project>` validates channel types, modes, retention bounds, and tap
  type assignability.
- Tap effects (`observe.record`, `observe.export`) appear in the function's row, so
  export is capability-gated.
- The local `.sema/collectors/<name>.json` artifact is always produced, so a run's
  telemetry is inspectable after the fact.

## See also

- [Monitors & Drift](/governance/monitor/) — the statistical channel that *can* react.
- [Events](/governance/events/) — the signal channel that drives control flow.
- [Effects & Capabilities](/governance/effects/) — `observe.record` / `observe.export`.
- [Policies](/governance/policy/) — gating export of trusted data.