Windowing

turbine.windowing ships three window primitives. They all consume the same aggregate functions (turbine.aggregates), offer the same pair of close callbacks (on_close_each= per pane, on_close= per batch), and support early-fire triggers with the default storage — they differ in how their boundaries are defined.

This page covers what is common to every window: the aggregate you compute, the close callbacks, early-fire, and the time model. The two companion pages cover the rest:

• Window types — Tumbling, Sliding and Session, with durable RocksDB-backed state and a code example for each.
• Windowing — advanced — failover, referencing one window's result from another (Ref / join_with), dynamic (deferred-construction) windows + the recovery hook, and sessions at high cardinality.

Primitive	Boundaries	Storage	Crash recovery
PersistentTumbling	Fixed size_ms	RocksDB by default; optional resident columns + checkpoints	✅ survives restart, full failover
PersistentSliding	Fixed size_ms / slide_ms	RocksDB by default; optional resident columns + checkpoints	✅ survives restart, full failover
PersistentSession	Data-driven (gap_ms)	RocksDB + checkpoints	✅ survives restart, full failover

Rules of thumb:

• Choose the shape by your boundaries: fixed buckets → Tumbling; overlapping/rolling → Sliding; activity-driven with no fixed boundary (user sessions, device bursts) → Session.
• The default RocksDB storage supports every window feature. Fixed event-time windows can instead keep compact aggregate columns resident in memory and recover them from checkpoints and input replay. Both are browsable in the console.

The former InMemoryTumbling, InMemorySliding and InMemorySession classes have been removed. Replace them with the corresponding Persistent* class, remove max_active_windows, and pass the handler's state to window.update. Give the handler a state argument when it previously accepted only a batch. The console now has one Windows view, and ClusterClient.windows() no longer accepts the storage-category kind filter.

When @app.subscribe(..., partition_key="...", parallelism=N) is set, every window inside that subscription must be keyed (at least) by the same column. Otherwise rows for a single window key would be split across shards and each shard would aggregate over a partial view. The SDK enforces this at construction — you get a clear error pointing at the offending key=. The partition_key itself is auto-injected: you never write it again on the window. See Partitioning for the full picture.

All window kinds use the same Rust mechanics, including custom Python accumulators. Python expressions and callbacks retain their SDK interfaces. Persistent windows require a root state store or a view with a matching declared partition_key; there is no Python engine opt-out.

This pre-deployment state-format migration requires fresh state/checkpoints for incompatible custom-session calculations. Upgrade both combine stages together and reset their internal topic and stage-2 state when the calculation payload changes. Keep application identities separate when replaying into fresh state.

Resident storage for fixed windows

Use storage="resident" on PersistentTumbling or PersistentSliding when many groups are open and the compact aggregates fit in memory. The default is storage="rocksdb". Both modes are durable; resident mode saves complete checkpoints and replays input after the saved position on restart. Close outputs are at-least-once and may repeat after a crash.

from pathlib import Path
import pyarrow as pa
from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState
from turbine import aggregates as agg, windowing as win

kafka: KafkaBroker = KafkaBroker(bootstrap="localhost:9092")
app: Turbine = Turbine(
    kafka,
    app_id="resident-scores-v1",
    state_dir="./resident-state",
    checkpoint_url=Path("./resident-checkpoints").resolve().as_uri(),
    from_earliest=True,
)

@app.subscribe(
    kafka.topic("events", event_time="event_ms"),
    name="scores",
    output=kafka.topic("scores"),
    partition_key=None,
)
class Scores:
    def __init__(self) -> None:
        self.window: win.PersistentTumbling = win.PersistentTumbling(
            self,
            name="per_user",
            storage="resident",
            size_ms=60_000,
            allowed_lateness_ms=2_000,
            key=["user_id"],
            value=agg.Mean("latency_ms"),
            on_close=self.close,
        )

    def process(self, batch: RecordBatch, state: TurbineState) -> None:
        self.window.update(batch, state)

    def close(self, state: TurbineState, panes: pa.Table) -> RecordBatch:
        return panes.combine_chunks().to_batches()[0]

if __name__ == "__main__":
    app.run()

The example uses local checkpoint files. For transfer between machines, configure a shared checkpoint URL as described in Deployment. Keep input records available from the last checkpoint: an input-retention gap stops the resident worker instead of discarding its aggregates. Starting-position overrides cannot skip a recovered checkpoint's input suffix.

