recon_gen.common.l2.deploy_pipeline

Studio “Deploy changes” pipeline (X.4.g + BS.4 architecture shift).

Four-step orchestration that takes the operator’s current cfg + L2Instance and refreshes the demo DB so the dashboards re-render against the new shape:

wipe — TRUNCATE <prefix>_transactions + <prefix>_daily_balances so step 2’s etl_hook + step 3’s generator both write into clean state.
etl_hook — run cfg.etl_hook as a subprocess. The hook is expected to write directly to demo_db (the BS.4 contract — see docs/audits/bs_4_arch_shift_spike.md). Non-zero exit halts the pipeline before steps 3-5 run; demo_db is left in whatever partial state the hook produced (operator re-runs after fixing the hook).
generator — emit_full_seed against the current cfg.test_generator knobs; always additive overlay on top of the etl_hook’s rows.
matview refresh — existing refresh_matviews_sql(instance).
reload — bump data_generation_id; Dashboards’ open page polls and reloads its current URL.

BS.4 (2026-05-29) collapsed the pre-existing upstream → demo_db → matview refresh model to truncate(demo_db) → ETL hook → matview refresh. The legacy step_2_pull (cross-dialect copy from cfg.etl_datasource) was deleted along with EtlDatasourceConfig; etl_hook is the only ETL-load contract now. The step ordering also flipped — wipe now runs BEFORE etl_hook (the hook writes into a clean demo_db, not into a parallel upstream that gets copied over).

This module is HTTP-free. The studio’s POST /deploy endpoint wires up a DevLogWriter that emits via _DEVLOG.info; tests wire one that appends to a list.

Functions

`get_data_generation_id`()	Read the current generation counter — used by the `GET /data_generation_id` endpoint Dashboards polls.
`run_deploy_pipeline`(cfg, instance, *[, ...])	Orchestrate the BS.4 4-step deploy pipeline (with the 3.5 derive sub-step).
`step_1_etl_hook`(cfg, *[, dev_log])	Run `cfg.etl_hook` as a subprocess; stream output to `dev_log`.
`step_2_wipe`(cfg, instance, *[, dev_log])	Empty `<prefix>_transactions` + `<prefix>_daily_balances`.
`step_3_5_derive_balances`(cfg, instance, *[, ...])	X.4.i.2 — re-derive `<prefix>_daily_balances` from `<prefix>_transactions` for the configured account roles.
`step_3_generator`(cfg, instance, *[, dev_log])	Run the synthetic-data generator, execute its SQL against the demo DB, return per-base-table row counts.
`step_4_matviews`(cfg, instance, *[, dev_log])	Run `refresh_matviews_sql(instance, dialect=cfg.dialect)` against the demo DB.
`step_5_reload`(*[, dev_log])	Bump `_data_generation_id` by one and emit the new value.

Classes

DeploySummary([halted, halt_reason, ...])

Structured per-step outcome of a run_deploy_pipeline call.

class recon_gen.common.l2.deploy_pipeline.DeploySummary(halted=False, halt_reason=None, step1_etl_hook_exit_code=0, step2_wipe_transactions_deleted=0, step2_wipe_daily_balances_deleted=0, step3_generator_transactions_after=0, step3_generator_daily_balances_after=0, step3_5_derived_balance_rows=0, step4_matviews_done=False, step5_data_generation_id=0, events=<factory>)[source]

Bases: object

Structured per-step outcome of a run_deploy_pipeline call.

Wraps the raw event stream the studio + tests collect. The studio’s POST /deploy serializes this dataclass straight to JSON; tests assert against the typed fields without re-parsing event payloads.

halted flips when step 1’s etl_hook returns non-zero exit (the only halt point — every other step runs unconditionally once we’re past step 1). Any halted summary leaves later steps’ fields at their default (zeros / False) — read halted first.

Parameters:

halted (bool)
halt_reason (str | None)
step1_etl_hook_exit_code (int)
step2_wipe_transactions_deleted (int)
step2_wipe_daily_balances_deleted (int)
step3_generator_transactions_after (int)
step3_generator_daily_balances_after (int)
step3_5_derived_balance_rows (int)
step4_matviews_done (bool)
step5_data_generation_id (int)
events (tuple[Mapping[str, object], ...])

events: tuple[Mapping[str, object], ...]

