streaming: expose block_bytes; buffer_granules default 50 -> 20 (issue #474) - #475
Conversation
| return {"buffer_granules": buffer_granules, "mode": mode} | ||
| block_bytes = block.get("block_bytes") | ||
| if block_bytes is not None: | ||
| if not isinstance(block_bytes, int) or block_bytes < 1: |
There was a problem hiding this comment.
🤖 from Claude (review)
(1) block_bytes: true passes this check — bool is an int subclass.
isinstance(True, int) is True, and True >= 1, so a YAML block_bytes: true (or yes / on, which PyYAML resolves to booleans) validates and reaches SpillAggregator, which does int(block_bytes) → 1 (src/zagg/processing/spill.py:589). A 1-byte threshold closes a block on every flush — the maximally degraded fold regime — from a knob whose name reads enough like a flag that block_bytes: true is a plausible operator mistake.
Verified on this branch:
>>> get_streaming(cfg({"mode": "spill", "block_bytes": True}))
{"buffer_granules": 20, "mode": "spill", "block_bytes": True}
One-clause fix on the new check:
if isinstance(block_bytes, bool) or not isinstance(block_bytes, int) or block_bytes < 1:buffer_granules has the identical hole (buffer_granules: true → 1, pre-existing at line 113) — worth closing in the same validator while it is open, or leaving with an explicit note. A parametrize case with True in test_bad_block_bytes_raises pins it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 10d8fdb — and yes, buffer_granules too, in the same validator.
Both checks now route through one helper, so neither knob can take a bool:
def _positive_int(value) -> bool:
return isinstance(value, int) and not isinstance(value, bool) and value >= 1True/False added to the parametrize lists of both test_bad_buffer_raises and test_bad_block_bytes_raises (tests/test_streaming.py). I judged the buffer_granules half in scope rather than scope creep: this PR already rewrites that line (the 50 → 20 default sits on it), the fix is the same one clause, and leaving one half of an identical hole open behind a shared helper would read as deliberate.
| if mode not in ("merge", "spill"): | ||
| raise ValueError(f"aggregation.streaming.mode must be 'merge' or 'spill' (got {mode!r})") | ||
| return {"buffer_granules": buffer_granules, "mode": mode} | ||
| block_bytes = block.get("block_bytes") |
There was a problem hiding this comment.
🤖 from Claude (review)
(2) An explicit block_bytes skips the 45%-of-free-/tmp cap the derived default exists to enforce — two blocks are resident at once.
_default_block_bytes deliberately clamps to 45% of the spill directory's free space:
tmp_cap = int(0.45 * st.f_bavail * st.f_frsize)
return max(1, min(int(0.8 * 0.75 * mem * n_partitions / _BUILD_MULT), tmp_cap))because under the overlap path a closing block coexists with the filling one (_close_block: "the /tmp working set stays closing + filling (the threshold formula's reservation)", spill.py:737-747).
The explicit branch takes a different route — it only requires the value itself to fit:
self.block_bytes = int(block_bytes)
check_tmp_headroom(max(_MIN_SPILL_BYTES, self.block_bytes), self.tmp_dir) # spill.py:589-590So any config value between ~50% and 100% of the variant's free /tmp passes the guard at construction and then runs out of disk mid-shard, ~15 minutes into a 900 s Lambda — the exact failure check_tmp_headroom was written to pre-empt. That was tolerable while block_bytes= was a test/ops kwarg; this PR makes it the knob operators are told to reach for.
Two ways to close it, pick one:
- (a) minimum: say it here (and in the
atl03_tdigest_healpix_hive.yamlcomment) — keepblock_bytesat or below ~45% of the tier's ephemeral storage, because the closing block is resident beside the filling one. - (b) stricter: headroom-check
2 * block_byteson the explicit branch, so the guard reserves what the run actually needs. Behavior-changing for the kwarg path, so it wants a line in the PR body if taken.
There was a problem hiding this comment.
🤖 from Claude
Took option (b), folded in e27f510 — plus the (a) documentation, since it costs nothing.
The explicit branch now reserves what the run actually holds:
self.block_bytes = int(block_bytes)
resident = self.block_bytes * (2 if overlap else 1)
check_tmp_headroom(max(_MIN_SPILL_BYTES, resident), self.tmp_dir, from_config=True)overlap=False still reserves 1x, since only one block is resident there. The ~50–100%-of-free-/tmp window you identified is now refused at construction instead of ENOSPC-ing 15 minutes in, and ~45% remains the largest value that passes — the same ceiling _default_block_bytes enforces on itself.
Test: tests/test_spill.py::TestSpillConfig::test_explicit_block_bytes_reserves_the_overlap_pair — statvfs faked to 1 GiB free, block_bytes at 60% of it raises under overlap and constructs cleanly with overlap=False.
The (a) half landed in 19f8568: the atl03_tdigest_healpix_hive.yaml comment now says to keep block_bytes at or below ~45% of the variant’s ephemeral storage, and why. The PR body carries the behavior change under "Approach".
| # Merge mode has no spill blocks; a silently-ignored knob is the | ||
| # #239 failure mode, so refuse it where it can do nothing. | ||
| raise ValueError( | ||
| "aggregation.streaming.block_bytes only applies to mode: spill " |
There was a problem hiding this comment.
🤖 from Claude (review)
(3) Both spill failure messages now quote a config-sourced number without naming the config knob.
With block_bytes reachable from config, the two errors an operator actually hits list every remedy except the one they now control:
check_tmp_headroom(spill.py:114-119): "...needs 1,073,741,824 bytes of free space in '/tmp' but only N are available; deploy on a function variant with larger ephemeral storage (the '-disk' variants...) or fall back to mode: merge." — verified live on this branch; the number is the operator's ownaggregation.streaming.block_bytesand the message never says so, so the obvious first fix (lower it) is not on the list.SpillOverflowError(spill.py:750-759): "Remedies: a bigger memory tier, a '-disk' function variant with more ephemeral storage, or a finer parent_order (smaller shards)." — when the crossing came from a config-set threshold, "raise or dropaggregation.streaming.block_bytes" belongs at the head of that list.
Both are one clause each. Cheapest version that stays honest: only mention the knob when it was set — the aggregator already knows (block_bytes is not None at construction), so store a flag and append "; the threshold came from aggregation.streaming.block_bytes" to each message.
There was a problem hiding this comment.
🤖 from Claude
Folded in 340207d, both messages, and only when the number is actually the operator’s.
check_tmp_headroomtakesfrom_config: bool = False; the explicit branch passesfrom_config=True, and the remedies then lead withlower aggregation.streaming.block_bytes (this requirement is derived from it), …before the-diskvariant andmode: merge. The derived-default call site is unchanged, so the disk-derived path never points at a knob nobody set.SpillAggregatorrecordsself._block_bytes_from_config = block_bytes is not None, and_close_blockprefixes theSpillOverflowErrorremedies witha larger aggregation.streaming.block_bytes (this threshold came from it),. Raising it is the remedy that keeps the shard in the exact single-block regime, so it heads the list; lowering never helps there.
Tests, both directions: TestTmpGuard::test_raises_naming_disk_variant_when_tmp_too_small asserts the knob appears with from_config=True and that block_bytes is absent from the default message; test_overflow_names_the_config_knob_that_set_the_threshold drives a config-set block_bytes: 1 through process_shard on a non-mergeable config, and test_non_mergeable_overflow_raises_loudly now asserts the derived-threshold overflow does not mention the knob.
| :mod:`zagg.processing.spill`). ``block_bytes`` (issue #474, spill only) | ||
| caps the spilled block before it closes and folds (issue #370) — the | ||
| principled bound on aggregate-phase memory for obs-dense shards. Absent it | ||
| stays ``None`` and the worker keeps the disk-derived default |
There was a problem hiding this comment.
🤖 from Claude (review)
(4) The spill.py side of this knob is now stale — it still says the only way in is the constructor kwarg.
This docstring points at _default_block_bytes, whose own docstring ends:
Injectable for tests and ops via
SpillAggregator(block_bytes=...).
(src/zagg/processing/spill.py:459). As of this PR it is also injectable from config, which is the route an operator will actually take — a reader who lands there from the fold-regime warning has no reason to discover aggregation.streaming.block_bytes. Same for the class docstring's multi-block bullet (spill.py:489-490), which names only "see :func:_default_block_bytes" as the source of the threshold.
Two one-line docstring edits in the files this PR already reaches into; cheap, and it keeps the knob discoverable from the module that raises about it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 19f8568 — both docstrings plus the template comment.
_default_block_bytes now ends:
Overridable from config —
aggregation.streaming.block_bytes(issue #474), the route operators take — and fromSpillAggregator(block_bytes=...)for tests and ops. An explicit value replaces this whole formula, 45% cap included, so the constructor headroom-checks the pair it will actually hold (2x under overlap); keep it at or below ~45% of the tier’s ephemeral storage.
The class docstring’s multi-block bullet now reads "bytes hit the threshold — aggregation.streaming.block_bytes when the config sets it, else _default_block_bytes", so a reader landing there from the fold-regime warning sees the config route. The atl03_tdigest_healpix_hive.yaml comment carries the same ~45% guidance (that is also finding (2)’s option (a)).
| def test_large_block_bytes_stays_single_block(self, monkeypatch): | ||
| # A generous config threshold keeps the exact single-block regime. | ||
| key = _shard_key() | ||
| spill = {"buffer_granules": 1, "mode": "spill", "block_bytes": 1 << 30} |
There was a problem hiding this comment.
🤖 from Claude (review)
(5) block_bytes: 1 << 30 makes this test demand 1 GiB free in the real $TMPDIR.
The explicit branch headroom-checks the value itself:
self.block_bytes = int(block_bytes)
check_tmp_headroom(max(_MIN_SPILL_BYTES, self.block_bytes), self.tmp_dir) # spill.py:589-590and this test's aggregator gets no tmp_dir, so self.tmp_dir = tempfile.gettempdir() — the process temp dir, not tmp_path. On any box with under 1 GiB free there (a small container, a full laptop disk, a TMPDIR on a tmpfs), the test fails with RuntimeError: aggregation.streaming.mode: spill needs 1,073,741,824 bytes of free space ... deploy on a function variant with larger ephemeral storage — an environmental failure wearing the costume of a spill bug, in a test whose subject is spill_blocks_closed == 0.
1 << 26 (64 MiB, exactly _MIN_SPILL_BYTES) keeps the headroom demand identical to the default path's floor while still sitting ~4 orders of magnitude above the ~6 KB this test actually spills (5 granules x 3 cells x 20 obs x 20 B/row), so the single-block assertion is just as strong.
There was a problem hiding this comment.
🤖 from Claude
Folded in 9bb969d — at 1 << 25, not 1 << 26, and only because of the fix for finding (2).
Your target was "the same headroom demand as the default path’s floor". With e27f510 the explicit branch now checks 2 * block_bytes under overlap (a closing block resident beside the filling one), so 1 << 26 would have demanded 128 MiB. 1 << 25 (32 MiB) doubles to exactly _MIN_SPILL_BYTES — the floor the default path checks — while still sitting ~4 orders of magnitude above the ~6 KB this test spills, so the spill_blocks_closed == 0 assertion is unchanged in strength. The comment on the test states both halves so the constant is not re-raised later.
| if block_bytes is not None: | ||
| assert seen["resolved"] == block_bytes | ||
| else: | ||
| assert seen["resolved"] > 0 # disk-derived default, unchanged |
There was a problem hiding this comment.
🤖 from Claude (review)
(6) The absent-block_bytes arm of this test is close to vacuous.
assert seen["resolved"] > 0 cannot fail: _default_block_bytes returns max(1, min(...)), so it is >= 1 by construction (spill.py:464). The arm therefore pins "some default was computed", not "the disk-derived default is unchanged", which is what the comment claims.
The spy already holds the aggregator; capturing it lets the assertion be exact:
def spy(agg, *args, **kwargs):
seen["block_bytes"] = kwargs.get("block_bytes")
real_init(agg, *args, **kwargs)
seen["resolved"] = agg.block_bytes
seen["agg"] = agg
...
else:
agg = seen["agg"]
assert seen["resolved"] == spill._default_block_bytes(agg._n_partitions, agg.tmp_dir)Second, smaller point on the same spy: kwargs.get("block_bytes") only observes the keyword form. If the worker ever passes the argument positionally, the None arm records None and still passes — a silent hole in exactly the seam this test exists to guard. inspect.signature(real_init).bind(agg, *args, **kwargs) (or asserting the call is keyword-shaped) closes it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 2096942 — both points, essentially your patch.
sig = inspect.signature(real_init)
def spy(agg, *args, **kwargs):
seen["block_bytes"] = sig.bind(agg, *args, **kwargs).arguments.get("block_bytes")
real_init(agg, *args, **kwargs)
seen["resolved"] = agg.block_bytes
seen["agg"] = agg
...
else:
agg = seen["agg"]
assert seen["resolved"] == _default_block_bytes(agg._n_partitions, agg.tmp_dir)sig.bind(...).arguments closes the positional hole (a positional block_bytes now lands in arguments, so the None arm would fail rather than silently pass), and the absent arm pins the disk-derived value exactly instead of the > 0 that max(1, …) made unfalsifiable. _default_block_bytes is imported from zagg.processing.spill alongside the other symbols the module already pulls in.
|
🤖 from Claude (review) (7) Two additions to the D19 report — neither asks for a ruling, both are evidence the ruling will want. (7a) The docstring is not the only place that says streaming is packaging. The PR body cites
So option (b) is not only "matches the docstring" — it is what the design record states as an acceptance property, and option (a) means amending that record in the same breath. (7b) Under status quo (a), That is exactly the motivating workflow of issue #474: the five CA o9 fat shards failed inside an existing store, and the rescue is a re-dispatch of those shards into that store. Under (a) an operator who adds Worth stating in the PR body under the D19 section so whichever option is picked is picked with the rescue path in view. |
|
🤖 from Claude Both addenda are now in the PR body’s D19 section; no code moved.
|
|
🤖 from Claude Decision record — the D19 question is ruled (espg, in-session, 2026-08-17): option (b) + the law-equivalence contract. Phase 3 (commit 3b0fd72) lands it:
Question 2 (merge-mode If any part of this record misstates the ruling, correct it here — this comment is the thread record the design doc points at. |
| within the documented approximation law of the pooled path — exact for | ||
| the single-block regime and the summation reducers, the | ||
| kway/`np.isclose` class across merge flushes and block closes — and a | ||
| mode that cannot maintain that law may not join the streaming block. |
There was a problem hiding this comment.
🤖 from Claude (review)
(1) The law-equivalence contract names two law classes, but the streaming block already admits two channels that are in neither — read literally, the contract excludes reducers the ruling keeps in.
The normative sentence is:
exact for the single-block regime and the summation reducers, the kway/
np.iscloseclass across merge flushes and block closes — and a mode that cannot maintain that law may not join the streaming block.
Two channels already shipping under mode: spill fold to a bounded/conservative law, not to np.isclose:
- the located companion — a block close coarsens each centroid's word toward the contributors' common ancestor (spec §9.1). The pin is an ancestor-or-equal hull assertion, not a closeness one —
tests/test_spill_crossblock.py:200(_assert_ancestor_or_equal), used at:436and:551. A multi-block located word is deliberately not the pooled word; - the packed composition word — presence (
lane > 0) is exact, counts are within one quantization of pooled:tests/test_spill_crossblock.py:700
tol = 1 + n / 255.0
assert np.abs(counts_from_composition(word_s, n) - truth).max() <= tolnp.isclose's default rtol=1e-5 is nowhere near that, so build_composition and located digests both fail the contract as spelled while validate_spill_fold admits them by design.
Suggested precision fix: scope the two named classes to the payload channel and name the companion/word law as a third class — something like "…and the conservative-bound class for the companion channels (located words fold to the contributors' common ancestor, §9.1; the packed composition word keeps presence exactly and counts within one quantization)". Otherwise the sentence that decides which modes may join the block is one a shipped reducer already violates.
The same sentence is mirrored verbatim in src/zagg/semantics.py:204-207, so both need the edit.
There was a problem hiding this comment.
🤖 from Claude
Folded in 5de8149 — the contract now names three law classes instead of two, in both the design doc and the mirrored comment.
docs/design/sparse_coverage.md: (1) exact (single-block regime + summation reducers), (2) kway/np.isclose for the payload channel across merge flushes and block closes, (3) the channel-specific documented bounds for the companion channels, "conservative rather than close" — the located word folding toward the contributors' common ancestor (ancestor-or-equal hull pin, §9.1) and the packed composition word keeping presence (lane > 0) exactly with counts within one lane quantization (tol = 1 + n / 255.0). The joining rule is now "a mode that cannot maintain the class its channels fall in may not join the streaming block".
src/zagg/semantics.py (the AGGREGATION_PACKAGING_KEYS comment) carries the same three-class list in condensed form.
Anchors checked against the branch: _assert_ancestor_or_equal at tests/test_spill_crossblock.py:200 (used at :436 and :551) and the tol = 1 + n / 255.0 assertion at :700.
| The contract, not the digest, is what makes one shared identity across | ||
| all streaming regimes honest; its enforcement is the cross-block suites | ||
| (`tests/test_spill_crossblock.py`, being extended to the temporal | ||
| channel by #477) and `tests/test_streaming.py`'s pooled-parity pins. |
There was a problem hiding this comment.
🤖 from Claude (review)
(2) The enforcement pointer misses the suite that pins the exactness half, and the code docstring's pointer disagrees with this one.
The paragraph claims enforcement is tests/test_spill_crossblock.py + tests/test_streaming.py's pooled-parity pins. But the first clause of the contract — "exact for the single-block regime" — is pinned in neither: it lives in tests/test_spill.py::TestSpillWorkerSingleBlock (tests/test_spill.py:637), specifically
test_full_reducer_matrix_byte_identical_to_pooled(:652)test_nan_cells_byte_identical_to_pooled(:661)test_k_gt_1_chunks_byte_identical_to_pooled(:678)test_capped_groups_byte_identical_to_pooled(:697)
test_spill_crossblock.py is by construction the multi-block suite (_force_tiny_blocks), and test_streaming.py's exact pin is merge-mode single-buffer only (test_single_buffer_is_byte_identical_to_pooled, :264). Adding tests/test_spill.py makes the pointer cover both halves of the contract it is supposed to enforce.
Second, src/zagg/semantics.py:208 names only tests/test_spill_crossblock.py while this paragraph names two files. Since the doc is the normative record and the docstring says "Enforcement is …", the two lists should be the same list.
There was a problem hiding this comment.
🤖 from Claude
Folded in 96f284a — both pointers now name the same enforcement set.
Doc paragraph: "its enforcement is the single-block exactness pins (tests/test_spill.py::TestSpillWorkerSingleBlock), the cross-block suites (tests/test_spill_crossblock.py, being extended to the temporal channel by issue #477), and tests/test_streaming.py's pooled-parity pins."
src/zagg/semantics.py: "Enforcement is tests/test_spill.py::TestSpillWorkerSingleBlock (the exactness half), tests/test_spill_crossblock.py, and tests/test_streaming.py's pooled-parity pins."
Verified the class exists at tests/test_spill.py:637 with the four byte-identical pins at :652/:661/:678/:697 (plus the tdigest multi-flush one at :669). Also spelled the #477 reference as "issue #477" in the doc so the bare #N reads unambiguously.
| :data:`GRID_LEAF_SHAPING_KEYS`), worker sizing, the whole | ||
| ``aggregation.streaming`` block (mode AND its sizing knobs, under the | ||
| law-equivalence contract — espg-ruled 2026-08-17 on the PR #475 D19 question; | ||
| see :data:`AGGREGATION_PACKAGING_KEYS`), read knobs, |
There was a problem hiding this comment.
🤖 from Claude (review)
(3) The module docstring is only half-updated: the included bullet still says the aggregation block comes in minus handoff alone.
This exclusion line is correct now, but its counterpart 50 lines up (src/zagg/semantics.py:12-14) was not touched and now contradicts it:
- the ``aggregation`` block — functions, params, dtypes, fills, ragged kinds,
declared coordinates, ``chunk_precompute`` — minus the ``handoff`` carrier
choice (arrow vs pandas is a worker-internal transport);
streaming is now the second subtraction, so that bullet should read "minus the handoff carrier choice and the streaming block". A reader who reads the Included list top-down gets the pre-ruling answer.
Same omission one level down: semantic_core's own docstring (src/zagg/semantics.py:552) enumerates the packaging classes as "orders, chunking, worker size, read machinery, carrier" — a list that was complete for AGGREGATION_PACKAGING_KEYS = ("handoff",) and no longer is. test_packaging_knobs_never_change_hash and the new pin both assert against that promise, so it is worth keeping literal.
There was a problem hiding this comment.
🤖 from Claude
Folded in b28d86c — both docstrings now list streaming as the second subtraction.
Module docstring (Included list): the aggregation bullet reads "…minus the handoff carrier choice (arrow vs pandas is a worker-internal transport) and the whole streaming block (every streaming regime shares one identity under the law-equivalence contract — see the exclusions below)", so a top-down reader gets the post-ruling answer without having to reach the Excluded paragraph.
semantic_core docstring: the packaging enumeration is now "orders, chunking, worker size, read machinery, carrier, the aggregation.streaming block".
No change to AGGREGATION_PACKAGING_KEYS membership or any behavior; tests/test_semantics.py stays green (49 passed).
| #: Enforcement is ``tests/test_spill_crossblock.py``. The D19 record stated | ||
| #: this acceptance property twice while the code hashed the block as spelled; | ||
| #: the ruling lands the code on the record's side before any long-lived store | ||
| #: carries a streaming-declared digest, so the epoch moves nothing that |
There was a problem hiding this comment.
🤖 from Claude (review)
(4) The docstring drops the qualifier that makes the store-compatibility premise true, and as written it is checkably wrong.
Not relitigating the ruling — the premise is espg's and the design doc states it defensibly: "no long-lived store carries a streaming-declared digest (the deployed stores age out on a 30-day cycle)" (docs/design/sparse_coverage.md:804-806). The qualifier is the whole load-bearing part.
The docstring keeps the claim and drops the qualifier — "before any long-lived store carries a streaming-declared digest" reads as no config has been declaring one, and two shipped configs have been declaring one on main before this PR:
$ git grep -ln '^ streaming:' main -- src/zagg/configs
src/zagg/configs/atl03_tdigest_healpix_hive.yaml
src/zagg/configs/gedi01b_waveform_healpix_hive.yamlatl03_tdigest_healpix_hive.yaml has carried streaming: {mode: spill} since before this branch (phases 1-2 touched only its comments), so every store built from either template does carry a digest computed with the block in it — those digests all move at this epoch. The doc's claim survives that because it is scoped to long-lived; the docstring's does not.
Suggest carrying the age-out clause across verbatim, e.g. "…before any long-lived store carries a streaming-declared digest (the deployed stores age out on a 30-day cycle), so the epoch moves nothing that outlives it." One clause, and the reader stops having to reconcile it with the shipped templates.
There was a problem hiding this comment.
🤖 from Claude
Folded in 7b33fd5 — the comment now carries the design doc's scoping verbatim:
#: the ruling lands the code on the record's side before any **long-lived**
#: store carries a streaming-declared digest (the deployed stores age out on
#: a 30-day cycle), so the epoch moves nothing that outlives it.
That matches docs/design/sparse_coverage.md ("no long-lived store carries a streaming-declared digest (the deployed stores age out on a 30-day cycle)") and stops the sentence from reading as a claim that no config declares streaming: — which the shipped atl03_tdigest_healpix_hive.yaml and gedi01b_waveform_healpix_hive.yaml templates on main would contradict. Ruling itself untouched.
1 participant
Closes #474
Espg-directed small-fix from the 2026-08-17 CA o9 fat-shard investigation (issue #474): worker memory tracks observation volume, the resident term is the pre-flush buffer, and the fold-regime threshold (
block_bytes) was unreachable from config.Phases
aggregation.streaming.block_bytes:get_streamingaccepts an optional positive int (spill-only; refused undermode: mergeso it can never be silently ignored — the raster configs silently ignore store_layout / coverage_moc (validate_config returns early) #239 discipline), the unknown-key refusal now names all three keys, and the worker threads it to theSpillAggregator(...)construction, whose signature already carriedblock_bytes=(src/zagg/processing/spill.pyline 520). Absent keeps the disk-derived_default_block_bytesbehavior byte-for-byte.buffer_granulesdefault 50 → 20 (get_streaming), with every pinned 50 updated (module example,tests/test_streaming.pydefault pin, theatl03_tdigest_healpix_hive.yamlcomment). The GEDI template pins 8 explicitly and is untouched.block_bytes/tmpguard now reserves the overlap pair; both spill failure messages name the knob when it set the number;spill.pydocstrings and the template comment point at the config route; two test-strength/portability fixes."streaming"joinsAGGREGATION_PACKAGING_KEYSso the whole block (mode,buffer_granules,block_bytes) leaves the semantic core; the law-equivalence contract is stated normatively indocs/design/sparse_coverage.mdbeside the D19 amendment record as the exclusion's condition;tests/test_semantics.pypins every spelling of the block to the bare config's hash; thesemantics.pydocstring/code contradiction is closed on the docstring's side.Approach
get_streamingstays the single validation point and now always returns{"buffer_granules", "mode", "block_bytes"}(block_bytes: None= disk-derived default), so the worker passes the key through unconditionally. The worker's aggregator dispatch was reshaped toif streaming_cfg is None / elif spill_mode / else— same behavior, and it lets mypy narrowstreaming_cfg(net mypy error count inworker.pygoes 16 → 15; the remaining 15 are pre-existing baseline).Two behavioral points came out of the fold:
bool.boolsubclassesint, soblock_bytes: true(PyYAML also resolvesyes/on) validated and ran as a 1-byte threshold — a block close on every flush, the maximally degraded fold regime, from a knob name that reads like a flag._positive_intinstreaming.pynow excludesboolforbuffer_granules(the same pre-existing hole) andblock_bytesalike.block_bytesis headroom-checked for the pair it actually holds._default_block_bytescaps itself at 45% of free/tmpprecisely because a closing block stays resident beside the filling one under overlap (_close_block); an explicit value skipped that cap and was checked 1x, so anything between ~50% and 100% of free/tmppassed construction and then ENOSPC'd mid-shard — the failurecheck_tmp_headroomexists to pre-empt. The explicit branch now checks2 * block_byteswhenoverlapis on (1x when off), and the template comment states the ~45% guidance.Tests
tests/test_streaming.py— default pin now 20 withblock_bytes: None; explicitblock_bytesround-trips; bad type/value refusals (0,-1,"1048576",2.5,True/False) on bothbuffer_granulesandblock_bytes; merge-mode refusal; unknown-key message names the full three-key grammar.tests/test_spill.py::TestSpillConfig::test_block_bytes_reaches_the_worker_constructor— seam-level spy onSpillAggregator.__init__throughprocess_shard, binding through the real signature (so a positional pass cannot slip the absent-knob arm): explicit value arrives and is honored; absent arrives asNoneand resolves exactly to_default_block_bytes(agg._n_partitions, agg.tmp_dir).tests/test_spill.py::TestSpillConfig::test_explicit_block_bytes_reserves_the_overlap_pair— at 60% of (faked) free/tmpthe construction raises under overlap and succeeds withoverlap=False.tests/test_spill.py—check_tmp_headroom(..., from_config=True)leads with "loweraggregation.streaming.block_bytes" and omits it otherwise;SpillOverflowErroroffers "a largeraggregation.streaming.block_bytes" only when the threshold came from config.tests/test_spill_crossblock.py::TestConfigBlockBytes— fold-regime smoke driven purely from config (no monkeypatched_default_block_bytes):block_bytes: 1+buffer_granules: 1closes a block per flush (spill_blocks_closed == 5), counts fold exactly vs pooled and digests carry exact total weight within t-digest accuracy (the issue spill: cross-block merge for located / strata / composition fields (lift the single-block restriction) #370 kway law); a generousblock_bytesstays in the exact single-block regime (spill_blocks_closed == 0). That "generous" value is 32 MiB, not 1 GiB: an explicit threshold is headroom-checked against the real$TMPDIR, so the old1 << 30demanded a free GiB of the developer's own disk.tests/test_semantics.py::TestCanonicalization::test_streaming_block_is_packaging_in_every_spelling(phase 3) — absent,{},{mode: spill},{mode: merge, buffer_granules: 7},{mode: spill, block_bytes: 1<<26}, and{buffer_granules: 20}all hash identically to the bare config, andstreamingnever appears in the core — the drift-proofing the review's D19 addendum flagged as missing. The existing golden hash pin (fb15224f…, no streaming declared) is untouched, as expected for a key-removal on configs that never carried the key.Local:
ruff check/ruff format --checkclean on every touched file;pytest tests/test_streaming.py tests/test_spill.py tests/test_spill_crossblock.py= 141 passed; everysemantic_hashconsumer suite green after phase 3 (test_semantics49, plustest_spec_conformance,test_hive,test_config,test_dispatch,test_dedup,test_lifecycle,test_hive_windows,test_sweep_overview,test_column,test_lambda_handler,test_sweep_stage,test_client= 1,381 passed / 1 skipped; no spec fixture declaresstreaming, so no fixture regeneration is triggered); fullpytest -qon the final tree = 4,449 passed, 38 skipped, 1 failed (test_lambda_build, environmental — see below).D19 / semantic-hash finding — RULED, option (b), implemented in phase 3
Ruling record (espg, in-session, 2026-08-17): option (b) plus the law-equivalence contract. The whole
aggregation.streamingblock is excluded from the semantic core (AGGREGATION_PACKAGING_KEYS = ("handoff", "streaming")), on these criteria as ruled:docs/design/sparse_coverage.md("Amended again — the streaming exclusion and its law-equivalence contract"): every streaming mode MUST land within the documented approximation law of the pooled path (exact single-block/summation, kway/np.iscloseacross flushes and block closes); a mode that cannot maintain the law may not join the block. Enforcement:tests/test_spill_crossblock.py(being extended to the temporal channel by spill: carry the temporal companion through the cross-block fold (block_bytes unusable on temporal stores) #477) andtests/test_streaming.py's pooled-parity pins.Phase 3 implements exactly this; the finding below stands as the record of what the code did before the ruling.
The finding as originally reported
aggregation.streamingDOES participate in the semantic core today — as declared (spelled), not resolved.semantic_corefoldsconfig.aggregationminus onlyAGGREGATION_PACKAGING_KEYS = ("handoff",)(src/zagg/semantics.pylines 195–197 and 562–565); nothing stripsstreaming. Verified empirically: absent block,{},{buffer_granules: 50},{buffer_granules: 20}, and{mode: spill}all produce five distinctsemantic_hashvalues.This contradicts the module's own docstring, which lists "streaming mode" among "Excluded as packaging" (
src/zagg/semantics.pyline 62) — and the D19 design record says it twice more, in stronger terms:docs/design/sparse_coverage.mdL727-735 — "Excluded as packaging: cell order …, parent/shard order,chunk_inner/sharded…, worker size, streaming mode (merge-vs-spill landsnp.iscloseand shares one store, with the actual mode recorded per-run), and read knobs".docs/design/sparse_coverage.mdL1190-L1195, the D19 acceptance list — "packaging-knob edits (orders, chunking, worker size, streaming mode) never change the hash".So option (b) below is not merely "matches the docstring" — it is what the design record states as an acceptance property, and option (a) means amending that record in the same breath. No test asserts the property today (
tests/test_semantics.pynever mentionsstreaming), which is how the code and the record drifted apart unnoticed.Consequences as the code stands:
This PR's default flip moves no existing hashes — defaults resolve in
get_streaming, not in the core, so a config that omitsbuffer_granuleshashes identically before and after. But undermode: mergethe flush cadence changes tdigest merge grouping, so the same identity can now produce (approximation-law-bounded) different bytes than a pre-streaming: expose block_bytes in config; drop buffer_granules default 50 -> 20 (obs-dense shard OOMs) #474 run — the bytes/identity tensionbuffer_granulesalways had, now exercised by a default change.Spelling an explicit default forks identity:
{buffer_granules: 20}hashes differently from the absent key even though the two are the same computation — the opposite of the resolved-not-spelled discipline the D19 hash epoch: granule_workers into DATA_SOURCE_PACKAGING_KEYS; semantic_core widened to the leaf-shaping output knobs #415 epoch applies to hashed knobs.block_bytesstraddles the packaging/identity line both ways: declaring it changes the hash even at values that keep the exact single-block regime (byte-identical to pooled), while an actually-crossed threshold changes bytes under issue spill: cross-block merge for located / strata / composition fields (lift the single-block restriction) #370's fold laws.Under status quo (a),
block_bytescannot be added to an existing store — the manifest guard refuses.semantic_hashis a FROZEN manifest key (src/zagg/hive.py_FROZEN_MANIFEST_KEYS, line 542–551) and_frozen_matchesrefuses an append when two hash-carrying manifests disagree (hive.pylines 565–581). Adding the knob changes the hash — measured on this branch against the shippedatl03_tdigest_healpix_hive.yaml, identical otherwise:That is exactly the motivating workflow of streaming: expose block_bytes in config; drop buffer_granules default 50 -> 20 (obs-dense shard OOMs) #474: the five CA o9 fat shards failed inside an existing store, and the rescue is a re-dispatch of those shards into that store. Under (a) an operator who adds
block_bytesto rescue them gets a frozen-key refusal instead, leaving only a fresh store or an out-of-band manifest edit.Options (key membership is espg-ruled per the #420 epoch precedent — no hash change is made in this PR):
"streaming"toAGGREGATION_PACKAGING_KEYS— matches the docstring and the design record's stated acceptance property; single-block spill is byte-identical to pooled so the packaging claim is honest there, but fold-regime and merge-mode runs would then produce different bytes under one identity (bounded by the t-digest/kway laws). A hash epoch: existing configs that declarestreamingchange digests. Allows the (4) rescue.mode, if a byte-affecting regime choice is deemed output-defining) and exclude the sizing knobs (buffer_granules,block_bytes) as packaging. Most precise, most machinery; the only option under which the (4) rescue is a config edit without a hash epoch for everystreaming-declaring config.Questions for review — both ruled
The D19 finding above — which option, (a)/(b)/(c)?Ruled: (b) + the law-equivalence contract (espg, in-session 2026-08-17; record above). Implemented in phase 3; the streaming: expose block_bytes in config; drop buffer_granules default 50 -> 20 (obs-dense shard OOMs) #474 rescue path (item (4)) is now a plain config edit on an existing store.block_bytesundermode: mergeis refused rather than ignored.Pre-existing failures observed (not touched, per §4)
ruff check src testson origin/main already fails:N818onsrc/zagg/registry.py:64(UnknownCapability), untouched here.ruff format --checkalready flagstests/data/benchmark/README.md(embedded snippet), untouched here.worker.py's count by one.test_lambda_build(environmental) andtest_client_transport(wall-clock flake) are known local failures unrelated to this change.