The initial resident scope requires one fixed window declared in the handler's __init__, one UTF-8 grouping column, event time and at-least-once processing. It supports the native Sum, Min, Max, Count, Mean, supported Variance/StdDev plans and compatible flat Multi bundles. Batch and per-group close callbacks retain their normal shapes. Multiple Kafka partitions are independent workers; keep each group's records on the same partition, as for ordinary window processing.

Use the default storage for sessions, custom/opaque calculations, processing time, early fire, ref values, dynamic windows, multiple window operators per subscription, partition_key/handler sharding, TTL, joins, repartition/combine, multi-input or exactly-once processing. Resident mode refuses incompatible configurations and offline rescale; it does not silently change storage or spill to disk.

The chosen storage mode belongs to the subscription's durable processing history. Restarting that subscription in the other mode is refused, including from a fresh local directory. A new subscription name starts a separate history; it does not migrate existing aggregate state. To start a new application history, also use fresh state and checkpoint locations: changing app_id alone does not reset the registry stored there. Keep the window definition compatible with its checkpoints and follow the coordinated cluster upgrade before enabling this mode. That runtime upgrade requires stopping all cluster nodes, including clusters using only ordinary storage.

Resident allocations share the process storage budget configured by TURBINE_BLOCK_CACHE_MB and may reserve at most a quarter of it; their reservations reduce the space available to the ordinary state cache. This is allocation accounting, not a whole-process RSS limit. Exhausting the resident allowance stops the worker without committing the rejected input. A checkpoint-capture refusal stops after its last acknowledged batch and recovers by replaying from the previous complete checkpoint. Increase the process budget or reduce retained groups, window duration or lateness; there is no automatic storage fallback.

Capture temporarily reserves the encoded image and bounded encoding scratch from the same resident allowance, in addition to open state. Leave headroom instead of sizing open state to the full quarter: for example, if capture needs another 65% of steady state, open state must stay below roughly 60% of the resident allowance. This is an example, not a fixed ceiling; key widths and aggregate plans change the overhead. A state can fit during ingestion yet stop at every checkpoint or graceful shutdown if capture has no room. Increase the process budget or reduce retained state before reaching that point.

Checkpoints currently save the full resident working set, so capture can pause input and transfer costs grow with retained state. On object storage, configure cleanup of abandoned multipart uploads; completed abandoned data may remain until a later successful checkpoint collects it.

The Windows view identifies resident storage and reads committed open groups, panes and finalized series directly from the live worker. Bounded responses report truncation. Metrics turbine_resident_groups, turbine_resident_panes, turbine_resident_open_windows and turbine_resident_memory_bytes expose occupancy. A resident close timer represents one temporal window shared by its groups; its timer count is not a count of group/window entries.

Aggregate functions

Pass any aggregate function from turbine.aggregates to a window's value= parameter. The common shapes — Count(), Sum("col"), Mean("col"), Max/Min, percentile sketches — all share the same merge-across-batches contract; the dedicated Aggregate Functions page lists every one.

Column arguments accept either a string or a function expression: agg.Sum(fn.col("bytes_in") + fn.col("bytes_out")) is equivalent to materialising the derived column upstream and passing the name. See Functions for the full catalog.

Multi-Metric Aggregation

To compute something like SUM(errors) / COUNT(*) per window — the stream-processing equivalent of a SQL SUM(a)/COUNT(*) — use agg.Multi(...). Each keyword argument becomes an attribute on the value delivered to on_close_each:

value=agg.Multi(
    err_sum=agg.Sum("is_error"),
    lat_sum=agg.Sum("latency_ms"),
    n=agg.Count(),
),
on_close_each=lambda state, window, v: pa.RecordBatch.from_pylist([{
    "region": window.group["region"],
    "error_rate": v.err_sum / v.n,
    "mean_latency_ms": v.lat_sum / v.n,
    "window_start_ms": window.start_ms,
}]),

Bundling sub-aggregates inside Multi computes several metrics in lockstep. If a batch is missing any input column required by a member, the window update raises an exception. Unless your handler catches it, the subscription's error policy applies to the entire input batch; the default policy stops the partition. To tolerate missing columns, validate or fill them before updating the window.

An output name (a Multi sub key, an .alias(...), or the implicit value) must not collide with one of the window's key columns or the bound column window_start_ms (or window_end_ms, on session windows) — and symmetrically, a key column must not be named window_start_ms (or window_end_ms, on session windows): the table delivered to the close callbacks merges columns by name, so a colliding column would silently overwrite the other — the window refuses it with a ValueError at construction instead.

