Skip to content

Streaming Core/Halo Pipeline

RFC 0008 streaming architecture: full-chip processing where layout growth increases tile count, not per-run memory. See architecture and rfcs/0008 for the design; this page is the API reference.

openlithohub.streaming.geometry

Streaming full-chip geometry primitives (RFC 0008).

Core/halo ownership model:

  • The core is the only region whose result may be committed to the final full-chip output.
  • The halo is read-region context that gives the forward model real neighbourhood content instead of zero-padded artefacts.
  • Adjacent cores must exactly cover the global domain (no gaps, no duplicate ownership); adjacent read regions may overlap freely.

BoundingBox dataclass

Half-open pixel rectangle [x0, x1) x [y0, y1) in global coords.

Source code in src/openlithohub/streaming/geometry.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class BoundingBox:
    """Half-open pixel rectangle ``[x0, x1) x [y0, y1)`` in global coords."""

    x0: int
    y0: int
    x1: int
    y1: int

    def __post_init__(self) -> None:
        if self.x1 <= self.x0 or self.y1 <= self.y0:
            raise ValueError(f"degenerate bbox: x=[{self.x0},{self.x1}) y=[{self.y0},{self.y1})")

    @property
    def width(self) -> int:
        return self.x1 - self.x0

    @property
    def height(self) -> int:
        return self.y1 - self.y0

    @property
    def area(self) -> int:
        return self.width * self.height

    def clipped_to(self, width: int, height: int) -> BoundingBox:
        """Clip to the global domain ``[0,width) x [0,height)``.

        Returns a bbox that may shrink; callers must treat an empty
        intersection (degenerate result) as "no overlap".
        """
        return BoundingBox(
            x0=max(0, self.x0),
            y0=max(0, self.y0),
            x1=min(width, self.x1),
            y1=min(height, self.y1),
        )

    def contains(self, other: BoundingBox) -> bool:
        return (
            self.x0 <= other.x0
            and self.y0 <= other.y0
            and self.x1 >= other.x1
            and self.y1 >= other.y1
        )

clipped_to(width, height)

Clip to the global domain [0,width) x [0,height).

Returns a bbox that may shrink; callers must treat an empty intersection (degenerate result) as "no overlap".

Source code in src/openlithohub/streaming/geometry.py 
44
45
46
47
48
49
50
51
52
53
54
55
def clipped_to(self, width: int, height: int) -> BoundingBox:
    """Clip to the global domain ``[0,width) x [0,height)``.

    Returns a bbox that may shrink; callers must treat an empty
    intersection (degenerate result) as "no overlap".
    """
    return BoundingBox(
        x0=max(0, self.x0),
        y0=max(0, self.y0),
        x1=min(width, self.x1),
        y1=min(height, self.y1),
    )

HaloSpec dataclass

Per-side halo widths requested for a tile read region.

Source code in src/openlithohub/streaming/geometry.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class HaloSpec:
    """Per-side halo widths requested for a tile read region."""

    left: int
    top: int
    right: int
    bottom: int

    @classmethod
    def uniform(cls, px: int) -> HaloSpec:
        if px < 0:
            raise ValueError(f"halo must be nonnegative, got {px}")
        return cls(left=px, top=px, right=px, bottom=px)

    @property
    def total(self) -> int:
        return self.left + self.top + self.right + self.bottom

core_grid_boxes(shape, core_size, *, core_stride=None)

Partition shape into an exact cover of axis-aligned core boxes.

The last column/row of cores is pulled back to the domain edge so cores cover [0,W) x [0,H) with no gap and no overlap. Returns boxes in row-major order.

Source code in src/openlithohub/streaming/geometry.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def core_grid_boxes(
    shape: tuple[int, int],
    core_size: int,
    *,
    core_stride: int | None = None,
) -> list[BoundingBox]:
    """Partition ``shape`` into an exact cover of axis-aligned core boxes.

    The last column/row of cores is pulled back to the domain edge so
    cores cover ``[0,W) x [0,H)`` with no gap and no overlap.  Returns
    boxes in row-major order.
    """
    h, w = shape
    if core_size <= 0:
        raise ValueError(f"core_size must be positive, got {core_size}")
    stride = core_stride if core_stride is not None else core_size
    if stride <= 0:
        raise ValueError(f"core_stride must be positive, got {stride}")

    xs = list(range(0, w, stride))
    ys = list(range(0, h, stride))
    boxes: list[BoundingBox] = []
    for y in ys:
        for x in xs:
            boxes.append(
                BoundingBox(
                    x0=x,
                    y0=y,
                    x1=min(x + core_size, w) if x + core_size > w else x + core_size,
                    y1=min(y + core_size, h) if y + core_size > h else y + core_size,
                )
            )
    return boxes

read_bbox_for(core, halo, width, height)

Expand a core by its halo, clipped to the global domain.

Source code in src/openlithohub/streaming/geometry.py 
125
126
127
128
129
130
131
132
def read_bbox_for(core: BoundingBox, halo: HaloSpec, width: int, height: int) -> BoundingBox:
    """Expand a core by its halo, clipped to the global domain."""
    return BoundingBox(
        x0=max(0, core.x0 - halo.left),
        y0=max(0, core.y0 - halo.top),
        x1=min(width, core.x1 + halo.right),
        y1=min(height, core.y1 + halo.bottom),
    )

halo_actual(core, read)

Recover the per-side halo actually realised after domain clipping.

Source code in src/openlithohub/streaming/geometry.py 
135
136
137
138
139
140
141
142
def halo_actual(core: BoundingBox, read: BoundingBox) -> HaloSpec:
    """Recover the per-side halo actually realised after domain clipping."""
    return HaloSpec(
        left=core.x0 - read.x0,
        top=core.y0 - read.y0,
        right=read.x1 - core.x1,
        bottom=read.y1 - core.y1,
    )

halo_overhead_stats(core_boxes, read_boxes)

Report halo duplicate-compute overhead (prompt §15).

eta_halo = ((C+2h)^2 - C^2) / C^2 generalised per tile with the realised (possibly clipped) halos.

Source code in src/openlithohub/streaming/geometry.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def halo_overhead_stats(
    core_boxes: list[BoundingBox],
    read_boxes: list[BoundingBox],
) -> dict[str, float]:
    """Report halo duplicate-compute overhead (prompt §15).

    ``eta_halo = ((C+2h)^2 - C^2) / C^2`` generalised per tile with the
    realised (possibly clipped) halos.
    """
    if len(core_boxes) != len(read_boxes):
        raise ValueError("core/read box lists must have the same length")
    core_px = sum(b.area for b in core_boxes)
    read_px = sum(b.area for b in read_boxes)
    if core_px == 0:
        return {
            "core_pixels": 0.0,
            "read_pixels": 0.0,
            "halo_pixels": 0.0,
            "halo_overhead_pct": 0.0,
            "n_tiles": float(len(core_boxes)),
        }
    return {
        "core_pixels": float(core_px),
        "read_pixels": float(read_px),
        "halo_pixels": float(read_px - core_px),
        "halo_overhead_pct": 100.0 * (read_px - core_px) / core_px,
        "n_tiles": float(len(core_boxes)),
    }

openlithohub.streaming.core_halo

Tile requests, planning, and the streaming scheduler (RFC 0008).

TileRequest pairs a trusted core with its context halo. Planning a request list from a layout shape guarantees an exact core cover; the scheduler consumes requests (possibly re-queuing refinements) one tile at a time so full-chip memory stays O(tile area + active batch).

TileRequest dataclass

Read read_bbox (core + halo) and commit only core_bbox.

Source code in src/openlithohub/streaming/core_halo.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class TileRequest:
    """Read ``read_bbox`` (core + halo) and commit only ``core_bbox``."""

    core_bbox: BoundingBox
    read_bbox: BoundingBox
    halo: HaloSpec
    tile_id: str

    @property
    def core_size(self) -> tuple[int, int]:
        return (self.core_bbox.height, self.core_bbox.width)

    @property
    def read_size(self) -> tuple[int, int]:
        return (self.read_bbox.height, self.read_bbox.width)

