recon_gen.common.l2.pipeline_overlays

BU.1.8 — Typed pipeline overlay primitives.

Replaces the round-1 indirection cfg.test_generator.scope = "uncovered_rails" (which means “after wipe + covered={} → baseline only”) with a typed surface that says what it means at the wiring site.

Two demo flows share the wipe+regenerate pipeline but differ on which overlay layers fire afterward:

  • ETL_DEBUG (/etl/run Refresh Data) — baseline + L1 invariant plants (drift/overdraft/etc) + BTa.8 L2 demo-gap overlay. Operator wants noise to debug against.

  • TRAINER_CLEAN (/training/reset) — baseline only. No overlays. Operator plants exactly one scenario via the Trainer + sees ONLY it surface on the dashboard tour.

  • LOCKED_SEED (used by tests + CLI data apply) — baseline + L1 plants, no L2 overlay. Matches build_full_seed_sql byte-output contract.

Each overlay layer is a typed OverlayLayer dataclass carrying its own apply function — adding a new overlay (e.g. a future Trainer-mode “amplify L1 plants by 5×” overlay) = one new singleton, no dispatch table to update. Mirrors the BU.0 Lock 7 PlantKindEntry “single source of truth” pattern but for the deploy-pipeline post-baseline layers.

The current scope-string surface (cfg.test_generator.scope) stays for CLI data apply backward compat; deploy_pipeline’s Studio-facing entry points switch to overlays= typed kwarg.

Module Attributes

ETL_DEBUG

/etl/run Refresh Data — baseline + L1 plants + L2 BTa.8 overlay.

TRAINER_CLEAN

/training/reset — baseline only.

LOCKED_SEED

Tests + CLI data apply — baseline + L1 plants, no L2 overlay.

Classes

OverlayContext(cfg, instance, dev_log)

The runtime context handed to each overlay layer's apply.

OverlayLayer(name, description, apply)

One typed overlay the deploy pipeline applies after baseline.

PipelineOverlays([layers])

The tuple of overlay layers the deploy pipeline applies after baseline.
recon_gen.common.l2.pipeline_overlays.ETL_DEBUG: Final[PipelineOverlays] = PipelineOverlays(layers=(OverlayLayer(name='l1_invariant_plants', description="Re-emit the L1 invariant violation plants (drift, overdraft, stuck_pending, etc) so the L1 Dashboard's exception sheets have demo content.", apply=<function _apply_l1_invariant_plants>), OverlayLayer(name='l2_demo_gap_overlay', description='Re-emit the BTa.8 phantom rail / phantom template / missing-metadata INSERTs + uncovered rail/template DELETEs so /etl/triage + /etl/run Coverage have demo content.', apply=<function _apply_l2_demo_gap_overlay>)))

/etl/run Refresh Data — baseline + L1 plants + L2 BTa.8 overlay. Operator wants noise to debug against.

recon_gen.common.l2.pipeline_overlays.LOCKED_SEED: Final[PipelineOverlays] = PipelineOverlays(layers=(OverlayLayer(name='l1_invariant_plants', description="Re-emit the L1 invariant violation plants (drift, overdraft, stuck_pending, etc) so the L1 Dashboard's exception sheets have demo content.", apply=<function _apply_l1_invariant_plants>),))

Tests + CLI data apply — baseline + L1 plants, no L2 overlay. Preserves byte-identity with build_full_seed_sql output (which predates the L2 BTa.8 overlay).

class recon_gen.common.l2.pipeline_overlays.OverlayContext(cfg, instance, dev_log)[source]

Bases: object

The runtime context handed to each overlay layer’s apply.

Carries the cfg + L2 instance + the shared event-emitter the deploy-pipeline tees through (so layer events land in the same DeploySummary.events stream as the baseline steps).

Parameters:

  • cfg (Config)

  • instance (L2Instance)

  • dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)

cfg: Config
dev_log: Callable[[Mapping[str, object]], Awaitable[None]] | None
instance: L2Instance
class recon_gen.common.l2.pipeline_overlays.OverlayLayer(name, description, apply)[source]

Bases: object

One typed overlay the deploy pipeline applies after baseline.

Each layer carries its own apply so adding a new layer = one singleton declaration + no dispatch-table changes. The layer name is the canonical key surfaced in DeploySummary.events + Lock 9 anti-drift tests.

Parameters:

  • name (str)

  • description (str)

  • apply (Callable[[OverlayContext], Awaitable[None]])

apply: Callable[[OverlayContext], Awaitable[None]]
description: str
name: str
class recon_gen.common.l2.pipeline_overlays.PipelineOverlays(layers=())[source]

Bases: object

The tuple of overlay layers the deploy pipeline applies after baseline. Order matters — layers fire in tuple order so a later layer sees the state earlier layers left behind.

Parameters:

layers (tuple[OverlayLayer, ...])

layers: tuple[OverlayLayer, ...]
names()[source]

Tuple of layer names for diagnostic logging.

Return type:

tuple[str, ...]

recon_gen.common.l2.pipeline_overlays.TRAINER_CLEAN: Final[PipelineOverlays] = PipelineOverlays(layers=())

/training/reset — baseline only. Operator plants exactly one scenario via the Trainer + sees ONLY it surface.