recon_gen.common.spine.scenario_context

ScenarioContext — composition safety + cleanup attribution (AV.5).

Promoted from the scenario-context-spike branch. The spike’s central finding (“daily_balances has no metadata column → use a side table”) was resolved by Phase AV.1 (the column was renamed from limits to metadata and demoted the per-rail caps to a nested key). Both base tables now carry the symmetric metadata column, so the production version tags PER ROW on both tables — no sidecar table.

What this module solves (the three user-flagged concerns from the 2026-05-23 spike planning):

1. Same-class collision. Two DriftGenerator instances on the same role plant into the same account_id ⇒ PK collision at INSERT time, silent data masking otherwise. ScenarioContext.compose() pre-checks pairwise disjoint claimed_accounts across the composed generators and raises a clear error naming both classes.

2. Cross-scenario interference. Apply scenario A; apply scenario B that claims one of A’s accounts ⇒ B would overwrite or compound A’s plant without warning. compose() runs a portable JSON_VALUE(metadata, '$.scenario_id') SELECT against both base tables and refuses if any of B’s claims overlap rows already tagged with a different scenario_id.

3. Cleanup attribution. Multiple scenarios in one DB need surgical tear-down. ScenarioContext.cleanup() deletes every row whose metadata.scenario_id matches — no sidecar bookkeeping, the tag itself IS the bookkeeping.

API shape (intentionally minimal):

• ScenarioContext(scenario_id="...") — frozen dataclass owning the scenario_id string.

• ctx.compose(conn, *generators, dialect=Dialect.DUCKDB) — runs the pre-checks, calls each generator’s emit(conn, scenario_id=ctx.scenario_id), commits.

• ctx.cleanup(conn, dialect=Dialect.DUCKDB) — DELETE FROM both base tables WHERE metadata.scenario_id matches; returns total rowcount.

• ClaimedAccountsGenerator Protocol — extension of ViolationGenerator adding claimed_accounts: frozenset[str] and the scenario_id kwarg on emit.

Backward compat — every existing call site stays untouched:

• generator.emit(conn) still works (scenario_id defaults to None).

• When scenario_id is None, the generator skips the metadata tagging entirely → byte-identical to pre-AV.5 emission. AS.5’s semantic_lock and AT.6’s training scenarios both keep working without changes.

• Only when a caller threads ScenarioContext does tagging happen.

Functions

dry_run_capture([dialect])	Return a fresh dry-run capture conn for the given dialect.
scenario_metadata(scenario_id, **extra)	Build the JSON metadata blob a tagged plant row carries.

Classes

ClaimedAccountsGenerator(*args, **kwargs)	Extension of ViolationGenerator: declares the account_ids the plant will touch + accepts a scenario_id kwarg on emit.
ScenarioContext(scenario_id[, prefix, dialect])	Owns a scenario_id; orchestrates a multi-generator plant with compose-time collision detection + per-row metadata tagging on both base tables for surgical cleanup.

class recon_gen.common.spine.scenario_context.ClaimedAccountsGenerator(*args, **kwargs)

Bases: Protocol

Extension of ViolationGenerator: declares the account_ids the plant will touch + accepts a scenario_id kwarg on emit.

The Protocol stays minimal — concrete generators add this property derived from their construction kwargs. emit’s scenario_id is keyword-only and defaults to None (preserves untagged behavior for existing call sites).

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)

Return type:

None

Parameters:

• conn (SyncConnection)

• scenario_id (str | None)

property intended: Violation | None

class recon_gen.common.spine.scenario_context.ScenarioContext(scenario_id, prefix='spec_example', dialect=Dialect.DUCKDB)

Bases: object

Owns a scenario_id; orchestrates a multi-generator plant with compose-time collision detection + per-row metadata tagging on both base tables for surgical cleanup.

AV.5 — promoted from the spike. The spike used a side-table (<prefix>_scenario_claims) because daily_balances had no metadata column; AV.1 fixed that, so the production version tags PER ROW on both tables and drops the sidecar entirely.

prefix is the deployment’s table prefix (matches cfg.db_table_prefix). dialect selects the SQL JSON path helper. Defaults match the in-process SQLite test harness — every AS/AT/AU unit test that builds a generator + composes via ScenarioContext can pass nothing.

Parameters:

• scenario_id (str)

• prefix (str)

• dialect (Dialect)

cleanup(conn)

Delete every row on either base table whose metadata.scenario_id matches this scenario. Returns the total rowcount across both tables.

No sidecar table to consult — the tag IS the bookkeeping (AV.1 unlocked this). Other scenarios on overlapping accounts survive because their rows carry a different scenario_id.

Return type:

int

Parameters:

conn (SyncConnection)

compose(conn, *generators, dry_run=False)

Pre-compose checks + emit all generators with the scenario_id threaded through + commit.

Two checks run before any emit fires:

1. Pairwise disjoint claims across the composed generators (catches “two DriftGenerators on the same role” at the wiring site, not at the DB-level PK violation).

2. Cross-scenario non-overlap against the data tables’ metadata.scenario_id (catches “scenario B overwrites scenario A’s rows” before any INSERT runs).

On success: each generator’s emit(conn, scenario_id=self.scenario_id) runs in order; the generator is responsible for tagging every row it writes.

AY.4.a — dry_run=True mode: conn must be a dry_run_capture(dialect) instance. The cross-scenario check is SKIPPED in dry-run (no real DB state exists to check against; the pairwise-disjoint check still fires since it’s a pure-data check on the generators themselves). Returns the captured [(sql, params), ...] list — pair with the AY.4.b renderer to produce static SQL text for the build_full_seed_sql path.

Live mode (default): returns None, commits the conn.

Return type:

list[tuple[str, tuple[object, ...]]] | None

Parameters:

• conn (SyncConnection)

• generators (ClaimedAccountsGenerator)

• dry_run (bool)

dialect: Dialect = 'duckdb'

prefix: str = 'spec_example'

scenario_id: str

recon_gen.common.spine.scenario_context.dry_run_capture(dialect=Dialect.DUCKDB)

Return a fresh dry-run capture conn for the given dialect.

The returned object satisfies the dbapi 2.0 shape insert_tx / insert_balance rely on: cursor() → cursor with execute(sql, params), commit() / close() no-op. captured accumulates every emitted (sql, params) pair in arrival order.

Pair with ScenarioContext.compose(conn=dry_run_capture(d), …, dry_run=True) (AY.4.a) + the AY.4.b renderer.

Return type:

_DryRunBase

Parameters:

dialect (Dialect)

recon_gen.common.spine.scenario_context.scenario_metadata(scenario_id, **extra)

Build the JSON metadata blob a tagged plant row carries.

Centralized so the cleanup query can extract via a portable JSON_VALUE(metadata, '$.scenario_id') and every tagging caller writes the same shape. Extra kwargs round-trip — useful for per-generator attribution (e.g. generator='DriftGenerator').

Return type:

str

Parameters:

• scenario_id (str)

• extra (object)