Read wire facts from dau-core and drop the undriven status bits

dau_driver/capability_decode.py

@@ -11,18 +11,21 @@
 
 from dau_core.capabilities import OperatorCapability
 from dau_core.capability_contract import CapabilityContract
-from dau_core.registers import DEFAULT_STREAM_JOB_REGISTER_CONTRACT, RegisterBlockSnapshot, RegisterMapVersion, RegisterOffset, StreamProtocolVersion
+from dau_core.registers import (
+    CAPABILITY_WORDS_REGISTER_MAP_VERSION,
+    DEFAULT_STREAM_JOB_REGISTER_CONTRACT,
+    JOB_CAPACITY_REGISTER_MAP_VERSION,
+    RegisterBlockSnapshot,
+    RegisterMapVersion,
+    RegisterOffset,
+    StreamProtocolVersion,
+)
 from dau_core.stream import LogicalType, OperationCode, output_type_for
 
 # the register-level type claim of the bitmap words — every
 # opcode-driven tile in the palette is INT32 (per-tile types travel in the
 # manifest capability contract, not in registers)
 REGISTER_BITMAP_INPUT_TYPES = (LogicalType.INT32,)
-CAPABILITY_WORDS_REGISTER_MAP_VERSION = RegisterMapVersion(0, 2)
-# job capacity arrived in 0.3. A 0.2 shell does not decode those offsets, so
-# they are not read from one -- it reports no capacity and the host refuses
-# to batch rather than guessing at depths it cannot see.
-JOB_CAPACITY_REGISTER_MAP_VERSION = RegisterMapVersion(0, 3)
 # identity window candidates: shell base, then the stream-job contract window
 REGISTER_WINDOW_OFFSETS = (0x0, DEFAULT_STREAM_JOB_REGISTER_CONTRACT.register_window_offset)
 

dau_driver/pipeline_execution.py

@@ -13,12 +13,21 @@
 from dataclasses import dataclass
 
 from dau_core.operators import FilterPredicate, GroupAggregate, MapConfig
+from dau_core.records import (
+    BAR_RECORD_BYTES,
+    BAR_RECORD_FIELDS,
+    BAR_ROW_BYTES,
+    BAR_ROW_FIELDS,
+    GROUP_RECORD_BYTES,
+    GROUP_RECORD_FIELDS,
+    KEYED_ROW_FIELDS,
+    struct_format,
+)
 from dau_core.registers import JobControlFlag, JobStatusFlag, RegisterOffset, StreamJobRegisterContract
 
 from .device import write_address
 from .errors import RegisterAccessError
 
-GROUP_RECORD_BYTES = 40
 DMA_ALIGN_BYTES = 64
 
 
@@ -71,14 +80,14 @@ def map_constant_word(self) -> int:
 def pack_keyed_rows(rows: list[tuple[int, int]]) -> bytes:
     """{value, key} pair rows on the 64-byte DMA grid; the logical length
     stays ``len(rows) * 8``."""
-    payload = b"".join(struct.pack("<iI", key, value & 0xFFFFFFFF) for key, value in rows)
+    payload = b"".join(struct.pack(struct_format(KEYED_ROW_FIELDS), key, value & 0xFFFFFFFF) for key, value in rows)
     return payload + b"\x00" * ((-len(payload)) % DMA_ALIGN_BYTES)
 
 
 def decode_group_records(data: bytes, count: int) -> list[GroupAggregate]:
     groups = []
     for offset in range(0, count * GROUP_RECORD_BYTES, GROUP_RECORD_BYTES):
-        key, first, last, minimum, maximum, _pad, total, rows = struct.unpack_from("<iiiiiiqQ", data, offset)
+        key, first, last, minimum, maximum, _pad, total, rows = struct.unpack_from(struct_format(GROUP_RECORD_FIELDS), data, offset)
         groups.append(GroupAggregate(key=key, first=first, last=last, minimum=minimum, maximum=maximum, total=total, count=rows))
     return groups
 
