Merge pull request #970 from PolicyEngine/codex/h5-migration-pr7-builder

Extract local H5 builder seams

Author: anth-volk

changelog.d/969.changed

@@ -0,0 +1 @@
+Extracted local H5 build-output seams and split US payload postprocessors for focused unit testing.

docs/engineering/skills/README.md

@@ -20,3 +20,7 @@ Current skills:
   pipeline status and durable error records.
 - `testing.md`: test layout, fixture scope, helper placement, and quality guard
   expectations.
+
+Stage-specific AI-facing engineering guides live under `docs/engineering/stages/`.
+Use them alongside these cross-cutting skills when modifying a stage-specific
+pipeline path.

docs/engineering/stages/build_outputs.md

@@ -0,0 +1,171 @@
+# Build Outputs Stage AI Guide
+
+This guide is for AI agents and maintainers modifying Stage 4
+(`4_build_outputs`) code. Stage 4 turns calibrated and staged pipeline artifacts
+into publishable outputs, including local-area H5 files, national H5 files,
+diagnostics, and release-staging artifacts.
+
+The active local H5 seams live under `policyengine_us_data/build_outputs/`.
+Treat this package as the place for reusable Stage 4 library boundaries. Keep
+Modal orchestration, worker entrypoints, and release promotion behavior outside
+these library seams unless a stage plan explicitly says otherwise.
+
+## Local H5 Build Path
+
+The transitional runtime entrypoint is still
+`policyengine_us_data.calibration.publish_local_area.build_h5()`. It should stay
+as a facade while Stage 4 is being migrated. New implementation logic should
+move behind narrower build-output library seams instead of growing this facade.
+
+The current in-memory local H5 path is:
+
+1. `AreaSelector` selects active clone-household rows from clone weights and
+   geography filters.
+2. `EntityReindexer` creates output household, person, and subentity IDs.
+3. `VariableCloner` copies allowed source variables into a period-grouped
+   payload.
+4. `LocalAreaDatasetBuilder` applies payload postprocessors in declared order.
+5. `H5Writer` writes the final `H5Payload` and verifies summary counts.
+
+When adding behavior to this path, decide whether it is a selection, reindexing,
+source-variable cloning, postprocessing, or writing concern. Do not place
+country-specific payload mutation in `build_h5()` when it can be represented as
+a postprocessor.
+
+## Payload Postprocessors
+
+Payload postprocessors are ordered, country- or product-specific transformations
+that consume an `H5Payload` and return either another `H5Payload` or a structured
+result object exposing a `.payload` attribute.
+
+Use a postprocessor when the operation:
+
+- Mutates or adds payload variables after generic source-variable cloning.
+- Depends on country-specific business rules.
+- Needs focused unit tests independent of the full H5 builder.
+- Should run after some other payload construction step.
+
+Do not use a postprocessor for:
+
+- Selecting active clones. Use `AreaSelector`.
+- Reindexing entities. Use `EntityReindexer`.
+- Copying source variables unchanged. Use `VariableCloner`.
+- Writing H5 files. Use `H5Writer`.
+- Modal orchestration, volume setup, or publication promotion.
+
+## Postprocessor Spec Contract
+
+Every postprocessor should expose a stable `spec`:
+
+```python
+spec = PayloadPostProcessorSpec(
+    key="stable_unique_key",
+    requires=("upstream_key",),
+)
+```
+
+The `key` is a durable identifier for the processing step. Prefer short,
+stage-specific names such as `us_entity`, `us_geography`, or `us_takeup`.
+Do not use display names, class names, or generated values as the key when the
+processor is part of a stable runtime path.
+
+The `requires` tuple lists postprocessor keys that must already have run. This
+declares ordering explicitly. It is not a substitute for validating the concrete
+payload fields the postprocessor consumes.
+
+`LocalAreaDatasetBuilder` validates the configured postprocessor sequence before
+building:
+
+- Duplicate `spec.key` values are rejected.
+- A postprocessor whose `requires` keys have not appeared earlier is rejected.
+- Processors without an explicit `spec` receive a fallback key based on class
+  name. This fallback is for tests or transitional code only; production
+  postprocessors should define stable keys.
+
+If a processor consumes fields written by an earlier processor, define both the
+dependency and a payload validation. The dependency catches bad builder
+configuration early; payload validation catches direct processor use and
+malformed payloads.
+
+## Current US Postprocessors
+
+The production US postprocessor sequence is defined by
+`default_us_postprocessors()`:
+
+1. `USEntityPostProcessor`
+   - Key: `us_entity`
+   - Dependencies: none
+   - Adds output entity IDs and `household_weight`.
+
+2. `USGeographyPostProcessor`
+   - Key: `us_geography`
+   - Dependencies: none
+   - Derives geography from selected block GEOIDs and writes geography
+     variables such as `state_fips`, `county_fips`, `zip_code`, and
+     `congressional_district_geoid`.
+
+3. `USTakeupPostProcessor`
+   - Key: `us_takeup`
+   - Dependencies: `us_entity`, `us_geography`
+   - Applies take-up draws and writes take-up variables.
+   - Validates that required reindexed subentities exist.
+   - Validates that `state_fips` exists in the payload.
+   - Validates that `person_tax_unit_id` and `tax_unit_id` exist when reported
+     ACA anchors are present.
+
+Keep this ordering unless you also update specs, structural validations, and
+unit tests.
+
+## Adding A Postprocessor
+
+When adding a postprocessor:
+
+1. Define a result dataclass if callers need metadata beyond the payload.
+2. Define a stable `PayloadPostProcessorSpec`.
+3. Add direct payload precondition checks for every field the processor consumes.
+4. Preserve the incoming payload's `time_period`, `entity_lengths`, and
+   `variable_entities` unless intentionally changing them.
+5. When adding variables with non-obvious entity lengths, update
+   `variable_entities` so `H5Payload` can validate their shapes.
+6. Add the postprocessor to the production factory only if it belongs in the
+   runtime path.
+7. Add unit tests for the processor in `tests/unit/build_outputs/`.
+8. Add or update a builder-order test if the processor has dependencies.
+
+Prefer dependency injection for expensive or external behavior. For example,
+`USTakeupPostProcessor` accepts a `takeup_applier` so unit tests can verify the
+contract without loading rates or running the full pipeline.
+
+## Testing Expectations
+
+Unit tests should cover each new postprocessor directly. At minimum, test:
+
+- The variables it writes.
+- The payload fields it consumes.
+- Its declared `spec.key` and `spec.requires` ordering.
+- Failure for missing required payload fields.
+- Failure for wrong-length generated arrays when the output entity is known.
+
+Builder tests should cover:
+
+- Missing dependency rejection.
+- Duplicate postprocessor key rejection.
+- Result recording through `PayloadPostProcessorRun`.
+
+Integration tests should only be added when the behavior crosses module or
+runtime boundaries that unit tests cannot represent. Do not add a second
+integration test that proves the same seam.
+
+## Documentation Expectations
+
+When Stage 4 behavior changes, update the durable documentation surface:
+
+- Add or update `@pipeline_node` metadata for new stable library seams.
+- Update `docs/pipeline_map.yaml` when the stage graph or durable artifacts
+  change.
+- Keep generated docs out of manual PR edits unless the repository workflow
+  specifically requires them.
+
+Do not put PR-specific rationale in docstrings. Put durable behavior in source
+docs and put review or migration rationale in PR descriptions, issues, or stage
+planning docs.

docs/pipeline_map.yaml

@@ -1167,10 +1167,6 @@ stages:
     label: Modal Worker Container
     node_type: external
     description: 16GB RAM, 1 CPU each, 8-hour timeout
-  - id: spm_recalc
-    label: SPM Threshold Recalculation
-    node_type: process
-    description: Local median rents, family composition, tenure type
   - id: takeup_apply
     label: Takeup Re-application
     node_type: process
@@ -1235,9 +1231,6 @@ stages:
     target: geo_derive
     edge_type: data_flow
   - source: geo_derive
-    target: spm_recalc
-    edge_type: data_flow
-  - source: spm_recalc
     target: takeup_apply
     edge_type: data_flow
   - source: modal_coord

policyengine_us_data/build_outputs/__init__.py

@@ -6,5 +6,7 @@
 fingerprinting, clone-weight shape contracts, worker partitioning, source
 dataset snapshot contracts, worker input normalization, worker-bootstrap
 artifacts, worker-scoped session and validation context setup, microsimulation
-access helpers, clone selection, entity reindexing, and source-variable cloning.
+access helpers, clone selection, entity reindexing, source-variable cloning,
+validated H5 payload contracts, ordered output postprocessing, one-area payload
+building, and H5 writing.
 """