Custom Multi subclasses. A bundle using dictionary state that overrides only some dictionary calculation methods (partial, partials_by, merge, finalize) cannot combine them with members whose Arrow calculation methods are customized: the window refuses this ambiguous composition at construction and names the affected bundle and member. Keep a standard Multi and transform its output in the callback, or implement all four dictionary methods and own the bundle's complete state contract. Standard Multi compositions and bundles of dictionary-only members remain supported.

Referencing another window's result

A window's aggregate expression can also pull in another window's last-closed result for the same key — for example, falling back to a rolling 15-minute baseline when the current minute is sparse. The referenced window names its output (with .alias("name") or Multi), and the reader declares join_with=[that_window] and reads win.Ref("name").field inside its value=. References are persistent-windows-only and resolve to the last closed value (null until the first close, so pair them with fn.coalesce). The full mechanics — key/partition rules, ordering, recovery and cost — are in Windowing — advanced.

Window close callback

You pick one of two callback shapes when constructing a window — they're mutually exclusive, and the SDK errors at construction if you set both or neither. The choice is the same for every window kind.

on_close_each — one call per pane (the common shape)

def _emit(self, state: TurbineState, window: win.Window, value: float) -> RecordBatch | None:
    ...

Called once per (group, window) pane when the window closes:

• state — the same TurbineState your process method receives. Use it to read/write auxiliary state (e.g. cooldown machines).
• window — a Window(start_ms, end_ms, group) dataclass. group is a dict mapping each key field to its value. For session windows end_ms is data-driven (last event + gap), not start + size.
• value — the aggregate's finalized result. Its type depends on the aggregate: float for Count (integer-valued) and Mean, bool for AnyTrue and AllTrue, a namespace for Multi, etc. See Aggregate Functions for each aggregate's result type.

Return a pyarrow.RecordBatch to publish, or None to skip emission for this pane.

Use this when each closed window produces an independent output row and the cost is dominated by the per-pane logic (typical for alerts, dashboards, per-key emissions).

on_close — one call per batch of panes (the vectorised shape)

def _emit(self, state: TurbineState, panes: pa.Table) -> pa.RecordBatch | None:
    ...

Called once per tick with all panes that closed at the same time packed into a single Arrow Table. The table carries one row per pane, with the key columns, window_start_ms, window_end_ms, and the finalized aggregate columns.

Use this when:

• Many panes close together and the per-pane computation is amenable to vectorised PyArrow compute (pc.divide, pc.if_else, …). Computing a score over 10 000 panes in one Arrow pass is dramatically faster than 10 000 Python calls.
• The downstream emission is a single RecordBatch regardless of how many panes contributed.

Return a pyarrow.RecordBatch (the merged emissions for this tick) or None to skip the tick entirely.

The two callbacks are mutually exclusive on a given window: declare on_close_each for the per-pane shape or on_close for the batched one, not both.

Early-fire triggers (provisional emissions)

By default a window emits once, at close. For a long-lived window — a 5-minute revenue rollup, an hour-long session still receiving events — you often want to see the current aggregate before the window closes (a live dashboard, a "latest" gauge). Set early_fire_interval_ms= together with a separate early callback and the window will also emit its current-so-far aggregate every interval while it is still open. The close emission is unchanged — it still fires once, at close, and remains authoritative.

It is opt-in and available on all three window kinds. When unset, behaviour is exactly the single emission at close.

Two callbacks, not a flag

Early emission gets its own callback, distinct from the close one:

@app.subscribe(kafka.topic("orders", event_time="event_ts_ms"), output=kafka.topic("revenue_5m"))
class revenue:
    def __init__(self) -> None:
        self._win = win.PersistentTumbling(
            self,
            name="revenue_5m",
            size_ms=5 * 60_000,
            key=["region"],
            value=agg.Sum("amount"),
            allowed_lateness_ms=5_000,
            on_close_each=self._commit,           # 1×, authoritative, at close
            early_fire_interval_ms=10_000,        # every 10s while the window is open
            on_early_fire_each=self._live_gauge,  # n×, provisional, current-so-far
        )

    def _commit(self, state, window, value):      # authoritative total → ledger
        return pa.RecordBatch.from_pylist([{
            **window.group, "revenue": float(value), "final": True,
            "window_start_ms": window.start_ms, "window_end_ms": window.end_ms,
        }])

    def _live_gauge(self, state, window, value):   # running total → live dashboard
        return pa.RecordBatch.from_pylist([{
            **window.group, "revenue": float(value), "final": False,
            "window_start_ms": window.start_ms, "window_end_ms": window.end_ms,
        }])

    def process(self, batch: RecordBatch, state: TurbineState) -> RecordBatch | None:
        self._win.update(batch, state)
        return None

The early fire typically pushes a provisional value somewhere cheap (a gauge, a "latest" topic), while the close triggers the authoritative effect (commit, raise an alert). Two callbacks put that intent in the signature instead of an if is_final: branch. If you genuinely want identical handling, pass the same function to both.

Form mirroring + the rules

• The early callback mirrors the close form: pair on_close= (batch: state, panes) with on_early_fire=, and on_close_each= (per pane: state, window, value) with on_early_fire_each=. Same data shape on both doors. Mixing forms is rejected at construction.
• early_fire_interval_ms and the early callback are co-required (one without the other is an error), and the interval must be > 0. All of this is validated when the window is built, never at fire time.
• The early callback returns a RecordBatch to publish, or None to skip — exactly like the close callback. Returned batches are published to output like any other output.

What's guaranteed (and what isn't)

• Provisional vs authoritative. Anything from the early callback is best-effort by construction; the close emission is the authoritative one. Tag your output (a final column as above) or route the two callbacks to different topics.
• Ordering. For a given window, every early fire strictly precedes the single close, and nothing is emitted after the close. An early fire never duplicates or races the close.
• Sessions can supersede an early value. Session windows merge: a bridging event can fuse two open sessions into one. An early fire emitted for a session that later merges describes an aggregate that, in hindsight, no longer exists on its own — the on_close total supersedes it. There are no retractions; if a downstream cannot tolerate provisional double-counting across a merge, either don't wire the early callback for sessions or upsert on a stable key. Tumbling and sliding windows have fixed bounds and never merge, so their early fires are monotone refinements of the same window.
• A closed window never re-opens. Once a window has closed, an event belonging to it is dropped — or routed to late_data=, if you asked for that; either way it stays out of the window (see lateness). A given (group, window) produces exactly one close, ever. A downstream keyed on the group alone can still see close then an early fire for the same key, but that is a different window (a later one), not the closed one coming back.

Timing

Persistent windows use wall-clock timers for early fires, including when time_column= is set. Their final close follows the window's time mode: the watermark in event time, the wall clock in processing time. Early timers retain their wall-clock cadence when the partition goes silent.

Under exactly-once, early fires are outputs like any other: they ride the same per-batch transaction and are never observable twice. They do multiply output volume, so size the interval to your downstream's tolerance rather than making it as small as possible.

Time model: processing time, event time, lateness

Every window runs in one of two time modes, selected the same way regardless of kind.

Processing time (default). With no time_column, every row in an incoming batch is treated as having happened now (now_ms()). Use it when your data carries no timestamp or when clock skew doesn't matter. For session windows this means the gap is measured on wall-clock inactivity.

Event time. Each row is placed by its own timestamp — a single batch can fan out across windows, and out-of-order / late events are handled correctly (for sessions, a late event can even bridge two open sessions). The window closes when the partition watermark reaches window_end + allowed_lateness_ms. Historical timestamps do not make a session close immediately against today's wall clock; late rows can still merge into it while its event-time deadline remains ahead of the watermark. There are two ways to select the timestamp.

The recommended way is to declare it once on the input topic with event_time="<column path>". Turbine reads that field from each record, converts it to an internal timestamp, and every window under that subscription automatically uses it — you don't repeat time_column= on each window:

@app.subscribe(kafka.topic("events", event_time="event_ts", event_time_unit="ms"), output=kafka.topic("counts"))
class count_per_region:
    def __init__(self) -> None:
        self._win = win.PersistentTumbling(
            self, name="events_per_region", size_ms=60_000, key=["region"],
            value=agg.Count(), on_close_each=self._emit,
            allowed_lateness_ms=5_000,   # event-time inferred from the subscription
        )
    ...

event_time accepts a dotted path to reach a nested field (event_time="meta.event_ts"), and the source field may be any of:

Source field type	Handling
Integer (epoch)	Scaled to milliseconds per event_time_unit ("ms" default, or "s", "us", "ns").
Float (epoch seconds)	Same, scaled per event_time_unit.
Timestamp	Converted directly; event_time_unit is ignored.
RFC 3339 / ISO-8601 string	Parsed (e.g. "2021-01-01T00:00:00Z"); event_time_unit is ignored.

A record whose field is missing or whose string can't be parsed is dropped from windowing (counted, never fatal) — one malformed record won't stall the stream.