@@ -192,36 +201,6 @@ def execute_keyed_pipeline_job(
 # meant a padding, signedness or field-order change could make the two halves
 # read one bitstream differently, so the shapes live here and both sides
 # derive from them.
-_STRUCT_CODE = {"i4": "i", "i8": "q", "u8": "Q"}
-
-BAR_ROW_FIELDS: tuple[tuple[str, str], ...] = (("key", "i4"), ("price", "i4"), ("shares", "i4"), ("frac", "i4"))
-BAR_RECORD_FIELDS: tuple[tuple[str, str], ...] = (
-    ("key", "i4"),
-    ("open", "i4"),
-    ("close", "i4"),
-    ("low", "i4"),
-    ("high", "i4"),
-    ("pad", "i4"),
-    ("shares", "i8"),
-    ("frac", "i8"),
-    ("count", "u8"),
-)
-
-
-def struct_format(fields: tuple[tuple[str, str], ...]) -> str:
-    """The little-endian ``struct`` format for a declared field layout."""
-    return "<" + "".join(_STRUCT_CODE[kind] for _name, kind in fields)
-
-
-def numpy_dtype_fields(fields: tuple[tuple[str, str], ...]) -> list[tuple[str, str]]:
-    """The same layout as numpy dtype pairs, for vectorized consumers."""
-    return [(name, "<" + kind) for name, kind in fields]
-
-
-BAR_ROW_BYTES = 16
-BAR_RECORD_BYTES = 48
-
-
 def pack_bar_rows(rows: list[tuple[int, int, int, int]]) -> bytes:
     """(key, price, shares, frac) rows for the fused bar tile, on the DMA
     grid; the logical length stays ``len(rows) * 16``."""
@@ -441,7 +420,9 @@ def device_job_capacity(device) -> tuple[int, int]:
     reflashed inside one session -- the flywheel does exactly that -- and a
     cached capacity would then describe the previous shell.
     """
-    from .capability_decode import JOB_CAPACITY_REGISTER_MAP_VERSION, decode_register_map_version
+    from dau_core.registers import JOB_CAPACITY_REGISTER_MAP_VERSION
+
+    from .capability_decode import decode_register_map_version
 
     version = decode_register_map_version(device.read_register(RegisterOffset.REGISTER_MAP_VERSION))
     if version < JOB_CAPACITY_REGISTER_MAP_VERSION:

dau_driver/status.py

@@ -2,33 +2,13 @@
 
 from dataclasses import dataclass
 
-LINK_READY_BIT = 1 << 0
-MIG_READY_BIT = 1 << 1
-OPERATOR_READY_BIT = 1 << 2
-RESET_ACTIVE_BIT = 1 << 3
-
 
 @dataclass(frozen=True)
 class DeviceStatus:
+    # STATUS as read. No shell assigns its bits yet, so it is not decoded
     raw_status: int
     last_error: int
 
-    @property
-    def link_ready(self) -> bool:
-        return bool(self.raw_status & LINK_READY_BIT)
-
-    @property
-    def mig_ready(self) -> bool:
-        return bool(self.raw_status & MIG_READY_BIT)
-
-    @property
-    def operator_ready(self) -> bool:
-        return bool(self.raw_status & OPERATOR_READY_BIT)
-
-    @property
-    def reset_active(self) -> bool:
-        return bool(self.raw_status & RESET_ACTIVE_BIT)
-
 
 @dataclass(frozen=True)
 class DeviceCounters:

dau_driver/stream_codec.py

@@ -6,6 +6,7 @@
 
 from dau_core.stream import (
     DEFAULT_RECORD_BATCH_LIMITS,
+    LOGICAL_TYPE_BYTE_WIDTHS,
     BatchHeader,
     ColumnDescriptor,
     InvalidStreamError,
@@ -17,18 +18,6 @@
 )
 
 _ALIGNMENT_BYTES = 64
-_SUPPORTED_WIDTHS_BYTES = {
-    LogicalType.INT8: 1,
-    LogicalType.INT16: 2,
-    LogicalType.INT32: 4,
-    LogicalType.INT64: 8,
-    LogicalType.UINT8: 1,
-    LogicalType.UINT16: 2,
-    LogicalType.UINT32: 4,
-    LogicalType.UINT64: 8,
-    LogicalType.FLOAT32: 4,
-    LogicalType.FLOAT64: 8,
-}
 _SCALAR_RESULT_STRUCTS = {
     LogicalType.INT8: struct.Struct("<b"),
     LogicalType.INT16: struct.Struct("<h"),
@@ -140,7 +129,7 @@ def encode_column(
         payload_bytes = bytes(payload)
         if row_count < 0:
             raise InvalidStreamError("row_count cannot be negative")
-        width_bytes = _SUPPORTED_WIDTHS_BYTES[normalized_type]
+        width_bytes = LOGICAL_TYPE_BYTE_WIDTHS[normalized_type]
         expected_payload_length = row_count * width_bytes
         if len(payload_bytes) != expected_payload_length:
             raise InvalidStreamError(
@@ -203,7 +192,7 @@ def encode_record_batch(
         offset = first_payload_offset
         for column in columns:
             normalized_type = _normalize_supported_type(column.logical_type)
-            width_bytes = _SUPPORTED_WIDTHS_BYTES[normalized_type]
+            width_bytes = LOGICAL_TYPE_BYTE_WIDTHS[normalized_type]
             payload_bytes = bytes(column.payload)
             if len(payload_bytes) != row_count * width_bytes:
                 raise InvalidStreamError(
@@ -264,7 +253,7 @@ def encode_arrow_batch(
                 raise InvalidStreamError(f"column {name!r} has nulls; record batches must be null-free")
             # zero-copy view of the primitive data buffer (offset-aware for
             # sliced arrays); one copy happens at stream assembly
-            width = _SUPPORTED_WIDTHS_BYTES[logical_type]
+            width = LOGICAL_TYPE_BYTE_WIDTHS[logical_type]
             start = array.offset * width
             payload = bytes(memoryview(array.buffers()[1])[start : start + len(array) * width])
             columns.append(RecordBatchColumn(field_id=field_id, logical_type=logical_type, payload=payload))
@@ -332,10 +321,10 @@ def decode_record_batch(data: bytes, *, limits: RecordBatchLimits = DEFAULT_RECO
             _normalize_supported_type(column.logical_type)
             if column.null_count != 0 or column.validity_length != 0:
                 raise InvalidStreamError("record-batch decode does not support validity bitmaps yet")
-            expected_width_bits = _SUPPORTED_WIDTHS_BYTES[column.logical_type] * 8
+            expected_width_bits = LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type] * 8
             if column.physical_width_bits != expected_width_bits:
                 raise InvalidStreamError(f"physical_width_bits must be {expected_width_bits} for {column.logical_type.name.lower()}")
-            if column.payload_length != column.row_count * _SUPPORTED_WIDTHS_BYTES[column.logical_type]:
+            if column.payload_length != column.row_count * LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type]:
                 raise InvalidStreamError(f"column {column.field_id} payload length does not match row_count")
             _validate_payload_window(column.payload_offset, column.payload_length, len(encoded))
         validate_record_batch(header, columns, limits=limits)
@@ -399,7 +388,7 @@ def _validate_header(header: BatchHeader, encoded_length: int) -> None:
 
 def _validate_column(header: BatchHeader, column: ColumnDescriptor, encoded_length: int) -> None:
     _normalize_supported_type(column.logical_type)
-    expected_width_bits = _SUPPORTED_WIDTHS_BYTES[column.logical_type] * 8
+    expected_width_bits = LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type] * 8
     if column.physical_width_bits != expected_width_bits:
         raise InvalidStreamError(f"physical_width_bits must be {expected_width_bits} for {column.logical_type.name.lower()}")
     if column.null_count != 0 or column.validity_length != 0:
@@ -408,7 +397,7 @@ def _validate_column(header: BatchHeader, column: ColumnDescriptor, encoded_leng
         raise InvalidStreamError("column row_count does not match batch header")
     if column.payload_offset != header.payload_offset or column.payload_length != header.payload_length:
         raise InvalidStreamError("column payload window does not match batch header")
-    expected_payload_length = column.row_count * _SUPPORTED_WIDTHS_BYTES[column.logical_type]
+    expected_payload_length = column.row_count * LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type]
     if column.payload_length != expected_payload_length:
         raise InvalidStreamError("column payload length does not match row_count")
     _validate_payload_window(column.payload_offset, column.payload_length, encoded_length)
@@ -435,7 +424,7 @@ def _normalize_supported_type(logical_type: LogicalType) -> LogicalType:
         normalized_type = LogicalType(logical_type)
     except ValueError as exc:
         raise InvalidStreamError(f"unsupported logical type {logical_type!r}") from exc
-    if normalized_type not in _SUPPORTED_WIDTHS_BYTES:
+    if normalized_type not in LOGICAL_TYPE_BYTE_WIDTHS:
         raise InvalidStreamError(f"unsupported logical type {normalized_type.name.lower()}")
     return normalized_type
 

dau_driver/tests/test_discovery.py

@@ -216,10 +216,7 @@ def test_device_decodes_status_and_counters_from_register_block(tmp_path: Path)
     status = device.read_status()
     counters = device.read_counters()
 
-    assert status.link_ready
-    assert status.mig_ready
-    assert status.operator_ready
-    assert not status.reset_active
+    assert status.raw_status == 0b0111
     assert status.last_error == 0x42
     assert counters.h2c_bytes == 0x1_0000_0040
     assert counters.c2h_bytes == 0x2_0000_0080

dau_driver/tests/test_no_board_geometry_constants.py

@@ -30,9 +30,6 @@
 # or RECORD shape -- fixed by the format, identical on every board.
 WIRE_SHAPE_CONSTANTS = {
     "DMA_ALIGN_BYTES": "the DMA engine's alignment, a property of the transfer format",
-    "GROUP_RECORD_BYTES": "bytes per grouped-aggregation record on the wire",
-    "BAR_ROW_BYTES": "bytes per bar input row on the wire",
-    "BAR_RECORD_BYTES": "bytes per bar output record on the wire",
 }
 
 # Constants that DO describe board or shell geometry and have not moved to

dau_driver/tests/test_pipeline_execution.py

@@ -14,6 +14,7 @@
 
 import pytest
 from dau_core.operators import BarAggregate, GroupAggregate
+from dau_core.records import BAR_RECORD_BYTES, GROUP_RECORD_BYTES
 from dau_core.registers import (
     JobStatusFlag,
     NocLaneRegisterOffset,
@@ -24,8 +25,6 @@
 )
 
 from dau_driver.pipeline_execution import (
-    BAR_RECORD_BYTES,
-    GROUP_RECORD_BYTES,
     PipelineJobError,
     bar_noc_lane_output_address,
     decode_bar_records,
@@ -387,35 +386,6 @@ def test_execute_bar_noc_packed_error(tmp_path):
     assert str(exc.value) == "bar NoC job failed: status=0xc last_error=1"
 
 
-def test_the_bar_layout_has_one_declaration() -> None:
-    """dau-polars builds numpy dtypes over these same bytes. When the layout
-    was declared in both places, a padding or field-order change could make
-    the two halves read one bitstream differently — so the derived formats
-    must reproduce the wire shapes exactly, and the byte sizes the job
-    contract advertises must follow from the fields rather than sit beside
-    them."""
-    import struct as _struct
-
-    from dau_driver.pipeline_execution import (
-        BAR_RECORD_BYTES,
-        BAR_RECORD_FIELDS,
-        BAR_ROW_BYTES,
-        BAR_ROW_FIELDS,
-        numpy_dtype_fields,
-        struct_format,
-    )
-
-    assert struct_format(BAR_ROW_FIELDS) == "<iiii"
-    assert struct_format(BAR_RECORD_FIELDS) == "<iiiiiiqqQ"
-    assert _struct.calcsize(struct_format(BAR_ROW_FIELDS)) == BAR_ROW_BYTES
-    assert _struct.calcsize(struct_format(BAR_RECORD_FIELDS)) == BAR_RECORD_BYTES
-
-    # the vectorized view must describe the identical bytes
-    numpy = pytest.importorskip("numpy")
-    assert numpy.dtype(numpy_dtype_fields(BAR_RECORD_FIELDS)).itemsize == BAR_RECORD_BYTES
-    assert [name for name, _ in BAR_RECORD_FIELDS] == ["key", "open", "close", "low", "high", "pad", "shares", "frac", "count"]
-
-
 def test_addresses_are_programmed_across_both_halves() -> None:
     """A wide design keeps the high half across a LOW write, so a host that
     wrote only LOW would inherit whatever the previous job left there and

docs/wiki/reference/Device-and-Discovery.md

@@ -169,8 +169,8 @@ direction's channel count is zero. `DauDevice` passes its own `memory_bytes`.
 | `raw_status` | `int` |
 | `last_error` | `int` |
 
-Properties decoding `raw_status`: `link_ready` (bit 0), `mig_ready` (bit 1),
-`operator_ready` (bit 2), `reset_active` (bit 3).
+`raw_status` is the STATUS word as read. No shell assigns its bits yet, so it
+is not decoded into named flags.
 
 ### `DeviceCounters`
 
@@ -196,7 +196,8 @@ hardware results. Host opcodes must never be submitted to the device. The
 composed-capability words (`LANE_COUNT`, `HOST_OPCODE_BITMAP`, `SORT_CAPACITY`)
 are only read on 0.2 and above; all-ones is normalized to not-advertised. The
 job-capacity words (`INPUT_ROW_CAPACITY`, `OUTPUT_RECORD_CAPACITY`) need 0.3 and
-above — `JOB_CAPACITY_REGISTER_MAP_VERSION`.
+above. Both gates, `CAPABILITY_WORDS_REGISTER_MAP_VERSION` and
+`JOB_CAPACITY_REGISTER_MAP_VERSION`, are defined in `dau_core.registers`.
 
 ## Storage
 

docs/wiki/reference/Execution-Contracts.md

@@ -257,9 +257,9 @@ identity.
 | `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)`.      |
 
-Constants: `GROUP_RECORD_BYTES = 40`, `DMA_ALIGN_BYTES = 64`. These are **wire
-and record shapes** — fixed by the format both sides compile against, not by
-the board.
+`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
@@ -270,10 +270,12 @@ module-level constants exactly, in both directions.
 
 ### Bar jobs
 
-The bar wire layout is declared once, in `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.
+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                                                                                                                     |
 | ------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
@@ -286,7 +288,7 @@ cannot make them read one bitstream differently.
 | `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.                                                                             |
 
-Constants: `BAR_ROW_BYTES = 16`, `BAR_RECORD_BYTES = 48`. Both are wire shapes.
+`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