# Execution contracts reference `dau_driver.stream_execution`, `dau_driver.scan_execution`, `dau_driver.pipeline_execution`, and `dau_driver.fleet`. Each module implements one shell's register contract; register offsets and flags come from `dau_core.registers`. Every runner polls `JOB_STATUS` through one shared wait, so timeout and error semantics are identical across shells: `PipelineJobError` on timeout, and `PipelineJobError` including `LAST_ERROR` when the status carries the error flag. ## Single stream job ### `execute_scalar_operation` ```python execute_scalar_operation( device: DauDevice, logical_type: LogicalType, payload: bytes, *, row_count: int, operation: OperationDescriptor, input_device_address: int, output_device_address: int, output_byte_count: int, trigger: StreamTrigger | None = None, ) -> DecodedStreamResult ``` Validate the operation against the device's capabilities, encode the payload as a single-column stream batch, write it H2C, invoke `trigger`, read the result C2H, and decode it as a scalar. Raises `StreamExecutionError` before any DMA when the device advertises no operator for `(opcode, input type)`, or when the advertised output type differs from the descriptor's. `StreamTrigger` is `Callable[[DauDevice, bytes], None]`, called with the device and the encoded input. ## Scan jobs ### `ScanBatch` | Field | Type | Description | | ----------- | ------- | -------------------------------------------------------- | | `payload` | `bytes` | The packed row payload, exactly `row_count * row_bytes`. | | `row_count` | `int` | Rows in the batch. | ### `run_scan_batches` ```python run_scan_batches( device, batches: Iterable[ScanBatch], *, per_key_state: bool, input_regions: tuple[int, int], row_bytes: int, collect: Callable[[Any], Any], timeout_seconds: float = 60.0, progress=None, poll_yield_seconds: float | None = None, ) -> list[Any] ``` Run one job per batch with ping-pong input staging: batch *i + 1* DMAs into the alternate region (and, for a lazy iterable, is produced) while job *i* computes. Returns the `collect(device)` values in batch order. Per batch the runner writes `INPUT_ADDRESS_LOW`/`HIGH` and `INPUT_LENGTH_LOW`, sets `JOB_CONTROL` to `START`, waits for DONE, then calls `collect`. The DMA is padded to the 64-byte grid while `INPUT_LENGTH` stays exact. | Parameter | Notes | | -------------------- | -------------------------------------------------------------------------------------------------------------- | | `per_key_state` | Required. When True, `batches` must be the result of `key_partitioned_scan_batches` or `ValueError` is raised. | | `input_regions` | Two distinct device addresses. Equal addresses raise `ValueError`. | | `row_bytes` | Must be positive. | | `collect` | Called once per batch after DONE, with the device. | | `progress` | Called with the running completion count after each batch. | | `poll_yield_seconds` | Sleep per status poll. `None` spins. | Raises `ValueError` for a zero-row batch and for a payload whose length is not `row_count * row_bytes`. The caller owns everything shell-specific: lane output addresses and static configuration are written before the run, and `collect` reads the results. Output regions are disjoint from input staging, so collection overlaps the next DMA. ### `run_resident_scan` ```python run_resident_scan( device, *, address: int, row_count: int, row_bytes: int, collect: Callable[[Any], Any], repeats: int = 1, timeout_seconds: float = 60.0, ) -> list[Any] ``` Run the job over rows already on the device, uploading nothing. Per repeat the runner writes `INPUT_ADDRESS_LOW`/`HIGH` and `INPUT_LENGTH_LOW`, sets `JOB_CONTROL` to `START`, waits for DONE, and calls `collect`. Returns one `collect(device)` per repeat rather than one merged answer, so a caller can check that every run returns identical results instead of assuming it. `row_count`, `row_bytes`, and `repeats` must be positive, and `address` must be on the `DMA_ALIGN_BYTES` grid, because an unaligned region returns wrong bytes on silicon without any error. All four raise `ValueError`. `address` is raw, and a raw address outlives the allocation it came from. `HandleTable.run_resident_scan` is this same run with a generation-checked handle in its place; see the handle table in [device and discovery](device-and-discovery.md). ### `key_partitioned_scan_batches` ```python key_partitioned_scan_batches( records: Iterable[Record], *, batch_count: int, key: Callable[[Record], Hashable], encode: Callable[[Iterable[Record]], bytes], ) -> Iterator[ScanBatch] ``` Build key-confined batches by longest-processing-time-first greedy assignment: the largest key group goes into the currently smallest batch. Records must be ordered by key: a key appearing in more than one run raises `ValueError`. Record order within each key is preserved. Creating the iterable consumes nothing. First iteration consumes one complete grouping pass, because LPT needs every group size before assignment; each batch's payload is built only when that batch is requested. Key groups are indivisible, so one hot key can dominate a batch. `batch_count` must be positive. The number of batches produced is `min(batch_count, number of key groups)`. ### `key_range_scan_batches` ```python key_range_scan_batches( records: Iterable[Record], *, key_capacity: int, key: Callable[[Record], Hashable], encode: Callable[[Iterable[Record]], bytes], ) -> Iterator[ScanBatch] ``` Split records into batches each holding at most `key_capacity` distinct keys, the host half of the hash-mode grouped tile's overflow contract. The sorted key space is partitioned into ranges of `key_capacity` keys and one batch is emitted per range. Records need not arrive in key order, and arrival order is preserved within each batch. Each key lands in exactly one batch, so combining the per-batch group records is a concatenation; use `dau_core.operators.merge_hash_grouped` for the general additive merge when batches were produced some other way. Costs a counting pass over every key before the first batch. Row counts may be uneven: key ranges are equal in keys, not rows. `key_capacity` must be positive. ## Fleet ### `run_scan_batches_fleet` ```python run_scan_batches_fleet( devices, batches: Iterable[ScanBatch], *, per_key_state: bool, input_regions: tuple[int, int], row_bytes: int, collect: Callable[[Any], Any], timeout_seconds: float = 60.0, progress=None, weights: Sequence[float] | None = None, poll_yield_seconds: float | None = None, ) -> list[FleetScanResult] ``` Run one independent `run_scan_batches` pipeline per device over one shared lazy input. Each batch is pulled exactly once by whichever pipeline demands it, so with `weights=None` scheduling is purely demand-driven and self-balances by device throughput. Results are returned in global batch order. `input_regions` is reused as device-local addresses on every card, never shared memory. `per_key_state` must be `False`; True raises `ValueError`. A stateful composition has no fleet form: a key's rows split across cards that never see each other's state. `weights` takes one positive, finite value per device and pre-biases each shared pull by weighted round-robin; batches remain lazy and are never partitioned or queued ahead. A length mismatch or a non-positive or non-finite weight raises `ValueError`. An empty batch iterable returns `[]`. An empty device sequence raises `ValueError`. A single device runs the same pipeline and adds ordered provenance. Known limitation: cancelling on a device failure wakes pipelines blocked pulling a batch, but one already inside its status wait runs to its own completion or timeout first, so an early failure can surface only after its peers finish. ### `FleetScanResult` | Field | Type | Description | | ------------- | ----- | ------------------------------------------------ | | `batch_index` | `int` | Global batch index. | | `device` | `Any` | The exact device object that produced the value. | | `value` | `Any` | The `collect` return value. | ### `FleetScanError` `RuntimeError` annotated with `batch_index` (`None` when the failure preceded any batch) and `device`. ## Keyed pipeline and bar jobs The `dau_mm_ohlcv_job` and bar shells run keyed `{value, key}` INT32 pair rows through filter, map, and ordered grouped aggregation, staging fixed-width records in the output region. ### `KeyedPipelineConfig` A host-side mirror of the `PIPELINE_*` configuration registers, as a dataclass. | Field | Type | Default | | --------------- | ------------------- | ------- | | `predicate` | `FilterPredicate` | — | | `filter_on_key` | `bool` | `False` | | `map_config` | `MapConfig \| None` | `None` | | `map_on_key` | `bool` | `False` | `filter_op_word()`, `map_op_word()`, and `map_constant_word()` produce the packed register values. A `None` `map_config` packs as ADD 0 on the value lane, the identity. `configure_pipeline(device, config)` writes all five `PIPELINE_*` registers. ### Keyed pipeline execution | Function | Behaviour | | ------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------- | | `pack_keyed_rows(rows)` | `(key, value)` pairs packed onto the 64-byte grid; the logical length stays `len(rows) * 8`. | | `decode_group_records(data, count)` | 40-byte group records as `GroupAggregate` values. | | `execute_keyed_pipeline_job(device, rows, *, timeout_seconds=5.0, contract=None, max_rows=None)` | One batch through a configured pipeline. Returns `(group records, selected row count)`. | `DMA_ALIGN_BYTES = 64` is a **wire shape**: fixed by the transfer format, not by the board. The record layouts and their sizes (`GROUP_RECORD_BYTES` is 40) are defined in `dau_core.records`, next to the tiles that produce them. Capacity is not a constant. `device_job_capacity(device)` reads the input row and output record capacities the shell advertises; exceeding them raises `PipelineJobError`. A module constant naming a capacity would be a second source of truth for something the platform already declares, and `dau_driver/tests/test_no_board_geometry_constants.py` asserts the allowed module-level constants exactly, in both directions. ### Bar jobs The bar wire layout is declared once, in `dau_core.records` as `BAR_ROW_FIELDS` and `BAR_RECORD_FIELDS`, with `struct_format(fields)` and `numpy_dtype_fields(fields)` deriving the two consumers' views so a padding, signedness, or field-order change cannot make them read one bitstream differently. dau-core's suite checks each layout's size against the tile's registry entry. | Function | Behaviour | | ------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | | `pack_bar_rows(rows)` | `(key, price, shares, frac)` rows on the DMA grid. | | `decode_bar_records(data, count)` | 48-byte records as `BarAggregate` values. | | `execute_bar_job(device, rows, *, timeout_seconds=5.0, contract=None, max_rows=None)` | One batch through the SRAM-staged fused bar shell. | | `stage_bar_ddr_input(device, payload, row_count, *, staging, input_address=None)` | DMA one pre-packed batch into a memory input region. | | `start_bar_ddr_job(device, row_count, *, staging, input_address=None, output_address=None)` | Program the addresses and length, then start. | | `finish_bar_ddr_job(device, *, staging, output_address=None, timeout_seconds=60.0)` | Wait and read back. Returns `(raw record bytes, count)`; decoding stays with the caller so hosts can use vectorized decoders. | | `execute_bar_ddr_packed(device, payload, row_count, *, staging, timeout_seconds=60.0)` | Stage, start, finish. | | `bar_ddr_max_rows_per_job(staging)` | The row budget the given staging layout supports. | `BAR_ROW_BYTES` (16) and `BAR_RECORD_BYTES` (48) come from the same module. `staging` is the board's `DdrStaging` (its input regions, output region, sizes, and alignment), passed in from the composed platform. It is a keyword argument rather than a module constant because addresses and region sizes are board geometry, and a package-level default is how one board's memory map ends up applied to another. Rows per job are bounded by the worst case of one record per row, so an output overrun is impossible by construction. Staging addresses must meet the layout's declared alignment or `PipelineJobError` is raised. ### Broadcast-NoC bar shell One scan computes every lane's partition concurrently, each lane writing its own slice of the output region. | Name | Behaviour | | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | | `device_lane_count(device)` | The lane count the shell advertises. | | `bar_noc_lane_output_bytes(staging, lanes)`, `bar_noc_max_rows_per_job(staging, lanes)` | The per-lane output slice and row budget for a staging layout. | | `bar_noc_lane_output_address(staging, lane, lanes)` | One lane's output base. | | `configure_bar_noc_lanes(device, *, staging, lanes=None)` | Point every lane's writer at its slice. | | `execute_bar_noc_packed(device, payload, row_count, *, staging, lanes=None, record_bytes=BAR_RECORD_BYTES, validate_record_count=True, timeout_seconds=60.0)` | One pre-packed batch through a configured NoC shell. Returns `(raw record bytes, count)` per lane. | A `lanes` of `None` asks the device how many it has, rather than assuming. A host that assumed four lanes on a shell advertising its own count is the reason the constant is gone. `record_bytes` defaults to the fused bar shell's record; key-partitioned scan shells reuse the runner with their own width. Pass `validate_record_count=False` for a lane whose count register latches an *input* row count and emits one fixed-width record. The `result_length == count * record_bytes` invariant does not hold there, so only `RESULT_LENGTH` is used.