diff --git a/docs/hive_layout.md b/docs/hive_layout.md index 445ba62c0..2dd9c6396 100644 --- a/docs/hive_layout.md +++ b/docs/hive_layout.md @@ -667,6 +667,14 @@ The example above is `zagg.hive.build_root_coverage` output for the shards `root_coverage_words`; the test suite parses it straight out of this file so the reference example can never drift from the implementation. +A temporal-declaring store adds one more key here: `temporal`, the +`zagg-coverage-toc/1` section (per-shard toc envelope words plus an optional +root time-digest) whose grammar is normative in +[`specification.md`](specification.md) §10 — one metadata GET then answers +"which shards hold data DURING my window" before any leaf is opened. A store +with no temporal channel carries no such key and its root object is +byte-identical to a pre-#480 one; absence is never a refusal. + A range is an inclusive run of same-order cells within one base cell, consecutive in digit-tail rank; endpoints are decimal **strings** (packed u64 words exceed 2^53 and raw JSON numbers get mangled by float-based diff --git a/docs/specification.md b/docs/specification.md index 82801a963..5bacd6813 100644 --- a/docs/specification.md +++ b/docs/specification.md @@ -3,8 +3,9 @@ This page is the **normative record** of zagg's byte-level store conventions: the ragged vlen-bytes layout, the t-digest payload bytes, the packed composition word, the pyramid/overview declarations, the O11 content-hash -recipe, and the temporal and located declarations on the word-typed -coordinate and companion arrays. It exists so an independent reader +recipe, the temporal and located declarations on the word-typed +coordinate and companion arrays, and the temporal section of the store-root +coverage sidecar. It exists so an independent reader ([moczarr](https://github.com/espg/moczarr)) can decode a zagg store from this page and the committed conformance fixtures alone — no zagg import, no reverse-engineering of `grids/base.py` @@ -56,7 +57,7 @@ text marked *informative* explains or motivates and binds nothing. explicit succession clause; readers add revisions, they never drop them. - The committed conformance fixtures (§7) are part of the contract: a reader implementation that reproduces the fixtures' expected decoded values and - content hashes conforms to §1–§3, §5, §8 and §9. zagg's own test suite asserts the + content hashes conforms to §1–§3, §5, §8, §9 and §10. zagg's own test suite asserts the same expectations (`tests/test_spec_conformance.py`), so the spec, the fixtures, and the shipping reader cannot drift apart silently. @@ -71,6 +72,7 @@ Contents: 7. [Conformance fixtures](#7-conformance-fixtures) 8. [`zagg-toc/1` — the temporal declaration](#8-zagg-toc1) 9. [`zagg-located/1` — the located declaration](#9-zagg-located1) +10. [`zagg-coverage-toc/1` — the root coverage temporal section](#10-zagg-coverage-toc1) --- @@ -1485,7 +1487,7 @@ openability). [`tests/data/spec/`](https://github.com/englacial/zagg/tree/main/tests/data/spec) are part of this specification: a reader implementation that reproduces their expected decoded values and content hashes conforms to §1–§3, §5, -§8 and §9. They are generated by +§8, §9 and §10. They are generated by [`tools/generate_spec_fixtures.py`](https://github.com/englacial/zagg/blob/main/tools/generate_spec_fixtures.py) through zagg's **production write path** (manifest, sharded leaf template, dense + ragged writes, coverage sidecar, commit stamp), so writer↔spec @@ -1596,6 +1598,22 @@ product is never sharded, §8/#247): `"per-cell"` shape's fold law is the grammar's join over a cell group rather than the field's own reducer, so it exists at native resolution only. + It is also the fixture set's only store with a **root `coverage.moc`**, and + so §10's golden: the object was written by the production sweep writer (the + MOC family's leaf read plus its finisher) and carries the + `zagg-coverage-toc/1` section — the shard's tier-1 envelope word and the + tier-2 root time-digest in the native ragged `(k, 2)` + word-sibling form. + `temporal.expected.json`'s `root_coverage` block records the tier-1 word + **derived from the generator's inputs** (the join over every per-centroid + word it fed the writer, so the writer is pinned rather than self-certified), + the decoded digest rows read back — the same exception `column/`'s group + values are — and `obs_total`, the cell plan's own observation count, which + §10.3's weight rule says the digest's total weight MUST equal. The other six + fixtures have **no root coverage object at all**: none of them declares a + temporal field, so a sweep of one produces no section, and their committed + trees are byte-identical to their pre-§10 selves — which is exactly §10's + absence rule, pinned as bytes. + `minimal/` and `kitchen_sink/` pin the layout edge cases a reader must handle (`column/`'s leaf is `minimal/`'s, so it pins them again): inner chunk ordinal 2 is **empty** (absent from the shard index — the §1.5 sentinel, and @@ -2166,3 +2184,256 @@ an undeclared one is legal **and** the result MAY carry the declaration: the words mean the same thing on both sides. A reader that cannot establish that — because §2.2 does not apply to the array in hand — MUST drop the declaration rather than assert it. + +--- + +## 10. `zagg-coverage-toc/1` + +**Status: contract** ([issue #480](https://github.com/englacial/zagg/issues/480)). + +The **temporal section of the store-root coverage sidecar** — the one object a +reader GETs to bootstrap discovery. Its carrier is the `morton-moc/1` root +envelope, whose body (`{spec, encoding: "ranges", order, source, +generated_at, ranges}`) is described in +[`hive_layout.md`](hive_layout.md#coverage) and whose word and decimal +grammars are mortie's; this page owns exactly one addition to it — a +**`temporal` key** — so that a spatiotemporal candidate query resolves from +metadata alone, before any leaf is opened. + +Two tiers, both derived from the §8.3 `"per-centroid"` companions the leaves +already carry: + +| tier | key | what | answers | +|---|---|---|---| +| 1 | `shards` | one toc word per populated shard — the join over that shard's sibling words | *which* shards hold data during a window | +| 2 | `digest` | a weighted t-digest over acquisition times — mass placed at the per-centroid toc envelope midpoints that are also its companion | *how much* data falls in a window | + +Neither tier is new information and neither is truth: like the spatial ranges +beside them they are a **regenerable accelerator** over the leaf arrays (§8.3, +D9), written at end of walk while leaves stamp continuously, so a reader MUST +treat them under the same staleness posture as the `ranges` — a shard the +section does not list is not proof the shard has no data in the window. + +**Absence is the whole-section rule.** A store with no temporal channel +carries no `temporal` key, and its root object is byte-identical to one +written before this revision. A reader MUST read that absence as "this store +publishes no temporal coverage" and MUST NOT refuse the store, the sidecar, or +a windowed query because of it — the standing absence posture of §8/§9, +restated here with its force. Absence of the `digest` sub-block alone says the +same thing one tier down: tier 1 stands without it. + +**Versioned key discipline.** The section carries its own `spec` marker, +independent of the carrier's. A reader MUST strict-check it. Unlike the +array-level sections of this page, an unknown revision here reads as +**absent**, not as a hard failure: the section is an accelerator whose truth +is elsewhere, and this matches the strict-gate-then-degrade rule the sibling +coverage envelopes already use (the leaf stamp's `coverage` payload, +`hive_layout.md`). A future revision is therefore a new `spec` string and a +new section here; keys are never repurposed in place. + +### 10.1 Section grammar + +**Contract.** + +```json +"temporal": { + "spec": "zagg-coverage-toc/1", + "source": "sweep", + "generated_at": "2026-08-17T22:59:35+00:00", + "fields": ["h_tdigest"], + "shards": {"11213": "10689250968998768172"}, + "digest": { + "delta": 64, + "weights": "counts", + "value": "toc-ns", + "element": {"dtype": "float32", "shape": [-1, 2]}, + "encoding": "base64", + "centroids": 35, + "weight_total": 346.0, + "payload": "…", + "times": "…" + } +} +``` + +- **`spec`** (required) — `"zagg-coverage-toc/1"`, gated as above. +- **`source`** (required) — which producer wrote the section: `"sweep"` (the + sweep's leaf walk) or `"refresh"` (the explicit whole-store rebuild). The + vocabulary is open, exactly as the carrier's `source` is. +- **`generated_at`** (required) — ISO-8601 UTC. The section's own clock: it + and the carrier's may differ, because a producer with no temporal + contribution rewrites the ranges and leaves the section standing (§10.4). +- **`fields`** (required) — the sorted payload field names whose §8.3 + companions the **`shards` map** was derived from. It is the map's + provenance, and it composes as a **union** across producers (§10.4): after a + merge it names every field any contributing producer read, which is not + necessarily the set the installed `digest` was built over. A reader MUST + therefore treat it as an upper bound when applying the once-per-field weight + rule of §10.3, not as a per-digest field list. Informative for tier 1 (the + words are already unioned across fields by construction). +- **`shards`** (required) — tier 1, below. +- **`digest`** (optional) — tier 2, below. + +A reader MUST ignore keys it does not recognize; a producer MUST NOT put +anything under `temporal` that is not defined here or by a later revision. + +### 10.2 Tier 1 — the per-shard envelope word map + +**Contract.** `shards` maps a **shard id** to one **toc word**: + +- keys are D1 decimal shard ids at the carrier's `order`, spelled exactly as + the `ranges` endpoints are; +- values are toc words as **decimal strings**, for the same reason the range + endpoints are strings: a `uint64` word exceeds 2^53 and a float-based JSON + parser would silently mangle a raw number; +- the word is the grammar's join (mortie `toc_reduce`) over **every** §8.3 + companion word the shard's leaves hold, **unioned across all of the store's + temporal-carrying fields** — coverage means "any data", not "data in this + field". Where a shard is split into window leaves (`morton-hive/2`), the one + word is the join over all of them: the map is shard-keyed, never + window-keyed. + +The word therefore **conservatively contains** every observation instant in +that shard, exactly as §8.2's per-cell envelope does one level down, and the +grammar's own predicates apply unchanged: `toc_overlaps` never under-reports +(every shard whose real content intersects the window tests true, possibly +with edge over-report of up to one quantum) and `toc_contains` never +over-reports. A reader MUST use those predicates on the words rather than +decoding to bounds and comparing itself. + +The map lists only shards a producer actually walked. Its key set therefore +need not equal the carrier's `ranges` — a shard listed by `ranges` but absent +from `shards` is one whose temporal contribution has not been rolled up yet +(or one carrying no companion at all), and MUST be treated as *unknown*, i.e. +a candidate, never as *empty*. + +That escape hatch is the *only* one: containment is a claim about a **listed** +shard, so a producer whose read of any input behind a shard's word FAILED — +one window leaf of several, one unreadable companion — MUST omit that shard +from `shards` rather than publish the word it managed to join over the rest. +Dropping the shard costs a reader one candidate it must open; publishing a +partial word costs it data it will never look for. A field a leaf simply does +not carry (one added to the store after that leaf was written) is absence, not +failure: it holds no observations, so the word over the remaining fields is +still whole. + +> **Open question, flagged not decided** ([#480](https://github.com/englacial/zagg/issues/480)): +> whether a multi-field store should additionally publish **per-field** maps +> instead of (or beside) this union. The union is the default because coverage +> is a "is there any data here" question; a store whose fields have genuinely +> different temporal extents would be better served by per-field maps. If that +> case earns it, it arrives as an additional key under `temporal` (e.g. +> `shards_by_field`) in a later revision — `shards` keeps this meaning +> unchanged. + +### 10.3 Tier 2 — the root time-digest + +**Contract, optional.** The `digest` block is a t-digest over acquisition +times, carried in the store's **native** forms so a reader needs no grammar it +does not already implement for the leaves: + +- **`payload`** is base64 of a §2.1 centroid array's bytes — the `(k, 2)` + little-endian C-order `float32` buffer of §1.4, exactly what one ragged + element holds — declared by `element` and `encoding` in the block. Rows MUST + be sorted ascending by mean, as §2.1 requires. +- **`times`** is base64 of the row-aligned §8.3 companion: `k` little-endian + `uint64` toc words, one per centroid, carrying the same claim §8.3 gives + them (a single-observation centroid an exact timestamp, a merged one the + `toc_merge` join over its members). `centroids` records `k`; a reader MUST + refuse a block whose two buffers disagree on it (§1.1's row alignment, + broken). +- **Column 0 is an instant on §8's internal nanosecond scale** (`value: + "toc-ns"`), directly comparable with `toc2time` output and needing no unit + conversion. It is **derived from the companion words, not measured from the + observations**: each contributing centroid enters the fold at the MIDPOINT + of its own §8.3 word's `toc2time` envelope, and a merged centroid's mean is + the weight-weighted mean of those midpoints. Two consequences a reader MUST + plan for: + - a **weight-1 centroid is exact**. Its word is a timestamp under §8.3's + kind-keyed semantics, `toc2time` returns `(t, t)`, and the midpoint is + that instant. This is the one exact arm of the value axis. + - every other mean is a **convex combination of envelope midpoints**, so it + lies inside that centroid's own word but at no particular observation. + The partition is the one the **value** distribution produced (zagg re-keys + each §8.3 companion's existing digest onto its words), so a centroid's + members are grouped by payload value, not by time: a heavy centroid whose + members straddle a campaign gap places all of its mass at a point inside + that gap, where the store may hold no data at all. + + Column 0 is also `float32`, carrying ~2^-24 **relative** precision — near + present-day magnitudes a quantum of roughly ten minutes, enough that both + statements above hold only up to that rounding. It is the smaller half of + the same approximation, and deliberate: the **companion word beside each + centroid is the exact temporal claim**, and a reader needing exactness MUST + use the words, never the means. +- **Column 1 is a weight under the §2.0 `"counts"` declaration** (`weights`, + restated in the block): observation counts, so `sum(weights)` — recorded as + `weight_total` — is the total number of temporal observations the listed + fields contributed, under §2.1's float32 representability bound. Where + `fields` names more than one field, an observation that contributes to + several of them is counted **once per field**; this is the same open + question §10.2 flags, seen from the weight side. `fields` bounds that set + from above rather than naming it exactly (§10.1): a digest installed by one + producer sits beside a field list unioned over all of them. +- **`delta`** records the compression budget the fold used (64 in zagg's + writer). It is provenance, not a promise about `k`. + +The digest MUST be produced by ONE k-way merge over its contributors +(zagg: `zagg.stats.tdigest.merge_tdigests_kway` with the `temporal` channel), +so that it is permutation-independent in the contributors' order and its +companion words describe **the centroid partition that merge produced** — the +§8.3 exactness-given-the-partition rule, which is why the payload and its +companion MUST come from one call and MUST NOT be folded in separate passes. + +Density over a window is then the existing algebra: a CDF difference over the +payload, with the companion words available to bound (and, near a window edge, +to correct) which centroids may legitimately contribute. Total weight is +**exact** — the fold conserves it, so `weight_total` is the observation count +however coarse the value axis is — while the placement of that weight along +the axis is only as time-resolved as the centroid partition above. + +Gaps between campaign clusters stay visible in the **envelope words**: `k` +centroids carry `k` words, and a gap between two clusters shows as the +absence of any word covering it. That is a claim about the `times` buffer, +not about column 0 — the means can and do land inside a gap when a single +centroid's members straddle one. A reader answering "is there data in this +window at all" MUST read the words; the CDF answers "roughly how much", +resolved to the partition, and nothing finer. + +### 10.4 Composition + +**Contract.** The root object is written GET-union-PUT (incremental runs +accumulate; concurrent runs race benignly), and the section composes across +that seam as follows: + +- **Tier 1 unions elementwise** under `toc_merge`: a shard on both sides + merges to the join of its two words, a shard on one side carries over + unchanged. The join is idempotent, so re-walking unchanged leaves reproduces + the identical map. +- **Tier 2 is never unioned.** Its weights are counts, and merging two digests + over overlapping shard sets would double-count them. It is **replaced**, and + only by a producer whose own map covered every shard the merged map lists; + a producer that covered only part of the store publishes no digest and + leaves the standing one alone, and a merge that can find no whole-covering + digest on either side drops the block rather than publish a partial one. +- **A producer with no temporal contribution at all leaves an existing section + untouched** — it is not evidence of absence, only of a walk that did not + look. Conversely a producer that overwrites the carrier wholesale (an + unparsable or incompatible existing root — the D9 regenerable-cache rule) + discards the stale section with it. +- **An existing section at an unknown revision is preserved verbatim, and + never downgraded.** The strict `spec` gate above makes such a section read + as absent, but composition is a *write*, and dropping the key is not the + same as ignoring it: a producer MUST copy an unreadable standing section + through unchanged, both when it has no section of its own and when the + section it holds is at an older revision. This is the page's standing + succession rule (readers add revisions, they never drop them) applied at + the one seam a mixed-version fleet actually meets — without it the oldest + producer in the fleet wins. An unmarked carrier (a `temporal` value with no + `spec` string at all) claims no revision and is debris a producer MAY + replace. + +Conformance for an external reader is §7's `temporal/` fixture: its root +`coverage.moc` carries this section, and the fixture's `temporal.expected.json` +records the shard word and the decoded digest so the containment and weight +claims above are pinned on committed bytes. diff --git a/src/zagg/coverage.py b/src/zagg/coverage.py index 3a2cc84e6..be42b49a1 100644 --- a/src/zagg/coverage.py +++ b/src/zagg/coverage.py @@ -255,7 +255,13 @@ def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: skipped when it carries one (never classified by position) — as is a ``{stem}.pyramid.zarr`` column (issue #383), the one derived family that lives at the leaf's OWN node. A - supersedes it). A successful refresh also re-arms the + supersedes it). A temporal-declaring store (spec §10, issue #480) also has + its ``zagg-coverage-toc/1`` section rebuilt from this same walk — + fail-open per SHARD, so an unreadable companion costs the section that + shard and never the refresh; and a walk that lost any shard COMPOSES its + rebuild with the standing section (§10.4) instead of replacing it, so the + escape hatch can never be the thing that deletes the section. A successful + refresh also re-arms the :func:`warn_if_stale` once-per-episode latch for this store. Returns the envelope written, or ``None`` — deleting any existing root object — when no stamped leaf exists (absence is truthful, a stale cache is not, and @@ -269,6 +275,12 @@ def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: import obstore from obstore.exceptions import NotFoundError + from zagg.coverage_toc import ( + build_temporal_section, + read_leaf_temporal, + temporal_cell_order, + temporal_fields, + ) from zagg.grids.morton import morton_words_from_decimals from zagg.store import open_store @@ -276,6 +288,20 @@ def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: if manifest is None: raise ValueError(f"no {MANIFEST_NAME} at {store_root} — not a hive store root") order = int(manifest["shard_order"]) + # The §10 temporal section (issue #480) is rebuilt from the SAME walk, so + # the escape hatch regenerates it rather than deleting it — and, because + # this walk is whole-store by construction, its root time-digest is the + # authoritative one (spec §10's whole-coverage rule). + toc_fields = temporal_fields(manifest) + cell_order = temporal_cell_order(manifest) + if toc_fields and cell_order is None: + logger.warning( + f"refresh: {store_root} declares temporal fields but carries no cell_order — " + f"rebuilding no §10 section rather than guessing a group" + ) + toc_fields = {} + contributions: dict[str, list] = {} + toc_failed: set[str] = set() store = open_object_store(store_root, **store_kwargs) root = store_root.rstrip("/") # Decimals accumulate through the walk and parse once at the end (issue @@ -352,6 +378,26 @@ def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: ) continue decimals.append(decimal) + if toc_fields and decimal not in toc_failed: + try: + got = read_leaf_temporal( + f"{root}/{rel}", cell_order, toc_fields, **store_kwargs + ) + except Exception as e: # fail-open: the section is a cache + # Shard-scoped, not leaf-scoped: §10.2's word must + # contain EVERY instant in a listed shard, which a + # word joined over the window leaves that happened to + # read cannot promise. Drop the shard — absent means + # "unknown", which stays a candidate. + logger.warning( + f"refresh: dropping shard {decimal} from the temporal section " + f"— leaf {rel} did not read ({e})" + ) + toc_failed.add(decimal) + contributions.pop(decimal, None) + got = None + if got is not None: + contributions.setdefault(decimal, []).append(got) # D15: windowed stamps carry the leaf's actual time range; # the rebuilt root summary re-derives the union from this # walk's stamps (truth), superseding any cached value. @@ -369,11 +415,32 @@ def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: except (FileNotFoundError, NotFoundError): pass return None + section = build_temporal_section(contributions, toc_fields, source="refresh") + if toc_failed: + # Fail-open per leaf is fail-DESTRUCTIVE in aggregate. This walk PUTs + # its envelope outright (no union, by design), so a section rebuilt + # from a partial read publishes the losses as fact, and an all-failed + # walk deletes the section entirely — at exactly the moment an + # operator reached for the escape hatch because something was already + # wrong. Compose with the standing section instead, through the same + # §10.4 seam the sweep writes across. + from zagg.coverage_toc import TEMPORAL_KEY, merge_temporal_sections + from zagg.hive import read_root_coverage + + standing = read_root_coverage(store_root, **store_kwargs) + section = merge_temporal_sections( + standing.get(TEMPORAL_KEY) if isinstance(standing, dict) else None, section + ) + logger.warning( + f"refresh: the temporal section is PARTIAL — {len(toc_failed)} shard(s) did " + f"not read; composing with the standing section rather than replacing it" + ) envelope = build_root_coverage( morton_words_from_decimals(decimals), order, source="refresh", time_range=union_time_range(*time_ranges), + temporal=section, ) obstore.put(store, ROOT_COVERAGE_NAME, json.dumps(envelope, indent=1).encode()) return envelope diff --git a/src/zagg/coverage_toc.py b/src/zagg/coverage_toc.py new file mode 100644 index 000000000..074ae593c --- /dev/null +++ b/src/zagg/coverage_toc.py @@ -0,0 +1,500 @@ +"""The root coverage sidecar's temporal section — ``zagg-coverage-toc/1`` (issue #480). + +Two tiers, both derived from the §8.3 per-centroid toc siblings a store +already carries, both riding the ONE object a reader already GETs to +bootstrap discovery (``{store_root}/coverage.moc``): + +1. **the per-shard envelope word map** — one toc word per populated shard + (``mortie.toc_reduce`` over that shard's sibling words, unioned across the + store's temporal-carrying fields). This is the spatiotemporal pruning + tier: "which shards hold data DURING my window" resolves from metadata, + before any leaf is opened; +2. **the optional root time-digest** — a t-digest whose weights are + observation counts and whose value axis is the §8.3 per-centroid toc + envelopes' MIDPOINTS (:func:`_centroid_times`), with those same envelopes + as its companion, in the store's NATIVE ragged ``(k, 2)`` + word-sibling + form (base64 of the §1.4 element bytes), so a reader needs no grammar it + does not already implement for the leaves. Its total weight is exact; the + placement of that weight is only as time-resolved as the leaf digest's own + centroid partition, which is a partition of VALUE (§10.3). + +Absence composes: a store with no temporal channel writes no section, and +every accessor here returns ``None`` for it. That is never a refusal — the +standing absence posture of the sidecar grammars. + +The writer rides :class:`zagg.sweep.MocFamily`, the walk that already visits +every leaf to roll the spatial coverage up to the root; this module owns the +per-leaf read, the fold, the section grammar, and the reader-side accessors +zagg's own tests need. The normative grammar is ``docs/specification.md`` +§10. +""" + +from __future__ import annotations + +import base64 +import logging + +import numpy as np + +logger = logging.getLogger(__name__) + +#: The versioned key discipline: the section carries its own spec marker, so +#: a reader gates on it exactly as it gates the enclosing ``morton-moc/1`` +#: envelope and refuses an unknown revision rather than guessing. +TEMPORAL_COVERAGE_SPEC = "zagg-coverage-toc/1" + +#: The section's key on the root coverage envelope. +TEMPORAL_KEY = "temporal" + +#: Compression budget for the root time-digest (the issue's δ≈64): coarse by +#: design — the digest answers "how MUCH data in this window", while the +#: per-centroid companion words carry the exact temporal claim. +ROOT_TOC_DELTA = 64 + +#: The §8.3 shape this section is derived from. A ``"per-cell"`` or +#: ``"coordinate"`` declaration is a different array grammar and contributes +#: nothing here (see the §10 open question). +PER_CENTROID = "per-centroid" + + +def temporal_fields(manifest: dict | None) -> dict[str, dict]: + """The store's temporal-carrying digest fields, from the manifest declaration. + + Keyed by payload field name, valued with the fold metadata plus the + resolved ``sibling`` name. Discovery is declaration-driven — the same + ``pyramid.overview.fields`` block :func:`zagg.sweep_overview.field_companions` + reads — never a member enumeration of the leaf (a foreign or orphaned + array prefix must not be able to steer this), and never a naming + convention reconstructed by the reader. A store that declares no + ``temporal`` field returns ``{}``, which is what makes the whole section + absent for it. + """ + from zagg.grids.base import ragged_times_name + + if not isinstance(manifest, dict): + return {} + decl = (manifest.get("pyramid") or {}).get("overview") + fields = (decl or {}).get("fields") if isinstance(decl, dict) else None + out: dict[str, dict] = {} + for name, meta in (fields or {}).items(): + if not isinstance(meta, dict) or meta.get(TEMPORAL_KEY) != PER_CENTROID: + continue + out[name] = {**meta, "sibling": ragged_times_name(name)} + return out + + +def temporal_cell_order(manifest: dict | None) -> int | None: + """The manifest's ``cell_order`` — the leaf group the arrays live under. + + ``None`` when the key is absent or unparsable, and a caller MUST read that + as "this store publishes no temporal coverage" rather than substitute a + default. ``cell_order`` is a required manifest key, and a silent ``0`` + either asks for a group that does not exist — surfacing as an ordinary + "no temporal contribution" log line, which hides the manifest problem — + or, on a store whose cell order really IS 0, reads the wrong group and + publishes words for it. + """ + if not isinstance(manifest, dict): + return None + try: + return int(manifest["cell_order"]) + except (KeyError, TypeError, ValueError): + return None + + +def _centroid_times(words: np.ndarray) -> np.ndarray: + """Representative instants (internal ns, float64) for toc words. + + ``mortie.toc2time`` decodes a timestamp to ``(t, t)`` and a range to its + conservative ``[start, end)`` envelope, so the midpoint is the instant + itself for a timestamp and the envelope's centre for a range. The + digest's value axis is approximate by construction (§10): the word + beside each centroid stays the exact claim. + """ + from mortie import toc2time + + start, end = toc2time(np.asarray(words, dtype=np.uint64)) + return (np.asarray(start, dtype=np.float64) + np.asarray(end, dtype=np.float64)) / 2.0 + + +def read_leaf_temporal(leaf_root: str, cell_order: int, fields: dict, **store_kwargs): + """One leaf's contribution: ``(envelope_word, digest, times)`` or ``None``. + + Reads each declared field's payload and its §8.3 sibling by NAME (never a + member enumeration), row-aligned per §1.1. The envelope word is + ``toc_reduce`` over every sibling word the leaf holds, unioned across the + declared fields — coverage as "any data", the issue's proposed default. + The digest is one flat k-way merge over the leaf's per-cell time digests, + each built from that cell's centroid instants (:func:`_centroid_times`) + weighted by the payload's own centroid weights, so its total weight is + the leaf's temporal observation count. ``None`` when the leaf holds no + temporal row at all (an unpopulated or pre-companion leaf), which is + absence, not failure. + + **Cost.** One leaf-sized read per declared field — ``payload[:]`` plus its + sibling — and a per-cell decode before the merge, the same shape as the + overview family's own leaf read (:func:`zagg.sweep_overview._fold_node`, + which reads ``arr[:]`` per field and accumulates per-cell centroid lists + the same way). What it returns is bounded regardless: the k-way merge here + compresses the whole leaf to ~``ROOT_TOC_DELTA`` centroids, so the caller + accumulating leaves (:func:`build_temporal_section`) holds + ``n_leaves × ~δ`` rows, not ``n_leaves × n_cells × k``. For a + 2,726-shard store that is ~1.4 MB of centroids plus ~1.4 MB of companion + words, which is why the root fold needs no chunk batching of its own. + """ + import zarr + from mortie import toc_reduce + + from zagg.stats.tdigest import merge_tdigests_kway + from zagg.store import open_store + from zagg.sweep_overview import decode_digest + + group = zarr.open_group( + open_store(leaf_root, **store_kwargs), path=str(cell_order), mode="r", zarr_format=3 + ) + all_words: list[np.ndarray] = [] + digests: list[np.ndarray] = [] + times: list[np.ndarray] = [] + for name in sorted(fields): + meta = fields[name] + try: + sibling, payload = group[meta["sibling"]], group[name] + except KeyError: + # Schema evolution: the field postdates this leaf. It contributes + # nothing, exactly as the pyramid fold treats the same gap. + logger.debug(f"coverage[toc]: leaf {leaf_root} lacks field {name!r}") + continue + dtype = meta.get("dtype") or "float32" + raw_words, raw_payload = sibling[:], payload[:] + if len(raw_words) != len(raw_payload): + # A SHORT companion aligns row for row over its own length, so the + # per-cell check below never fires: the leaf would publish a word + # joined over a PREFIX of its cells and be listed as complete. This + # is the truncated-array shape the read path refuses everywhere + # else (issue #452); refuse it here too, per shard (§10.2). + raise ValueError( + f"{meta['sibling']} has {len(raw_words)} rows for a " + f"{len(raw_payload)}-row {name} payload — the companion must be " + f"row-aligned with its digest (spec §1.1)" + ) + for i, row in enumerate(raw_words): + if row is None or not len(row): + continue + words = decode_digest(row, "uint64", ()) + cell = decode_digest(raw_payload[i], dtype, (2,)) + if len(cell) != len(words): + raise ValueError( + f"{meta['sibling']} cell {i} has {len(words)} words for a " + f"{len(cell)}-centroid payload — the companion must be row-aligned " + f"with its digest (spec §1.1)" + ) + all_words.append(words) + t = _centroid_times(words) + # Sort into value (= time) order: the stored rows are in the + # payload's own value order (§8.3), which is not time order, and a + # digest's rows MUST ascend by mean (§2.1). The word breaks ties so + # the per-cell array is a function of its contents, not of row order. + order = np.lexsort((words, t)) + arr = np.empty((len(t), 2), dtype=np.float32) + arr[:, 0] = t[order] + arr[:, 1] = cell[order, 1] + digests.append(arr) + times.append(np.asarray(words, dtype=np.uint64)[order]) + if not all_words: + return None + word = int(toc_reduce(np.concatenate(all_words))) + digest, folded = merge_tdigests_kway(digests, delta=ROOT_TOC_DELTA, temporal=times) + return word, digest, folded + + +def build_temporal_section(contributions: dict, fields, *, source: str = "sweep") -> dict | None: + """The ``zagg-coverage-toc/1`` section from per-leaf contributions. + + ``contributions`` maps a shard's D1 decimal id to the LIST of + ``(word, digest, times)`` triples :func:`read_leaf_temporal` returned for + it — one per window leaf, so a windowed shard's several leaves reduce to + the one envelope word the shard-keyed map holds. Returns ``None`` for an + empty map: a store with no temporal channel gets no section, and its root + object stays byte-identical to a pre-#480 one. + + The root digest is ONE flat k-way merge over the per-leaf digests + (:func:`zagg.stats.tdigest.merge_tdigests_kway` with the ``temporal`` + channel), so it is permutation-independent in the leaf order and its + companion words are the envelopes of the centroid partition that merge + produced — the shipped law, not a second pass over it. + """ + from mortie import toc_reduce + + from zagg.hive import _utcnow + from zagg.stats.tdigest import merge_tdigests_kway + + if not contributions: + return None + shards: dict[str, int] = {} + digests, times = [], [] + for decimal in sorted(contributions): + parts = contributions[decimal] + shards[decimal] = int(toc_reduce(np.asarray([p[0] for p in parts], dtype=np.uint64))) + for _word, digest, folded in parts: + if len(digest): + digests.append(np.asarray(digest, dtype=np.float32)) + times.append(np.asarray(folded, dtype=np.uint64)) + section = { + "spec": TEMPORAL_COVERAGE_SPEC, + "source": source, + "generated_at": _utcnow(), + "fields": sorted(fields), + "shards": {d: str(w) for d, w in sorted(shards.items())}, + } + if digests: + payload, words = merge_tdigests_kway(digests, delta=ROOT_TOC_DELTA, temporal=times) + section["digest"] = _encode_digest_block(payload, words) + return section + + +def _encode_digest_block(payload: np.ndarray, words: np.ndarray) -> dict: + """The tier-2 block: the native ragged ``(k, 2)`` + word sibling, base64'd. + + The bytes are exactly what the same digest would occupy as a + ``zagg-ragged/1`` element and its §8.3 companion row — little-endian + C-order at the declared dtype (§1.4) — so a reader decodes them with the + leaf decoder it already has, base64 being the only wrapper a JSON + carrier forces. + """ + from zagg.sweep_overview import encode_digest + + payload = np.asarray(payload, dtype=np.float32) + words = np.asarray(words, dtype=np.uint64) + return { + "delta": ROOT_TOC_DELTA, + "weights": "counts", + "value": "toc-ns", + "element": {"dtype": "float32", "shape": [-1, 2]}, + "encoding": "base64", + "centroids": int(len(payload)), + "weight_total": float(payload[:, 1].sum()) if len(payload) else 0.0, + "payload": base64.b64encode(encode_digest(payload, "float32")).decode("ascii"), + "times": base64.b64encode(encode_digest(words, "uint64")).decode("ascii"), + } + + +def merge_temporal_sections(existing, incoming) -> dict | None: + """Compose two temporal sections for the root object's GET-union-PUT. + + Tier 1 unions elementwise under the grammar's ``toc_merge`` join — + idempotent and exact, so a re-sweep of unchanged leaves reproduces the + same words. Tier 2 is deliberately NOT unioned: its weights are + observation counts, and merging two digests over overlapping shard sets + would double-count them. It is REPLACED instead, and only by a producer + that covered every shard the merged map lists; a partial producer drops + it and leaves tier 1 standing (§10). + + ``fields`` unions with tier 1, because it is the provenance of the SHARD + MAP and the map is the thing that unions. It is deliberately not narrowed + to the surviving digest's own fields: doing so would describe the map with + a list that no longer covers it. §10.1 says so, and makes the list an + upper bound for the once-per-field weight rule rather than an exact + description of the installed digest. + + An unknown-spec section on the INCOMING side contributes nothing — the + same strict gate the enclosing envelope uses. On the EXISTING side it is + kept **verbatim** instead: this function is the write composer, and a + ``None`` return deletes the key. A section at a revision this zagg cannot + read was written by a producer that knew more than this one, so it is + neither dropped by a producer with no contribution nor downgraded by one + carrying a ``zagg-coverage-toc/1`` section — the page's standing rule that + readers add revisions and never drop them (§10.4). + """ + from mortie import toc_merge + + a, b = _usable(existing), _usable(incoming) + if b is None: + return dict(a) if a is not None else _preserved(existing) + if a is None: + newer = _preserved(existing) + if newer is not None: + logger.warning( + f"coverage[toc]: keeping the standing {newer.get('spec')!r} section — " + f"{TEMPORAL_COVERAGE_SPEC} does not read it and MUST NOT downgrade it" + ) + return newer + return dict(b) + shards = {d: int(w) for d, w in (a.get("shards") or {}).items()} + for d, w in (b.get("shards") or {}).items(): + prior = shards.get(d) + shards[d] = int(w) if prior is None else int(toc_merge(prior, int(w))) + merged = { + "spec": TEMPORAL_COVERAGE_SPEC, + "source": b.get("source", a.get("source")), + "generated_at": b.get("generated_at", a.get("generated_at")), + "fields": sorted(set(a.get("fields") or []) | set(b.get("fields") or [])), + "shards": {d: str(w) for d, w in sorted(shards.items())}, + } + # Whole-coverage test, newest producer first: only a section whose own map + # listed every shard in the union can vouch for a store-wide digest. + listed = set(merged["shards"]) + for side in (b, a): + if side.get("digest") is not None and set(side.get("shards") or {}) >= listed: + merged["digest"] = side["digest"] + break + return merged + + +def section_unchanged(existing, incoming) -> bool: + """Whether composing ``incoming`` into ``existing`` would change nothing. + + The temporal half of the MOC family's skip-if-current test: an unchanged + re-sweep of a temporal store must write no root object either. The test is + on what would actually be **written** — :func:`merge_temporal_sections`'s + own output — compared against the standing section on CONTENT (the shard + words, the digest block, the field list), never on the whole section: + ``source`` and ``generated_at`` churn per pass by construction. + + Testing the merge rather than the inputs is what makes it converge. A + producer that walked only part of the store always builds a digest, and + §10.4 always drops that digest at the seam; a test asking "does the + standing section already carry everything this one holds" therefore + answers *no* forever on any store with more than one shard, and every + incremental sweep re-PUTs a byte-identical object. + """ + merged = merge_temporal_sections(existing, incoming) + return _content(merged) == _content(existing if isinstance(existing, dict) else None) + + +def _content(section) -> tuple | None: + """The part of a section a no-op pass must reproduce exactly.""" + if section is None: + return None + return ( + {d: str(w) for d, w in (section.get("shards") or {}).items()}, + section.get("digest"), + sorted(section.get("fields") or []), + ) + + +def _usable(section) -> dict | None: + """A section dict at the spec revision this module implements, else ``None``.""" + if not isinstance(section, dict) or section.get("spec") != TEMPORAL_COVERAGE_SPEC: + if section is not None: + logger.debug("coverage[toc]: ignoring a section with an unknown spec") + return None + return section + + +def _preserved(section) -> dict | None: + """A standing section carrying a spec MARKER this revision does not implement. + + Distinguished from plain malformation: a marked section is some future + revision's, and §10.4 keeps it verbatim rather than clobber it. An + unmarked carrier (no ``spec``, or a non-string one) claims no revision, + so it is debris a producer may legitimately replace — otherwise one bad + write would wedge the key shut forever. + """ + if not isinstance(section, dict): + return None + spec = section.get("spec") + if isinstance(spec, str) and spec and spec != TEMPORAL_COVERAGE_SPEC: + return dict(section) + return None + + +# --------------------------------------------------------------------------- +# Reader side — the minimum zagg's own tests (and demos) need. The external +# reader's `coverage_toc` / `when=` surface is espg/moczarr#45, decoded from +# the spec text and the §7 fixtures alone. +# --------------------------------------------------------------------------- + + +def load_temporal_coverage(envelope) -> dict | None: + """The temporal section of a root coverage envelope, or ``None``. + + Absence — no section, an unknown spec revision, a malformed carrier — all + read as ``None``. A store with no temporal channel is the common case and + is never an error. + """ + if not isinstance(envelope, dict): + return None + return _usable(envelope.get(TEMPORAL_KEY)) + + +def coverage_toc(envelope) -> dict[str, int] | None: + """The per-shard envelope word map, ``{shard decimal: toc word}``. + + ``None`` when the store carries no temporal section. Words come back as + Python ints (the JSON carries decimal STRINGS — a uint64 exceeds 2^53 and + a float-based parser would mangle a raw number, the same rule the spatial + ranges follow). + """ + section = load_temporal_coverage(envelope) + if section is None: + return None + return {d: int(w) for d, w in (section.get("shards") or {}).items()} + + +def coverage_toc_digest(envelope): + """The root time-digest as ``(payload, words)``, or ``None``. + + ``payload`` is the §2.1 ``(k, 2)`` float32 centroid array whose value + column is an instant on the §8 internal-ns scale and whose weight column + is an observation count; ``words`` is its row-aligned §8.3 companion. + + §10.3's MUST-check is on all THREE: the two buffers against each other and + both against the block's declared ``centroids``. A ``k`` that agrees with + neither is a broken block, not a decorative field — and a reference + accessor that skipped the check would leave the external reader + (moczarr) implementing one zagg does not. + """ + section = load_temporal_coverage(envelope) + block = (section or {}).get("digest") + if not isinstance(block, dict): + return None + from zagg.sweep_overview import decode_digest + + payload = decode_digest(base64.b64decode(block["payload"]), "float32", (2,)) + words = decode_digest(base64.b64decode(block["times"]), "uint64", ()) + if len(payload) != len(words) or block.get("centroids") != len(payload): + raise ValueError( + f"root time-digest declares {block.get('centroids')!r} centroids and decodes " + f"{len(payload)} of them with {len(words)} companion words — the companion " + f"must be row-aligned with its digest at the declared k (spec §1.1, §10.3)" + ) + return payload, words + + +def shards_overlapping(envelope, q_start_ns: int, q_end_ns: int) -> list[str] | None: + """Shard ids whose envelope word intersects ``[q_start_ns, q_end_ns)``. + + The tier-1 pruning answer, on the grammar's own predicate + (``mortie.toc_overlaps``): conservative — it never under-reports, and may + over-report by up to one quantum at a window edge. ``None`` when the + store carries no temporal section, which a caller MUST read as "no + temporal information", never as "no shards". + """ + from mortie import toc_overlaps + + words = coverage_toc(envelope) + if words is None: + return None + keys = sorted(words) + if not keys: + return [] + hit = toc_overlaps(np.asarray([words[k] for k in keys], dtype=np.uint64), q_start_ns, q_end_ns) + return [k for k, ok in zip(keys, np.atleast_1d(hit), strict=True) if bool(ok)] + + +__all__ = [ + "PER_CENTROID", + "ROOT_TOC_DELTA", + "TEMPORAL_COVERAGE_SPEC", + "TEMPORAL_KEY", + "build_temporal_section", + "coverage_toc", + "coverage_toc_digest", + "load_temporal_coverage", + "merge_temporal_sections", + "read_leaf_temporal", + "section_unchanged", + "shards_overlapping", + "temporal_cell_order", + "temporal_fields", +] diff --git a/src/zagg/hive.py b/src/zagg/hive.py index 2d26dde29..9dc36c5a9 100644 --- a/src/zagg/hive.py +++ b/src/zagg/hive.py @@ -953,7 +953,12 @@ def _rank_tail(rank: int, depth: int) -> str: def build_root_coverage( - shard_keys, order: int, *, source: str = "dispatcher", time_range: tuple | list | None = None + shard_keys, + order: int, + *, + source: str = "dispatcher", + time_range: tuple | list | None = None, + temporal: dict | None = None, ) -> dict: """Store-root coverage envelope from completed shard keys (issue #200 phase 3). @@ -973,6 +978,12 @@ def build_root_coverage( CACHE, never truth (the per-leaf stamps are the truth; the walk and the sweep regenerate this). Omitted for unwindowed stores, keeping their root object byte-identical to pre-#246 runs. + + ``temporal`` (issue #480): the spec §10 ``zagg-coverage-toc/1`` section — + the per-shard toc envelope word map plus the optional root time-digest, + built by :func:`zagg.coverage_toc.build_temporal_section`. ``None`` for a + store with no temporal channel, which keeps ITS root object byte-identical + to a pre-#480 one; absence of the section is never a refusal. """ from zagg.grids.morton import to_morton_array @@ -1008,6 +1019,10 @@ def build_root_coverage( } if time_range is not None: envelope["time_range"] = [str(t) for t in time_range] + if temporal is not None: + from zagg.coverage_toc import TEMPORAL_KEY + + envelope[TEMPORAL_KEY] = temporal return envelope @@ -1056,6 +1071,13 @@ def write_root_coverage(store_root: str, envelope: dict, **store_kwargs) -> dict re-unions — accepted under D9/O7 (a missing listing degrades to "reader doesn't see the newest run", never a wrong answer; do NOT add a lock). Returns the payload actually written. + + The spec §10 temporal section (issue #480) composes across the same seam + (:func:`zagg.coverage_toc.merge_temporal_sections`): its per-shard word + map unions elementwise under the grammar's join, its digest is replaced + rather than unioned (weights are counts), and a producer carrying no + section leaves an existing one standing. Two stores with no temporal + channel at all still write byte-identical bytes to a pre-#480 zagg. """ import obstore @@ -1098,6 +1120,17 @@ def write_root_coverage(store_root: str, envelope: dict, **store_kwargs) -> dict f"existing {ROOT_COVERAGE_NAME} at {store_root} has an incompatible " f"envelope; overwriting (regenerable cache)" ) + from zagg.coverage_toc import TEMPORAL_KEY, merge_temporal_sections + + # A rebuilt `merged` dropped the existing carrier's extra keys with it; + # an OVERWRITE (`merged is envelope`) deliberately discards the stale + # section too, exactly as it discards the stale ranges. + carried = isinstance(existing, dict) and merged is not envelope + section = merge_temporal_sections( + existing.get(TEMPORAL_KEY) if carried else None, envelope.get(TEMPORAL_KEY) + ) + if section is not None: + merged[TEMPORAL_KEY] = section obstore.put(store, ROOT_COVERAGE_NAME, json.dumps(merged, indent=1).encode()) return merged diff --git a/src/zagg/sweep.py b/src/zagg/sweep.py index 7c49aa2aa..f827fb9c7 100644 --- a/src/zagg/sweep.py +++ b/src/zagg/sweep.py @@ -171,10 +171,81 @@ class MocFamily(SweepFamily): already covers the folded words and time range (sweep idempotence). The O8 in-leaf bitmap contract is untouched: this family reads only the stamp envelope, never the cell-order bitmap sidecar. + + The spec §10 TEMPORAL section (issue #480) rides this same walk: every + stamped leaf of a temporal-declaring store also yields its §8.3 toc + envelope word and a per-leaf time digest + (:func:`zagg.coverage_toc.read_leaf_temporal`), accumulated on the family + INSTANCE — one per run, since :func:`get_family` constructs a fresh one — + and folded into the section :meth:`finish` writes. It stays OUT of the + per-node rollup payloads on purpose: those are the skip-if-current + currency, compared byte for byte, and a per-node k-way fold would + describe a different centroid partition at every node (the fold-tree + caveat on :func:`zagg.stats.tdigest.merge_tdigests_kway`). A store + declaring no temporal field accumulates nothing, and its root object is + byte-identical to a pre-#480 one. """ name = "moc" + def __init__(self): + #: ``{shard decimal: [(word, digest, times), ...]}`` — one entry per + #: window leaf this run visited (issue #480). + self._temporal: dict[str, list] = {} + #: Shards whose temporal read failed: dropped from the map entirely, + #: never published from the window leaves that did read (issue #480). + self._temporal_failed: set[str] = set() + #: Resolved once, on the first leaf read; ``None`` until then. + self._temporal_fields: dict | None = None + self._cell_order = 0 + + def _accumulate_temporal(self, store_root, decimal, leaf, store_kwargs) -> None: + """Read one leaf's §10 temporal contribution; fail-open per SHARD (D9). + + A store declaring no temporal field short-circuits after one manifest + read. An unreadable companion is logged and skipped rather than + failing the leaf: the temporal section is a regenerable accelerator, + and the spatial rollup this walk exists for must not die on it. + + The fail-open is shard-scoped, not leaf-scoped. A windowed shard has + several leaves behind ONE map entry, and §10.2 promises that a LISTED + shard's word conservatively contains every instant in it — a word + joined over whichever windows happened to read is not that. So one + failed window drops its whole shard from the map, permanently for this + run; a shard absent from ``shards`` reads as *unknown* (still a + candidate), which is always safe, while a shard listed with a partial + word is not. + """ + from zagg.coverage_toc import read_leaf_temporal, temporal_cell_order, temporal_fields + + if self._temporal_fields is None: + from zagg.hive import read_manifest + + manifest = read_manifest(store_root, **store_kwargs) or {} + self._temporal_fields = temporal_fields(manifest) + cell_order = temporal_cell_order(manifest) + if self._temporal_fields and cell_order is None: + logger.warning( + f"sweep[moc]: {store_root} declares temporal fields but carries no " + f"cell_order — publishing no §10 section rather than guessing a group" + ) + self._temporal_fields = {} + self._cell_order = cell_order or 0 + if not self._temporal_fields or decimal in self._temporal_failed: + return + try: + got = read_leaf_temporal(leaf, self._cell_order, self._temporal_fields, **store_kwargs) + except Exception as e: + logger.warning( + f"sweep[moc]: dropping shard {decimal} from the temporal section — " + f"leaf {leaf} did not read ({e})" + ) + self._temporal_failed.add(decimal) + self._temporal.pop(decimal, None) + return + if got is not None: + self._temporal.setdefault(decimal, []).append(got) + def read_leaf(self, store_root, decimal, window, spec, store_kwargs): # ``spec`` is unused here: leaf PATHS are the frozen /1-/2 grammar # (shard_leaf_path); the D23 /3 leaf naming has no writer yet, and @@ -187,6 +258,7 @@ def read_leaf(self, store_root, decimal, window, spec, store_kwargs): stamp = read_commit(open_store(leaf, **store_kwargs)) if stamp is None: return None # absent leaf or unstamped debris (D4) + self._accumulate_temporal(store_root, decimal, leaf, store_kwargs) payload = _moc_payload([morton_word(decimal)], stamp.get("time_range")) return payload, stamp.get("written_at") @@ -205,12 +277,17 @@ def finish(self, store_root, tops, shard_order, store_kwargs) -> dict: Unions with the existing root object (the sweep may cover only the dirty subtrees — untouched bases must keep their listing), via the same :func:`zagg.hive.write_root_coverage` transport the runner uses. - No PUT when the existing root already lists every folded word and - covers the folded time range, so an unchanged tree re-sweep is a - no-op here too. + No PUT when the existing root already lists every folded word, + covers the folded time range, and would be UNCHANGED by composing + this run's §10 temporal section into it + (:func:`zagg.coverage_toc.section_unchanged` — the test is on the + merge's output rather than on the raw section, because §10.4 drops a + partial producer's digest at the seam), so an unchanged tree re-sweep + is a no-op here too, on a multi-shard store as much as a one-shard one. """ import numpy as np + from zagg.coverage_toc import TEMPORAL_KEY, build_temporal_section, section_unchanged from zagg.hive import ( build_root_coverage, read_root_coverage, @@ -221,6 +298,9 @@ def finish(self, store_root, tops, shard_order, store_kwargs) -> dict: if not tops: return {"root_moc_written": False} + section = build_temporal_section( + self._temporal, self._temporal_fields or {}, source="sweep" + ) words = np.unique(np.concatenate([root_coverage_words(t["payload"]) for t in tops])) time_range = union_time_range(*(t["payload"].get("time_range") for t in tops)) try: @@ -234,16 +314,26 @@ def finish(self, store_root, tops, shard_order, store_kwargs) -> dict: union_time_range(existing.get("time_range"), time_range) == existing.get("time_range") ) + covered = covered and section_unchanged(existing.get(TEMPORAL_KEY), section) if covered: return {"root_moc_written": False} except (KeyError, TypeError, ValueError): pass # malformed cache cannot vouch for coverage -> rewrite write_root_coverage( store_root, - build_root_coverage(words, shard_order, source="sweep", time_range=time_range), + build_root_coverage( + words, shard_order, source="sweep", time_range=time_range, temporal=section + ), **store_kwargs, ) - return {"root_moc_written": True} + out = {"root_moc_written": True} + if section is not None: + # THIS pass's contribution, not the store's total. The section + # composes with the standing one at the seam (§10.4), so the map + # that landed lists at least these shards and usually more; the + # run summary reports what the run did. + out["temporal_shards"] = len(section["shards"]) + return out def _moc_payload(words, time_range) -> dict: diff --git a/tests/data/spec/temporal.expected.json b/tests/data/spec/temporal.expected.json index 0ea588dc2..f4e71a348 100644 --- a/tests/data/spec/temporal.expected.json +++ b/tests/data/spec/temporal.expected.json @@ -9,6 +9,199 @@ "chunks_per_shard": 4, "empty_chunk": 2, "delta": 16, + "root_coverage": { + "object": "coverage.moc", + "spec": "zagg-coverage-toc/1", + "fields": [ + "h_tdigest" + ], + "shards": { + "11213": "10689250968998768172" + }, + "obs_total": 346, + "digest": { + "delta": 64, + "centroids": [ + [ + 5.344625417305194e+18, + 1.0 + ], + [ + 5.344625417305194e+18, + 1.0 + ], + [ + 5.344625417305194e+18, + 1.0 + ], + [ + 5.344625417305194e+18, + 1.0 + ], + [ + 5.344625417305194e+18, + 1.0 + ], + [ + 5.344625417305194e+18, + 2.0 + ], + [ + 5.344625417305194e+18, + 2.0 + ], + [ + 5.344625417305194e+18, + 2.0 + ], + [ + 5.344625417305194e+18, + 2.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 3.0 + ], + [ + 5.344625417305194e+18, + 5.0 + ], + [ + 5.344625967061008e+18, + 4.0 + ], + [ + 5.344626516816822e+18, + 2.0 + ], + [ + 5.344626516816822e+18, + 7.0 + ], + [ + 5.344626516816822e+18, + 12.0 + ], + [ + 5.344626516816822e+18, + 17.0 + ], + [ + 5.344626516816822e+18, + 21.0 + ], + [ + 5.344626516816822e+18, + 25.0 + ], + [ + 5.344626516816822e+18, + 27.0 + ], + [ + 5.344626516816822e+18, + 29.0 + ], + [ + 5.344627066572636e+18, + 7.0 + ], + [ + 5.344627066572636e+18, + 10.0 + ], + [ + 5.344627066572636e+18, + 16.0 + ], + [ + 5.344627066572636e+18, + 20.0 + ], + [ + 5.344627066572636e+18, + 24.0 + ], + [ + 5.344627066572636e+18, + 26.0 + ], + [ + 5.344627066572636e+18, + 28.0 + ], + [ + 5.344627066572636e+18, + 29.0 + ] + ], + "times": [ + "10689250971274671232", + "10689250978569638528", + "10689251193943820928", + "10689251199091304576", + "10689251206386271872", + "10689250981883669650", + "10689250994768571540", + "10689251166567263400", + "10689251179452165289", + "10689251007653473430", + "10689251024833342616", + "10689251042013211802", + "10689251059193080988", + "10689251076372950174", + "10689251093552819360", + "10689251115027655842", + "10689251132207525029", + "10689251149387394328", + "10689252141524839707", + "10689252717050457436", + "10689252725640392033", + "10689252768590065002", + "10689252841604509045", + "10689252944683724164", + "10689253069237775766", + "10689253219561631144", + "10689253382770388413", + "10689254473692081708", + "10689254413562539560", + "10689254319073259041", + "10689254198814174741", + "10689254052785286663", + "10689253898166463991", + "10689253730662739428", + "10689253554569080273" + ] + } + }, "declarations": { "observed": { "spec": "zagg-toc/1", diff --git a/tests/data/spec/temporal/1/1/2/1/3/11213.zarr/zarr.json b/tests/data/spec/temporal/1/1/2/1/3/11213.zarr/zarr.json index cb4bb833f..920b6cb2f 100644 --- a/tests/data/spec/temporal/1/1/2/1/3/11213.zarr/zarr.json +++ b/tests/data/spec/temporal/1/1/2/1/3/11213.zarr/zarr.json @@ -5,7 +5,7 @@ "complete": true, "cells_with_data": 4, "granule_count": 1, - "written_at": "2026-08-17T15:01:20+00:00", + "written_at": "2026-08-18T07:47:02+00:00", "coverage": { "spec": "morton-moc/1", "box": [ diff --git a/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.stats.json b/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.stats.json index 7f04a25d1..92d383577 100644 --- a/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.stats.json +++ b/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.stats.json @@ -1 +1 @@ -{"schema_version": 1, "shard_key": 1233986297899515908, "window": null, "run_id": null, "semantic_hash": null, "zagg_version": "0.46.1.dev20+gd35e576c2", "n_shards": 1, "n_granules": 1, "granules_sha256": null, "content_hashes": {"arrays": {"4/count": "71f269d578d413efdf624d699f2d8430be995bd75dba859bd9a782e7c89c3118", "4/h_tdigest": "5d0ed5f48cd76f4ba11b8eea66e9b30b1a8c3aebc266164d95c6c8e9129d97a3", "4/h_tdigest_locations": "19a334339a7f39cb61b79065841fcdac2cb94423e872bcb571543cfafea215c1", "4/h_tdigest_times": "f4a1fa9471b1fbb9b20397074195e5df3248a083ecd1f59eae2b57f89e4cb65d", "4/morton": "92cedd52796c6c0844af1ce3e7e7636ed28804f6bf8f9d10ceab93edf1a29805", "5/count": "6380596d3ca826d3ccda1b07f368ca3b43b486b7aaa0ca17e055ac8bf0b2c064", "5/h_tdigest": "990da3f3e099e188748a022e7205d25b3b33114ccb7188245aa456a9c4133497", "5/h_tdigest_locations": "f0f2ef5c0dee840a58c1bde560d39ca265016df125ef01dae816c3a5e6898f40", "5/h_tdigest_times": "9594edbc3a78644568e4838474e739d48292b836a16ebbfc664e6d20adcc068e", "5/morton": "2d7a07b3e2db265b2050fedfd968fb6763f5b012c87033526852bffc90b059d2"}, "combined": "b6859b89aa7ee03ba856fc446c829aaeee06cdf269faf91bbeebd9ceddf6bf45"}, "n_obs": 0, "n_obs_read": null, "cells_with_data": 3, "phase_timings": {}, "duration_s": 0.0, "spill_bytes": null, "spill_blocks_closed": null, "raster_bytes_read": null, "raster_px_decoded": null, "raster_px_sampled": null, "leaf_column": null, "gb_seconds": null, "est_cost_usd": null, "max_memory_mb": null, "container_hwm_mb": null, "lambda": null, "timestamp": "2026-08-17T15:01:20+00:00", "success": true, "error": null, "invoked_by": null} \ No newline at end of file +{"schema_version": 1, "shard_key": 1233986297899515908, "window": null, "run_id": null, "semantic_hash": null, "zagg_version": "0.46.1.dev131+gd52e30631.d20260817", "n_shards": 1, "n_granules": 1, "granules_sha256": null, "content_hashes": {"arrays": {"4/count": "71f269d578d413efdf624d699f2d8430be995bd75dba859bd9a782e7c89c3118", "4/h_tdigest": "5d0ed5f48cd76f4ba11b8eea66e9b30b1a8c3aebc266164d95c6c8e9129d97a3", "4/h_tdigest_locations": "19a334339a7f39cb61b79065841fcdac2cb94423e872bcb571543cfafea215c1", "4/h_tdigest_times": "f4a1fa9471b1fbb9b20397074195e5df3248a083ecd1f59eae2b57f89e4cb65d", "4/morton": "92cedd52796c6c0844af1ce3e7e7636ed28804f6bf8f9d10ceab93edf1a29805", "5/count": "6380596d3ca826d3ccda1b07f368ca3b43b486b7aaa0ca17e055ac8bf0b2c064", "5/h_tdigest": "990da3f3e099e188748a022e7205d25b3b33114ccb7188245aa456a9c4133497", "5/h_tdigest_locations": "f0f2ef5c0dee840a58c1bde560d39ca265016df125ef01dae816c3a5e6898f40", "5/h_tdigest_times": "9594edbc3a78644568e4838474e739d48292b836a16ebbfc664e6d20adcc068e", "5/morton": "2d7a07b3e2db265b2050fedfd968fb6763f5b012c87033526852bffc90b059d2"}, "combined": "b6859b89aa7ee03ba856fc446c829aaeee06cdf269faf91bbeebd9ceddf6bf45"}, "n_obs": 0, "n_obs_read": null, "cells_with_data": 3, "phase_timings": {}, "duration_s": 0.0, "spill_bytes": null, "spill_blocks_closed": null, "raster_bytes_read": null, "raster_px_decoded": null, "raster_px_sampled": null, "leaf_column": null, "gb_seconds": null, "est_cost_usd": null, "max_memory_mb": null, "container_hwm_mb": null, "lambda": null, "timestamp": "2026-08-18T07:47:02+00:00", "success": true, "error": null, "invoked_by": null} \ No newline at end of file diff --git a/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.zarr/zarr.json b/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.zarr/zarr.json index d1b42aa60..e647b42f8 100644 --- a/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.zarr/zarr.json +++ b/tests/data/spec/temporal/1/1/2/1/3/all.pyramid.zarr/zarr.json @@ -37,14 +37,14 @@ } }, "cells_with_data_order": 5, - "generated_at": "2026-08-17T15:01:20+00:00" + "generated_at": "2026-08-18T07:47:02+00:00" }, "morton_hive_commit": { "spec": "morton-hive/1", "complete": true, "cells_with_data": 3, "granule_count": 1, - "written_at": "2026-08-17T15:01:20+00:00" + "written_at": "2026-08-18T07:47:02+00:00" } }, "zarr_format": 3, diff --git a/tests/data/spec/temporal/coverage.moc b/tests/data/spec/temporal/coverage.moc new file mode 100644 index 000000000..647bbce42 --- /dev/null +++ b/tests/data/spec/temporal/coverage.moc @@ -0,0 +1,41 @@ +{ + "spec": "morton-moc/1", + "encoding": "ranges", + "order": 4, + "source": "sweep", + "generated_at": "2026-08-18T07:47:02+00:00", + "ranges": [ + [ + "11213", + "11213" + ] + ], + "temporal": { + "spec": "zagg-coverage-toc/1", + "source": "sweep", + "generated_at": "2026-08-18T07:47:02+00:00", + "fields": [ + "h_tdigest" + ], + "shards": { + "11213": "10689250968998768172" + }, + "digest": { + "delta": 64, + "weights": "counts", + "value": "toc-ns", + "element": { + "dtype": "float32", + "shape": [ + -1, + 2 + ] + }, + "encoding": "base64", + "centroids": 35, + "weight_total": 346.0, + "payload": "2VeUXgAAgD/ZV5ReAACAP9lXlF4AAIA/2VeUXgAAgD/ZV5ReAACAP9lXlF4AAABA2VeUXgAAAEDZV5ReAAAAQNlXlF4AAABA2VeUXgAAQEDZV5ReAABAQNlXlF4AAEBA2VeUXgAAQEDZV5ReAABAQNlXlF4AAEBA2VeUXgAAQEDZV5ReAABAQNlXlF4AAKBA2leUXgAAgEDbV5ReAAAAQNtXlF4AAOBA21eUXgAAQEHbV5ReAACIQdtXlF4AAKhB21eUXgAAyEHbV5ReAADYQdtXlF4AAOhB3FeUXgAA4EDcV5ReAAAgQdxXlF4AAIBB3FeUXgAAoEHcV5ReAADAQdxXlF4AANBB3FeUXgAA4EHcV5ReAADoQQ==", + "times": "gHTT0R/ZV5SA0qOEIdlXlIAK8alT2VeUgGjB3FTZV5SAxpGPVtlXlJLsK0oi2VeUlOwrSiXZV5So7CtKTdlXlKnsK0pQ2VeUluwrSijZV5SY7CtKLNlXlJrsK0ow2VeUnOwrSjTZV5Se7CtKONlXlKDsK0o82VeUouwrSkHZV5Sl7CtKRdlXlBjtK0pJ2VeUG+0rSjDaV5Rc7StKttpXlGHtK0q42leUau0rSsLaV5R17StK09pXlITtK0rr2leUlu0rSgjbV5So7StKK9tXlL3tK0pR21eULO4rSk/cV5Qo7itKQdxXlCHuK0or3FeUFe4rSg/cV5QH7itK7dtXlPftK0rJ21eU5O0rSqLbV5TR7StKedtXlA==" + } + } +} \ No newline at end of file diff --git a/tests/data/spec/temporal/morton_hive.json b/tests/data/spec/temporal/morton_hive.json index af61e5fb9..a632f103d 100644 --- a/tests/data/spec/temporal/morton_hive.json +++ b/tests/data/spec/temporal/morton_hive.json @@ -78,5 +78,5 @@ } } }, - "generated_at": "2026-08-17T15:01:20+00:00" + "generated_at": "2026-08-18T07:47:02+00:00" } \ No newline at end of file diff --git a/tests/test_coverage_toc.py b/tests/test_coverage_toc.py new file mode 100644 index 000000000..ff11bb89c --- /dev/null +++ b/tests/test_coverage_toc.py @@ -0,0 +1,599 @@ +"""The root coverage sidecar's temporal section — spec §10, issue #480. + +Three things are asserted here that the §7 conformance suite cannot: the +order-independence of the root fold, the composition rules the GET-union-PUT +seam applies, and the byte-identity of a NON-temporal store's root object — +the promise that a store with no temporal channel is untouched by this +revision. The committed ``temporal/`` fixture is the real-store end of it; +``minimal/`` is the absence end. +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path + +import numpy as np +import pytest +from mortie import span2toc, time2toc, toc_merge, toc_overlaps, toc_reduce + +from zagg.coverage import refresh_root_coverage +from zagg.coverage_toc import ( + ROOT_TOC_DELTA, + TEMPORAL_COVERAGE_SPEC, + build_temporal_section, + coverage_toc, + coverage_toc_digest, + load_temporal_coverage, + merge_temporal_sections, + section_unchanged, + shards_overlapping, + temporal_fields, +) +from zagg.hive import build_root_coverage, read_root_coverage, write_root_coverage + +SPEC_DATA = Path(__file__).parent / "data" / "spec" +#: A day on the toc scale, in internal ns — enough to keep the synthetic +#: leaves below in visibly distinct campaign clusters. +DAY_NS = 86_400 * 10**9 +#: An arbitrary but realistic base instant on the §8 internal-ns scale. +BASE_NS = 5_344_000_000_000_000_000 + + +def _leaf(seed: int, n: int = 12): + """A synthetic per-leaf contribution: ``(word, digest, times)``. + + Shaped exactly like :func:`zagg.coverage_toc.read_leaf_temporal`'s return + — a valid ``(k, 2)`` digest sorted by mean, its row-aligned toc words, and + the join over them. + """ + rng = np.random.default_rng(seed) + starts = np.sort(BASE_NS + seed * 40 * DAY_NS + rng.integers(0, 30 * DAY_NS, n)).astype( + np.uint64 + ) + # Both variants, deliberately: a single-instant centroid keeps its exact + # timestamp word, a spanning one gets a conservative range. + words = np.array( + [ + int(time2toc(int(t))) if i % 3 == 0 else int(span2toc(int(t), int(t) + 3600 * 10**9)) + for i, t in enumerate(starts) + ], + dtype=np.uint64, + ) + digest = np.empty((n, 2), dtype=np.float32) + digest[:, 0] = starts.astype(np.float64) + digest[:, 1] = rng.integers(1, 20, n).astype(np.float64) + order = np.lexsort((words, digest[:, 0])) + return int(toc_reduce(words)), digest[order], words[order] + + +def _contributions(seeds): + return {f"1121{i}": [_leaf(s)] for i, s in enumerate(seeds)} + + +class TestSectionGrammar: + """§10.1 — what a built section carries, and what it refuses to carry.""" + + def test_required_keys_and_string_words(self): + section = build_temporal_section(_contributions([1, 2, 3]), ["h_tdigest"]) + assert section["spec"] == TEMPORAL_COVERAGE_SPEC + assert set(section) == {"spec", "source", "generated_at", "fields", "shards", "digest"} + assert section["fields"] == ["h_tdigest"] + assert all(isinstance(w, str) and w.isdigit() for w in section["shards"].values()) + assert section["digest"]["delta"] == ROOT_TOC_DELTA + assert section["digest"]["element"] == {"dtype": "float32", "shape": [-1, 2]} + + def test_an_empty_walk_builds_no_section(self): + assert build_temporal_section({}, []) is None + assert build_temporal_section({}, ["h_tdigest"]) is None + + def test_several_window_leaves_reduce_to_one_shard_word(self): + a, b = _leaf(1), _leaf(2) + section = build_temporal_section({"11213": [a, b]}, ["h_tdigest"]) + assert set(section["shards"]) == {"11213"} + assert int(section["shards"]["11213"]) == int(toc_merge(a[0], b[0])) + + def test_weight_conservation(self): + contributions = _contributions([1, 2, 3]) + section = build_temporal_section(contributions, ["h_tdigest"]) + total = sum( + float(part[1][:, 1].sum()) for parts in contributions.values() for part in parts + ) + payload, _words = coverage_toc_digest({"temporal": section}) + assert float(payload[:, 1].sum()) == pytest.approx(total) + assert section["digest"]["weight_total"] == pytest.approx(total) + + def test_the_root_words_reduce_to_the_join_of_every_shard_word(self): + section = build_temporal_section(_contributions([1, 2, 3]), ["h_tdigest"]) + _payload, words = coverage_toc_digest({"temporal": section}) + assert int(toc_reduce(words)) == int( + toc_reduce(np.array([int(w) for w in section["shards"].values()], dtype=np.uint64)) + ) + + +class TestOrderIndependence: + """§10.3 — the fold is ONE k-way merge, so leaf order cannot matter.""" + + def test_permuting_the_leaves_reproduces_the_section(self): + contributions = _contributions([4, 7, 11, 13, 17]) + forward = build_temporal_section(contributions, ["h_tdigest"]) + keys = list(contributions) + orders = [list(reversed(keys)), [keys[i] for i in (2, 0, 4, 1, 3)]] + for order in orders: + other = build_temporal_section({k: contributions[k] for k in order}, ["h_tdigest"]) + assert other["shards"] == forward["shards"] + # Byte-for-byte: both the digest and its companion words, which is + # what "permutation-independent in every channel" means. + assert other["digest"]["payload"] == forward["digest"]["payload"] + assert other["digest"]["times"] == forward["digest"]["times"] + + def test_the_fold_compresses(self): + # δ is provenance, not a promise about k (§10.3) — the k1 budget is + # scale-free, not a hard cap — but the fold must actually compress: + # a root digest the size of its inputs would be no summary at all. + contributions = _contributions(range(1, 12)) + rows = sum(len(part[1]) for parts in contributions.values() for part in parts) + section = build_temporal_section(contributions, ["h_tdigest"]) + assert 0 < section["digest"]["centroids"] < rows + assert section["digest"]["delta"] == ROOT_TOC_DELTA + + +class TestComposition: + """§10.4 — how two sections meet at the GET-union-PUT seam.""" + + def test_tier_one_unions_elementwise(self): + a = build_temporal_section({"11211": [_leaf(1)], "11212": [_leaf(2)]}, ["h_tdigest"]) + b = build_temporal_section({"11212": [_leaf(3)], "11213": [_leaf(4)]}, ["h_tdigest"]) + merged = merge_temporal_sections(a, b) + assert set(merged["shards"]) == {"11211", "11212", "11213"} + assert int(merged["shards"]["11212"]) == int( + toc_merge(int(a["shards"]["11212"]), int(b["shards"]["11212"])) + ) + # The join is idempotent: re-merging changes nothing. + assert merge_temporal_sections(merged, merged)["shards"] == merged["shards"] + + def test_a_partial_producer_drops_the_digest(self): + whole = build_temporal_section({"11211": [_leaf(1)], "11212": [_leaf(2)]}, ["h_tdigest"]) + partial = build_temporal_section({"11213": [_leaf(3)]}, ["h_tdigest"]) + merged = merge_temporal_sections(whole, partial) + # Neither side's map covers the union, so neither digest can vouch for + # the store — tier 1 stands, tier 2 goes. + assert set(merged["shards"]) == {"11211", "11212", "11213"} + assert "digest" not in merged + + def test_a_whole_covering_producer_replaces_the_digest(self): + old = build_temporal_section({"11211": [_leaf(1)]}, ["h_tdigest"]) + new = build_temporal_section({"11211": [_leaf(9)]}, ["h_tdigest"]) + merged = merge_temporal_sections(old, new) + assert merged["digest"]["payload"] == new["digest"]["payload"] + + def test_a_producer_with_no_section_leaves_the_standing_one_alone(self): + standing = build_temporal_section(_contributions([1, 2]), ["h_tdigest"]) + assert merge_temporal_sections(standing, None) == standing + assert merge_temporal_sections(None, standing) == standing + assert merge_temporal_sections(None, None) is None + + def test_an_unknown_revision_on_the_standing_side_is_preserved(self): + """§10.4: readers add revisions, they never drop them. + + The merge is the WRITE composer — a ``None`` return deletes the key — + so an unreadable standing section must survive both a producer with + nothing to say and one carrying this revision's section. Otherwise the + older zagg in a mixed fleet is the one that wins. + """ + good = build_temporal_section(_contributions([1]), ["h_tdigest"]) + future = {**good, "spec": "zagg-coverage-toc/2", "shards": {"99999": "1"}} + assert merge_temporal_sections(future, None) == future + assert merge_temporal_sections(future, good) == future + # Incoming side: this revision cannot read it, so it contributes + # nothing and the standing section stands. + assert merge_temporal_sections(good, future) == good + # Unmarked debris claims no revision and does not wedge the key shut. + assert merge_temporal_sections({}, good) == good + assert merge_temporal_sections({"shards": {"1": "2"}}, good) == good + + def test_section_unchanged(self): + a = build_temporal_section({"11211": [_leaf(1)], "11212": [_leaf(2)]}, ["h_tdigest"]) + assert section_unchanged(a, None) + assert section_unchanged(a, a) + assert not section_unchanged(None, a) + assert not section_unchanged( + build_temporal_section({"11211": [_leaf(1)]}, ["h_tdigest"]), a + ) + # A standing section this revision cannot read is preserved verbatim + # by the merge, so composing over it changes nothing either — the + # skip test must not churn the object on a mixed-version store. + assert section_unchanged({"spec": "zagg-coverage-toc/2"}, a) + + def test_a_partial_producer_converges_instead_of_re_putting_forever(self): + """The composed digest, not the built one, is what the skip test sees. + + A producer that walked one shard of a two-shard store always builds a + digest, and §10.4 always drops it at the seam. Comparing the built + section against the standing one therefore never converges; comparing + the MERGE against it does, on the very next pass. + """ + first = build_temporal_section({"11211": [_leaf(1)]}, ["h_tdigest"]) + second = build_temporal_section({"11212": [_leaf(2)]}, ["h_tdigest"]) + standing = merge_temporal_sections(first, second) + assert "digest" not in standing # neither producer covered the store + assert second.get("digest") is not None # ... yet the producer built one + assert section_unchanged(standing, second) + assert section_unchanged(standing, first) + + +class TestAbsence: + """§10's standing posture: absence composes, and is never a refusal.""" + + @pytest.mark.parametrize( + "envelope", + [ + None, + {}, + {"spec": "morton-moc/1", "encoding": "ranges"}, + {"temporal": None}, + {"temporal": {"spec": "zagg-coverage-toc/2"}}, + "not a dict", + ], + ) + def test_readers_return_none_cleanly(self, envelope): + assert load_temporal_coverage(envelope) is None + assert coverage_toc(envelope) is None + assert coverage_toc_digest(envelope) is None + assert shards_overlapping(envelope, 0, 10**18) is None + + def test_a_block_whose_buffers_disagree_with_k_is_refused(self): + """§10.3's MUST-check, on all three of the block's shape claims.""" + section = build_temporal_section(_contributions([1, 2]), ["h_tdigest"]) + block = section["digest"] + for bad in ( + {"centroids": block["centroids"] + 1}, + {"centroids": None}, + {"times": build_temporal_section(_contributions([3]), ["h"])["digest"]["times"]}, + ): + envelope = {"temporal": {**section, "digest": {**block, **bad}}} + with pytest.raises(ValueError, match="row-aligned"): + coverage_toc_digest(envelope) + + def test_a_section_without_a_digest_still_prunes(self): + section = build_temporal_section(_contributions([1, 2]), ["h_tdigest"]) + section.pop("digest") + envelope = {"temporal": section} + assert coverage_toc_digest(envelope) is None + assert set(coverage_toc(envelope)) == set(section["shards"]) + + def test_a_store_declaring_no_temporal_field_has_no_fields(self): + manifest = json.loads((SPEC_DATA / "minimal" / "morton_hive.json").read_text()) + assert temporal_fields(manifest) == {} + assert temporal_fields(None) == {} + assert temporal_fields({}) == {} + + def test_a_temporal_store_declares_its_sibling(self): + manifest = json.loads((SPEC_DATA / "temporal" / "morton_hive.json").read_text()) + fields = temporal_fields(manifest) + assert set(fields) == {"h_tdigest"} + assert fields["h_tdigest"]["sibling"] == "h_tdigest_times" + + +class TestPartialReadsDropTheShard: + """§10.2 — a LISTED shard's word contains every instant in that shard. + + A word joined over whichever window leaves happened to read does not, so + a failed read costs the shard its map entry. Absent reads as *unknown* + (still a candidate); listed-but-partial reads as a promise the section + cannot keep. + """ + + def test_a_failed_window_leaf_drops_its_whole_shard(self, monkeypatch): + import zagg.coverage_toc as toc_module + from zagg.sweep import MocFamily + + def reader(leaf, *args, **kwargs): + if leaf.endswith("_2020.zarr"): + raise OSError("truncated companion") + return _leaf(1) + + monkeypatch.setattr(toc_module, "read_leaf_temporal", reader) + family = MocFamily() + family._temporal_fields = {"h_tdigest": {"sibling": "h_tdigest_times"}} + family._accumulate_temporal("root", "11213", "root/11213_2019.zarr", {}) + assert "11213" in family._temporal + family._accumulate_temporal("root", "11213", "root/11213_2020.zarr", {}) + assert "11213" not in family._temporal + # A later window that DOES read cannot resurrect a half-read shard. + family._accumulate_temporal("root", "11213", "root/11213_2021.zarr", {}) + assert "11213" not in family._temporal + # ... and the failure is scoped to its own shard. + family._accumulate_temporal("root", "11214", "root/11214_2019.zarr", {}) + assert "11214" in family._temporal + + +class TestPruning: + """§10.2 — the tier-1 predicate, conservative by the grammar's own law.""" + + def test_windows_select_the_right_shards(self): + contributions = _contributions([1, 5]) + section = build_temporal_section(contributions, ["h_tdigest"]) + envelope = {"temporal": section} + for shard, parts in contributions.items(): + word = np.array([parts[0][0]], dtype=np.uint64) + lo = int(np.asarray(parts[0][1][:, 0]).min()) - DAY_NS + hi = int(np.asarray(parts[0][1][:, 0]).max()) + DAY_NS + assert bool(np.asarray(toc_overlaps(word, lo, hi))[0]) + assert shard in shards_overlapping(envelope, lo, hi) + + def test_a_window_past_every_shard_selects_none(self): + section = build_temporal_section(_contributions([1, 5]), ["h_tdigest"]) + far = BASE_NS + 10_000 * DAY_NS + assert shards_overlapping({"temporal": section}, far, far + DAY_NS) == [] + + +class TestOnCommittedStores: + """End to end, on the §7 fixtures: the writer, and the absence pin.""" + + def _copy(self, tmp_path, name): + dst = tmp_path / name + shutil.copytree(SPEC_DATA / name, dst) + return str(dst) + + def _clone_shard(self, root, src="11213", dst="11214"): + """Give ``root`` a second shard, cloned from its committed leaf. + + The §7 ``temporal/`` fixture ships ONE shard, which is exactly the + shape that hides the composition seam: a single-shard producer is + always whole-covering, so its digest survives the merge and the skip + test converges by accident. + """ + from zagg.grids.morton import morton_word + from zagg.hive import shard_leaf_path + + src_leaf = Path(shard_leaf_path(root, int(morton_word(src)))) + dst_leaf = Path(shard_leaf_path(root, int(morton_word(dst)))) + dst_leaf.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(src_leaf, dst_leaf) + return [(int(morton_word(src)), None)], [(int(morton_word(dst)), None)] + + def test_refresh_rebuilds_the_section_from_its_own_walk(self, tmp_path): + root = self._copy(tmp_path, "temporal") + committed = json.loads((SPEC_DATA / "temporal" / "coverage.moc").read_text()) + envelope = refresh_root_coverage(root) + assert envelope["temporal"]["source"] == "refresh" + # Same walk, same words: only the carrier's provenance differs. + assert envelope["temporal"]["shards"] == committed["temporal"]["shards"] + assert ( + envelope["temporal"]["digest"]["payload"] == committed["temporal"]["digest"]["payload"] + ) + + def test_a_sweep_writes_the_section_the_fixture_committed(self, tmp_path): + from zagg.grids.morton import morton_word + from zagg.sweep import run_sweep + + root = self._copy(tmp_path, "temporal") + (Path(root) / "coverage.moc").unlink() + leaves = [(int(morton_word("11213")), None)] + summary = run_sweep(root, leaves, families=["moc"], record=False) + assert summary["families"]["moc"]["temporal_shards"] == 1 + committed = json.loads((SPEC_DATA / "temporal" / "coverage.moc").read_text()) + written = read_root_coverage(root) + assert written["temporal"]["shards"] == committed["temporal"]["shards"] + assert ( + written["temporal"]["digest"]["payload"] == committed["temporal"]["digest"]["payload"] + ) + # Idempotence: a second pass over an unchanged tree writes nothing. + again = run_sweep(root, leaves, families=["moc"], record=False) + assert again["families"]["moc"]["root_moc_written"] is False + + def test_a_truncated_companion_is_refused(self, tmp_path): + """§1.1 row alignment, at ARRAY level (issue #452's failure shape). + + A companion with fewer rows than its payload aligns row for row over + its own length, so the per-cell check never fires — the leaf would + join a prefix of its cells and be published as whole. + """ + import zarr + + from zagg.coverage_toc import read_leaf_temporal + from zagg.grids.morton import morton_word + from zagg.hive import shard_leaf_path + + root = self._copy(tmp_path, "temporal") + manifest = json.loads((Path(root) / "morton_hive.json").read_text()) + fields = temporal_fields(manifest) + leaf = shard_leaf_path(root, int(morton_word("11213"))) + group = zarr.open_group(leaf, path=str(manifest["cell_order"]), mode="a", zarr_format=3) + rows = group["h_tdigest_times"].shape[0] + group["h_tdigest_times"].resize((rows - 1,)) + with pytest.raises(ValueError, match="row-aligned"): + read_leaf_temporal(leaf, int(manifest["cell_order"]), fields) + + def test_refresh_drops_only_the_shard_whose_leaf_failed(self, tmp_path, monkeypatch): + import zagg.coverage_toc as toc_module + + root = self._copy(tmp_path, "temporal") + self._clone_shard(root) + real = toc_module.read_leaf_temporal + + def reader(leaf, *args, **kwargs): + if "11214" in leaf: + raise OSError("truncated companion") + return real(leaf, *args, **kwargs) + + monkeypatch.setattr(toc_module, "read_leaf_temporal", reader) + envelope = refresh_root_coverage(root) + # The spatial walk still lists both shards; the temporal map lists + # only the one it could read whole (§10.2's unknown-not-empty rule). + assert set(envelope["temporal"]["shards"]) == {"11213"} + + def test_the_shard_word_unions_across_every_temporal_field(self, tmp_path): + """§10.2's headline rule: coverage is "any data", not "data in field X". + + The committed fixture declares ONE temporal field, so the union is + invisible on it. A second field is grafted onto a copy of the leaf — + the same payload rows under a companion whose words sit in a different + campaign — and the shard word must be the join across both. + """ + import zarr + from mortie import time2toc + + from zagg.coverage_toc import read_leaf_temporal + from zagg.grids.morton import morton_word + from zagg.hive import shard_leaf_path + + root = self._copy(tmp_path, "temporal") + manifest = json.loads((Path(root) / "morton_hive.json").read_text()) + order = int(manifest["cell_order"]) + leaf = shard_leaf_path(root, int(morton_word("11213"))) + group = zarr.open_group(leaf, path=str(order), mode="a", zarr_format=3) + payload, sibling = group["h_tdigest"], group["h_tdigest_times"] + # A companion of the same per-row width, so §1.1 alignment holds, but + # carrying instants a whole campaign away from the committed ones. + far = np.empty(sibling.shape[0], dtype=object) + for i, row in enumerate(sibling[:]): + width = 0 if row is None else len(row) // 8 + far[i] = np.array( + [int(time2toc(BASE_NS + 20_000 * DAY_NS + j * DAY_NS)) for j in range(width)], + dtype="= 2 + + def test_refresh_never_deletes_the_section_when_every_leaf_fails(self, tmp_path, monkeypatch): + """The escape hatch must not be the thing that destroys the section. + + Fail-open per leaf is fail-DESTRUCTIVE in aggregate: refresh PUTs its + envelope outright, so an all-failed walk would publish a root object + with the ``temporal`` key gone — during exactly the incident an + operator reached for refresh to repair. + """ + import zagg.coverage_toc as toc_module + + root = self._copy(tmp_path, "temporal") + standing = json.loads((Path(root) / "coverage.moc").read_text())["temporal"] + + def reader(*args, **kwargs): + raise OSError("credentials expired mid-walk") + + monkeypatch.setattr(toc_module, "read_leaf_temporal", reader) + envelope = refresh_root_coverage(root) + assert envelope["ranges"] # the spatial refresh still succeeded + assert envelope["temporal"] == standing + assert coverage_toc(envelope) == coverage_toc({"temporal": standing}) + + def test_refresh_composes_a_partial_rebuild_with_the_standing_section( + self, tmp_path, monkeypatch + ): + import zagg.coverage_toc as toc_module + + root = self._copy(tmp_path, "temporal") + self._clone_shard(root) + refresh_root_coverage(root) # both shards land in the standing section + real = toc_module.read_leaf_temporal + + def reader(leaf, *args, **kwargs): + if "11214" in leaf: + raise OSError("truncated companion") + return real(leaf, *args, **kwargs) + + monkeypatch.setattr(toc_module, "read_leaf_temporal", reader) + envelope = refresh_root_coverage(root) + # The shard the walk could not read keeps the word the last whole walk + # published: a partial rebuild composes, it does not overwrite. + assert set(envelope["temporal"]["shards"]) == {"11213", "11214"} + + def test_a_second_pass_over_a_multi_shard_store_writes_nothing(self, tmp_path): + """Sweep idempotence where the seam actually bites (§10.4). + + Two shards, two incremental sweeps: neither producer covers the store, + so the composed section carries no digest while every pass keeps + building one. The skip test has to converge on what was WRITTEN, or + the fleet re-PUTs a byte-identical root object forever. + """ + from zagg.sweep import run_sweep + + root = self._copy(tmp_path, "temporal") + (Path(root) / "coverage.moc").unlink() + a, b = self._clone_shard(root) + for leaves in (a, b): + summary = run_sweep(root, leaves, families=["moc"], record=False) + assert summary["families"]["moc"]["root_moc_written"] is True + written = read_root_coverage(root) + assert set(written["temporal"]["shards"]) == {"11213", "11214"} + assert "digest" not in written["temporal"] + for leaves in (b, a): + again = run_sweep(root, leaves, families=["moc"], record=False) + assert again["families"]["moc"]["root_moc_written"] is False + assert read_root_coverage(root)["temporal"] == written["temporal"] + + def test_a_non_temporal_store_writes_byte_identical_bytes(self, tmp_path): + """The §10 absence promise, as bytes. + + A store declaring no temporal field must produce EXACTLY the root + object a pre-#480 zagg produced: no key added, no key reordered. + """ + from zagg.grids.morton import morton_word + from zagg.sweep import run_sweep + + root = self._copy(tmp_path, "minimal") + summary = run_sweep( + root, [(int(morton_word("11213")), None)], families=["moc"], record=False + ) + assert "temporal_shards" not in summary["families"]["moc"] + raw = (Path(root) / "coverage.moc").read_bytes() + envelope = json.loads(raw) + assert "temporal" not in envelope + assert coverage_toc(envelope) is None + # The reference: the same carrier built with the §10 parameter omitted + # entirely — the pre-#480 call — serialized the pre-#480 way. Equal + # bytes is the whole promise. + reference = build_root_coverage([morton_word("11213")], 4, source="sweep") + reference["generated_at"] = envelope["generated_at"] + assert json.dumps(reference, indent=1).encode() == raw + # And a re-write through the GET-union-PUT seam stays temporal-free. + assert "temporal" not in write_root_coverage(root, reference) + assert "temporal" not in json.loads((Path(root) / "coverage.moc").read_bytes()) diff --git a/tests/test_spec_conformance.py b/tests/test_spec_conformance.py index ed85d7d85..ca12d95f1 100644 --- a/tests/test_spec_conformance.py +++ b/tests/test_spec_conformance.py @@ -17,6 +17,7 @@ and spec text are pinned against each other through the committed bytes. """ +import base64 import hashlib import json import struct @@ -28,6 +29,7 @@ from numcodecs import Zstd from zarr.storage import LocalStore +from zagg.coverage_toc import coverage_toc, coverage_toc_digest from zagg.readers.tdigest_tensor import read_cell, read_locations from zagg.stats.composition import counts_from_composition, unpack_composition @@ -1703,3 +1705,162 @@ def test_unknown_declaration_is_refused(self): for bad in ({"spec": "zagg-located/2"}, {"shape": "per-cell"}, {"grammar": "geohash/1"}): with pytest.raises(ValueError): located_declaration({"located": {**block, **bad}}) + + +class TestRootCoverageTemporalSection: + """§10 — the ``zagg-coverage-toc/1`` section, on the committed sidecar. + + ``temporal/`` is the only fixture with a root coverage object at all: + every other fixture store declares no temporal field, so a sweep of one + produces no section. Their *lack* of the object is §10's absence rule + pinned as bytes, asserted below. + """ + + def _envelope(self): + return json.loads((SPEC_DATA / "temporal" / "coverage.moc").read_text()) + + def test_the_carrier_still_reads_as_a_plain_root_moc(self): + # §10 adds ONE key to the `morton-moc/1` carrier; a reader that knows + # nothing about §10 must decode the object exactly as before. + from zagg.coverage import load_coverage + from zagg.grids.morton import morton_word + from zagg.hive import root_coverage_words + + envelope = self._envelope() + assert envelope["spec"] == "morton-moc/1" + assert envelope["encoding"] == "ranges" + assert envelope["ranges"] == [[_expected("temporal")["shard"]] * 2] + assert load_coverage(str(SPEC_DATA / "temporal")) == envelope + assert set(root_coverage_words(envelope)) == { + int(morton_word(_expected("temporal")["shard"])) + } + + def test_section_grammar(self): + exp = _expected("temporal")["root_coverage"] + section = self._envelope()["temporal"] + assert section["spec"] == exp["spec"] == "zagg-coverage-toc/1" + assert section["source"] == "sweep" + assert section["fields"] == exp["fields"] + # Words are DECIMAL STRINGS: a uint64 exceeds 2^53 and a float-based + # JSON parser would mangle a raw number, exactly as for the ranges. + assert section["shards"] == exp["shards"] + assert all(isinstance(w, str) for w in section["shards"].values()) + digest = section["digest"] + assert digest["element"] == {"dtype": "float32", "shape": [-1, 2]} + assert (digest["encoding"], digest["weights"], digest["value"]) == ( + "base64", + "counts", + "toc-ns", + ) + assert digest["delta"] == exp["digest"]["delta"] + + def test_digest_decodes_through_the_native_grammars(self): + """§10.3: the payload is §2.1 bytes and the sibling is §8.3 words. + + Decoded here with the SPEC-TEXT recipe (base64, then the §1.4 raw + little-endian buffer at the declared dtype) — no zagg decoder — so the + "zero new grammar" claim is what is being asserted. + """ + exp = _expected("temporal")["root_coverage"]["digest"] + block = self._envelope()["temporal"]["digest"] + payload = np.frombuffer( + base64.b64decode(block["payload"]), dtype=np.dtype("float32").newbyteorder("<") + ).reshape(-1, 2) + words = np.frombuffer( + base64.b64decode(block["times"]), dtype=np.dtype("uint64").newbyteorder("<") + ) + assert len(payload) == len(words) == block["centroids"] + np.testing.assert_array_equal(payload, np.array(exp["centroids"], dtype=np.float32)) + np.testing.assert_array_equal(words, np.array(exp["times"], dtype=np.uint64)) + # §2.1: rows ascend by mean, and every weight is a positive count. + assert (np.diff(payload[:, 0]) >= 0).all() + assert (payload[:, 1] > 0).all() + + def test_weight_conservation(self): + """§10.3: `sum(weights)` is the store's temporal observation count.""" + exp = _expected("temporal") + block = self._envelope()["temporal"]["digest"] + payload, _words = coverage_toc_digest(self._envelope()) + total = exp["root_coverage"]["obs_total"] + assert total == sum(cell["count"] for cell in exp["cells"]) + assert float(payload[:, 1].sum()) == block["weight_total"] == float(total) + + def test_shard_word_conservatively_contains_every_instant(self): + """§10.2's whole claim, on committed bytes. + + Every observation instant in the shard falls inside that shard's one + envelope word, and the grammar's own overlap predicate says so for + each instant without any decoding by the caller. + """ + from mortie import from_datetime64, to_datetime64, toc2time, toc_overlaps + + exp = _expected("temporal") + word = coverage_toc(self._envelope())[exp["shard"]] + lo, hi = (int(np.asarray(to_datetime64(b)).astype("int64")) for b in toc2time(word)) + instants = np.array( + [int(t) for cell in exp["cells"] for t in cell["obs_span_ns"]], dtype="int64" + ) + assert (lo <= instants).all() and (instants < hi).all() + internal = np.asarray(from_datetime64(instants.astype("datetime64[ns]")), dtype=np.uint64) + for t in internal: + assert bool(np.asarray(toc_overlaps(np.array([word]), int(t), int(t) + 1))[0]) + + def test_the_value_axis_is_the_envelope_midpoint(self): + """§10.3: column 0 is derived from the words, not from the observations. + + Each contributing centroid enters the fold at the midpoint of its own + §8.3 envelope, so a weight-1 centroid — whose word is a timestamp and + whose ``toc2time`` envelope is a point — carries that EXACT instant, + and every other mean is a convex combination of midpoints and so lies + inside its own centroid's word. Both hold to the float32 quantum the + section documents, which is why the words, never the means, are the + exact temporal claim. + """ + from mortie import toc2time + + payload, words = coverage_toc_digest(self._envelope()) + start, end = (np.asarray(b, dtype=np.float64) for b in toc2time(np.asarray(words))) + means = payload[:, 0].astype(np.float64) + # The one exact arm: weight-1 rows sit on their word's instant. + single = payload[:, 1] == 1 + assert single.any() + assert (start[single] == end[single]).all() + np.testing.assert_array_equal(payload[single, 0], start[single].astype(np.float32)) + # Everything else: inside its own envelope, up to float32 rounding + # (~2^-24 relative — roughly ten minutes at present-day magnitudes). + quantum = np.abs(means) * 2.0**-23 + assert ((means >= start - quantum) & (means <= end + quantum)).all() + + def test_the_root_words_reduce_to_the_shard_word(self): + # The tier-2 companion and the tier-1 map are two views of the same + # join: reducing the digest's per-centroid envelopes reproduces the + # shard's envelope word exactly (mortie's semilattice, spec §8.4). + from mortie import toc_reduce + + exp = _expected("temporal") + _payload, words = coverage_toc_digest(self._envelope()) + assert int(toc_reduce(words)) == coverage_toc(self._envelope())[exp["shard"]] + + def test_the_shard_word_is_the_join_of_the_committed_leaf_words(self): + # Derived from the LEAF bytes, not from the sidecar: a writer that + # folded the wrong thing fails here instead of agreeing with itself. + from mortie import toc_reduce + + exp = _expected("temporal") + leaf = np.concatenate( + [np.array(cell["h_tdigest_times"], dtype=np.uint64) for cell in exp["cells"]] + ) + assert int(toc_reduce(leaf)) == coverage_toc(self._envelope())[exp["shard"]] + + @pytest.mark.parametrize( + "name", ["minimal", "kitchen_sink", "column", "flux", "raster_toc", "pyramid"] + ) + def test_non_temporal_fixtures_carry_no_root_coverage_object(self, name): + # §10's absence rule, pinned as bytes — WITH its precondition, which + # is what makes the missing object evidence of the rule rather than + # of the generator simply never having been pointed at these trees. + from zagg.coverage_toc import temporal_fields + + manifest = json.loads((SPEC_DATA / name / "morton_hive.json").read_text()) + assert temporal_fields(manifest) == {} + assert not (SPEC_DATA / name / "coverage.moc").exists() diff --git a/tools/generate_spec_fixtures.py b/tools/generate_spec_fixtures.py index f3a8c36d7..fe368b4f1 100644 --- a/tools/generate_spec_fixtures.py +++ b/tools/generate_spec_fixtures.py @@ -83,6 +83,12 @@ — the aggregation kernel that will produce them is #410's next PR — so the expectations stay input-derived. ``kitchen_sink/``, committed before §9 and not regenerated, is the absent-``located`` ⇒ §2.2 pin. + ``temporal/`` is ALSO the only fixture carrying a **root ``coverage.moc``** + (issue #480): the §10 ``zagg-coverage-toc/1`` section, written here by the + production sweep writer (``MocFamily``'s leaf read + finisher). The other + six declare no temporal field, so a sweep of one produces no section — + leaving them without a root coverage object IS §10's absence rule, and + keeps those trees byte-identical. STALE BY DESIGN: ``minimal/`` and ``kitchen_sink/`` were committed before issue #382 and their ``morton_hive.json`` still carries the pre-#382 @@ -1161,6 +1167,39 @@ def build_temporal(out: Path) -> None: processing.process_shard = original assert meta.get("error") is None, meta + # The §10 root coverage sidecar, through the PRODUCTION writer: the MOC + # family's own leaf read plus its finisher, which is exactly what a sweep + # runs. This is the only fixture that gets a root coverage.moc — every + # other fixture store declares no temporal field, so a sweep of one would + # produce no section at all, and writing a bare carrier there would churn + # four committed trees for nothing (§10's absence rule, pinned as byte + # identity by the conformance suite). + from mortie import toc_reduce + + from zagg.coverage_toc import coverage_toc, coverage_toc_digest + from zagg.sweep import MocFamily + + family = MocFamily() + contribution, _written_at = family.read_leaf(root, SHARD_KEY, None, "morton-hive/1", {}) + family.finish(root, [{"payload": contribution}], 4, {}) + envelope = hive.read_root_coverage(root) + root_digest, root_words = coverage_toc_digest(envelope) + # The shard envelope word is DERIVED from the generator's inputs — the + # join over every per-centroid word it handed the writer — so a writer + # that folds the wrong thing fails here instead of certifying itself. The + # digest rows are the writer's committed output read back (pinned the way + # column/'s group values are); the claims that matter over them — weight + # conservation and per-centroid containment — are derived, from the cell + # plan's own observation counts and the instants recorded per cell. + shard_word = int( + toc_reduce( + np.concatenate( + [cell["h_tdigest"][2] for cells in by_chunk.values() for cell in cells.values()] + ).astype(np.uint64) + ) + ) + assert coverage_toc(envelope) == {SHARD_KEY: shard_word}, coverage_toc(envelope) + leaf_rel = hive.shard_leaf_path("", shard).lstrip("/") expected = { "shard": SHARD_KEY, @@ -1173,6 +1212,21 @@ def build_temporal(out: Path) -> None: "chunks_per_shard": grid.chunks_per_shard, "empty_chunk": EMPTY_CHUNK, "delta": DELTA, + # The §10 root coverage temporal section: the tier-1 word (derived), + # the tier-2 digest (the writer's, read back) and the weight total the + # cell plan says it must carry. + "root_coverage": { + "object": "coverage.moc", + "spec": "zagg-coverage-toc/1", + "fields": ["h_tdigest"], + "shards": {SHARD_KEY: str(shard_word)}, + "obs_total": sum(c["count"] for c in expected_cells), + "digest": { + "delta": envelope["temporal"]["digest"]["delta"], + "centroids": [[float(m), float(w)] for m, w in root_digest], + "times": [str(int(w)) for w in root_words], + }, + }, # The declarations the conformance tests assert against the committed # attrs — each on the array that HOLDS the words (§8/§9), and the # payload's binding, which is a sibling key of the ragged block. @@ -1194,7 +1248,10 @@ def build_temporal(out: Path) -> None: "content_hashes": _o11_hashes(str(out / leaf_rel)), } (out.parent / f"{out.name}.expected.json").write_text(json.dumps(expected, indent=1) + "\n") - print(f"{out.name}: leaf {leaf_rel}, {len(expected_cells)} populated cells, both toc variants") + print( + f"{out.name}: leaf {leaf_rel}, {len(expected_cells)} populated cells, both toc " + f"variants, root coverage.moc with {len(root_digest)} digest centroids" + ) def main() -> None: