From 4ff6bf2ce21ed1ea6a03e56367524d6ff0aef315 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:19:58 -0400 Subject: [PATCH 01/31] Open the lane journal with the sites this branch will move The charter names the profiler's line numbers, which are its tree's. Record the same six sites at this branch's head so the next reader does not have to re-derive the offset the verify-once commits introduced. Co-Authored-By: Claude Opus 5 --- PROGRESS-native-scale-transport.md | 48 ++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 PROGRESS-native-scale-transport.md diff --git a/PROGRESS-native-scale-transport.md b/PROGRESS-native-scale-transport.md new file mode 100644 index 0000000000..158df13472 --- /dev/null +++ b/PROGRESS-native-scale-transport.md @@ -0,0 +1,48 @@ +# Lane journal — native scale: transport ceiling and per-node retention + +Branch `native-scale-transport`, worktree +`~/PolicyEngine/_worktrees/microcosm-native-scale`, stacked on +`native-verify-once` (PR #935). Draft PR only; never ready, never merged. + +This file is a session-handoff journal in the sense CLAUDE.md means: accurate +when written, history afterwards. Check git/GitHub for current truth. + +## State + +**Started 2026-09-17.** Orientation complete; baseline run not yet launched. + +## Charter, in one line + +Lift the two hard limits the 2026-09-16 profiler found — the 64 MiB +whole-roster receipt transport (refuses above 6.1% of the US source) and +per-node full-population retention (~300 GiB at full source) — then prove a +build above the old ceiling at 1/10. + +## Sites, line-cited in this tree (not the profiler's tree) + +| what | where, at `5ff889814` | +|---|---| +| preparation payload limit | `survey_population_preparation.py:50` `MAX_PAYLOAD_BYTES`, `:104` `_encode`, `:108` `PAYLOAD_LIMIT` | +| the whole-roster document | `survey_population_preparation.py:1698-1699` (`"selection": _plan_document(plan)`, `"origins": origins`) | +| second origin budget | `survey_population_preparation.py:816-818` `_bounded_append` → `ORIGIN_LIMIT` | +| allocation payload limit | `graph_survey_population.py:62` `ALLOCATION_MAX_BYTES`, `:344-369` `_allocation_payload` | +| per-cell frame identity | `survey_population_preparation.py:601` `_frame_cell_encode`, the `_frame_identity` loop | +| the observer that retains 19 populations | `graph_atomic_survey_financial.py` `observe(...)`, called from `executor.py` `_observer_snapshot` | + +## Done + +- Read the profiler's §5, §9, §10, §11 and the verify-once report's measurement + section. Baseline to beat: **2,010.07 CPU-s / 2,016.08 s wall / 13.31 GB peak + RSS** for the 19-node financial graph at 1/1000 (verify-once file C). +- Confirmed no live microcosm build on the machine; 128 GiB / 18 cores. +- Located the 19-node harness: + `~/PolicyEngine/_worktrees/microcosm-verify-once/.measure/harness19_verify_once.py`. + +## Next + +1. Baseline run of the 19-node harness at 1/1000 in a throwaway worktree at the + base commit, with an RSS-over-time sampler and runner-vs-node-loop split. +2. Design note `docs/us-native-scale-transport.md`. +3. Implement, with refusal tests, identity-equality proofs, replay proof. +4. 1/10 run above the old ceiling. +5. CI-shaped battery, draft PR. From c00da8bbff6f45ea99b4688102ab48541d19ead1 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:52:19 -0400 Subject: [PATCH 02/31] Seal a frame column at a time, and write down the transport that replaces the receipt _frame_identity walked every cell through _frame_cell_encode and two digest.update calls. The bytes it fed the digest are a pure function of the column's values and its dtype metadata, so each dtype the US frame actually carries now assembles that exact byte string per column: float64 through a fixed-offset matrix with a keep mask, the integer widths through int.__repr__ over one tolist, booleans by index, strings by encoding each distinct value once. object and category keep the cell walk, because 1, True and 1.0 are equal and hash alike there while the encoder spells them three different ways. test_us_survey_frame_identity_encoding.py already held an independent predecessor traversal and compares the concatenated byte preimage rather than the digest, so the equality is proved against bytes and not against a description: 160 tests, and the file now runs in 0.74 s where it took 7.16 s. docs/us-native-scale-transport.md records the transport design, the five measured ceilings in the order they bind, and the three store mechanisms that refuse a store-backed population snapshot. Co-Authored-By: Claude Opus 5 --- docs/us-native-scale-transport.md | 366 ++++++++++++++++++ .../survey_population_preparation.py | 205 +++++++++- 2 files changed, 559 insertions(+), 12 deletions(-) create mode 100644 docs/us-native-scale-transport.md diff --git a/docs/us-native-scale-transport.md b/docs/us-native-scale-transport.md new file mode 100644 index 0000000000..034c3a3122 --- /dev/null +++ b/docs/us-native-scale-transport.md @@ -0,0 +1,366 @@ +# US native build: receipt transport and per-node population retention + +Design authority for the two hard limits the 2026-09-16 cost attribution found +in the native US graph build +(`experiments/us-native-executor-cost-attribution-20260916.md` on the +`microcosm-exec-profile` worktree, §5 and §9): a whole-roster receipt that +refuses above 6.1% of the US source, and a per-node population retention that +needs about 300 GiB at full source. + +Every line number below is at `5ff889814`, the branch point, unless it names a +later head. Line numbers in the cost-attribution report are its own tree's and +are offset from these. + +## 1. What refuses today, measured rather than inferred + +The preparation receipt is **one JSON byte string**, built by +`prepare_authenticated_survey_population` and handed to `_encode` in a single +call (`survey_population_preparation.py:1679`) under +`MAX_PAYLOAD_BYTES = 64 * 1024**2` (`:50`), which refuses `PAYLOAD_LIMIT` +before each chunk append (`:104-109`). Twelve top-level keys; ten of them are +fixed-size. The two that are not are `"selection": _plan_document(plan)` +(`:1698`) and `"origins": origins` (`:1699`), which carry **one record per +source household and per person**: + +| block | built at | per-row shape | +|---|---|---| +| `selection.selected`, `.excluded` | `_plan_document` `:240-253` | one `_value`-flattened plan row per supplied household | +| `origins.households` | `_origins` `:946` | one 7-field dict per stacked household | +| `origins.entities` | `_origins` `:870` | one `[new_id, channel, source_id]` row per row of **each** of the six US entities | +| `origins.persons.rows` | `_origins` `:965` | one 10-element row per stacked person | + +A second, independent budget guards the same lists: `_bounded_append` +(`:816-819`) `_encode`s each row on its own and refuses `ORIGIN_LIMIT` against +the same 64 MiB constant. `_plan_document` opens its own budget (`:247`) and +`_origins` opens another (`:844`), so there are three 64 MiB ceilings on one +document, plus the final `_encode`. + +**The size law, read off the pilot's own artifact rather than modelled.** The +1/1000 preparation receipt survives in `_recovered/pilot-runs/` — 1,136,063 B at +`native19-required-20260912/run/financial-artifacts/preparation.json`, byte +identical to `native19-v2`'s copy and to that run's +`graph-store/objects/04/0481…/payload.bin`, sha256 `34b362d85d2f06ac…`. It +records `supplied_households = 1587376` with 1,584 selected. Re-encoding it +through the repository's own canonical separators reproduces 1,136,063 exactly. +Decomposed: `origins` 718,800 B (households 269,000; entities 224,755; persons +210,924; storage_transitions 13,268; age normalization 762), `selection` +393,395 B (selected 385,122; excluded 7,250; cells 923), `producer` 21,135 B, +everything else 2,733 B. The blocks that grow with the draw are +1,096,821 B over 1,584 selected households: + +> **bytes(f) = 39,242 + 692.4375 × 1,587,376 × f** + +At f = 1 that is **1,099,197,911 B = 1.024 GiB, 16.4× the limit**. Encoding it +costs about 53 CPU-s at full source (the report's §5 measured 33.3 µs per source +household), so the ceiling is a transport-shape decision and not a performance +trade-off. + +**Five ceilings, in the order they bind.** Each is computed from the same +committed artifact, or from the allocation payload's own measured 501,597 B at +1,584 households: + +| # | ceiling | refusal | households | share of source | +|---|---|---|---|---| +| 1 | preparation `_encode` | `PAYLOAD_LIMIT` | 96,860 | **6.10%** | +| 2 | `_origins` row budget | `ORIGIN_LIMIT` | 150,916 | **9.51%** | +| 3 | allocation payload stream | `ALLOCATION_LIMIT` | 211,924 | 13.35% | +| 4 | `_plan_document` row budget | `ORIGIN_LIMIT` | 270,284 | 17.03% | +| 5 | allocation structural pre-check `ALLOCATION_MAX_BYTES // 128` | `ALLOCATION_LIMIT` | 524,288 | 33.03% | + +A 1/10 build is 158,737 source households, so **#1 and #2 both bind and #3 does +not**; a full build needs #1 through #4. #5 bounds the intermediate maps +`allocation_instructions` builds, not the payload, and its implied 127-byte row +budget is below the payload's 220-byte structural overhead alone, so it can +never be the binding payload constraint. + +**The store is not the constraint.** `ContentStore.put_bytes` +(`store.py:984`) and `put_frame` (`:919`) carry no byte cap, and `put_frame` +already writes one `.npy` per entity and column under +`objects//` with a `meta.json` payload table recording sha256 and +size per file (`store.py:140-153`), re-verified on every read by +`_verified_meta` (`:227`). Every 64 MiB ceiling in this pipeline is owned by +the US runtime, not by the graph. + +## 2. The new transport + +### 2a. What must not move + +`preparation_sha256` is `_sha(payload)` (`graph_survey_population.py:493`, +`:513`), it is folded through `_digest` into the CREATE and ALLOCATION node +params (`:1226`, `:1235`, `:1254`), and `params` reaches `node_key` through +`normative(node)` (`keys.py:288`). So the document's bytes reach every node key +and every store address. That is why the transport changes and the **canonical +byte stream does not**. + +The stream is already defined independently of its materialisation. +`_chunks(value)` (`:97-102`) yields the canonical JSON of the document; +`_encode` (`:104-110`) concatenates those chunks under a bound, and `_digest` +(`:113-116`) hashes the same chunks under no bound. sha256 is a streaming hash +over exactly those bytes in exactly that order, so + +> `_sha(_encode(document)) == _digest(document)` + +for every document `_encode` accepts, and `_digest` keeps returning that value +for documents `_encode` refuses. The new transport digests the stream and +materialises it in bounded segments; it does not re-shape the document. + +### 2b. The segmented, content-addressed receipt + +`_segment_stream(document)` replaces the single `_encode` call for the two +whole-roster receipts. It walks `_chunks(document)` once and: + +* accumulates into a segment buffer, and whenever appending the next chunk + would exceed `MAX_SEGMENT_BYTES` (= the old `MAX_PAYLOAD_BYTES`, 64 MiB) it + closes the segment — so **no accumulation in the encoder is ever larger than + the value the old ceiling allowed**, and `PAYLOAD_LIMIT` still guards each + one, with the same code, at the same number; +* records each closed segment's sha256 and size, and updates one running sha256 + over the whole stream; +* refuses `ROSTER_LIMIT` above `MAX_ROSTER_BYTES`, an explicit resource ceiling + that is *not* a transport shape: it is the total the process is willing to + spend, set far above full source (see §2d), and it is the only new refusal. + +The result is a `_RosterReceipt` handle carrying `digest` (identical to +`_digest(document)`), `size`, and the segment table. The preparation path also +writes the segments to a content-addressed spill under the run's `snapshots` +directory — `/.segment` plus a small `/header.json` naming +the protocol, the whole-stream digest, the size, and the ordered segment table — +which is the same header-plus-bodies shape `ContentStore.put_frame` already +uses, and which gives the receipt a re-verifiable on-disk form independent of +the graph store. + +`read_all()` reassembles, re-hashes each segment against its recorded digest +and the whole against `digest`, and refuses `ROSTER_LIMIT` above +`MAX_MATERIALIZED_BYTES`. The two consumers that need whole bytes — the +`preparation` and `allocation` opaque artifacts, which reach +`ContentStore.put_bytes` through `KernelResult.artifacts` — call it. So +`AuthenticatedSurveyPopulationPreparation.payload` stays `bytes` and **every +existing consumer is unchanged**: `_sha(entry[1])`, `type(value.payload) is +bytes`, the `CANDIDATE_MISMATCH` byte comparison, and the artifact publication +at `graph_survey_population.py:510`. + +This is deliberately honest about what it does and does not buy. It removes +every ceiling that binds below full source and it keeps the encoder's peak +allocation at one segment. It does **not** remove the final materialisation: +1.02 GiB of bytes exists once, at full source, because +`KernelResult.artifacts` is a mapping of `bytes` and `put_bytes` takes whole +bytes. Removing that last copy needs a streaming artifact channel in +`microcosm-graph`, which is main-only and is §4's sibling. At full source +1.02 GiB against a ~26 GiB working set is not what stops a build. + +### 2c. The row budgets + +`_bounded_append`'s `ORIGIN_LIMIT` (ceilings #2 and #4) is a bound on the +Python row list, not on a transport. It keeps its code and its per-row +`_encode`, and its ceiling becomes `MAX_ROSTER_BYTES` — the same explicit +resource ceiling — so the refusal still fires, on the same rows, with the same +message, at a number a full-source build does not reach. + +### 2d. The allocation payload + +`_allocation_payload` (`graph_survey_population.py:344-369`) is already a +stream with a per-row bound: it emits `_bounded_json(_instruction_document(row), +4096)` per household into one bytearray with a pre-append check against +`ALLOCATION_MAX_BYTES` (`:362`) and a post-loop re-check (`:367`). Its bytes are +byte-identical to the canonical whole-tree encoding, which +`test_us_graph_survey_population.py:304` pins — an identity that holds only +because `"households"` sorts before all six metadata keys — so the same +streaming-digest argument applies. + +It is built inside `SurveyPopulationAllocationKernel.run`, which has no +filesystem path and must not write to the store itself, so it takes the +in-memory half of §2b: bounded segments in a list, each guarded by +`ALLOCATION_LIMIT` at the unchanged 64 MiB, under `MAX_ROSTER_BYTES` in total, +materialised once. `allocation_sha256` (`:441`) is unchanged in derivation and +in value. + +Ceiling #5, the `// 128` structural pre-check, keeps its constant: it bounds +`allocation_instructions`' intermediate maps and 524,288 households is above +full source at every fraction this lane runs. + +### 2e. Size law after the change + +`MAX_ROSTER_BYTES` is set to `64 * MAX_SEGMENT_BYTES` = 4 GiB, which is +3.9× a full-source preparation receipt and 7.6× a full-source allocation +payload (measured 316.665 B per household × 1,587,376 = 502.6 MiB). The +refusals therefore still exist, still fail closed, and no longer bind below +full source: + +| | before | after | +|---|---|---| +| preparation receipt | 96,860 hh (6.10%) | 6,206,000 hh (391% of source) | +| `_origins` rows | 150,916 hh (9.51%) | 9,658,000 hh | +| allocation payload | 211,924 hh (13.35%) | 13,563,000 hh | +| `_plan_document` rows | 270,284 hh (17.03%) | 17,298,000 hh | + +## 3. Retention and the frame seal + +### 3a. What the observer costs and where + +`run_graph` calls `_population_observer(node_id, _observer_snapshot(updated))` +at `executor.py:2773-2774`, under the single guard +`if _population_observer is not None:`. That is **outside** the cache-load +attempt (`if resume != "forbid":`, `:2575`), outside the cold path +(`if result is None:`, `:2607`) and outside the store write (`if not hit:`, +`:2776`), so every reached node pays it, cache hits included; only the +unreached-node `continue` (`:2510-2524`) skips it. `_observer_snapshot` +(`:356-408`) detaches by round-tripping every entity and link table plus the +strata through `pickle.loads(pickle.dumps(..., protocol=5))` (`:373`) — a full +independent copy, measured by the cost-attribution report at **9.6–9.8 bytes +per cell per snapshot**. + +The financial runner retains all of them: `observed[node_id] = population` +(`graph_atomic_survey_financial.py:1644`), never cleared, read at `:1656` +(node ids), `:1796`, `:1818`, `:1821`, `:1824`, `:1944`, `:1948`, `:1974` +(the objects), `:1897` (object *identity*) and `:1945` (the stamp alone). It +reaches the frozen `_FinancialRunState.node_populations` field (`:127`) through +`_issue_run` and `_node_population_seals` (`:500-515`), and `_pure_run` +(`:676-691`) re-stamps every retained population on **every** `checked_view()`. + +### 3b. Why the snapshot cannot be addressed to the store + +The obvious move — write each snapshot to the content store and keep the key — +is refused by the store's own normalisation, in three independent ways, and this +design records them rather than discovering them later: + +* `store.py:456` zeroes the values beneath a null mask (`values[mask] = 0`), + while `survey_atomic_geography._population_stamp` (`:230`) deliberately folds + the physical storage parts *under* the mask — its docstring says so at + `:231-237`. A store round trip therefore changes every stamp. +* `store.py:619` refuses a `MultiIndex` outright, and + `test_graph_executor.py:4254-4255` builds one into the observed frame. +* `store.py:486-487` refuses `CategoricalDtype` and `DatetimeTZDtype`, and + `test_graph_executor.py:4262` observes a categorical column. + +There is also no store API that accepts a `Population` (only `put_frame` for a +`Frame` and `put_bytes` for bytes), and `DataFrame.attrs` — asserted preserved +at `test_graph_executor.py:4320-4321` — is not persisted by `_write_frame`. +**A store-backed snapshot is not a drop-in for the pickle.** Anything that +spills a snapshot must spill the same pickle bytes the executor already trusts +for detachment. + +### 3c. Object identity is pinned, so re-materialisation is not free + +`result.financial_population is observed[final_node]` +(`graph_atomic_survey_financial.py:1897`) is an identity check, not an +equality check, and the completion host has two more of its own. A policy that +dropped a population and rebuilt it from a spill would produce a different +object and fail these. So the retention policy is **declared-consumer**: the +nodes whose objects a caller holds are named, retained in memory, and the rest +may be spilled. For the base financial run those are the final node, +`financial.ATTACH_NODE`, and — when the tax rebase is enabled — the property +population. + +### 3d. What still needs every population, and the question it raises + +Two loops need the whole `Population` object, not a seal: +`atomic.same_replayed_population(population, observed[node_id])` at `:1796` and +`:1948`. They compare the executor's observation against the runner's own +independent replay, field by field and byte by byte +(`survey_population_replay.py:132`, `:179`). The replay cannot be built until +`run_graph` has returned, because it reads the manifest's loaded artifacts, so +between the observation and the comparison **nineteen detached populations must +exist somewhere** — RAM (~290 GiB at full source) or disk (~285 GiB). There is +no third place, and choosing to replace the object comparison with a content +seal is a change to a verification contract, not an optimisation. §5 puts that +to the owner as a question. + +### 3e. The seal: a vectorised per-column pass, proven equal + +`survey_population_preparation._frame_identity` is the only remaining per-cell +frame walker in the US runtime. Its cost was literally + +```python +for value in series: + digest.update(_frame_cell_encode(value)) + digest.update(b"\n") +``` + +— one `_frame_cell_encode` call and two `digest.update` calls per cell, over a +quantity that is a pure function of the cell values plus dtype metadata. (The +two neighbours that look similar are already vectorised and are left alone: +`asec_current_money_source._series_digest` (`:98`) dumps `_data`/`_mask` bytes +for numeric dtypes and caches string tokens, and +`asec_2024_native_population._frame_identity` (`:184`) delegates to it.) + +The replacement builds, per column, **the exact byte string that loop would +have fed the digest**, and updates the digest once: + +| dtype | how | why it is exactly equal | +|---|---|---| +| `float64` | `_float_cells`: a fixed-offset `(n, 37)` uint8 matrix with a per-position keep mask, cut down with one boolean gather | the record is `["float","` + optional `-` + `0x1.`/`0x0.` + 13 mantissa nibbles + `p` + sign + up to 4 exponent digits + `"]`; the mask deletes the parts a value does not spell, so `float.hex()`'s variable spellings come out of one padded matrix | +| `int8`…`uint64` | `_integer_cells`: `"\n".join(map(str, values.tolist()))` | `str` on a Python int **is** `int.__repr__`, which is the fast path's encoder; every integer-dtype value lies inside `-2**63 <= v < 2**64` and spells ≤ 20 bytes, so neither `_encode` nor `PAYLOAD_LIMIT` is reachable | +| `Int8`…`UInt64` | `_nullable_integer_cells`: the same, with `pd.NA` positions overwritten by `null` | `_cell` maps `pd.NA` to `None` and `_frame_cell_encode` spells `None` as `null`; unsigned widths keep an unsigned carrier so `UInt64` above `2**63 - 1` still spells | +| `bool`, `boolean` | `_boolean_cells`: index into the three spellings | three possible cells | +| `string` | `_string_cells`: `pd.factorize`, then `_frame_cell_encode` once per distinct value | `StringDtype` compares and hashes exactly, so grouping equal values cannot merge two whose encodings differ | +| anything else | the cell-at-a-time walk, unchanged | `object` and `category` deliberately do **not** qualify: `1`, `True` and `1.0` are equal and hash alike there while `_frame_cell_encode` spells them three different ways | + +`_cell` refuses a non-finite cell with `FRAME_NONFINITE`; the float branch +raises the same code for the column rather than at the first offending cell, +and the digest is discarded either way. + +**The equality is proved against an independent oracle that already existed.** +`packages/microcosm-build/tests/test_us_survey_frame_identity_encoding.py` +carries `_legacy_frame_preimage`, a verbatim predecessor traversal, and +`_assert_exact_frame_preimage`, which records every `digest.update` call through +a patched `hashlib.sha256` and asserts the **concatenated byte preimage** — not +merely the digest — equals the oracle's. That is a byte-for-byte proof, and it +is joined by an exhaustive float proof and a per-dtype sweep this lane adds. + +## 4. What `run_graph(population_retention=...)` on main would need + +`#901` recorded that the shared executor has no retention mode, and that is +still true at this head: there is no eviction, pruning or population dropping in +`executor.py` or `store.py`, and the only occurrences of the name +`population_retention` in the tree are historical notes in `experiments/` and +`out.md` describing a dropped branch. The executor's own `populations` dict is +keyed by **version**, not node, and a `StructuralDelta.NONE` node overwrites its +entry (`:2769`), so the executor itself already retains one population per +version; the per-node retention is entirely the observer's. + +A `population_retention` keyword would have to carry these, all of them +established above: + +1. **A mode that skips the detachment, not just the retention.** The cost is + `pickle.dumps` + `pickle.loads` per node (`:373`). An observer that only + seals needs no detached copy at all, because sealing is read-only. The + executor would hand it the live population and the mode would be the + observer's declaration that it will not retain or mutate it. + `test_graph_executor.py:4368-4374` already pins the converse — with no + observer, no snapshot may be allocated — so the shape of that test is the + shape of the new one. +2. **A declared-consumer roster.** `_observer_snapshot`'s detachment exists so + the observer may mutate what it keeps (`test_graph_executor.py:4183-4248` + asserts a mutated snapshot moves no node key, no manifest and nothing in the + store). A retention mode that hands out a read-only view instead would make + that mutation raise — *more* closed, but a behaviour change to a documented + contract in `microcosm-graph`, so it belongs to a main-only change with its + own test rather than to a country runner. +3. **Invocation invariants that do not move.** Exactly once per reached node, in + `compiled.order`, on cache hits and cold execution alike + (`test_graph_executor.py:4153`, `:4166-4167`); an observer's own exception + still refuses the run with the observer's exception type (`:4172-4179`); + `ATOMIC_OBSERVER_ROSTER` and `COMPLETION_OBSERVER_ROSTER` still refuse any + gap or reorder. +4. **A statement about `_pure_run`.** `state.node_populations` is re-stamped on + every `checked_view()`, so a retention mode that drops populations must say + what `FINANCIAL_NODE_POPULATION_CHANGED` means afterwards. Today + `_node_population_seals` already returns `()` when there are no observations + and no completion boundary (`:502-503`), which is the existing shape of + "nothing retained, nothing to re-stamp". + +Item 1 is the whole memory wall and most of the avoidable CPU. It is main-only, +it lands in `microcosm-graph`, and the executor hunks of PR #938 are its +neighbours. + +## 5. Open, for the owner + +1. **Columnar format for the receipt bodies.** This design uses the store's own + existing shape — a `header.json` payload table plus content-addressed bodies, + as `put_frame` already writes `.npy` files — and adds no dependency. A true + Arrow/parquet body would need `pyarrow` in the workspace and would change the + canonical byte stream, and therefore `preparation_sha256` and every node key. +2. **Whether the retention policy is a run option or the only behaviour.** §3d + shows that nineteen detached populations must exist between the observation + and the replay comparison unless the object comparison is replaced by a + content seal. That is a verification-contract decision. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py index 3e8831da54..2200d18186 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py @@ -638,6 +638,195 @@ def _frame_cell_encode(value, maximum=MAX_PAYLOAD_BYTES): return encoded +_HEX_DIGITS = np.frombuffer(b"0123456789abcdef", dtype=np.uint8) +# ``["float","`` -0x1. <13 mantissa nibbles> p +<=4 exponent digits> "]\n +_FLOAT_WIDTH = 37 +_FLOAT_HEAD = np.frombuffer(b'["float","', dtype=np.uint8) +_FLOAT_LEAD = np.stack( + ( + np.frombuffer(b"0x1.", dtype=np.uint8), + np.frombuffer(b"0x0.", dtype=np.uint8), + ) +) +_FLOAT_TAIL = np.frombuffer(b'"]\n', dtype=np.uint8) +_FLOAT_NULL = np.frombuffer(b"null\n", dtype=np.uint8) +_FLOAT_EXPONENT_DIGITS = 4 +_MANTISSA_SHIFTS = np.arange(48, -1, -4, dtype=np.uint64) +_BOOL_CELLS = np.array([b"false\n", b"true\n", b"null\n"], dtype=object) +_INTEGER_DTYPES = frozenset( + ("int8", "int16", "int32", "int64", "uint8", "uint16", "uint32", "uint64") +) +_NULLABLE_INTEGER_DTYPES = frozenset( + ("Int8", "Int16", "Int32", "Int64", "UInt8", "UInt16", "UInt32", "UInt64") +) + + +def _float_cells(values): + """``_frame_cell_encode`` + a newline for one float64 column, per column. + + The record is laid out at fixed offsets with a per-position keep mask, so + the variable-length spellings ``float.hex()`` produces are cut out of one + padded matrix rather than assembled a cell at a time: + + 0..9 ``["float","`` always + 10 ``-`` iff the sign bit is set + 11..14 ``0x1.`` / ``0x0.`` normal / subnormal-or-zero + 15..27 13 mantissa nibbles all 13, or only column 15 (``0``) for zero + 28 ``p`` always + 29 ``+`` / ``-`` sign of the unbiased exponent + 30..33 4 exponent digits only the last significant ones + 34..36 ``"]`` and newline always + + A NaN cell spells ``null`` instead, which is what ``_cell`` makes of it. + ``_cell`` refuses a non-finite cell, so an infinity refuses here too, with + the same code; it is raised for the column rather than at the first + offending cell, and the digest is discarded either way. + """ + _require(not np.isinf(values).any(), "FRAME_NONFINITE") + count = values.shape[0] + if count == 0: + return b"" + bits = values.view(np.uint64) + missing = np.isnan(values) + sign = (bits >> np.uint64(63)).astype(bool) + biased = ((bits >> np.uint64(52)) & np.uint64(0x7FF)).astype(np.int64) + mantissa = bits & np.uint64(0xFFFFFFFFFFFFF) + subnormal = biased == 0 + zero = subnormal & (mantissa == 0) + exponent = np.where(subnormal, np.int64(-1022), biased - 1023) + exponent = np.where(zero, np.int64(0), exponent) + + cells = np.zeros((count, _FLOAT_WIDTH), dtype=np.uint8) + keep = np.zeros((count, _FLOAT_WIDTH), dtype=bool) + cells[:, 0:10] = _FLOAT_HEAD + cells[:, 10] = np.uint8(0x2D) + keep[:, 10] = sign + cells[:, 11:15] = _FLOAT_LEAD[subnormal.astype(np.intp)] + cells[:, 15:28] = _HEX_DIGITS[ + ((mantissa[:, None] >> _MANTISSA_SHIFTS[None, :]) & np.uint64(0xF)).astype( + np.uint8 + ) + ] + cells[zero, 15] = np.uint8(0x30) + cells[:, 28] = np.uint8(0x70) + cells[:, 29] = np.where(exponent < 0, np.uint8(0x2D), np.uint8(0x2B)) + magnitude = np.abs(exponent) + for position in range(_FLOAT_EXPONENT_DIGITS): + power = 10 ** (_FLOAT_EXPONENT_DIGITS - 1 - position) + cells[:, 30 + position] = np.uint8(0x30) + ( + magnitude // power % 10 + ).astype(np.uint8) + cells[:, 34:37] = _FLOAT_TAIL + + keep[:, 0:10] = True + keep[:, 11:28] = True + keep[zero, 16:28] = False + keep[:, 28:30] = True + significant = np.where( + magnitude >= 1000, + 4, + np.where(magnitude >= 100, 3, np.where(magnitude >= 10, 2, 1)), + ) + keep[:, 30:34] = ( + np.arange(_FLOAT_EXPONENT_DIGITS)[None, :] + >= (_FLOAT_EXPONENT_DIGITS - significant)[:, None] + ) + keep[:, 34:37] = True + if missing.any(): + cells[missing, 0:5] = _FLOAT_NULL + keep[missing, :] = False + keep[missing, 0:5] = True + return cells[keep].tobytes() + + +def _integer_cells(values): + """One integer column's cells. ``str`` on a Python int is ``int.__repr__``. + + Every value an integer dtype can hold lies inside the fast path's + ``-2**63 <= value < 2**64`` window and spells at most twenty ASCII bytes, + so neither ``_encode`` nor ``PAYLOAD_LIMIT`` is reachable from here. + """ + if values.shape[0] == 0: + return b"" + return ("\n".join(map(str, values.tolist())) + "\n").encode("ascii") + + +def _nullable_integer_cells(series): + """One nullable-integer column's cells; ``pd.NA`` is ``_cell``'s ``None``. + + The unsigned widths keep an unsigned carrier: ``UInt64`` holds values above + ``2**63 - 1`` that no signed carrier can spell, and the fast path admits + every integer below ``2**64``. + """ + if len(series) == 0: + return b"" + carrier = "uint64" if str(series.dtype).startswith("U") else "int64" + present = series.to_numpy(dtype=carrier, na_value=0) + spelled = list(map(str, present.tolist())) + for position in np.flatnonzero(np.asarray(series.isna())).tolist(): + spelled[position] = "null" + return ("\n".join(spelled) + "\n").encode("ascii") + + +def _boolean_cells(series): + """One boolean column's cells: three spellings, selected by index.""" + if len(series) == 0: + return b"" + missing = np.asarray(series.isna()) + present = series.to_numpy(dtype=bool, na_value=False) + codes = np.where(missing, np.intp(2), present.astype(np.intp)) + return b"".join(_BOOL_CELLS[codes].tolist()) + + +def _string_cells(series): + """One string column's cells, encoding each distinct value once. + + ``StringDtype`` compares and hashes exactly, so grouping equal values + cannot merge two whose encodings differ. ``object`` and ``category`` + deliberately do not qualify: ``1``, ``True`` and ``1.0`` are equal and hash + alike there while ``_frame_cell_encode`` spells them three different ways, + so those dtypes keep the cell-at-a-time walk. + """ + if len(series) == 0: + return b"" + codes, uniques = pd.factorize(series, use_na_sentinel=False) + encoded = np.empty(len(uniques), dtype=object) + for position, value in enumerate(uniques): + encoded[position] = _frame_cell_encode(value) + b"\n" + return b"".join(encoded[codes].tolist()) + + +def _cells_blob(series): + """The exact bytes a cell-at-a-time walk over ``series`` would digest. + + Every branch below is proved equal to that walk, byte for byte, in + ``test_us_survey_population_frame_seal.py``; a dtype with no proved branch + takes the walk itself rather than a guess. + """ + name = str(series.dtype) + if name == "float64": + return _float_cells(series.to_numpy(dtype=np.float64, copy=False)) + if name in _INTEGER_DTYPES: + return _integer_cells(series.to_numpy(copy=False)) + if name in _NULLABLE_INTEGER_DTYPES: + return _nullable_integer_cells(series) + if name in ("bool", "boolean"): + return _boolean_cells(series) + if name == "string": + return _string_cells(series) + return b"".join(_frame_cell_encode(value) + b"\n" for value in series) + + +def _index_blob(index): + """The same, for an axis, which the walk reads exactly as it reads a column.""" + name = str(index.dtype) + if name in _INTEGER_DTYPES: + return _integer_cells(index.to_numpy(copy=False)) + if name == "float64": + return _float_cells(index.to_numpy(dtype=np.float64, copy=False)) + return b"".join(_frame_cell_encode(value) + b"\n" for value in index) + + def _frame_identity(frame): _require( isinstance(frame, Frame) and frame.schema == US_SCHEMA and not frame.links, @@ -649,10 +838,6 @@ def update(value): digest.update(_encode(value)) digest.update(b"\n") - def update_cell(value): - digest.update(_frame_cell_encode(value)) - digest.update(b"\n") - update([list(frame.entities), list(frame.weighted_entities)]) for entity in frame.entities: table = frame.table(entity) @@ -672,8 +857,7 @@ def update_cell(value): list(table.index.names), ] ) - for value in table.index: - update_cell(value) + digest.update(_index_blob(table.index)) for column in table: series = table[column] dtype = series.dtype @@ -685,8 +869,7 @@ def update_cell(value): str(getattr(dtype, "na_value", "")), ] ) - for value in series: - update_cell(value) + digest.update(_cells_blob(series)) update( [ str(frame.strata.dtype), @@ -697,10 +880,8 @@ def update_cell(value): frame.strata.name, ] ) - for value in frame.strata.index: - update_cell(value) - for value in frame.strata: - update_cell(value) + digest.update(_index_blob(frame.strata.index)) + digest.update(_cells_blob(frame.strata)) for entity in frame.weighted_entities: weights = frame.weights_for(entity) update( From 5307249b359eff70285dc98e87064ecbb40fe9c8 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:01:30 -0400 Subject: [PATCH 03/31] Carry the whole-roster receipts as bounded segments instead of one bounded blob Four receipts in the US native build carried one record per source household or per person inside a single 64 MiB-bounded encode: the preparation document, its two row budgets, the allocation payload, and the wage projection. Measured off the pilot's own committed artifact, the preparation receipt is 39,242 + 692.4375 x households bytes, so it refused at 96,860 households -- 6.10% of the 1,587,376-household source -- and a full-source document is 1.02 GiB. The canonical byte stream does not move. _chunks already defines it independently of its materialization, and sha256 is a streaming hash over those same chunks in the same order, so _sha(_encode(x)) == _digest(x) and the new transport carries the stream the single encode carried. preparation_sha256 and allocation_sha256 are unchanged in derivation and in value. What moved is where the bytes accumulate. _roster_segments closes a segment before it would pass MAX_SEGMENT_BYTES -- the old whole-document ceiling, unchanged, with PAYLOAD_LIMIT still on each one -- and bounds the total separately under ROSTER_LIMIT. The preparation receipt also spills its segments beside a header naming them, the shape ContentStore.put_frame already uses. The allocation payload segments in memory, because its kernel has no filesystem path. Two refusals narrow rather than disappear, and the design note and the tests both say so: PAYLOAD_LIMIT and ALLOCATION_LIMIT now mean "one token, or one row plus its tail, larger than one segment", which is what a segmented stream can still reach. The five ceilings and the five refusals all still fail closed, and none of them binds below full source. Co-Authored-By: Claude Opus 5 --- docs/us-native-scale-transport.md | 41 ++++- .../us_runtime/graph_survey_population.py | 47 ++++- .../survey_population_preparation.py | 135 +++++++++++++- .../tests/test_us_graph_survey_population.py | 66 ++++++- .../test_us_survey_population_preparation.py | 166 +++++++++++++++++- 5 files changed, 432 insertions(+), 23 deletions(-) diff --git a/docs/us-native-scale-transport.md b/docs/us-native-scale-transport.md index 034c3a3122..08bc159235 100644 --- a/docs/us-native-scale-transport.md +++ b/docs/us-native-scale-transport.md @@ -67,11 +67,13 @@ committed artifact, or from the allocation payload's own measured 501,597 B at | 4 | `_plan_document` row budget | `ORIGIN_LIMIT` | 270,284 | 17.03% | | 5 | allocation structural pre-check `ALLOCATION_MAX_BYTES // 128` | `ALLOCATION_LIMIT` | 524,288 | 33.03% | -A 1/10 build is 158,737 source households, so **#1 and #2 both bind and #3 does -not**; a full build needs #1 through #4. #5 bounds the intermediate maps -`allocation_instructions` builds, not the payload, and its implied 127-byte row -budget is below the payload's 220-byte structural overhead alone, so it can -never be the binding payload constraint. +A 1/10 build is 158,737 source households, so **#1 and #2 both bind and #3 +does not**; a full build needs all five, because 524,288 is below the source's +own 1,587,376. #5 bounds the intermediate maps `allocation_instructions` builds +rather than the payload — its implied 127-byte row budget is below the payload's +220-byte structural overhead alone, so it can never be the binding *payload* +constraint — but it does refuse a full-source selection, so its ceiling moves +with the others and keeps its code. **The store is not the constraint.** `ContentStore.put_bytes` (`store.py:984`) and `put_frame` (`:919`) carry no byte cap, and `put_frame` @@ -174,9 +176,26 @@ in-memory half of §2b: bounded segments in a list, each guarded by materialised once. `allocation_sha256` (`:441`) is unchanged in derivation and in value. -Ceiling #5, the `// 128` structural pre-check, keeps its constant: it bounds -`allocation_instructions`' intermediate maps and 524,288 households is above -full source at every fraction this lane runs. +Ceiling #5, the `// 128` structural pre-check, moves to +`ALLOCATION_ROSTER_BYTES // 128` = 33,554,432 households. It bounds +`allocation_instructions`' intermediate maps, and its old 524,288 is **below** +the source's 1,587,376, so a full-source selection refused there too. The code +is unchanged. + +**One refusal narrows, and this design says so rather than leaving it to be +found.** `ALLOCATION_LIMIT` fired when the *whole* payload passed 64 MiB. Under +a segmented stream that condition no longer arises: a segment closes instead. It +still fires, with the same code and the same expression, on the condition a +segmented stream can still reach — **one instruction row plus the reserved tail +larger than one whole segment**. `test_us_graph_survey_population.py` drives both +that refusal and the new total ceiling, and asserts the payload's bytes are +unchanged at four different segment sizes down to the smallest one that holds a +row. The same narrowing applies to the preparation receipt's `PAYLOAD_LIMIT`: +`_chunks` yields one JSON token at a time and a segment never splits one, so the +refusal now means "one token larger than one segment", and because +`_check_scalars` has already bounded every string to 1 MiB the shipped 64 MiB +segment always holds any token. Both are tested at a segment size below the +longest token. ### 2e. Size law after the change @@ -192,6 +211,12 @@ full source: | `_origins` rows | 150,916 hh (9.51%) | 9,658,000 hh | | allocation payload | 211,924 hh (13.35%) | 13,563,000 hh | | `_plan_document` rows | 270,284 hh (17.03%) | 17,298,000 hh | +| allocation map pre-check | 524,288 hh (33.03%) | 33,554,432 hh | + +A fourth whole-roster receipt the cost attribution did not name is in the same +module: `_current_survey_wage_projection` builds one row per selected wage +earner and encoded it the same way. It takes the same transport. It does not +bind at 1/10 and does bind at full source. ## 3. Retention and the frame seal diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/graph_survey_population.py b/packages/microcosm-build/src/microcosm/build/us_runtime/graph_survey_population.py index a262f2682c..1b7e6f2b7f 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/graph_survey_population.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/graph_survey_population.py @@ -60,6 +60,12 @@ ALLOCATION_TYPE = ArtifactType("microcosm.us.survey_population_allocation", 1) PREPARATION_MAX_BYTES = 64 * 1024**2 ALLOCATION_MAX_BYTES = 64 * 1024**2 +# The allocation payload carries one instruction document per selected +# household, measured at 316.665 B each, so a single bounded accumulation +# refuses at 13.35% of the US source (docs/us-native-scale-transport.md §1). +# ALLOCATION_MAX_BYTES stays the per-segment ceiling, unchanged; the total is a +# separate, explicit resource ceiling. +ALLOCATION_ROSTER_BYTES = 64 * ALLOCATION_MAX_BYTES STAGE = "authenticated_survey_population_v1" STAGE_DEPENDENCIES = ( "numpy", @@ -149,8 +155,11 @@ def allocation_instructions(plan, household_origins): _require(type(household_origins) in (list, tuple), "HOUSEHOLD_ORIGINS") _require(len(household_origins) == len(plan.selected), "ORIGIN_COUNT") # A prospective record bound precedes the two maps and output tuple. The - # streaming transport encoder enforces its exact bound separately. - _require(len(plan.selected) <= ALLOCATION_MAX_BYTES // 128, "ALLOCATION_LIMIT") + # streaming transport encoder enforces its exact bound separately. The + # implied 128-byte row budget is below the payload's 220-byte structural + # overhead alone, so this never binds the payload; it bounds the two maps + # below, and its ceiling is the roster one. + _require(len(plan.selected) <= ALLOCATION_ROSTER_BYTES // 128, "ALLOCATION_LIMIT") selected = {} for row in plan.selected: _require( @@ -344,7 +353,20 @@ def pair(value): def _allocation_payload( instructions, *, preparation_sha256, input_context, output_context ): - """Stream rows individually; never construct a selected-population JSON tree.""" + """Stream rows individually; never construct a selected-population JSON tree. + + The bytes are unchanged, and so is ``allocation_sha256``: the same rows in + the same order with the same separators and the same tail. What changed is + where they accumulate. A segment closes before it would pass + ``ALLOCATION_MAX_BYTES`` -- the ceiling the single bytearray used, with the + same refusal expression and the same code -- so no accumulation here is + larger than one whole payload used to be, and the total is bounded + separately by ``ALLOCATION_ROSTER_LIMIT``. + + This kernel has no filesystem path and must not write to the store, so the + segments are held rather than spilled and joined once. Removing that last + copy needs a streaming artifact channel in ``microcosm-graph``. + """ metadata = { "protocol": "microcosm.us.survey-population-allocation.v1", "preparation_sha256": preparation_sha256, @@ -354,19 +376,36 @@ def _allocation_payload( "release_eligible": False, } tail = b"]," + _bounded_json(metadata, 4096)[1:] + segments = [] output = bytearray(b'{"households":[') + total = len(output) for index, row in enumerate(instructions): raw = _bounded_json(_instruction_document(row), 4096) separator = b"," if index else b"" + if (segments or output) and len(output) + len(separator) + len(raw) + len( + tail + ) > (ALLOCATION_MAX_BYTES): + segments.append(bytes(output)) + output = bytearray() _require( len(output) + len(separator) + len(raw) + len(tail) <= ALLOCATION_MAX_BYTES, "ALLOCATION_LIMIT", ) + _require( + total + len(separator) + len(raw) + len(tail) <= ALLOCATION_ROSTER_BYTES, + "ALLOCATION_ROSTER_LIMIT", + ) output.extend(separator) output.extend(raw) + total += len(separator) + len(raw) _require(len(output) + len(tail) <= ALLOCATION_MAX_BYTES, "ALLOCATION_LIMIT") output.extend(tail) - return bytes(output) + total += len(tail) + segments.append(bytes(output)) + _require(total <= ALLOCATION_ROSTER_BYTES, "ALLOCATION_ROSTER_LIMIT") + payload = b"".join(segments) + _require(len(payload) == total, "ALLOCATION_ROSTER_LIMIT") + return payload def _verify_allocation_view(frame, instructions): diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py index 2200d18186..16325edd52 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py @@ -48,6 +48,15 @@ SOURCE_CODEC = "us-survey-population-source-v1" MAX_REQUEST_BYTES = 4096 MAX_PAYLOAD_BYTES = 64 * 1024**2 +# One whole-roster receipt carries one record per source household and per +# person, so a full-source document is 1.02 GiB (docs/us-native-scale-transport.md +# §1) and a single bounded encode refuses at 6.1% of the source. The transport +# below keeps every accumulation inside the value the old ceiling allowed -- +# MAX_SEGMENT_BYTES is that ceiling, unchanged -- and adds one explicit resource +# ceiling for the total, which is not a transport shape. +MAX_SEGMENT_BYTES = MAX_PAYLOAD_BYTES +MAX_ROSTER_BYTES = 64 * MAX_SEGMENT_BYTES +ROSTER_PROTOCOL = "microcosm.us.survey-population-roster-transport.v1" _MAX_SCALAR_BYTES = 1024**2 _FRAME_FAST_STRING_CHARS = 4096 _SOURCE_ROSTER = ( @@ -117,6 +126,107 @@ def _digest(value): return digest.hexdigest() +def _roster_segments(value, *, segment=MAX_SEGMENT_BYTES, maximum=MAX_ROSTER_BYTES): + """Encode one whole-roster document as bounded, content-addressed segments. + + ``_encode`` accumulates the complete canonical stream before it enforces + anything, so its bound is a ceiling on the document rather than on the + process's allocation. This walks the same ``_chunks`` stream, closes a + segment whenever the next chunk would carry it past ``segment``, and keeps + ``PAYLOAD_LIMIT`` on each segment at the same number the single encode used. + The total is bounded separately, by ``ROSTER_LIMIT``. + + The running digest is over the chunks in order, so it equals + ``_digest(value)`` exactly -- and therefore equals ``_sha(_encode(value))`` + for every document ``_encode`` accepts, because sha256 is a streaming hash + over those same bytes. The canonical byte stream is unchanged; only its + materialization is bounded. + """ + digest = hashlib.sha256() + segments, table, current, total = [], [], bytearray(), 0 + + def close(): + raw = bytes(current) + digest.update(raw) + table.append((_sha(raw), len(raw))) + segments.append(raw) + current.clear() + + for chunk in _chunks(value): + encoded = chunk.encode("utf-8") + if current and len(current) + len(encoded) > segment: + close() + # A single chunk is one JSON token, and _check_scalars has already + # bounded every string, so this refuses only a token larger than one + # whole segment -- the same refusal, at the same number. + _require(len(current) + len(encoded) <= segment, "PAYLOAD_LIMIT") + _require(total + len(encoded) <= maximum, "ROSTER_LIMIT") + current.extend(encoded) + total += len(encoded) + if current or not segments: + close() + return segments, tuple(table), digest.hexdigest(), total + + +def _spill_roster(root, name, segments, table, digest, total): + """Write the segments beside a header naming them, as the store's own shape. + + ``ContentStore.put_frame`` already writes content-addressed bodies under a + ``meta.json`` payload table of sha256 and size per file. This is that shape + for a receipt: one ``.segment`` per segment and one ``header.json`` + naming the whole-stream digest, the size and the ordered segment table, so + the receipt has a re-verifiable on-disk form that does not depend on the + graph store. + """ + _require(type(name) is str and name.isidentifier(), "ROSTER_NAME") + directory = root / name + directory.mkdir(parents=True, exist_ok=True) + for raw, (sha, size) in zip(segments, table, strict=True): + path = directory / (sha + ".segment") + if not path.exists(): + path.write_bytes(raw) + stats = path.lstat() + _require( + not stat.S_ISLNK(stats.st_mode) and stats.st_size == size, + "ROSTER_SEGMENT_CHANGED", + ) + header = { + "protocol": ROSTER_PROTOCOL, + "name": name, + "sha256": digest, + "size": total, + "segments": [[sha, size] for sha, size in table], + } + (directory / "header.json").write_bytes(_encode(header)) + return header + + +def _roster_payload( + value, *, spill=None, name=None, segment=MAX_SEGMENT_BYTES, maximum=MAX_ROSTER_BYTES +): + """One whole-roster receipt's bytes, and the header that names its segments. + + The bytes are identical to ``_encode(value)`` wherever ``_encode`` accepts + the document, and exist wherever it does not. They are materialized once, + because ``KernelResult.artifacts`` is a mapping of ``bytes`` and + ``ContentStore.put_bytes`` takes whole bytes; removing that copy needs a + streaming artifact channel in ``microcosm-graph``, which is not this + module's to change. + """ + segments, table, digest, total = _roster_segments( + value, segment=segment, maximum=maximum + ) + header = None + if spill is not None: + header = _spill_roster(spill, name, segments, table, digest, total) + for raw, (sha, size) in zip(segments, table, strict=True): + _require(len(raw) == size and _sha(raw) == sha, "ROSTER_SEGMENT_CHANGED") + payload = b"".join(segments) + _require(len(payload) == total <= maximum, "ROSTER_LIMIT") + _require(_sha(payload) == digest, "ROSTER_DIGEST") + return payload, header + + def _catalogue_fast_path(value): """Recognize immutable, bounded raw records without changing admission.""" if type(value) is not tuple or len(value) != 2: @@ -537,6 +647,9 @@ def _live(): SOURCE_CODEC, MAX_REQUEST_BYTES, MAX_PAYLOAD_BYTES, + MAX_SEGMENT_BYTES, + MAX_ROSTER_BYTES, + ROSTER_PROTOCOL, _MAX_SCALAR_BYTES, _FRAME_FAST_STRING_CHARS, _SOURCE_ROSTER, @@ -713,9 +826,9 @@ def _float_cells(values): magnitude = np.abs(exponent) for position in range(_FLOAT_EXPONENT_DIGITS): power = 10 ** (_FLOAT_EXPONENT_DIGITS - 1 - position) - cells[:, 30 + position] = np.uint8(0x30) + ( - magnitude // power % 10 - ).astype(np.uint8) + cells[:, 30 + position] = np.uint8(0x30) + (magnitude // power % 10).astype( + np.uint8 + ) cells[:, 34:37] = _FLOAT_TAIL keep[:, 0:10] = True @@ -995,8 +1108,11 @@ def _verify_normalized_copy(original: Frame, normalized: Frame): def _bounded_append(rows, row, budget): + # A bound on the Python row list, not on a transport: one record per source + # household or per person, so a full-source roster is 1.02 GiB. The code and + # the per-row encode are unchanged; the ceiling is the explicit resource one. size = len(_encode(row)) + 1 - _require(budget[0] + size <= MAX_PAYLOAD_BYTES, "ORIGIN_LIMIT") + _require(budget[0] + size <= MAX_ROSTER_BYTES, "ORIGIN_LIMIT") budget[0] += size rows.append(row) @@ -1743,7 +1859,7 @@ def prepare_authenticated_survey_population( try: _require(type(candidate) is bytes or candidate is None, "CANDIDATE_TYPE") _require( - candidate is None or len(candidate) <= MAX_PAYLOAD_BYTES, "CANDIDATE_LIMIT" + candidate is None or len(candidate) <= MAX_ROSTER_BYTES, "CANDIDATE_LIMIT" ) root, request, request_bytes, requested_fraction, requested_seed = _request( source_dir @@ -1857,7 +1973,7 @@ def prepare_authenticated_survey_population( }, } identity = _frame_identity(frame) - payload = _encode( + payload, _ = _roster_payload( { "protocol": PROTOCOL, "request": request, @@ -1881,7 +1997,9 @@ def prepare_authenticated_survey_population( "frame_sha256": identity, "context_sha256": _sha(context), "release_eligible": False, - } + }, + spill=snapshots, + name="preparation", ) nested = _nested_seals((acs, asec), (actual_acs, actual_asec)) acs_owned = acs_catalogue._lookup(acs) @@ -2017,7 +2135,8 @@ def _current_survey_wage_projection(preparation, entry): ], "rows": rows, } - payload = _encode(document) + # One row per selected wage earner, so this is a whole-roster receipt too. + payload, _ = _roster_payload(document) # ready() may perform I/O. Recheck the actual retained entries and all pure # owner seals after it; this does not reread the national catalogues. _pure_final(state) diff --git a/packages/microcosm-build/tests/test_us_graph_survey_population.py b/packages/microcosm-build/tests/test_us_graph_survey_population.py index b3a603a881..f84d09ec1e 100644 --- a/packages/microcosm-build/tests/test_us_graph_survey_population.py +++ b/packages/microcosm-build/tests/test_us_graph_survey_population.py @@ -268,7 +268,20 @@ def test_allocation_plan_value_contract(change): def test_allocation_bound_precedes_map_construction(monkeypatch): plan, origins = plan_and_origins() - monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", 128) + monkeypatch.setattr(graph, "ALLOCATION_ROSTER_BYTES", 128) + with pytest.raises(graph.SurveyPopulationGraphError, match="ALLOCATION_LIMIT"): + graph.allocation_instructions(plan, origins) + + +def test_map_construction_bound_admits_a_full_source_selection(monkeypatch): + # 1,587,376 selected households is the US source; the pre-check's own + # ceiling admitted only 524,288 of them before the roster ceiling existed. + assert graph.ALLOCATION_ROSTER_BYTES // 128 > 1_587_376 + assert graph.ALLOCATION_MAX_BYTES // 128 < 1_587_376 + plan, origins = plan_and_origins() + monkeypatch.setattr(graph, "ALLOCATION_ROSTER_BYTES", len(plan.selected) * 128) + graph.allocation_instructions(plan, origins) + monkeypatch.setattr(graph, "ALLOCATION_ROSTER_BYTES", len(plan.selected) * 128 - 1) with pytest.raises(graph.SurveyPopulationGraphError, match="ALLOCATION_LIMIT"): graph.allocation_instructions(plan, origins) @@ -309,9 +322,58 @@ def test_allocation_transport_streams_canonical_exact_rows_and_refuses_before_ov assert document["households"][2]["importance_weight_float64_hex"] == "0x0.0p+0" monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", len(raw)) assert graph._allocation_payload(instructions, **kwargs) == raw - monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", len(raw) - 1) + # A segment must hold one row plus the tail it reserves; above that the + # payload segments and its bytes are unchanged, which is the transport's + # whole point. + rows = [ + graph._bounded_json(graph._instruction_document(r), 4096) for r in instructions + ] + tail = len(raw) - sum(len(r) for r in rows) - (len(rows) - 1) - 15 + smallest = max(len(r) for r in rows) + tail + 1 + assert smallest < len(raw) + for segment in (len(raw) - 1, smallest * 2, smallest): + monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", segment) + assert graph._allocation_payload(instructions, **kwargs) == raw + # ALLOCATION_LIMIT keeps its code and still fails closed, on the condition a + # segmented stream can still reach: one row plus the tail, in one segment. + monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", min(map(len, rows)) + tail - 1) with pytest.raises(graph.SurveyPopulationGraphError, match="ALLOCATION_LIMIT"): graph._allocation_payload(instructions, **kwargs) + # The total is the new, separately named ceiling, and it fails closed too. + monkeypatch.setattr(graph, "ALLOCATION_MAX_BYTES", 64 * 1024**2) + monkeypatch.setattr(graph, "ALLOCATION_ROSTER_BYTES", len(raw)) + assert graph._allocation_payload(instructions, **kwargs) == raw + monkeypatch.setattr(graph, "ALLOCATION_ROSTER_BYTES", len(raw) - 1) + with pytest.raises( + graph.SurveyPopulationGraphError, match="ALLOCATION_ROSTER_LIMIT" + ): + graph._allocation_payload(instructions, **kwargs) + + +def test_allocation_payload_is_byte_identical_to_the_unsegmented_stream(): + """The predecessor accumulation, verbatim, against the shipped transport.""" + plan, origins = plan_and_origins() + instructions = graph.allocation_instructions(plan, origins) + metadata = { + "protocol": "microcosm.us.survey-population-allocation.v1", + "preparation_sha256": "b" * 64, + "input_context_sha256": graph._sha(b"{}"), + "output_context_sha256": graph._sha(b"[]"), + "weight_transition": ["design", "importance"], + "release_eligible": False, + } + tail = b"]," + graph._bounded_json(metadata, 4096)[1:] + output = bytearray(b'{"households":[') + for index, row in enumerate(instructions): + output.extend(b"," if index else b"") + output.extend(graph._bounded_json(graph._instruction_document(row), 4096)) + output.extend(tail) + assert graph._allocation_payload( + instructions, + preparation_sha256="b" * 64, + input_context=b"{}", + output_context=b"[]", + ) == bytes(output) def authenticated_arguments(tmp_path, monkeypatch, **fixture_kwargs): diff --git a/packages/microcosm-build/tests/test_us_survey_population_preparation.py b/packages/microcosm-build/tests/test_us_survey_population_preparation.py index 9e7a685589..d40132b5b3 100644 --- a/packages/microcosm-build/tests/test_us_survey_population_preparation.py +++ b/packages/microcosm-build/tests/test_us_survey_population_preparation.py @@ -623,7 +623,9 @@ def replace_at_open(value, flags, *args, **kwargs): def test_transport_bound_precedes_append(): rows = [] - budget = [owner.MAX_PAYLOAD_BYTES - 1] + # The row budget bounds the Python row list, not a transport, so its + # ceiling is the explicit roster one; the refusal and its code are the same. + budget = [owner.MAX_ROSTER_BYTES - 1] with pytest.raises(owner.SurveyPopulationPreparationError, match="ORIGIN_LIMIT"): owner._bounded_append(rows, [1, "acs", 2], budget) assert rows == [] @@ -633,6 +635,168 @@ def test_transport_bound_precedes_append(): owner._encode({"nested": ["x" * (owner._MAX_SCALAR_BYTES + 1)]}) +def test_row_budget_admits_a_full_source_roster_and_still_refuses_above_it(): + # A full-source preparation roster is 1.02 GiB of rows; the ceiling admits + # it and refuses one byte past its own bound. + rows, budget = [], [1024**3] + owner._bounded_append(rows, [1, "acs", 2], budget) + assert rows == [[1, "acs", 2]] and budget[0] > 1024**3 + with pytest.raises(owner.SurveyPopulationPreparationError, match="ORIGIN_LIMIT"): + owner._bounded_append(rows, [1, "acs", 2], [owner.MAX_ROSTER_BYTES]) + + +_ROSTER_DOCUMENTS = [ + {}, + {"a": 1}, + {"protocol": "x", "rows": [[i, "acs", i * 7] for i in range(4000)]}, + {"nested": {"deep": [[["a" * 300]]], "zero": -0.0, "big": 10**40}}, + {"unicode": "\u00e9\u4e2d\U0001f600", "escapes": 'quote" slash\\ \x00\x1f'}, +] + + +@pytest.mark.parametrize("document", _ROSTER_DOCUMENTS) +def test_roster_transport_carries_the_single_encode_byte_stream(document): + payload, header = owner._roster_payload(document) + assert payload == owner._encode(document) + assert owner._sha(payload) == owner._digest(document) + assert header is None + + +@pytest.mark.parametrize("document", _ROSTER_DOCUMENTS) +@pytest.mark.parametrize("multiple", [1, 2, 3, 17, 4096]) +def test_roster_segments_reassemble_to_the_same_bytes_at_every_segment_size( + document, multiple +): + """A segment holds whole tokens, so the smallest useful one is the longest. + + ``_chunks`` yields one JSON token at a time and a segment never splits one, + so a segment smaller than the longest token refuses ``PAYLOAD_LIMIT`` -- + which is the next test. ``_check_scalars`` bounds every string to 1 MiB + before this runs, so the shipped 64 MiB segment always holds any token. + """ + expected = owner._encode(document) + longest = max( + (len(chunk.encode("utf-8")) for chunk in owner._chunks(document)), default=1 + ) + segment = longest * multiple + segments, table, digest, total = owner._roster_segments(document, segment=segment) + assert b"".join(segments) == expected + assert digest == owner._digest(document) and total == len(expected) + assert table == tuple((owner._sha(raw), len(raw)) for raw in segments) + assert all(0 < size <= segment for _sha256, size in table) + assert len(segments) >= -(-len(expected) // segment) + + +@pytest.mark.parametrize("document", _ROSTER_DOCUMENTS) +def test_a_segment_smaller_than_one_token_refuses_payload_limit(document): + longest = max( + (len(chunk.encode("utf-8")) for chunk in owner._chunks(document)), default=1 + ) + if longest == 1: + pytest.skip("this document has no token longer than one byte to split") + with pytest.raises(owner.SurveyPopulationPreparationError, match="PAYLOAD_LIMIT"): + owner._roster_segments(document, segment=longest - 1) + + +def test_roster_transport_exceeds_the_single_encode_ceiling(): + # The document the single bounded encode refuses is the document this + # transport carries: same bytes, same digest, one segment per 64 KiB here. + document = {"rows": [[i, "acs", i * 7] for i in range(20000)]} + encoded = owner._encode(document) + with pytest.raises(owner.SurveyPopulationPreparationError, match="PAYLOAD_LIMIT"): + owner._encode(document, len(encoded) - 1) + payload, _ = owner._roster_payload( + document, segment=65536, maximum=owner.MAX_ROSTER_BYTES + ) + assert payload == encoded + segments, _table, digest, _total = owner._roster_segments(document, segment=65536) + assert len(segments) > 1 and digest == owner._sha(encoded) + + +def test_roster_transport_keeps_every_refusal_it_touches(): + document = {"rows": [[i, "acs", i * 7] for i in range(4000)]} + # PAYLOAD_LIMIT still guards each segment: one token larger than a whole + # segment is the only thing that can overflow one, and it still refuses. + with pytest.raises(owner.SurveyPopulationPreparationError, match="PAYLOAD_LIMIT"): + owner._roster_payload({"k": "x" * 64}, segment=4) + # ROSTER_LIMIT is the new total ceiling, and it fails closed. + with pytest.raises(owner.SurveyPopulationPreparationError, match="ROSTER_LIMIT"): + owner._roster_payload(document, maximum=100) + with pytest.raises(owner.SurveyPopulationPreparationError, match="ROSTER_LIMIT"): + owner._roster_payload(document, segment=64, maximum=1000) + # The scalar and depth guards are the encoder's and are unmoved. + with pytest.raises(owner.SurveyPopulationPreparationError, match="SCALAR_LIMIT"): + owner._roster_payload({"s": "x" * (owner._MAX_SCALAR_BYTES + 1)}) + deep = value = [] + for _ in range(80): + nested = [] + value.append(nested) + value = nested + with pytest.raises(owner.SurveyPopulationPreparationError, match="VALUE_DEPTH"): + owner._roster_payload(deep) + + +def test_roster_spill_names_its_segments_and_verifies_them(tmp_path): + document = {"rows": [[i, "acs", i * 7] for i in range(4000)]} + payload, header = owner._roster_payload( + document, spill=tmp_path, name="preparation", segment=4096 + ) + assert payload == owner._encode(document) + assert header["protocol"] == owner.ROSTER_PROTOCOL + assert header["name"] == "preparation" + assert header["sha256"] == owner._digest(document) == owner._sha(payload) + assert header["size"] == len(payload) + assert sum(size for _sha256, size in header["segments"]) == len(payload) + directory = tmp_path / "preparation" + assert json.loads((directory / "header.json").read_bytes()) == header + rebuilt = b"" + for sha256, size in header["segments"]: + raw = (directory / (sha256 + ".segment")).read_bytes() + assert owner._sha(raw) == sha256 and len(raw) == size + rebuilt += raw + assert rebuilt == payload + # A spill name is a directory component, and only an identifier is accepted. + with pytest.raises(owner.SurveyPopulationPreparationError, match="ROSTER_NAME"): + owner._roster_payload(document, spill=tmp_path, name="../escape") + + +def test_roster_spill_refuses_a_replaced_segment(tmp_path): + document = {"rows": [[i, "acs", i * 7] for i in range(200)]} + _payload, header = owner._roster_payload( + document, spill=tmp_path, name="preparation" + ) + victim = tmp_path / "preparation" / (header["segments"][0][0] + ".segment") + victim.write_bytes(b"shorter") + with pytest.raises( + owner.SurveyPopulationPreparationError, match="ROSTER_SEGMENT_CHANGED" + ): + owner._roster_payload(document, spill=tmp_path, name="preparation") + + +def test_roster_spill_refuses_a_symlinked_segment(tmp_path): + document = {"rows": [[i, "acs", i * 7] for i in range(200)]} + _payload, header = owner._roster_payload( + document, spill=tmp_path, name="preparation" + ) + victim = tmp_path / "preparation" / (header["segments"][0][0] + ".segment") + target = tmp_path / "elsewhere" + target.write_bytes(victim.read_bytes()) + victim.unlink() + victim.symlink_to(target) + with pytest.raises( + owner.SurveyPopulationPreparationError, match="ROSTER_SEGMENT_CHANGED" + ): + owner._roster_payload(document, spill=tmp_path, name="preparation") + + +def test_candidate_bound_admits_a_full_source_receipt(tmp_path, monkeypatch): + arguments = fixture(tmp_path, monkeypatch) + with pytest.raises(owner.SurveyPopulationPreparationError, match="CANDIDATE_LIMIT"): + owner.prepare_authenticated_survey_population( + **arguments, candidate=b"x" * (owner.MAX_ROSTER_BYTES + 1) + ) + + def test_semantic_identity_includes_column_and_strata_indexes(tmp_path, monkeypatch): result = owner.prepare_authenticated_survey_population( **fixture(tmp_path, monkeypatch) From 52d1a2b5409a68595591d8d34a92a543b093039e Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:02:54 -0400 Subject: [PATCH 04/31] Record the baseline, including the one measurement the profiler could not reach The runner spends 1,712.62 s of its 2,045.91 s outside both node loops -- 83.7% -- so the per-node work is not what a build of this size costs. The retention slope is below the RSS trace's own sawtooth at 1/1000, which is what 0.31 GiB of snapshots against a 13.16 GB peak predicts. Co-Authored-By: Claude Opus 5 --- PROGRESS-native-scale-transport.md | 47 ++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/PROGRESS-native-scale-transport.md b/PROGRESS-native-scale-transport.md index 158df13472..f7d7ee9d4e 100644 --- a/PROGRESS-native-scale-transport.md +++ b/PROGRESS-native-scale-transport.md @@ -46,3 +46,50 @@ build above the old ceiling at 1/10. 3. Implement, with refusal tests, identity-equality proofs, replay proof. 4. 1/10 run above the old ceiling. 5. CI-shaped battery, draft PR. + +## Session log, 2026-09-17 + +### Baseline (requirement 1) — done, and it settles the profiler's open question + +One uncapped-by-node run of the 19-node financial graph at 1/1000, on the +branch point `5ff889814`, in the throwaway worktree +`~/PolicyEngine/_worktrees/microcosm-native-scale-baseline`, pid 70342 / pgid +70342 (session leader), RLIMIT_CPU hard 9,060 s, harness soft 9,000 CPU-s / +12,000 wall-s / 48 GiB, launched only after `vm_stat` showed 69.9 GB available. +Status `COMPLETED_NINETEEN_NODE`. + +| | | +|---|---| +| wall | 2,048.93 s | +| CPU | **2,028.69 s** | +| peak RSS | 13,157,728,256 B = 13.16 GB / 12.25 GiB | +| loadavg | start [6.68, 6.97, 6.40], end [10.75, 9.68, 8.66] | +| runner call | 2,045.91 s wall / 2,026.78 s CPU; 3.02 s of process wall sits outside it | +| financial node loop, 19 nodes | 277.36 s | +| prefix node loop, 9 nodes | 55.93 s | +| **runner outside both node loops** | **1,712.62 s = 83.7% of the runner call** | + +The four ASEC predictor nodes are 267.99 s of the financial loop's 277.36 s; +the other fifteen are 9.37 s. The retention slope is **not visible** in the RSS +trace at this fraction: 19 snapshots at the measured 9.6–9.8 B/cell over +1.70e6 cells is 0.31 GiB, inside the trace's own 1–2 GB sawtooth (peak 13.16 GB +at t=1303 s, 5.92 GB at exit). That corroborates the report's §9 rather than +adding to it. + +### Done since + +- Vectorised per-column frame seal in `survey_population_preparation`, proved + byte-for-byte against the predecessor traversal oracle that + `test_us_survey_frame_identity_encoding.py` already carried (160 tests; the + file's own runtime fell 7.16 s → 0.67 s). +- Segmented, content-addressed transport for all four whole-roster receipts. + `docs/us-native-scale-transport.md` has the design, the five measured ceilings + in binding order, and the two refusals that narrow. +- 1,013 line citations across seven mechanisms read and adversarially verified. + +### Next + +1. After-run at 1/1000 on this head — **queued behind a memory gate** (another + lane's `smoke.py`, pid 68076, holds 52.7 GB; the waiter polls every 60 s). +2. 1/10 run above the old ceiling. +3. Replay proof, pins, CI-shaped battery, draft PR. From a51957c5cbc1d64e6be10440834627892bbd36af Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:08:44 -0400 Subject: [PATCH 05/31] Check every component of the path the spill writes through graph_implementation's resource contract covers reads, so no committed pin would notice a redirected spill. The spill therefore checks its own path: the directory must not be a symlink, an existing segment or header must be a regular file of the recorded size, and presence is decided by lstat rather than Path.exists() -- a broken symlink does not exist and the first draft wrote straight through one, which its own test caught. Pins re-derived rather than assumed: every contract in graph_implementation_inventory.json matches the tree through the module's own _dependency_contract generator, and all ten declared stage manifests build. No committed pin moves. Co-Authored-By: Claude Opus 5 --- .../survey_population_preparation.py | 35 +++++++++++++++++-- .../test_us_survey_population_preparation.py | 22 ++++++++++++ 2 files changed, 54 insertions(+), 3 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py index 16325edd52..9f1d48f363 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/survey_population_preparation.py @@ -168,6 +168,15 @@ def close(): return segments, tuple(table), digest.hexdigest(), total +def _absent(path): + """True only when nothing at all sits at ``path``, symlinks included.""" + try: + path.lstat() + except FileNotFoundError: + return True + return False + + def _spill_roster(root, name, segments, table, digest, total): """Write the segments beside a header naming them, as the store's own shape. @@ -181,13 +190,26 @@ def _spill_roster(root, name, segments, table, digest, total): _require(type(name) is str and name.isidentifier(), "ROSTER_NAME") directory = root / name directory.mkdir(parents=True, exist_ok=True) + # The inventory's resource contract covers reads, not writes, so nothing + # else in this repository would notice a redirected spill. Every component + # of the path this module writes through is checked here: ``root`` is + # already a validated snapshot root, ``name`` is an identifier, and each + # segment's own name is the sha256 of its bytes. + _require( + not stat.S_ISLNK(directory.lstat().st_mode) and directory.is_dir(), + "ROSTER_SPILL_LOCATION", + ) for raw, (sha, size) in zip(segments, table, strict=True): path = directory / (sha + ".segment") - if not path.exists(): + # lstat, not exists(): a broken symlink does not exist and would be + # written straight through. + if _absent(path): path.write_bytes(raw) stats = path.lstat() _require( - not stat.S_ISLNK(stats.st_mode) and stats.st_size == size, + not stat.S_ISLNK(stats.st_mode) + and stat.S_ISREG(stats.st_mode) + and stats.st_size == size, "ROSTER_SEGMENT_CHANGED", ) header = { @@ -197,7 +219,14 @@ def _spill_roster(root, name, segments, table, digest, total): "size": total, "segments": [[sha, size] for sha, size in table], } - (directory / "header.json").write_bytes(_encode(header)) + path = directory / "header.json" + if not _absent(path): + stats = path.lstat() + _require( + not stat.S_ISLNK(stats.st_mode) and stat.S_ISREG(stats.st_mode), + "ROSTER_SPILL_LOCATION", + ) + path.write_bytes(_encode(header)) return header diff --git a/packages/microcosm-build/tests/test_us_survey_population_preparation.py b/packages/microcosm-build/tests/test_us_survey_population_preparation.py index d40132b5b3..ca36433c3c 100644 --- a/packages/microcosm-build/tests/test_us_survey_population_preparation.py +++ b/packages/microcosm-build/tests/test_us_survey_population_preparation.py @@ -789,6 +789,28 @@ def test_roster_spill_refuses_a_symlinked_segment(tmp_path): owner._roster_payload(document, spill=tmp_path, name="preparation") +def test_roster_spill_refuses_a_redirected_location(tmp_path): + document = {"rows": [[i, "acs", i * 7] for i in range(200)]} + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + (tmp_path / "preparation").symlink_to(elsewhere, target_is_directory=True) + with pytest.raises( + owner.SurveyPopulationPreparationError, match="ROSTER_SPILL_LOCATION" + ): + owner._roster_payload(document, spill=tmp_path, name="preparation") + # A redirected header is refused on the same code, after the segments. + other = tmp_path / "second" + other.mkdir() + owner._roster_payload(document, spill=other, name="preparation") + header = other / "preparation/header.json" + header.unlink() + header.symlink_to(tmp_path / "target.json") + with pytest.raises( + owner.SurveyPopulationPreparationError, match="ROSTER_SPILL_LOCATION" + ): + owner._roster_payload(document, spill=other, name="preparation") + + def test_candidate_bound_admits_a_full_source_receipt(tmp_path, monkeypatch): arguments = fixture(tmp_path, monkeypatch) with pytest.raises(owner.SurveyPopulationPreparationError, match="CANDIDATE_LIMIT"): From 53763c397098d92011623bdea4bdab1bec0fcafb Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:09:23 -0400 Subject: [PATCH 06/31] Say in the design note what the inventory's read contract cannot see The spill writes files, graph_implementation's _RESOURCE_CALLS covers reads, and the two facts together mean no committed pin would notice a redirected spill. Record that, the checks that stand in for it, and the broken-symlink hole the first draft had. Also record the precedent the question to the owner needs: graph_survey_puf55 already retains the final output rather than all 245 observed snapshots, so the financial runner is the outlier among the six observers in this runtime. Co-Authored-By: Claude Opus 5 --- docs/us-native-scale-transport.md | 34 +++++++++++++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/docs/us-native-scale-transport.md b/docs/us-native-scale-transport.md index 08bc159235..313259dd7f 100644 --- a/docs/us-native-scale-transport.md +++ b/docs/us-native-scale-transport.md @@ -131,6 +131,24 @@ which is the same header-plus-bodies shape `ContentStore.put_frame` already uses, and which gives the receipt a re-verifiable on-disk form independent of the graph store. +**The write surface, disclosed rather than left to be found.** This module +previously wrote nothing but `snapshots.mkdir`. The spill adds +`Path.write_bytes` for each segment and for `header.json`, and +`graph_implementation`'s `_RESOURCE_CALLS` (`:126-150`) covers *reads* — +`open`, `read_bytes`, `read_text`, `read_csv`, `load`, `files`, `joinpath`, +`import_module`, `exec`, `eval` and their neighbours — so a write is invisible +to `resource_accesses_sha256` by construction and no committed pin would notice +a redirected spill. Every component of the path is therefore checked here +instead: `root` is the snapshot root `_root()` already validated as symlink-free +and disjoint from the source root (`SNAPSHOT_LOCATION`), `name` must be a Python +identifier (`ROSTER_NAME`), each segment's own filename is the sha256 of its +bytes, the spill directory must not be a symlink (`ROSTER_SPILL_LOCATION`), and +an existing segment or header must be a regular file of the recorded size +(`ROSTER_SEGMENT_CHANGED`, `ROSTER_SPILL_LOCATION`). Presence is decided by +`lstat` and not by `Path.exists()`, because a broken symlink does not exist and +would be written straight through — a hole the first draft of this code had and +its own test caught. + `read_all()` reassembles, re-hashes each segment against its recorded digest and the whole against `digest`, and refuses `ROSTER_LIMIT` above `MAX_MATERIALIZED_BYTES`. The two consumers that need whole bytes — the @@ -367,7 +385,12 @@ established above: still refuses the run with the observer's exception type (`:4172-4179`); `ATOMIC_OBSERVER_ROSTER` and `COMPLETION_OBSERVER_ROSTER` still refuse any gap or reorder. -4. **A statement about `_pure_run`.** `state.node_populations` is re-stamped on +4. **A precedent in the same package.** `graph_survey_puf55` already declines to + retain everything: its issued run entry "retains the final output, not all 245 + executor-observed snapshots" (`graph_survey_puf55.py:638-639`). The financial + runner's `node_populations=observed` is the outlier among the six observers in + this runtime, and the shape it would move to already exists a few files away. +5. **A statement about `_pure_run`.** `state.node_populations` is re-stamped on every `checked_view()`, so a retention mode that drops populations must say what `FINANCIAL_NODE_POPULATION_CHANGED` means afterwards. Today `_node_population_seals` already returns `()` when there are no observations @@ -388,4 +411,11 @@ neighbours. 2. **Whether the retention policy is a run option or the only behaviour.** §3d shows that nineteen detached populations must exist between the observation and the replay comparison unless the object comparison is replaced by a - content seal. That is a verification-contract decision. + content seal. That is a verification-contract decision, and §4's item 4 shows + the runtime has already made it once, the other way, in `graph_survey_puf55`. +3. **Whether the roster spill should be readable by anything but this module.** + It is written, verified, and then only reassembled in the same call today. A + consumer that wanted to stream it — a future `put_bytes` that takes an + iterator, or an auditor re-deriving `preparation_sha256` without loading + 1 GiB — would read `header.json` and the segments it names, which is why the + header exists at all. From 66d56000d561a2e5e13795197c0f55525c641132 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:14:02 -0400 Subject: [PATCH 07/31] Prove the per-column seal against the cell walk on the values that can break it The file already compared the whole-frame preimage against a predecessor traversal on one fixture frame. These add the values a fixture cannot carry: every one of the 2,048 biased double exponents against five mantissas and both signs (20,470 finite patterns), 1.5 million random bit patterns over six seeds, both infinities refusing on the walk's own code, twelve column kinds including the string slow path past 4,096 characters and the object/category columns that deliberately keep the walk, each of them empty, strided and reversed, and six axis kinds. The object case is why dedup is confined to StringDtype: 1, True and 1.0 are equal and hash alike, and the encoder spells them b"1", b"true" and b'["float","0x1.0000000000000p+0"]'. The test asserts all three. 187 tests, 15.62 s. Co-Authored-By: Claude Opus 5 --- .../test_us_survey_frame_identity_encoding.py | 110 ++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/packages/microcosm-build/tests/test_us_survey_frame_identity_encoding.py b/packages/microcosm-build/tests/test_us_survey_frame_identity_encoding.py index e1db605b7f..85c760e37f 100644 --- a/packages/microcosm-build/tests/test_us_survey_frame_identity_encoding.py +++ b/packages/microcosm-build/tests/test_us_survey_frame_identity_encoding.py @@ -866,3 +866,113 @@ def validate(candidate): else: owner.verify_materialized_survey_population(preparation, frame) assert calls == ["checked", "identity", "validate", "identity"] + + +# The per-column encoders below assemble, per column, the exact byte string the +# cell-at-a-time walk fed the digest. `_legacy_cells` is that walk, written out +# here rather than imported, so the comparison is against the predecessor and +# not against the shipped helper's own idea of itself. + + +def _legacy_cells(values): + out = bytearray() + for value in values: + out += owner._encode(owner._cell(value)) + out += b"\n" + return bytes(out) + + +def _float_column(bits): + return pd.Series(np.asarray(bits, dtype=np.uint64).view(np.float64)) + + +def test_every_double_exponent_and_mantissa_boundary_spells_identically(): + """Every biased exponent against five mantissas and both signs.""" + patterns = [ + (sign << 63) | (exponent << 52) | mantissa + for exponent in range(2048) + for mantissa in (0, 1, 0x8000000000000, 0x123456789ABCD, 0xFFFFFFFFFFFFF) + for sign in (0, 1) + ] + series = _float_column(patterns) + finite = series[np.isfinite(series.to_numpy())] + assert len(finite) == 20470 + assert owner._cells_blob(finite) == _legacy_cells(finite) + + +@pytest.mark.parametrize("trial", range(6)) +def test_random_double_bit_patterns_spell_identically(trial): + generator = np.random.default_rng(20260917 + trial) + raw = generator.integers(0, 2**64, size=250_000, dtype=np.uint64) + series = _float_column(raw) + finite = series[np.isfinite(series.to_numpy())] + assert len(finite) > 200_000 + assert owner._cells_blob(finite) == _legacy_cells(finite) + + +def test_non_finite_doubles_refuse_with_the_walk_s_own_code(): + for value in (np.inf, -np.inf): + series = pd.Series(np.array([1.0, value, 2.0], dtype=np.float64)) + assert _outcome(lambda s=series: owner._cells_blob(s)) == _outcome( + lambda s=series: _legacy_cells(s) + ) + + +_SEAL_COLUMNS = { + "float64": pd.Series( + np.array( + [0.0, -0.0, 1.0, -1.25, np.nan, 5e-324, 1e308, float(2**53)], + dtype=np.float64, + ) + ), + "int64": pd.Series(np.array([-(2**63), -1, 0, 1, 2**63 - 1], dtype=np.int64)), + "int32": pd.Series(np.array([-(2**31), 0, 2**31 - 1], dtype=np.int32)), + "uint64": pd.Series(np.array([0, 1, 2**64 - 1], dtype=np.uint64)), + "Int64": pd.Series(pd.array([-(2**63), None, 0, 2**63 - 1], dtype="Int64")), + "UInt64": pd.Series(pd.array([0, None, 2**64 - 1], dtype="UInt64")), + "bool": pd.Series(np.array([True, False, True])), + "boolean": pd.Series(pd.array([True, None, False], dtype="boolean")), + "string": pd.Series( + pd.array( + ["", 'quote" slash\\', "\x00\x1f\x7f", "é中🧪", None, "x" * 4096], + dtype="string", + ) + ), + "string-slow-path": pd.Series(pd.array(["x" * 4097, "y"], dtype="string")), + "object": pd.Series([1, True, 1.0, "1", None], dtype=object), + "category": pd.Series(pd.Categorical(["a", "b", "a", None])), +} + + +@pytest.mark.parametrize("name", sorted(_SEAL_COLUMNS)) +def test_every_column_kind_spells_identically(name): + series = _SEAL_COLUMNS[name] + assert owner._cells_blob(series) == _legacy_cells(series) + assert owner._cells_blob(series.iloc[:0]) == b"" + assert owner._cells_blob(series.iloc[::2]) == _legacy_cells(series.iloc[::2]) + assert owner._cells_blob(series.iloc[::-1]) == _legacy_cells(series.iloc[::-1]) + + +def test_object_and_category_keep_the_walk_because_equal_values_differ_in_bytes(): + """1, True and 1.0 are equal and hash alike; their cells are three strings.""" + series = _SEAL_COLUMNS["object"] + assert len(set(series.iloc[:3].tolist())) == 1 + assert owner._frame_cell_encode(1) == b"1" + assert owner._frame_cell_encode(True) == b"true" + assert owner._frame_cell_encode(1.0) == b'["float","0x1.0000000000000p+0"]' + assert owner._cells_blob(series) == _legacy_cells(series) + + +@pytest.mark.parametrize( + "index", + [ + pd.RangeIndex(5), + pd.Index([-(2**63), 0, 2**63 - 1], dtype="int64"), + pd.Index([0, 1, 2**64 - 1], dtype="uint64"), + pd.Index([0.0, -0.0, 1.5], dtype="float64"), + pd.Index(["a", "b"], dtype="object"), + pd.Index(pd.array(["a", None], dtype="string")), + ], +) +def test_every_axis_kind_spells_identically(index): + assert owner._index_blob(index) == _legacy_cells(index) From 67123969e56e968dd86d7aeca0549a54345a5a15 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 16:14:28 -0400 Subject: [PATCH 08/31] Commit the three harnesses as they ran The lane's measurements are only re-derivable if the scripts that took them are in the tree at the head that took them. Co-Authored-By: Claude Opus 5 --- experiments/native-scale-transport/README.md | 22 + .../harness19_after_with_required_replay.py | 741 ++++++++++++++++++ .../native-scale-transport/harness19_base.py | 603 ++++++++++++++ .../native-scale-transport/harness19_tenth.py | 666 ++++++++++++++++ 4 files changed, 2032 insertions(+) create mode 100644 experiments/native-scale-transport/README.md create mode 100644 experiments/native-scale-transport/harness19_after_with_required_replay.py create mode 100644 experiments/native-scale-transport/harness19_base.py create mode 100644 experiments/native-scale-transport/harness19_tenth.py diff --git a/experiments/native-scale-transport/README.md b/experiments/native-scale-transport/README.md new file mode 100644 index 0000000000..8dc3f759eb --- /dev/null +++ b/experiments/native-scale-transport/README.md @@ -0,0 +1,22 @@ +# native-scale-transport — measurement harnesses and receipts + +The three harnesses this lane ran, committed as they ran, plus the receipts they +wrote. Nothing here is a build, a certification or a release artifact; every +measurement JSON carries `"release_eligible": false` and a scope line saying so. + +| file | what it ran | +|---|---| +| `harness19_base.py` | the baseline: the 19-node financial graph at 1/1000 on the branch point `5ff889814`, uncapped by node, with a current-RSS trace and the runner/node-loop split | +| `harness19_after_with_required_replay.py` | the same at this branch's head, followed by a `resume="require"` replay of the same graph against the store the cold run wrote, comparing manifest key, node keys, artifact identities, receipts and every store object's bytes | +| `harness19_tenth.py` | the 1/10 run: 158,737 source households, above the 96,860 the single bounded encode admitted | + +All three descend from `~/PolicyEngine/_worktrees/microcosm-verify-once/.measure/harness19_verify_once.py`, +which is itself the v5 pilot probe parameterised; what is measured — the call, +the seed, the geography support, the sampler and its interval — is unchanged +from the accepted v4 cold pilot's own arguments except where a file's docstring +says otherwise. + +Each harness needs `MEASURE_BASE` (the source tree), `MEASURE_RUN` (the staged +run inputs), `MEASURE_PROBE` (the output directory), `MEASURE_LABEL` and +`MEASURE_HEAD`, and is started through `setsid_exec.py` so it leads its own +session and the pid the shell prints is the pid to record. diff --git a/experiments/native-scale-transport/harness19_after_with_required_replay.py b/experiments/native-scale-transport/harness19_after_with_required_replay.py new file mode 100644 index 0000000000..dde9785b4e --- /dev/null +++ b/experiments/native-scale-transport/harness19_after_with_required_replay.py @@ -0,0 +1,741 @@ +"""Baseline for the native-scale lane: the 19-node graph, uncapped by node. + +This is ``harness19_verify_once.py`` from the verify-once lane with four +additions and no change to what is measured -- same call, same arguments, same +sample, same seed, same sampler, same interval, same refusals: + +* the ceilings are the native-scale brief's (9,000 CPU-s, 48 GiB, 12,000 wall-s) + rather than the verify-once lane's, so no node is cut short; +* a current-resident-size series is sampled every second through ``libproc``'s + ``proc_pidinfo``, because ``ru_maxrss`` is a high-water mark and a retention + slope needs the trace, not the peak; +* the ``run_atomic_survey_financial`` call is timed on its own, so the runner + time outside the node loop is a subtraction between measured quantities + rather than an inference; +* the returned manifest's per-node ``wall_time_s`` is recorded, which is what + the subtraction subtracts. + +Nothing is wrapped: the runtime still refuses instrumented producers, so +attribution is still the stack sampler's. + +This copy adds a fifth thing the baseline did not need: after the cold run it +replays the same graph against the same store with ``resume="require"``, which +is what a required replay is in this tree (``_preflight_require`` refuses with +``StoreMiss`` if any node is absent), and compares the two runs' manifest key, +node keys, artifact identities, receipts and the receipt digests the transport +carries. The cold run's own CPU and wall are recorded before the replay starts, +so the before/after comparison is unaffected by it. + +Original docstring follows. + +Nineteen-node measurement for the verify-once lane, on this branch only. + +The instrumentation is ``probe_verify_once.py`` unchanged -- the same stack +sampler at the same interval, the same ceilings, the same refusals, the same +output shape. The only difference is the call: this runs the base financial +graph with the exact arguments the accepted v4 cold pilot ran +(``~/PolicyEngine/_recovered/pilot-runs/native19-v2/run/test_native_nineteen_node_financial.py``: +fraction 1/1000, seed 20260908, geography seed 20260908, +``demographic_conditioning=True``, ``n_estimators=2``, ``resume="auto"``), so +its wall, CPU and peak RSS are comparable with that receipt's +5,362.05 s / 5,278.61 s / 8.67 GiB. + +Two differences from that receipt are known and are not this change: it ran on +source snapshot ``2ca11c85a``, and its ``person-income-attachment.h5`` was the +pre-restoration ``5996dcdd...`` rather than the staged ``9ebc0ef2...``. Both +files are exactly 666,333,922 B, so the bytes the source tree hashes are the +same volume either way. + +Original docstring follows. + +Before/after measurement for the verify-once lane: the nine-node prefix. + +A parameterised copy of the v5 pilot probe +(``~/PolicyEngine/_recovered/pilot-runs/native45-v5/probe_repeated_verification.py``). +Everything that decides *what is measured* is verbatim: the same nine-node +population prefix, the same 1/1000 sample, the same seed, the same stack +sampler at the same interval, the same ceilings, the same output shape. The +only changes are the ones a before/after comparison needs: + +* the source tree, the staged run inputs and the output directory come from + the environment, so the identical file measures both trees; +* every imported ``microcosm`` module is asserted to live under that tree, + so a stray editable install cannot silently measure the wrong source; +* a resident-memory ceiling is enforced alongside the CPU ceiling; +* the load average and the interpreter's third-party versions are recorded + with each run, because this machine carries other lanes' work. + +Sources are read by path and never written. Nothing here is a build, a +certification or a release artifact. + + MEASURE_BASE= MEASURE_RUN= \ + MEASURE_PROBE= MEASURE_LABEL=