halt_reason: str | None = None

halted: bool = False

step1_etl_hook_exit_code: int = 0

step2_wipe_daily_balances_deleted: int = 0

step2_wipe_transactions_deleted: int = 0

step3_5_derived_balance_rows: int = 0

step3_generator_daily_balances_after: int = 0

step3_generator_transactions_after: int = 0

step4_matviews_done: bool = False

step5_data_generation_id: int = 0

to_json()[source]

Serialize to a JSON-safe dict for POST /deploy responses.

Return type:: dict[str, object]

recon_gen.common.l2.deploy_pipeline.get_data_generation_id()[source]

Read the current generation counter — used by the GET /data_generation_id endpoint Dashboards polls.

Return type:: int

async recon_gen.common.l2.deploy_pipeline.run_deploy_pipeline(cfg, instance, *, dev_log=None, overlays=None)[source]

Orchestrate the BS.4 4-step deploy pipeline (with the 3.5 derive sub-step). Order: wipe → etl_hook → generator → overlays → matviews → reload.

BS.4 (2026-05-29) reordered + dropped the legacy step_2_pull. The wipe now runs FIRST (clean slate for etl_hook + generator to write into); etl_hook writes directly to demo_db (no parallel upstream); pull is gone.

BU.1.8 — overlays is the typed surface for post-baseline plant layers (replaces the round-1 cfg.test_generator.scope = "uncovered_rails" indirection). When None, defaults are dialect-aware: ETL_DEBUG (full noise) for studio Refresh Data callers, TRAINER_CLEAN for Trainer reset, LOCKED_SEED for tests. See common/l2/pipeline_overlays.py for the named flows.

When overlays is provided AND cfg.test_generator.scope is the default "full", the generator step emits the BASELINE-only SQL (emit_baseline_seed) + each overlay layer applies after. This separates baseline-vs-plants so the Trainer can plant on a quiet baseline without the L1-plant overlay firing.

Callers passing overlays=None get the legacy behavior (build_full_seed_sql via scope-string dispatch) for CLI data apply + locked-seed test back-compat.

Halt contract: step 2’s etl_hook exit code gates steps 3-5. Non-zero ⇒ stop. demo_db is left in whatever state the hook produced (the wipe ran; the hook may have written some / no / partial rows). Operators are expected to wrap their hook in a transaction so a failure rolls back to the post-wipe empty state.

Every step shares one event-collecting writer that fans out to the caller’s dev_log AND captures the events on the returned DeploySummary.events tuple — so the studio’s POST /deploy can render a “what happened” timeline even if dev_log is off.

Return type:

DeploySummary

Parameters:

cfg (Config)
instance (L2Instance)
dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)
overlays (PipelineOverlays | None)

async recon_gen.common.l2.deploy_pipeline.step_1_etl_hook(cfg, *, dev_log=None)[source]

Run cfg.etl_hook as a subprocess; stream output to dev_log.

Returns the subprocess exit code, OR 0 when cfg.etl_hook is unset / empty (no-op skip). Caller checks the return value and halts the pipeline if non-zero — steps 3-5 (generator overlay, matview refresh, reload) MUST NOT run when the operator’s ETL failed.

BS.4 (2026-05-29) reordered the pipeline so the wipe runs BEFORE this step (the hook writes directly to demo_db, not to a parallel upstream that gets copied over). On etl_hook failure, demo_db is left in whatever partial state the hook produced — operators re-run after fixing the hook (recommend wrapping hook writes in a transaction so partial writes roll back automatically).

The command is shlex.split then run via asyncio.create_subprocess_exec (NOT shell=True). Stdout and stderr stream line-by-line as deploy:step1:stdout / deploy:step1:stderr events so the operator watches progress in the studio’s dev_log overlay rather than waiting for the subprocess to drain.

A missing binary (FileNotFoundError from create_subprocess_exec) propagates — the caller surfaces it as an actionable error, NOT a silent skip. The whole point of declaring an etl_hook is that it MUST run.

Return type:

int

Parameters:

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

async recon_gen.common.l2.deploy_pipeline.step_2_wipe(cfg, instance, *, dev_log=None)[source]

Empty <prefix>_transactions + <prefix>_daily_balances.