RefinementRequest dataclass

Verifier-agnostic advice to re-run a tile with more context.

Produced by verification plugins (prompt §12); the scheduler never needs to know why a tile was inconclusive — only which knob to turn.

Source code in src/openlithohub/streaming/core_halo.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class RefinementRequest:
    """Verifier-agnostic advice to re-run a tile with more context.

    Produced by verification plugins (prompt §12); the scheduler never
    needs to know *why* a tile was inconclusive — only which knob to turn.
    """

    tile_id: str
    action: str = "increase_halo"  # increase_halo | subdivide
    extra_halo_px: int = 0
    subdivision: int = 2

    def __post_init__(self) -> None:
        if self.action not in ("increase_halo", "subdivide"):
            raise ValueError(f"unknown refinement action: {self.action!r}")
        if self.action == "increase_halo" and self.extra_halo_px <= 0:
            raise ValueError("increase_halo refinement needs extra_halo_px > 0")
        if self.action == "subdivide" and self.subdivision < 2:
            raise ValueError("subdivide refinement needs subdivision >= 2")

TileScheduler dataclass

Sequential streaming scheduler with verifier-driven refinement.

The consumer pulls one request at a time; refinement requests re-enqueue work without the scheduler knowing anything about the verifier's maths. domain (global (H, W)) lets grown halos clip to the real layout instead of the tile's previous read region; when omitted, the previous read extent is used as the clip bound.

Source code in src/openlithohub/streaming/core_halo.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@dataclass
class TileScheduler:
    """Sequential streaming scheduler with verifier-driven refinement.

    The consumer pulls one request at a time; refinement requests re-enqueue
    work without the scheduler knowing anything about the verifier's maths.
    ``domain`` (global ``(H, W)``) lets grown halos clip to the real layout
    instead of the tile's previous read region; when omitted, the previous
    read extent is used as the clip bound.
    """

    requests: list[TileRequest]
    domain: tuple[int, int] | None = None
    _pending: list[TileRequest] = field(default_factory=list, init=False)

    def __post_init__(self) -> None:
        self._pending = list(self.requests)

    def __iter__(self) -> Iterator[TileRequest]:
        return self

    def __next__(self) -> TileRequest:
        if not self._pending:
            raise StopIteration
        return self._pending.pop(0)

    def enqueue(self, requests: list[TileRequest]) -> None:
        self._pending.extend(requests)

    def enqueue_refinement(
        self, request: TileRequest, refinement: RefinementRequest
    ) -> list[TileRequest]:
        """Re-queue refined work for an inconclusive tile."""
        if refinement.action == "subdivide":
            refined = subdivide_request(request, refinement.subdivision)
        else:
            grown = HaloSpec.uniform(
                max(
                    request.halo.left,
                    request.halo.top,
                    request.halo.right,
                    request.halo.bottom,
                )
                + refinement.extra_halo_px
            )
            if self.domain is not None:
                clip_w, clip_h = self.domain
            else:
                clip_w = max(request.read_bbox.x1, request.core_bbox.x1)
                clip_h = max(request.read_bbox.y1, request.core_bbox.y1)
            read = read_bbox_for(request.core_bbox, grown, clip_w, clip_h)
            refined = [
                TileRequest(
                    core_bbox=request.core_bbox,
                    read_bbox=read,
                    halo=halo_actual(request.core_bbox, read),
                    tile_id=request.tile_id,
                )
            ]
        self.enqueue(refined)
        return refined

enqueue_refinement(request, refinement)

Re-queue refined work for an inconclusive tile.

Source code in src/openlithohub/streaming/core_halo.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def enqueue_refinement(
    self, request: TileRequest, refinement: RefinementRequest
) -> list[TileRequest]:
    """Re-queue refined work for an inconclusive tile."""
    if refinement.action == "subdivide":
        refined = subdivide_request(request, refinement.subdivision)
    else:
        grown = HaloSpec.uniform(
            max(
                request.halo.left,
                request.halo.top,
                request.halo.right,
                request.halo.bottom,
            )
            + refinement.extra_halo_px
        )
        if self.domain is not None:
            clip_w, clip_h = self.domain
        else:
            clip_w = max(request.read_bbox.x1, request.core_bbox.x1)
            clip_h = max(request.read_bbox.y1, request.core_bbox.y1)
        read = read_bbox_for(request.core_bbox, grown, clip_w, clip_h)
        refined = [
            TileRequest(
                core_bbox=request.core_bbox,
                read_bbox=read,
                halo=halo_actual(request.core_bbox, read),
                tile_id=request.tile_id,
            )
        ]
    self.enqueue(refined)
    return refined

plan_tile_requests(shape, core_size, halo_px, *, core_stride=None, prefix='tile')

Plan core-exact-cover tile requests for a layout shape.

Source code in src/openlithohub/streaming/core_halo.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def plan_tile_requests(
    shape: tuple[int, int],
    core_size: int,
    halo_px: int | HaloSpec,
    *,
    core_stride: int | None = None,
    prefix: str = "tile",
) -> list[TileRequest]:
    """Plan core-exact-cover tile requests for a layout shape."""
    halo = HaloSpec.uniform(halo_px) if isinstance(halo_px, int) else halo_px
    h, w = shape
    requests: list[TileRequest] = []
    for idx, core in enumerate(core_grid_boxes(shape, core_size, core_stride=core_stride)):
        read = read_bbox_for(core, halo, w, h)
        requests.append(
            TileRequest(
                core_bbox=core,
                read_bbox=read,
                halo=halo_actual(core, read),
                tile_id=f"{prefix}_{idx}",
            )
        )
    return requests

subdivide_request(request, subdivision)

Split a tile's core into subdivision^2 smaller cores.