All event-time windows under one subscription must use its canonical clock, the clock driving its watermark. Declare the payload field through the input topic's event_time= and omit time_column= on the windows. If the watermark comes from Kafka timestamps instead, explicitly set time_column="_kafka_ts_ms"; see Driving windows off the Kafka timestamp. Omitting time_column in that Kafka-only configuration keeps processing-time windows.

A time_column that differs from the canonical column is refused at construction, for every window kind. Independent window clocks do not yet have their own recovery frontiers; using a shared frontier could reject valid rows or reopen closed windows after a restart. Use separate subscriptions with their own input-topic event_time= declarations if you need different clocks. The remaining key-group rescaling limitation is described under Recovery scope.

allowed_lateness_ms is how long after a window's end Turbine keeps accepting events that belong to it. It delays the close: the window stays open until window_end + allowed_lateness_ms in event time, and an out-of-order event arriving before that is an ordinary in-window event — it is merged normally and the window still emits exactly one authoritative result.

Once that deadline passes, the window closes and its state is released. An event that belongs to it and arrives after that point is dropped by default, and counted (see below). Turbine does not re-open a closed window: doing so would publish a second result for the same window carrying only the straggler, contradicting the one already sent — and it would put no bound at all on how far back a single old record could reach. So allowed_lateness_ms is the whole of your tolerance budget: size it for the worst upstream stall you want to absorb.

A window closed because its partition went quiet rejects stragglers just like one closed by data — Turbine tracks how far closing has actually got, so a straggler cannot re-open a window that already published. Rows rejected only for that reason are counted separately on turbine_late_rows_idle_closed_total; a non-zero value means part of what you are losing depends on your traffic pattern rather than on the data's own timestamps.

After a restart, persistent windows retain this protection on the same partition and the subscription's canonical event-time axis. Key-group rescaling is outside this recovery guarantee; see Recovery scope.

Late drops are never silent — they increment turbine_late_rows_dropped_total, labelled by worker and window name. A non-zero, growing value means your allowed_lateness_ms is smaller than your real out-of-orderness, and you are losing data you asked to keep. Alert on it.

Keeping late rows instead of dropping them

Dropping is the default, not the only option. Set on_late="route" on the subscription and name a destination with late_data= — every row past the deadline is published there instead of being discarded:

@app.subscribe(
    kafka.topic("events", event_time="ts"),
    output=kafka.topic("counts"),
    on_late="route",
    late_data=kafka.topic("events-late", message_key="tenant_id"),
)

What you get and what you give up:

• Nothing is lost, and no result is contradicted. The straggler leaves on its own topic; the window it belonged to keeps the single authoritative result it already published. This is the reason routing exists rather than re-opening: both avoid losing the row, but only one of them avoids emitting two conflicting answers for the same window.
• The routed record is the input row as Turbine decoded it, encoded like any other output, plus three columns describing the rejection: _late_window (the window's declared name), _late_window_end_ms (the end of the last window that could still have owned it) and _late_watermark_ms (how far the data's clock had already advanced). With several windows on one subscription, _late_window is what tells them apart. Turbine's internal columns are stripped.
• late_data must be its own topic. Pointing it at the subscription's input loops forever; pointing it at output= merges rejected rows into your result stream. Both are refused at startup.
• Delivery follows the subscription's guarantee. The routed rows are produced with the batch they were rejected from — under exactly_once they commit inside the same transaction as that batch's offsets, so a crash cannot leave a row routed-but-not-committed or committed-but-not-routed.
• Routed rows are counted separately, on turbine_late_rows_routed_total{worker, window} — turbine_late_rows_dropped_total stays for the rows a "drop" subscription discards, so switching policy reads as one series going quiet and another waking up.
• Session windows use a different, more permissive rule — see below. on_late applies to them too.

late_data is one of the extra destinations a subscription can declare; the others (side_outputs= for imperative ctx.emit, branch= for a declared split) work the same way and are covered in Side outputs.

Lateness for session windows

A session has no fixed end: activity keeps extending it. So "is this row late?" is decided on the session the row would land in after merging, not on the row's own timestamp plus the gap.

The practical consequence is worth stating plainly: a session window accepts rows far older than allowed_lateness_ms, as long as they fall inside a session that is still open. If a session has been kept alive for an hour by continuous activity, an event from forty minutes ago still belongs to it and is merged normally. Only once that session has closed do rows belonging to it become late.

This is deliberate, and it is what gap_ms means: a session is bounded by inactivity, not by the clock. Judging a straggler on timestamp + gap_ms would drop exactly the out-of-order rows that bridging exists to absorb.

