tg_model.analysis

Multi-run orchestration: sweeps, variant comparison, value-graph propagation (Phase 5).

impact / dependency_impact() report dependency reachability on the value graph only — not program-wide “engineering impact” (see tg_model.analysis.impact).

class tg_model.analysis.CapturedSlotOutput(value, present_in_run_outputs)

Bases: object

Resolved output for one output_paths entry in VariantComparisonRow.

Parameters:

• value (Any | None)

• present_in_run_outputs (bool)

value: Any | None

present_in_run_outputs: bool

True iff the slot’s stable_id was present in RunResult.outputs.

property realized: bool

exception tg_model.analysis.CompareVariantsValidationError(label, failures)

Bases: Exception

Raised when validate_graph() fails for one variant.

Parameters:

• label (str)

• failures (list[str])

Return type:

None

class tg_model.analysis.ImpactReport(changed_paths, upstream_slot_ids, downstream_slot_ids)

Bases: object

Value-graph reachability summary from a set of changed slots.

Parameters:

• changed_paths (tuple[str, ...])

• upstream_slot_ids (frozenset[str])

• downstream_slot_ids (frozenset[str])

changed_paths: tuple[str, ...]

upstream_slot_ids: frozenset[str]

downstream_slot_ids: frozenset[str]

class tg_model.analysis.SweepRecord(index, inputs, result)

Bases: object

One row from sweep() / sweep_async().

Parameters:

• index (int)

• inputs (dict[str, Any])

• result (RunResult)

index: int

inputs: dict[str, Any]

result: RunResult

class tg_model.analysis.VariantComparisonRow(label, outputs, result)

Bases: object

One scenario row from compare_variants().

Parameters:

• label (str)

• outputs (dict[str, CapturedSlotOutput])

• result (RunResult)

label: str

outputs: dict[str, CapturedSlotOutput]

result: RunResult

tg_model.analysis.compare_variants(*, scenarios, output_paths, validate_before_run=True, require_same_root_definition_type=False)

Evaluate each (label, configured_model, inputs) with a fresh graph and context.

Parameters:

• scenarios (Sequence[tuple[str, ConfiguredModel, Mapping[str, Any]]]) – Sequence of (label, configured_model, inputs) tuples.

• output_paths (Sequence[str]) – Dotted paths that must resolve to ValueSlot.

• validate_before_run (bool, default True) – Run validate_graph() before each evaluation.

• require_same_root_definition_type (bool, default False) – When True, all roots must share the same Python type.

Returns:

One row per scenario in input order.

Return type:

list of VariantComparisonRow

Raises:

• CompareVariantsValidationError – When validation fails for a labeled scenario.

• ValueError – When require_same_root_definition_type is violated.

• TypeError – When an output_paths entry does not resolve to a ValueSlot.

Notes

inputs maps ValueSlot.stable_id strings to values, same as evaluate(). CapturedSlotOutput distinguishes absent outputs from None values via present_in_run_outputs.

async tg_model.analysis.compare_variants_async(*, scenarios, output_paths, validate_before_run=True, require_same_root_definition_type=False)

Async variant of compare_variants() (uses evaluate_async per scenario).

Raises:

CompareVariantsValidationError, ValueError, TypeError – Same families as compare_variants().

Parameters:

• scenarios (Sequence[tuple[str, ConfiguredModel, Mapping[str, Any]]])

• output_paths (Sequence[str])

• validate_before_run (bool)

• require_same_root_definition_type (bool)

Return type:

list[VariantComparisonRow]

tg_model.analysis.dependency_impact(graph, changed, *, upstream=True, downstream=True)

Return other value slots reachable from changed on the value graph.

Parameters:

• graph (DependencyGraph) – Compiled graph for the configuration under study.

• changed (sequence of ValueSlot) – Slots whose perturbation you want to analyze.

• upstream (bool, default True) – Include reachability in each direction.

• downstream (bool, default True) – Include reachability in each direction.

Returns:

Excludes the changed slots’ own stable_id values from the sets.

Return type:

ImpactReport

Raises:

ValueError – If a changed slot does not map to a val:<path> node in graph.

Notes

This is dependency reachability only, not full engineering impact (see module docstring).

tg_model.analysis.impact(graph, changed, *, upstream=True, downstream=True)

Return other value slots reachable from changed on the value graph.

Parameters:

• graph (DependencyGraph) – Compiled graph for the configuration under study.

• changed (sequence of ValueSlot) – Slots whose perturbation you want to analyze.

• upstream (bool, default True) – Include reachability in each direction.

• downstream (bool, default True) – Include reachability in each direction.

Returns:

Excludes the changed slots’ own stable_id values from the sets.

Return type:

ImpactReport

Raises:

ValueError – If a changed slot does not map to a val:<path> node in graph.

Notes

This is dependency reachability only, not full engineering impact (see module docstring).

tg_model.analysis.sweep(*, graph, handlers, parameter_values, configured_model=None, prune_to_slots=None, collect=True, sink=None)

Cartesian product over parameter_values; one synchronous evaluation per tuple.

Parameters:

• graph (DependencyGraph) – From compile_graph().

• handlers (dict[str, Any]) – From compile_graph().

• parameter_values (Mapping[ValueSlot, Sequence[Any]]) – Maps each parameter ValueSlot to a sequence of values (axes). Dimension order is sorted by stable_id.

• configured_model (ConfiguredModel, optional) – When passed, asserts sweep slots match the graph (coherence check).

• prune_to_slots (sequence of ValueSlot, optional) – Restricts to upstream closure of these slots (see module warnings).

• collect (bool, default True) – When False, return an empty list and stream via sink.

• sink (callable, optional) – Receives each SweepRecord when provided.

Returns:

All samples when collect is True.

Return type:

list of SweepRecord

Raises:

ValueError – If collect=False without sink, or prune targets are not graph nodes.

async tg_model.analysis.sweep_async(*, configured_model, graph, handlers, parameter_values, prune_to_slots=None, collect=True, sink=None)

Like sweep() but awaits evaluate_async().

Parameters:

• configured_model (ConfiguredModel) – Required for async externals and always used for coherence checks.

• graph (DependencyGraph) – Same semantics as sweep().

• handlers (dict[str, Any]) – Same semantics as sweep().

• parameter_values (Mapping[ValueSlot, Sequence[Any]]) – Same semantics as sweep().

• prune_to_slots (Sequence[ValueSlot] | None) – Same semantics as sweep().

• collect (bool) – Same semantics as sweep().

• sink (Callable[[SweepRecord], None] | None) – Same semantics as sweep().

Returns:

Same as sweep().

Return type:

list of SweepRecord

Raises:

ValueError – Same as sweep().

tg_model.analysis.value_graph_propagation(graph, changed, *, upstream=True, downstream=True)

Return other value slots reachable from changed on the value graph.

Parameters:

• graph (DependencyGraph) – Compiled graph for the configuration under study.

• changed (sequence of ValueSlot) – Slots whose perturbation you want to analyze.

• upstream (bool, default True) – Include reachability in each direction.

• downstream (bool, default True) – Include reachability in each direction.

Returns:

Excludes the changed slots’ own stable_id values from the sets.

Return type:

ImpactReport

Raises:

ValueError – If a changed slot does not map to a val:<path> node in graph.

Notes

This is dependency reachability only, not full engineering impact (see module docstring).