BS.4 (2026-05-29) — runs FIRST in the pipeline so the etl_hook (step 1) and the synthetic-data generator (step 3) both write into clean state. Pre-BS.4 this ran AFTER step 1’s etl_hook (the hook was assumed to write to upstream and step_2_pull copied to demo); the pull step is gone and the order swapped accordingly. The matview re-derive is step 4’s job.

Returns (transactions_deleted, daily_balances_deleted) row counts so the caller can surface “wiped 12,345 transactions” in the deploy summary.

Sync DB-API 2.0 work runs in asyncio.to_thread so it doesn’t block the asyncio loop — the studio’s POST /deploy endpoint otherwise stalls all other requests for the wipe duration on a multi-million-row demo DB.

Return type:

tuple[int, int]

Parameters:

cfg (Config)
instance (L2Instance)
dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)

async recon_gen.common.l2.deploy_pipeline.step_3_5_derive_balances(cfg, instance, *, dev_log=None)[source]

X.4.i.2 — re-derive <prefix>_daily_balances from <prefix>_transactions for the configured account roles.

No-op when cfg.test_generator.derive_balances is False (the default). When enabled, computes money = SUM(amount_money) per (account_id, business_day_end) for accounts whose account_role matches cfg.test_generator.derive_balances_account_roles (or the default control-account set when None) and UPSERTs into the daily_balances table. Existing rows for those roles are overwritten in-place; rows for other roles are untouched.

The drift invariant is what this is “running forward”: auditing money == SUM(amount_money) would always pass for derived rows since they were just computed that way. That’s the point — operators use this when they want planted scenarios to reconcile cleanly against derived balances (e.g. test the dashboard renders against a known-clean control set), or when their ETL provides only transactions and balances must be back-filled.

Returns the number of (account_id, business_day) rows inserted / updated. dev_log receives lifecycle events deploy:step3_5:derive:start and deploy:step3_5:derive:done (with rows count + account_roles for visibility).

Return type:

int

Parameters:

cfg (Config)
instance (L2Instance)
dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)

async recon_gen.common.l2.deploy_pipeline.step_3_generator(cfg, instance, *, dev_log=None)[source]

Run the synthetic-data generator, execute its SQL against the demo DB, return per-base-table row counts.

Return type:

tuple[int, int]

Parameters:

cfg (Config)
instance (L2Instance)
dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)

X.4.g.7 — scaffolding. Honors cfg.test_generator:

enabled = False ⇒ skip event + return (0, 0).
scope = "full" ⇒ X.4.g.8 — build_full_seed_sql (today’s behavior). Byte-identical to the locked seeds when etl_hook is absent (or a no-op) AND knobs at defaults.
scope = "exceptions_only" ⇒ ships in X.4.g.9.
scope = "uncovered_rails" ⇒ ships in X.4.g.10.

Always additive — runs after step 2’s wipe + optional pull, so the generator’s INSERTs land on top of whatever step 2 left in the base tables. The returned counts are the post-step-3 totals (not the delta), so the deploy summary can report “ended with 12,345 transactions”.

async recon_gen.common.l2.deploy_pipeline.step_4_matviews(cfg, instance, *, dev_log=None)[source]

Run refresh_matviews_sql(instance, dialect=cfg.dialect) against the demo DB.

Return type:

None

Parameters:

cfg (Config)
instance (L2Instance)
dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)

The schema helper picks the right shape per dialect:

PG / Oracle: REFRESH MATERIALIZED VIEW + ANALYZE per name.
SQLite (matview-as-table): DROP TABLE + CREATE TABLE … AS per name (re-runs the matview body).

No-op safe — the SQL is dependency-ordered + idempotent at the schema level. Sync DB-API work runs in asyncio.to_thread so the studio’s POST /deploy doesn’t block other requests for the refresh duration (matview refresh is the slowest pipeline step on a multi-million-row demo).

async recon_gen.common.l2.deploy_pipeline.step_5_reload(*, dev_log=None)[source]

Bump _data_generation_id by one and emit the new value.

Returns the post-bump value so the deploy summary can surface “data_generation_id: 7”. This is the cheapest pipeline step — no DB access, no I/O, just an integer increment under the asyncio event loop’s single-threaded guarantee.

Return type:: int
Parameters:: dev_log (Callable[[Mapping[str, object]], Awaitable[None]] | None)