What you get is the same guarantee as the fixed windows: a (group, session) produces exactly one result, and a row arriving after that result is dropped or routed — never merged into a second, overlapping session for the same key.

allowed_lateness_ms also doubles as the failover-completeness budget for persistent windows (see Failover-resilient windows).

How an event-time window decides to close

In event-time mode a window closes when the data's own clock reaches its end — not when your wall clock does. Turbine tracks, per partition, the largest event timestamp it has seen (its watermark); a window [start, end) closes once that watermark passes end + allowed_lateness_ms. Two consequences worth knowing:

• Out-of-order and historical data just work. Replaying a day-old backfill closes its windows as the watermark sweeps through the old timestamps — at full speed, not pinned to real time. A burst of future-dated events closes every earlier window up to their timestamp in one step.
• A window still closes when the stream goes quiet. If events stop arriving, the watermark can't advance from data — so Turbine lets it drift forward at real time from the last event seen. In practice a window then closes after roughly the part of it not yet covered by data, plus allowed_lateness_ms, of real silence. Example: a 60-second window whose last event landed 20 seconds in waits about 40 more seconds of silence before closing. A window that's already fully covered by data closes immediately. This guarantees no window stays open forever on an idle partition, while never closing one early just because the wall clock moved on.

Driving windows off the Kafka timestamp

If your producer doesn't embed a timestamp in the payload, use the Kafka broker timestamp. Opt in on the input topic with with_kafka_timestamp=True; Turbine injects an Int64 _kafka_ts_ms column on every batch, which you then pass as time_column="_kafka_ts_ms".

@app.subscribe(kafka.topic("events", with_kafka_timestamp=True), output=kafka.topic("counts"))
class count_per_region:
    def __init__(self) -> None:
        self._win = win.PersistentTumbling(
            self, name="events_per_region", size_ms=60_000, key=["region"],
            value=agg.Count(), on_close_each=self._emit,
            time_column="_kafka_ts_ms", allowed_lateness_ms=5_000,
        )
    ...

Two broker-side configs shape what _kafka_ts_ms means:

Topic config	Source of timestamp
CreateTime (default)	Producer clock at send. Reflects event time if the producer is close to the source, but vulnerable to producer clock skew.
LogAppendTime	Broker clock at append. Monotone per-partition, but closer to processing time than true event time.

Use LogAppendTime when you don't trust producer clocks; use CreateTime (or a payload column) when the producer is authoritative.

Naming a window: name= vs label=

Every window takes a name=. It is an identity, not a title: it prefixes every state and timer key the window owns, so changing it abandons the partial aggregates, panes and timers already on disk — the window starts from nothing, and whatever was mid-flight is never emitted.

That is fine when you write the name yourself. It is a problem when windows come from configuration — one per rule, per tenant, per customer — because the only stable thing to name them after is an id, and an id is exactly what nobody can read. A console listing alert-rule:17761784b6 a hundred times tells you nothing about what any of them watches.

label= separates the two. It is display metadata: it never appears in a key, it can be anything, and it can change at any time.

self._win = win.PersistentSliding(
    self,
    name=f"alert-rule:{rule.id}",   # identity — stable for the life of the state
    label=rule.title,                # what a human reads: "Net latency per runner"
    size_ms=600_000, slide_ms=60_000,
    key=["tenant_id", "runner_id"],
    value=agg.Mean("latency_ms"),
    on_close_each=self._emit,
)

The console shows the label as the window's title and keeps the name visible underneath — the name is what your logs, alert payloads and API calls carry, so it must stay findable. GET /windows returns both.

If your configuration is reloaded while the app runs, call window.set_label(rule.title) when a title changes. It republishes the metadata — no key is touched, nothing is lost, and the console catches up on its next poll instead of at the next restart.

Reading a window while it is open

A window's value is normally something you receive: the close callback fires and the result goes downstream. But while a window is open — and especially when you are asking why one fired and its neighbour didn't — you often want the numbers behind it.

GET /windows/{name}/series answers exactly that for one group: the value of each pane a sliding window is aggregating (or of each window a group has open), plus the aggregate they currently roll up to. Values come back finalized — a mean is a number — so nothing on the reading side needs to know how the aggregate function stores itself. The REST API reference has the full shape, and the console plots the same data as a curve.

The series covers what is still open: a sliding window's held panes (the last size_ms), or the windows not yet closed. Closed windows leave state as they emit; their history lives in the output stream you sent them to.