Source code in src/openlithohub/streaming/core_halo.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def subdivide_request(request: TileRequest, subdivision: int) -> list[TileRequest]:
    """Split a tile's core into ``subdivision^2`` smaller cores."""
    ch, cw = request.core_size
    sub_h = max(1, ch // subdivision)
    sub_w = max(1, cw // subdivision)
    out: list[TileRequest] = []
    core = request.core_bbox
    idx = 0
    for y0 in range(core.y0, core.y1, sub_h):
        for x0 in range(core.x0, core.x1, sub_w):
            sub_core = BoundingBox(
                x0=x0,
                y0=y0,
                x1=min(x0 + sub_w, core.x1),
                y1=min(y0 + sub_h, core.y1),
            )
            if sub_core.width <= 0 or sub_core.height <= 0:
                continue
            extra = HaloSpec.uniform(
                max(request.halo.left, request.halo.top, request.halo.right, request.halo.bottom)
            )
            w = request.read_bbox.x1
            h = request.read_bbox.y1
            read = read_bbox_for(
                sub_core,
                extra,
                width=max(w, sub_core.x1),
                height=max(h, sub_core.y1),
            )
            out.append(
                TileRequest(
                    core_bbox=sub_core,
                    read_bbox=read,
                    halo=halo_actual(sub_core, read),
                    tile_id=f"{request.tile_id}_s{idx}",
                )
            )
            idx += 1
    return out

tiling_overhead(requests)

Aggregate core/halo duplicate-compute statistics for a plan.

Source code in src/openlithohub/streaming/core_halo.py 
193
194
195
196
197
198
199
def tiling_overhead(requests: list[TileRequest]) -> dict[str, float]:
    """Aggregate core/halo duplicate-compute statistics for a plan."""
    cores = [r.core_bbox for r in requests]
    reads = [r.read_bbox for r in requests]
    stats = halo_overhead_stats(cores, reads)
    stats["estimated_duplicate_compute_pct"] = stats["halo_overhead_pct"]
    return stats

plan_tiling(layout_shape, *, memory_budget_bytes, bytes_per_pixel=4.0, halo_px=0, min_core_size=32, max_core_size=4096, batch=1)

Pick the largest core size that fits the per-tile memory budget.

Approximates peak per-tile memory as bytes_per_pixel * (C + 2h)^2 * batch and maximises core efficiency C^2 / (C + 2h)^2 subject to the budget (prompt §16). Callers may pass an explicit HaloPolicy-derived halo; the planner itself stays policy-agnostic.

Source code in src/openlithohub/streaming/core_halo.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def plan_tiling(
    layout_shape: tuple[int, int],
    *,
    memory_budget_bytes: int,
    bytes_per_pixel: float = 4.0,
    halo_px: int = 0,
    min_core_size: int = 32,
    max_core_size: int = 4096,
    batch: int = 1,
) -> list[TileRequest]:
    """Pick the largest core size that fits the per-tile memory budget.

    Approximates peak per-tile memory as ``bytes_per_pixel * (C + 2h)^2 *
    batch`` and maximises core efficiency ``C^2 / (C + 2h)^2`` subject to
    the budget (prompt §16).  Callers may pass an explicit
    ``HaloPolicy``-derived halo; the planner itself stays policy-agnostic.
    """
    if memory_budget_bytes <= 0:
        raise ValueError("memory_budget_bytes must be positive")
    if min_core_size <= 0 or max_core_size < min_core_size:
        raise ValueError(f"invalid core size bounds: {min_core_size}..{max_core_size}")

    best_core = min_core_size
    for core in range(min_core_size, max_core_size + 1, 8):
        read = core + 2 * halo_px
        if bytes_per_pixel * read * read * batch <= memory_budget_bytes:
            best_core = core
    h, w = layout_shape
    if best_core >= min(h, w):
        # Small layout: a single core covering the whole domain needs no halo.
        best_core = max(min(h, w), min_core_size)
        return plan_tile_requests(layout_shape, best_core, HaloSpec.uniform(0))
    return plan_tile_requests(layout_shape, best_core, HaloSpec.uniform(halo_px))

openlithohub.streaming.sources

Streaming tile sources (RFC 0008, prompt §2).

A TileSource produces the read region tensor for a tile request on demand. Implementations must never require the whole layout as a dense host tensor:

  • :class:TensorTileSource wraps an already-materialised tensor (compatibility, tests, small benchmarks).
  • :class:MemmapTensorTileSource reads windows out of an on-disk np.memmap raster (out-of-core big raster).
  • :class:VectorLayoutTileSource rasterizes only the objects that intersect the requested window of a vector GDS/OASIS layout through a :class:SpatialLayoutIndex.

TileSource

Bases: Protocol

Read-only windowed view over a full-chip layout.

Source code in src/openlithohub/streaming/sources.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@runtime_checkable
class TileSource(Protocol):
    """Read-only windowed view over a full-chip layout."""

    @property
    def shape(self) -> tuple[int, int]:
        """Global raster shape ``(H, W)`` in pixels."""
        ...

    @property
    def pixel_size_nm(self) -> float: ...

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        """Return a ``(H, W)`` float tensor for the half-open bbox."""
        ...

shape property

Global raster shape (H, W) in pixels.

read_window(bbox)

Return a (H, W) float tensor for the half-open bbox.

Source code in src/openlithohub/streaming/sources.py 
39
40
41
def read_window(self, bbox: BoundingBox) -> torch.Tensor:
    """Return a ``(H, W)`` float tensor for the half-open bbox."""
    ...

TensorTileSource

Wrap an existing dense tensor. Zero-copy via narrow/as_strided.

Source code in src/openlithohub/streaming/sources.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class TensorTileSource:
    """Wrap an existing dense tensor. Zero-copy via narrow/as_strided."""

    def __init__(self, tensor: torch.Tensor, pixel_size_nm: float = 1.0) -> None:
        t = tensor.detach()
        if t.ndim != 2:
            raise ValueError(f"expected a 2-D layout tensor, got shape {tuple(t.shape)}")
        self._t = t
        self._pixel_nm = float(pixel_size_nm)

    @property
    def shape(self) -> tuple[int, int]:
        return (self._t.shape[0], self._t.shape[1])

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        return self._t[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1].float()

MemmapTensorTileSource

Out-of-core windowed reads over an on-disk raster via np.memmap.

The backing file may be far larger than RAM; only requested windows are paged in. Accepts a .npy/raw path plus dtype/shape, or an existing np.memmap.

Source code in src/openlithohub/streaming/sources.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class MemmapTensorTileSource:
    """Out-of-core windowed reads over an on-disk raster via ``np.memmap``.

    The backing file may be far larger than RAM; only requested windows are
    paged in.  Accepts a ``.npy``/raw path plus dtype/shape, or an existing
    ``np.memmap``.
    """

    def __init__(
        self,
        source: str | Path | np.memmap,
        *,
        dtype: np.dtype[Any] | None = None,
        shape: tuple[int, int] | None = None,
        pixel_size_nm: float = 1.0,
    ) -> None:
        if isinstance(source, (str, Path)):
            if dtype is not None and shape is not None:
                # Raw raster: caller must supply the on-disk layout.
                self._map: np.memmap = np.memmap(source, dtype=dtype, mode="r", shape=shape)
            else:
                # .npy path: np.load with mmap_mode pages windows from disk.
                self._map = np.load(source, mmap_mode="r")
        else:
            self._map = source
        if self._map.ndim != 2:
            raise ValueError(f"expected a 2-D raster, got shape {self._map.shape}")
        self._pixel_nm = float(pixel_size_nm)

    @property
    def shape(self) -> tuple[int, int]:
        return (int(self._map.shape[0]), int(self._map.shape[1]))

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        window = np.asarray(self._map[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1])
        return torch.from_numpy(np.ascontiguousarray(window)).float()

SpatialLayoutIndex

Minimal grid-bucket spatial index over vector layout boxes.

Each layout object is an axis-aligned bounding box plus optional fill (1.0 foreground). Objects are bucketed on a coarse uniform grid so a window query only visits intersecting buckets. This is the smallest useful realisation of the SpatialLayoutIndex concept from RFC 0008; a klayout-Region-backed implementation can drop in later behind the same interface.

Source code in src/openlithohub/streaming/sources.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class SpatialLayoutIndex:
    """Minimal grid-bucket spatial index over vector layout boxes.

    Each layout object is an axis-aligned bounding box plus optional fill
    (1.0 foreground).  Objects are bucketed on a coarse uniform grid so a
    window query only visits intersecting buckets.  This is the smallest
    useful realisation of the ``SpatialLayoutIndex`` concept from RFC 0008;
    a klayout-Region-backed implementation can drop in later behind the
    same interface.
    """

    def __init__(
        self,
        boxes: list[tuple[int, int, int, int]],
        raster_shape: tuple[int, int],
        bucket: int = 256,
    ) -> None:
        if not boxes:
            raise ValueError("SpatialLayoutIndex needs at least one layout box")
        h, w = raster_shape
        self._shape = raster_shape
        self._bucket = max(1, int(bucket))
        nbx = max(1, (w + self._bucket - 1) // self._bucket)
        nby = max(1, (h + self._bucket - 1) // self._bucket)
        self._buckets: dict[tuple[int, int], list[tuple[int, int, int, int]]] = {}
        for x0, y0, x1, y1 in boxes:
            if x1 <= x0 or y1 <= y0:
                continue
            for by in range(y0 // self._bucket, min((y1 - 1) // self._bucket, nby - 1) + 1):
                for bx in range(x0 // self._bucket, min((x1 - 1) // self._bucket, nbx - 1) + 1):
                    self._buckets.setdefault((bx, by), []).append((x0, y0, x1, y1))

    def query(self, bbox: BoundingBox) -> list[tuple[int, int, int, int]]:
        found: list[tuple[int, int, int, int]] = []
        seen: set[tuple[int, int, int, int]] = set()
        nbx = max(1, (self._shape[1] + self._bucket - 1) // self._bucket)
        nby = max(1, (self._shape[0] + self._bucket - 1) // self._bucket)
        for by in range(bbox.y0 // self._bucket, min((bbox.y1 - 1) // self._bucket, nby - 1) + 1):
            for bx in range(
                bbox.x0 // self._bucket, min((bbox.x1 - 1) // self._bucket, nbx - 1) + 1
            ):
                for box in self._buckets.get((bx, by), ()):  # may repeat across buckets
                    if box in seen:
                        continue
                    seen.add(box)
                    ix0, iy0, ix1, iy1 = box
                    ox0, oy0 = max(ix0, bbox.x0), max(iy0, bbox.y0)
                    ox1, oy1 = min(ix1, bbox.x1), min(iy1, bbox.y1)
                    if ox1 > ox0 and oy1 > oy0:
                        found.append((ox0, oy0, ox1, oy1))
        return found

VectorLayoutTileSource

Rasterize only the objects intersecting each requested window.

Source code in src/openlithohub/streaming/sources.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class VectorLayoutTileSource:
    """Rasterize only the objects intersecting each requested window."""

    def __init__(
        self,
        index: SpatialLayoutIndex,
        shape: tuple[int, int],
        pixel_size_nm: float = 1.0,
        background: float = 0.0,
        fill: float = 1.0,
    ) -> None:
        self._index = index
        self._shape = shape
        self._pixel_nm = float(pixel_size_nm)
        self._background = float(background)
        self._fill = float(fill)

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        window = np.full((bbox.height, bbox.width), self._background, dtype=np.float32)
        for x0, y0, x1, y1 in self._index.query(bbox):
            window[y0 - bbox.y0 : y1 - bbox.y0, x0 - bbox.x0 : x1 - bbox.x0] = self._fill
        return torch.from_numpy(window)

openlithohub.streaming.sinks

Streaming tile sinks (RFC 0008, prompt §3).

A TileSink receives each tile's trusted-core result. Implementations decide the output materialisation:

  • :class:TensorTileSink — assemble a full dense tensor (small layouts, backward-compatible behaviour).
  • :class:MemmapTileSink — page the output to an on-disk memmap (out-of-core big rasters).
  • :class:MetricOnlyTileSink — keep only per-tile metrics/aggregates and never materialise a raster at all (the QDM-critical mode).

Because only trusted cores are ever written, no weight map / blend pass is required: cores are an exact partition of the output domain.

CertifiedCommitNotRepresentableError

Bases: ValueError

A sink was asked to commit a certified core it cannot represent.

R17 C2: a verifier PASS (or a screening fact) without a known output tensor is a verification-only disposition. Metric-only sinks can record it; tensor-materializing sinks must reject it, because verifier PASS does not imply a known output tensor.

Source code in src/openlithohub/streaming/sinks.py 
45
46
47
48
49
50
51
52
class CertifiedCommitNotRepresentableError(ValueError):
    """A sink was asked to commit a certified core it cannot represent.

    R17 C2: a verifier PASS (or a screening fact) without a known output
    tensor is a *verification-only* disposition.  Metric-only sinks can
    record it; tensor-materializing sinks must reject it, because
    ``verifier PASS does not imply a known output tensor``.
    """

TensorTileSink

Assemble the full dense output tensor in host memory.

Source code in src/openlithohub/streaming/sinks.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class TensorTileSink:
    """Assemble the full dense output tensor in host memory."""

    def __init__(self, shape: tuple[int, int], dtype: torch.dtype = torch.float32) -> None:
        self._shape = shape
        self.output = torch.zeros(shape, dtype=dtype)
        self._written: set[str] = set()

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if tile_id in self._written:
            raise ValueError(f"duplicate core write for {tile_id!r}")
        self._written.add(tile_id)
        expected = (bbox.height, bbox.width)
        if tuple(tensor.shape) != expected:
            raise ValueError(f"core tensor shape {tuple(tensor.shape)} != bbox {expected}")
        self.output[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1] = tensor

    def record_certified_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        *,
        exact_fill: float | None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """R17 C2 certified commit: synthesize the exact fill, or reject."""
        if exact_fill is None:
            raise CertifiedCommitNotRepresentableError(
                "verification-only skip cannot be represented in a tensor "
                "output sink; verifier PASS does not imply a known tensor"
            )
        fill = torch.full((bbox.height, bbox.width), float(exact_fill), dtype=self.output.dtype)
        self.write_core(tile_id, bbox, fill, metadata)

    def finalize(self) -> torch.Tensor:
        return self.output

record_certified_core(tile_id, bbox, *, exact_fill, metadata=None)

R17 C2 certified commit: synthesize the exact fill, or reject.

Source code in src/openlithohub/streaming/sinks.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def record_certified_core(
    self,
    tile_id: str,
    bbox: BoundingBox,
    *,
    exact_fill: float | None,
    metadata: dict[str, Any] | None = None,
) -> None:
    """R17 C2 certified commit: synthesize the exact fill, or reject."""
    if exact_fill is None:
        raise CertifiedCommitNotRepresentableError(
            "verification-only skip cannot be represented in a tensor "
            "output sink; verifier PASS does not imply a known tensor"
        )
    fill = torch.full((bbox.height, bbox.width), float(exact_fill), dtype=self.output.dtype)
    self.write_core(tile_id, bbox, fill, metadata)

MemmapTileSink

Page trusted cores straight into an on-disk memmap raster.

Source code in src/openlithohub/streaming/sinks.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class MemmapTileSink:
    """Page trusted cores straight into an on-disk memmap raster."""

    def __init__(
        self,
        shape: tuple[int, int],
        path: str | Path,
        dtype: np.dtype[Any] | None = None,
    ) -> None:
        self._shape = shape
        self.path = Path(path)
        if dtype is None:
            dtype = np.dtype(np.float32)
        self._map: np.memmap = np.memmap(self.path, dtype=dtype, mode="w+", shape=shape)
        self._written: set[str] = set()

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if tile_id in self._written:
            raise ValueError(f"duplicate core write for {tile_id!r}")
        self._written.add(tile_id)
        window = tensor.detach().cpu().numpy().astype(self._map.dtype, copy=False)
        self._map[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1] = window

    def record_certified_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        *,
        exact_fill: float | None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """R17 C2 certified commit: synthesize the exact fill, or reject."""
        if exact_fill is None:
            raise CertifiedCommitNotRepresentableError(
                "verification-only skip cannot be represented in a memmap "
                "output sink; verifier PASS does not imply a known tensor"
            )
        fill = torch.full((bbox.height, bbox.width), float(exact_fill), dtype=torch.float32)
        self.write_core(tile_id, bbox, fill, metadata)

    def finalize(self) -> np.memmap:
        self._map.flush()
        return self._map

record_certified_core(tile_id, bbox, *, exact_fill, metadata=None)

R17 C2 certified commit: synthesize the exact fill, or reject.

Source code in src/openlithohub/streaming/sinks.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def record_certified_core(
    self,
    tile_id: str,
    bbox: BoundingBox,
    *,
    exact_fill: float | None,
    metadata: dict[str, Any] | None = None,
) -> None:
    """R17 C2 certified commit: synthesize the exact fill, or reject."""
    if exact_fill is None:
        raise CertifiedCommitNotRepresentableError(
            "verification-only skip cannot be represented in a memmap "
            "output sink; verifier PASS does not imply a known tensor"
        )
    fill = torch.full((bbox.height, bbox.width), float(exact_fill), dtype=torch.float32)
    self.write_core(tile_id, bbox, fill, metadata)

MetricOnlyTileSink

Aggregate per-tile results without ever materialising a raster.

Reducers receive one tile at a time, so peak memory is O(1) beyond the running aggregates. This is the sink mode QDM-style verification needs (certificates, worst bounds, violation lists, provenance).

Source code in src/openlithohub/streaming/sinks.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
class MetricOnlyTileSink:
    """Aggregate per-tile results without ever materialising a raster.

    Reducers receive one tile at a time, so peak memory is O(1) beyond the
    running aggregates.  This is the sink mode QDM-style verification needs
    (certificates, worst bounds, violation lists, provenance).
    """

    def __init__(self, shape: tuple[int, int]) -> None:
        self._shape = shape
        self.tiles: list[str] = []
        self.covered_pixels = 0
        self.metrics: dict[str, float] = {}
        self.metadata: dict[str, Any] = {}
        self.certified_records: dict[str, str] = {}
        self.certified_pixels = 0

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        self.tiles.append(tile_id)
        self.covered_pixels += bbox.area
        self.metadata[tile_id] = metadata or {}
        for key, value in (metadata or {}).items():
            if isinstance(value, (int, float)):
                prev = self.metrics.get(f"max_{key}", float("-inf"))
                self.metrics[f"max_{key}"] = max(prev, float(value))

    def record_certified_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        *,
        exact_fill: float | None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """R17 C2 tensor-free certified commit (the QDM-critical mode).

        No raster is materialised for a certified core — with or without an
        exact output fill.  ``exact_fill`` is recorded as provenance only.
        """
        disposition = "EXACT_OUTPUT_SKIP" if exact_fill is not None else "VERIFICATION_ONLY_SKIP"
        meta = dict(metadata or {})
        meta.setdefault("screen_disposition", disposition)
        if exact_fill is not None:
            meta.setdefault("certified_exact_fill", float(exact_fill))
        self.write_core(tile_id, bbox, torch.empty(0), meta)
        self.certified_records[tile_id] = disposition
        self.certified_pixels += bbox.area

    def finalize(self) -> dict[str, Any]:
        return {
            "n_tiles": len(self.tiles),
            "covered_pixels": self.covered_pixels,
            "metrics": dict(self.metrics),
        }

record_certified_core(tile_id, bbox, *, exact_fill, metadata=None)

R17 C2 tensor-free certified commit (the QDM-critical mode).

No raster is materialised for a certified core — with or without an exact output fill. exact_fill is recorded as provenance only.

Source code in src/openlithohub/streaming/sinks.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def record_certified_core(
    self,
    tile_id: str,
    bbox: BoundingBox,
    *,
    exact_fill: float | None,
    metadata: dict[str, Any] | None = None,
) -> None:
    """R17 C2 tensor-free certified commit (the QDM-critical mode).

    No raster is materialised for a certified core — with or without an
    exact output fill.  ``exact_fill`` is recorded as provenance only.
    """
    disposition = "EXACT_OUTPUT_SKIP" if exact_fill is not None else "VERIFICATION_ONLY_SKIP"
    meta = dict(metadata or {})
    meta.setdefault("screen_disposition", disposition)
    if exact_fill is not None:
        meta.setdefault("certified_exact_fill", float(exact_fill))
    self.write_core(tile_id, bbox, torch.empty(0), meta)
    self.certified_records[tile_id] = disposition
    self.certified_pixels += bbox.area

covered_pixels(sink)

Total trusted-core pixels written (for gap/overlap audits).

Source code in src/openlithohub/streaming/sinks.py 
224
225
226
227
228
229
230
231
232
def covered_pixels(sink: TileSink) -> int:
    """Total trusted-core pixels written (for gap/overlap audits)."""
    written = getattr(sink, "covered_pixels", None)
    if written is not None:
        return int(written)
    written = getattr(sink, "_written", None)
    if isinstance(written, Iterable):
        return len(written)  # type: ignore[arg-type]
    return -1

openlithohub.streaming.halo_policy

Halo policies (RFC 0008, prompt §4-6).

A HaloPolicy answers "how much context halo does this tile need?" and must always explain itself: every answer carries the halo, its provenance, an optional rigorous error bound, and a certification status. Policies composing means taking the max halo; the strictest provenance/certification status wins.

HaloContext dataclass

Everything a policy may need to size a halo.

Source code in src/openlithohub/streaming/halo_policy.py 
85
86
87
88
89
90
91
92
93
94
@dataclass(frozen=True)
class HaloContext:
    """Everything a policy may need to size a halo."""

    pixel_nm: float
    tile_core_px: int
    node: ProcessNodeConfig | None = None
    model: LithographyModel | None = None
    kernel: torch.Tensor | None = None
    kernel_tolerance: float | None = None

LegacyFixedHaloPolicy

Reproduce the pre-RFC-0005 fixed halo (default 128 px).

Source code in src/openlithohub/streaming/halo_policy.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class LegacyFixedHaloPolicy:
    """Reproduce the pre-RFC-0005 fixed halo (default 128 px)."""

    name = "legacy_fixed"

    def __init__(self, halo_px: int = DEFAULT_HALO_PX) -> None:
        if halo_px < 0:
            raise ValueError(f"halo_px must be nonnegative, got {halo_px}")
        self._halo_px = halo_px

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        return HaloRequirement(
            halo_px=min(self._halo_px, max(context.tile_core_px - 1, 0)),
            provenance="legacy_fixed",
            error_bound=None,
            certified=False,
            reason=f"fixed legacy halo of {self._halo_px} px (backward compatibility)",
            status=HaloStatus.FIXED,
        )

PhysicalInteractionHaloPolicy

h = max(OIR_px, receptive-field px), RFC-0005 logic moved here.

Source code in src/openlithohub/streaming/halo_policy.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class PhysicalInteractionHaloPolicy:
    """``h = max(OIR_px, receptive-field px)``, RFC-0005 logic moved here."""

    name = "physical_interaction"

    def __init__(self, round_up_to: int = _HALO_ROUND_PX) -> None:
        self._round = max(1, round_up_to)

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        parts: list[str] = []
        px = 0
        if context.node is not None and context.pixel_nm > 0:
            oir_px = math.ceil(context.node.optical_radius_nm / context.pixel_nm)
            parts.append(f"optical interaction radius {oir_px} px")
            px = max(px, oir_px)
        if context.model is not None:
            rf = getattr(context.model, "RECEPTIVE_FIELD_PX", 0) or 0
            if rf:
                parts.append(f"model receptive field {rf} px")
            px = max(px, rf)
        halo = _round_up(px) if self._round > 1 else px
        return HaloRequirement(
            halo_px=min(halo, max(context.tile_core_px - 1, 0)),
            provenance="physical_interaction",
            error_bound=None,
            certified=False,
            reason="max(" + (", ".join(parts) if parts else "no physics source") + ")",
            status=HaloStatus.PHYSICS_ESTIMATED,
        )

KernelTailHaloPolicy

Pick the smallest halo whose kernel tail mass <= tolerance.

T(h) = sum_{|x|>h} |K(x)| <= eps over the sampled spatial kernel. When the sample truly bounds the full kernel, the emitted halo is CERTIFIED_SUFFICIENT with tail mass as the error bound; if the caller cannot vouch for the sample, pass trusted_sample=False and the answer stays PHYSICS_ESTIMATED (never pretend certification).

Source code in src/openlithohub/streaming/halo_policy.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class KernelTailHaloPolicy:
    """Pick the smallest halo whose kernel tail mass <= tolerance.

    ``T(h) = sum_{|x|>h} |K(x)| <= eps`` over the sampled spatial kernel.
    When the sample truly bounds the full kernel, the emitted halo is
    CERTIFIED_SUFFICIENT with ``tail mass`` as the error bound; if the
    caller cannot vouch for the sample, pass ``trusted_sample=False`` and
    the answer stays PHYSICS_ESTIMATED (never pretend certification).
    """

    name = "kernel_tail"

    def __init__(self, tolerance: float = 1e-3, trusted_sample: bool = True) -> None:
        if tolerance <= 0.0:
            raise ValueError(f"tolerance must be positive, got {tolerance}")
        self._tolerance = float(tolerance)
        self._trusted = bool(trusted_sample)

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        kernel = context.kernel
        if kernel is None:
            return HaloRequirement(
                halo_px=DEFAULT_HALO_PX,
                provenance="kernel_tail",
                error_bound=None,
                certified=False,
                reason="no kernel sample available; fell back to legacy default",
                status=HaloStatus.INCONCLUSIVE,
            )
        k = kernel.detach().abs()
        total = float(k.sum())
        if total <= 0.0:
            return HaloRequirement(
                halo_px=DEFAULT_HALO_PX,
                provenance="kernel_tail",
                error_bound=None,
                certified=False,
                reason="degenerate (all-zero) kernel sample",
                status=HaloStatus.INCONCLUSIVE,
            )
        # Box-aligned tail sweep: the halo keeps a (2h+1)^2 central box, so
        # the tail at halo h is exactly kernel_tail_mass(k, h). Find the
        # smallest h whose tail meets the tolerance.
        max_r = min(k.shape[0], k.shape[1]) // 2
        best_h: int | None = None
        for h in range(0, max_r + 1):
            if kernel_tail_mass(k, h) <= self._tolerance:
                best_h = h
                break
        if best_h is None:
            # even keeping the whole sampled kernel leaves the tolerance
            # unmet only when the sample is unnormalizable noise; report the
            # tail at the largest box.
            tail = kernel_tail_mass(k, max_r)
            return HaloRequirement(
                halo_px=_round_up(max_r),
                provenance="kernel_tail",
                error_bound=tail,
                certified=False,
                reason=(
                    f"kernel tail {tail:.3e} still exceeds tolerance "
                    f"{self._tolerance:.3e} at the largest sampled radius"
                ),
                status=HaloStatus.INCONCLUSIVE,
            )
        halo = min(_round_up(best_h), max(context.tile_core_px - 1, 0))
        return HaloRequirement(
            halo_px=halo,
            provenance="kernel_tail",
            error_bound=self._tolerance if self._trusted else None,
            certified=self._trusted,
            reason=(f"smallest sampled radius with kernel tail mass <= {self._tolerance:.3e}"),
            status=(
                HaloStatus.CERTIFIED_SUFFICIENT if self._trusted else HaloStatus.PHYSICS_ESTIMATED
            ),
        )

combine_requirements(*requirements)

Max-combine halo requirements; strictest provenance wins.

Any non-certified participant demotes the combination, and any INCONCLUSIVE participant makes the whole answer inconclusive — a halo is only as trustworthy as its weakest contributor.

Source code in src/openlithohub/streaming/halo_policy.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def combine_requirements(*requirements: HaloRequirement | None) -> HaloRequirement:
    """Max-combine halo requirements; strictest provenance wins.

    Any non-certified participant demotes the combination, and any
    INCONCLUSIVE participant makes the whole answer inconclusive — a halo
    is only as trustworthy as its weakest contributor.
    """
    live = [r for r in requirements if r is not None]
    if not live:
        raise ValueError("combine_requirements needs at least one requirement")
    halo = max(r.halo_px for r in live)
    provenance = "+".join(r.provenance for r in live)
    bounds = [r.error_bound for r in live if r.error_bound is not None]
    error_bound = max(bounds) if bounds else None
    if any(r.status is HaloStatus.INCONCLUSIVE for r in live):
        status = HaloStatus.INCONCLUSIVE
        certified = False
    elif all(r.status is HaloStatus.CERTIFIED_SUFFICIENT for r in live):
        status = HaloStatus.CERTIFIED_SUFFICIENT
        certified = all(r.certified for r in live)
    else:
        status = HaloStatus.PHYSICS_ESTIMATED
        certified = False
    reason = "; ".join(r.reason for r in live)
    return HaloRequirement(
        halo_px=halo,
        provenance=provenance,
        error_bound=error_bound,
        certified=certified,
        reason=reason,
        status=status,
    )

kernel_tail_mass(kernel, halo_px)

Fraction of |kernel| mass outside the central (2h+1)^2 box.

Source code in src/openlithohub/streaming/halo_policy.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def kernel_tail_mass(kernel: torch.Tensor, halo_px: int) -> float:
    """Fraction of |kernel| mass outside the central ``(2h+1)^2`` box."""
    k = kernel.detach().abs()
    total = float(k.sum())
    if total <= 0.0:
        return 0.0
    cy, cx = k.shape[0] // 2, k.shape[1] // 2
    # Clamp so a halo larger than the kernel radius cannot wrap via
    # negative slicing — beyond the kernel radius the tail is exactly 0.
    y0, y1 = max(0, cy - halo_px), min(k.shape[0], cy + halo_px + 1)
    x0, x1 = max(0, cx - halo_px), min(k.shape[1], cx + halo_px + 1)
    keep = torch.zeros_like(k, dtype=torch.bool)
    keep[y0:y1, x0:x1] = True
    return float(k[~keep].sum()) / total

estimate_minimum_halo(forward_fn, *, core_bbox_px, full_context, candidate_halos=(0, 8, 16, 32, 64, 128), tolerance=0.001)

Adaptive empirical halo sizing (prompt §6).

Compares the forward result restricted to a fixed core as the halo grows. The first candidate whose discrepancy with the final (largest-halo) reference drops under tolerance — and stays there for the remaining candidates — is returned as EMPIRICALLY_STABLE. This is explicitly not a mathematical certificate; a rigorous bound needs an analytic kernel-tail or QDM argument.

Source code in src/openlithohub/streaming/halo_policy.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def estimate_minimum_halo(
    forward_fn: Callable[[torch.Tensor], torch.Tensor],
    *,
    core_bbox_px: tuple[int, int, int, int],
    full_context: torch.Tensor,
    candidate_halos: Iterable[int] = (0, 8, 16, 32, 64, 128),
    tolerance: float = 1e-3,
) -> HaloRequirement:
    """Adaptive empirical halo sizing (prompt §6).

    Compares the forward result restricted to a fixed core as the halo
    grows.  The first candidate whose discrepancy with the final
    (largest-halo) reference drops under ``tolerance`` — and stays there
    for the remaining candidates — is returned as EMPIRICALLY_STABLE.
    This is explicitly *not* a mathematical certificate; a rigorous bound
    needs an analytic kernel-tail or QDM argument.
    """
    x0, y0, x1, y1 = core_bbox_px
    core_h, core_w = y1 - y0, x1 - x0
    candidates = sorted({int(h) for h in candidate_halos if h >= 0})
    if not candidates:
        raise ValueError("need at least one candidate halo")

    core_errs: list[tuple[int, torch.Tensor]] = []
    for halo in candidates:
        tile = full_context[max(0, y0 - halo) : y1 + halo, max(0, x0 - halo) : x1 + halo]
        out = forward_fn(tile)
        # Extract the core response from the (possibly clipped) read region.
        oy = y0 - max(0, y0 - halo)
        ox = x0 - max(0, x0 - halo)
        core_out = out[oy : oy + core_h, ox : ox + core_w]
        core_errs.append((halo, core_out))

    reference_h, reference = core_errs[-1]
    stable_from: int | None = None
    for i in range(len(core_errs) - 1):
        h_i, out_i = core_errs[i]
        disc = float(torch.max(torch.abs(out_i - reference)).item())
        if disc <= tolerance and stable_from is None:
            stable_from = h_i
        elif disc > tolerance:
            stable_from = None
    if stable_from is None:
        return HaloRequirement(
            halo_px=reference_h,
            provenance="empirical_sweep",
            error_bound=None,
            certified=False,
            reason=f"core discrepancy never stabilised under tolerance {tolerance:g}",
            status=HaloStatus.INCONCLUSIVE,
        )
    return HaloRequirement(
        halo_px=stable_from,
        provenance="empirical_sweep",
        error_bound=tolerance,
        certified=False,
        reason=(
            f"core discrepancy <= {tolerance:g} for all halos >= {stable_from} px "
            f"(reference halo {reference_h} px)"
        ),
        status=HaloStatus.EMPIRICALLY_STABLE,
    )

openlithohub.streaming.pipeline

Streaming full-chip pipeline (RFC 0008, prompt §11).

Runs the per-tile loop::

optional certified pre-forward screen
    -> TileSource.read_window(core+halo) only for survivors
    -> forward model
    -> optional VerificationPlugin(s)
    -> TileSink.write_core(trusted core only)
    -> discard tile tensors

Peak memory is O(tile area + active batch), never O(full-chip raster). Every core is written exactly once; an inconclusive verdict re-runs the same core with a larger halo (bounded by max_requeues) before the best-effort result is committed.

Increment 28 integrates work accounting into the actual pipeline and adds a fail-closed screening hook. Screening may skip expensive work only when its decision is certified; with verification plugins present, the screen must also explicitly certify those verifiers or the pipeline falls back to the ordinary active path.

BoundingBox dataclass

Half-open pixel rectangle [x0, x1) x [y0, y1) in global coords.

Source code in src/openlithohub/streaming/geometry.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class BoundingBox:
    """Half-open pixel rectangle ``[x0, x1) x [y0, y1)`` in global coords."""

    x0: int
    y0: int
    x1: int
    y1: int

    def __post_init__(self) -> None:
        if self.x1 <= self.x0 or self.y1 <= self.y0:
            raise ValueError(f"degenerate bbox: x=[{self.x0},{self.x1}) y=[{self.y0},{self.y1})")

    @property
    def width(self) -> int:
        return self.x1 - self.x0

    @property
    def height(self) -> int:
        return self.y1 - self.y0

    @property
    def area(self) -> int:
        return self.width * self.height

    def clipped_to(self, width: int, height: int) -> BoundingBox:
        """Clip to the global domain ``[0,width) x [0,height)``.

        Returns a bbox that may shrink; callers must treat an empty
        intersection (degenerate result) as "no overlap".
        """
        return BoundingBox(
            x0=max(0, self.x0),
            y0=max(0, self.y0),
            x1=min(width, self.x1),
            y1=min(height, self.y1),
        )

    def contains(self, other: BoundingBox) -> bool:
        return (
            self.x0 <= other.x0
            and self.y0 <= other.y0
            and self.x1 >= other.x1
            and self.y1 >= other.y1
        )

clipped_to(width, height)

Clip to the global domain [0,width) x [0,height).

Returns a bbox that may shrink; callers must treat an empty intersection (degenerate result) as "no overlap".

Source code in src/openlithohub/streaming/geometry.py 
44
45
46
47
48
49
50
51
52
53
54
55
def clipped_to(self, width: int, height: int) -> BoundingBox:
    """Clip to the global domain ``[0,width) x [0,height)``.

    Returns a bbox that may shrink; callers must treat an empty
    intersection (degenerate result) as "no overlap".
    """
    return BoundingBox(
        x0=max(0, self.x0),
        y0=max(0, self.y0),
        x1=min(width, self.x1),
        y1=min(height, self.y1),
    )

HaloSpec dataclass

Per-side halo widths requested for a tile read region.

Source code in src/openlithohub/streaming/geometry.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class HaloSpec:
    """Per-side halo widths requested for a tile read region."""

    left: int
    top: int
    right: int
    bottom: int

    @classmethod
    def uniform(cls, px: int) -> HaloSpec:
        if px < 0:
            raise ValueError(f"halo must be nonnegative, got {px}")
        return cls(left=px, top=px, right=px, bottom=px)

    @property
    def total(self) -> int:
        return self.left + self.top + self.right + self.bottom

run_streaming(source, sink, forward_fn, *, core_size, halo_policy=None, verifiers=(), screening_policy=None, work_accounting=None, max_halo_px=1024, pixel_nm=1.0, max_requeues=4, tolerance_nm=None)

Process a full chip tile-by-tile under core/halo ownership.

tolerance_nm (R17 C5) is a legacy convenience: it flows down only when there is at most one verifier. With several verifiers the error budget gate requires per-verifier tolerances on each verifier's own :class:MetricDescriptor — a scalar cannot be a shared tolerance for heterogeneous metrics.

Source code in src/openlithohub/streaming/pipeline.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def run_streaming(
    source: TileSource,
    sink: TileSink,
    forward_fn: Callable[[torch.Tensor], torch.Tensor],
    *,
    core_size: int,
    halo_policy: HaloPolicy | None = None,
    verifiers: Iterable[VerificationPlugin] = (),
    screening_policy: TileScreeningPolicy | None = None,
    work_accounting: WorkAccounting | None = None,
    max_halo_px: int = 1024,
    pixel_nm: float = 1.0,
    max_requeues: int = 4,
    tolerance_nm: float | None = None,
) -> StreamingRunReport:
    """Process a full chip tile-by-tile under core/halo ownership.

    ``tolerance_nm`` (R17 C5) is a legacy convenience: it flows down only
    when there is at most one verifier.  With several verifiers the error
    budget gate requires per-verifier tolerances on each verifier's own
    :class:`MetricDescriptor` — a scalar cannot be a shared tolerance for
    heterogeneous metrics.
    """
    policy = halo_policy or LegacyFixedHaloPolicy()
    verifier_list = list(verifiers)

    vctx = VerificationContext(model=forward_fn, pixel_nm=pixel_nm)
    plugin_requirements = []
    for verifier in verifier_list:
        requirement = verifier.required_halo(vctx)
        if requirement is not None:
            plugin_requirements.append(requirement)

    physical = policy.required_halo(HaloContext(pixel_nm=pixel_nm, tile_core_px=core_size))
    if plugin_requirements:
        from .halo_policy import combine_requirements

        requirement = combine_requirements(physical, *plugin_requirements)
    else:
        requirement = physical
    halo_px = min(requirement.halo_px, max_halo_px)

    # R17 C1a: one verifier instance → one session → one reducer.  Verifier
    # results never share reduction state; the pipeline only schedules and
    # folds the per-verifier outcomes.
    if tolerance_nm is not None and len(verifier_list) > 1:
        raise ValueError(
            "tolerance_nm is a single-verifier legacy convenience; attach a "
            "MetricDescriptor with its own tolerance to each verifier instead"
        )
    sessions: list[VerifierSession] = [
        make_verifier_session(verifier, vctx, fallback_tolerance=tolerance_nm)
        for verifier in verifier_list
    ]
    report = StreamingRunReport(halo_requirement=requirement)
    accounting = work_accounting or WorkAccounting()
    full_pixels = int(source.shape[0] * source.shape[1])
    if accounting.full_chip_pixels == 0:
        accounting.full_chip_pixels = full_pixels
    elif accounting.full_chip_pixels != full_pixels:
        raise ValueError("WorkAccounting.full_chip_pixels does not match the source domain")

    for verifier in verifier_list:
        verifier.prepare(vctx)

    ownership = OwnershipTree()
    for base in plan_tile_requests(source.shape, core_size, halo_px):
        ownership.add_root(base.tile_id, base.core_bbox)
        # R17 C3a: explicit work stack.  increase_halo re-plans one leaf;
        # subdivide retires the leaf and pushes a true area-conserving child
        # partition whose read windows are reconstructed against the GLOBAL
        # domain (never the parent's read bbox).
        pending: list[TileRequest] = [base]
        refinements_left = max_requeues
        while pending:
            current = pending.pop()
            screen_decision = None
            if screening_policy is not None:
                accounting.record_screen_query()
                screen_decision = screening_policy.screen(source, current)

            tile_meta_rejected: dict[str, Any] = {}
            if screen_decision is not None and screen_decision.status == "SCREENED_OUT":
                # R17 C1b: the screen produces typed proof facts; it has no
                # authority over verifiers.  A tile may skip the expensive
                # path only when (a) there are no verifiers to discharge, or
                # (b) every attached verifier accepts the facts itself.
                exact_output, proof_facts, deprecated_authority = screen_decision_to_facts(
                    screen_decision
                )

                accepted: list[TileVerificationResult] = []
                discharged = True
                for session in sessions:
                    accept = getattr(session.plugin, "accept_screen_facts", None)
                    if not callable(accept):
                        discharged = False
                        break
                    fact_ctx = TileContext(
                        tile_id=current.tile_id,
                        core_bbox=current.core_bbox,
                        read_bbox=current.read_bbox,
                        halo=current.halo,
                        tensor=None,
                        metadata={
                            "screening_status": screen_decision.status,
                            "screening_reason": screen_decision.reason,
                        },
                    )
                    verdict = accept(fact_ctx, proof_facts)
                    if verdict is None or verdict.status != "PASS":
                        discharged = False
                        break
                    accepted.append(verdict)

                if discharged:
                    tile_meta: dict[str, Any] = {
                        "screen_disposition": "EXACT_OUTPUT_SKIP",
                        "screening_status": "SCREENED_OUT",
                        "screening_policy": getattr(
                            screening_policy, "name", type(screening_policy).__name__
                        ),
                        "screening_reason": screen_decision.reason,
                        "screening_certified": True,
                        "screening_certificate_ref": screen_decision.certificate_ref,
                        **screen_decision.metrics,
                    }
                    if deprecated_authority:
                        tile_meta["screen_authority"] = (
                            "deprecated-blanket-input-recorded-not-honored"
                        )

                    for session, verdict in zip(sessions, accepted, strict=True):
                        session.add(verdict)

                    # R17 C2: prefer the tensor-free certified commit; legacy
                    # sinks without the capability fall back to a synthesized
                    # exact-fill write_core.
                    recorder = getattr(sink, "record_certified_core", None)
                    if callable(recorder):
                        recorder(
                            current.tile_id,
                            current.core_bbox,
                            exact_fill=exact_output.fill_value,
                            metadata=tile_meta,
                        )
                    else:
                        core_result = torch.full(
                            (current.core_bbox.height, current.core_bbox.width),
                            exact_output.fill_value,
                            dtype=torch.float32,
                        )
                        sink.write_core(current.tile_id, current.core_bbox, core_result, tile_meta)
                    accounting.record_screened_out(current.tile_id, current.core_bbox.area)
                    ownership.set_terminal(current.tile_id, TERMINAL_EXACT_SKIP)
                    report.n_tiles += 1
                    continue

                # Not discharged: fall through to the ordinary active path,
                # recording the rejected legacy authority as provenance.
                tile_meta_rejected = {
                    "screening_status": "SCREENED_OUT_NOT_DISCHARGED",
                    "screening_reason": screen_decision.reason,
                }
                if deprecated_authority:
                    tile_meta_rejected["screen_authority"] = (
                        "deprecated-blanket-input-recorded-not-honored"
                    )

            accounting.record_active_core(current.tile_id, current.core_bbox)
            tile = source.read_window(current.read_bbox)
            accounting.record_read_window(current.read_bbox.area)

            source_meta: dict[str, Any] = {}
            metadata_fn = getattr(source, "verification_metadata", None)
            if callable(metadata_fn):
                source_meta = dict(
                    metadata_fn(
                        current.read_bbox,
                        tile_id=current.tile_id,
                        core_bbox=current.core_bbox,
                    )
                )
            if screen_decision is not None:
                source_meta["screening_status"] = screen_decision.status
                source_meta["screening_reason"] = screen_decision.reason

            result = forward_fn(tile)
            accounting.record_forward(current.read_bbox.area)
            accounting.record_forward_core(current.core_bbox)
            ownership.mark_forward(current.tile_id)
            ys, xs = _core_slices(current)
            core_result = result[ys, xs]

            # Screen-fact provenance (e.g. a rejected SCREENED_OUT) rides on
            # the committed core metadata.
            tile_meta = dict(tile_meta_rejected)
            refinement: RefinementRequest | None = None
            for session in sessions:
                verifier = session.plugin
                tctx = TileContext(
                    tile_id=current.tile_id,
                    core_bbox=current.core_bbox,
                    read_bbox=current.read_bbox,
                    halo=current.halo,
                    tensor=tile,
                    metadata=source_meta,
                )
                verdict = verifier.verify_tile(tctx)
                session.add(verdict)
                tile_meta[f"{verifier.name}_status"] = verdict.status
                if verdict.status == "INCONCLUSIVE" and refinement is None:
                    if refinements_left > 0:
                        refinement = _default_refinement(current, verifier)
                    else:
                        tile_meta[f"{verifier.name}_note"] = (
                            "inconclusive; refinement budget exhausted"
                        )

            if refinement is None:
                ownership.set_terminal(current.tile_id, TERMINAL_ACTIVE)
                sink.write_core(
                    current.tile_id,
                    current.core_bbox,
                    core_result,
                    tile_meta,
                )
                report.n_tiles += 1
                continue

            refinements_left -= 1
            report.n_requeued += 1
            accounting.record_refinement(
                current.tile_id,
                current.tile_id,
                current.core_bbox.area,
            )
            if refinement.action == "subdivide":
                # R17 C3a: retire the parent everywhere and push a true
                # area-conserving partition.  Child read windows are
                # reconstructed against the GLOBAL domain (F3/§10.2), never
                # clipped to the parent read bbox.
                child_cores = [
                    (f"{current.tile_id}#q{index}", box)
                    for index, box in enumerate(
                        partition_core(current.core_bbox, refinement.subdivision)
                    )
                ]
                ownership.subdivide(current.tile_id, child_cores)
                accounting.record_subdivision(current.core_bbox)
                for session in sessions:
                    # R17 C3b: no automatic restriction theorem — every
                    # verifier must reach a fresh verdict on every child,
                    # including verifiers that had PASSed the parent.
                    session.retire(current.tile_id)
                for child_id, child_core in child_cores:
                    child_read = read_bbox_for(
                        child_core,
                        current.halo,
                        width=source.shape[1],
                        height=source.shape[0],
                    )
                    pending.append(
                        TileRequest(
                            core_bbox=child_core,
                            read_bbox=child_read,
                            halo=halo_actual(child_core, child_read),
                            tile_id=child_id,
                        )
                    )
            else:
                pending.append(_grown_request(current, refinement, source.shape))
            del tile, result, core_result

    # R17 C4a: the terminal coverage ledger must partition the chip exactly
    # — gap, overlap, unterminated or double-disposition leaves fail closed.
    terminal = ownership.verify_total_coverage(int(source.shape[0] * source.shape[1]))
    report.ownership = ownership
    report.overhead = tiling_overhead(plan_tile_requests(source.shape, core_size, halo_px))
    if sessions:
        report.verification_results = {session.identity: session.finalize() for session in sessions}
        report.metric_descriptors = {
            session.identity: session.metric_descriptor for session in sessions
        }
        if len(sessions) == 1:
            report.verification = next(iter(report.verification_results.values()))
        else:
            # R17 C1a: multi-verifier facade carries cross-metric-safe data
            # only — never a generic numeric upper bound.
            report.verification = fold_global(report.verification_results)
    summary = accounting.summary()
    summary.update(
        {
            "terminal_active_pixels": terminal[TERMINAL_ACTIVE],
            "terminal_exact_skip_pixels": terminal[TERMINAL_EXACT_SKIP],
            "terminal_verification_skip_pixels": terminal[TERMINAL_VERIFY_SKIP],
        }
    )
    report.work_accounting = summary
    return report