From df47c8850130ffa33c94a85c9c8ee4c1b795e485 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 21 May 2026 10:10:53 -0400 Subject: [PATCH 1/3] test(dataflow): failing tests for cascade observability (#822) Add tests that assert the contract of the cache_timing channel: one log line per cascade observer per fire, every line carries a per-cascade correlation id, all lines from one cascade share that id, ids are monotonic across cascades, the channel is at DEBUG level (not INFO), and the per-line format is uniform. These tests fail on main because no cascade observer currently emits on buckaroo.dataflow.cache_timing. --- .../dataflow/cascade_observability_test.py | 198 ++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 tests/unit/dataflow/cascade_observability_test.py diff --git a/tests/unit/dataflow/cascade_observability_test.py b/tests/unit/dataflow/cascade_observability_test.py new file mode 100644 index 000000000..06f70c1df --- /dev/null +++ b/tests/unit/dataflow/cascade_observability_test.py @@ -0,0 +1,198 @@ +"""Tests for cascade observability — issue #822. + +The dataflow cascade (``_operation_result`` → ``cleaned`` → +``_processed_result`` → ``_summary_sd`` → ``_merged_sd`` → +``_populate_sd_cache``) is the hot path for every state_change. We want +one log line per observer per fire, with elapsed_ms, the trait change +that triggered it, and a per-cascade correlation id, on a dedicated +``buckaroo.dataflow.cache_timing`` channel at DEBUG level. + +These tests assert the observability surface, not cascade behavior — +they only inspect ``caplog`` records on the cache_timing channel. +""" +import logging +import re + +import pandas as pd +import pytest + +from buckaroo import BuckarooWidget +from buckaroo.customizations.analysis import DefaultSummaryStats, PdCleaningStats +from buckaroo.customizations.pandas_commands import ( + DropCol, FillNA, GroupBy, NoOp, SafeInt, Search) +from buckaroo.customizations.pd_autoclean_conf import NoCleaningConf +from buckaroo.dataflow.autocleaning import AutocleaningConfig, PandasAutocleaning +from buckaroo.pluggable_analysis_framework.col_analysis import ColAnalysis + + +CACHE_TIMING_CHANNEL = "buckaroo.dataflow.cache_timing" + +# Observers covered by the cascade observability layer. Every member of +# this set must emit at least one log line during a normal state_change +# cycle. +CASCADE_OBSERVERS = {'_sampled_df', '_operation_result', '_processed_result', '_summary_sd', '_merged_sd', + '_populate_sd_cache', '_widget_config', '_handle_widget_change'} + +# Match the per-step log format. Keep the regex strict so format drift +# trips a test rather than silently degrading the log channel. +LINE_RE = re.compile( + r'^cascade ' + r'cid=(?P\d+) ' + r'observer=(?P[A-Za-z_][A-Za-z0-9_]*) ' + r'trait=(?P[A-Za-z_][A-Za-z0-9_]*) ' + r'elapsed_ms=(?P\d+\.\d+)$') + + +class _CleaningGenOps(ColAnalysis): + requires_summary = ['int_parse_fail', 'int_parse'] + provides_defaults = {'cleaning_ops': []} + + @classmethod + def computed_summary(kls, column_metadata): + if column_metadata['int_parse'] > 0.3: + return { + 'cleaning_ops': [ + {'symbol': 'safe_int', 'meta': {'auto_clean': True}}, + {'symbol': 'df'}, + ], + 'add_orig': True, + } + return {'cleaning_ops': []} + + +class _Conf(AutocleaningConfig): + autocleaning_analysis_klasses = [DefaultSummaryStats, _CleaningGenOps, PdCleaningStats] + command_klasses = [DropCol, FillNA, GroupBy, NoOp, SafeInt, Search] + quick_command_klasses = [Search] + name = 'default' + + +class _CascadeWidget(BuckarooWidget): + autocleaning_klass = PandasAutocleaning + autoclean_conf = (_Conf, NoCleaningConf) + + +@pytest.fixture +def dirty_df(): + return pd.DataFrame({'a': [10, 20, 30, 40, 10, 20.3, 5, None, None, None], + 'b': ['3', '4', 'a', '5', '5', 'b', 'b', None, None, None]}) + + +def _parse_lines(records): + parsed = [] + for rec in records: + if rec.name != CACHE_TIMING_CHANNEL: + continue + m = LINE_RE.match(rec.getMessage()) + assert m is not None, ( + f"cache_timing log line does not match expected format: {rec.getMessage()!r}" + ) + parsed.append({'cid': int(m.group('cid')), 'observer': m.group('observer'), 'trait': m.group('trait'), + 'elapsed_ms': float(m.group('elapsed')), 'level': rec.levelname}) + return parsed + + +def test_cache_timing_channel_emits_at_debug_level(dirty_df, caplog): + """All cache_timing log lines should be DEBUG level (channel must + not flood at INFO during normal operation).""" + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): + bw = _CascadeWidget(dirty_df, debug=False) + bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['needle']}} + + parsed = _parse_lines(caplog.records) + assert parsed, "expected cache_timing log lines but got none" + assert all(p['level'] == 'DEBUG' for p in parsed), ( + f"all cache_timing lines must be DEBUG, got levels " + f"{sorted({p['level'] for p in parsed})}" + ) + + +def test_cache_timing_channel_silent_at_info_level(dirty_df, caplog): + """The cache_timing channel must not emit at INFO — it would flood + normal widget operation otherwise.""" + with caplog.at_level(logging.INFO, logger=CACHE_TIMING_CHANNEL): + _CascadeWidget(dirty_df, debug=False) + + info_lines = [r for r in caplog.records + if r.name == CACHE_TIMING_CHANNEL and r.levelno >= logging.INFO] + assert info_lines == [], ( + f"cache_timing channel emitted at INFO+: {[r.getMessage() for r in info_lines]}" + ) + + +def test_every_cascade_observer_logs_during_state_change(dirty_df, caplog): + """Each cascade observer must emit at least one cache_timing line + during a state_change cycle.""" + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): + bw = _CascadeWidget(dirty_df, debug=False) + # A filter flip exercises the full cascade. + bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['needle']}} + + parsed = _parse_lines(caplog.records) + observers_seen = {p['observer'] for p in parsed} + missing = CASCADE_OBSERVERS - observers_seen + assert not missing, ( + f"cascade observers missing cache_timing coverage: {sorted(missing)}; " + f"saw {sorted(observers_seen)}" + ) + + +def test_one_cascade_shares_one_correlation_id(dirty_df, caplog): + """A single ``buckaroo_state`` flip triggers one cascade — every log + line emitted by it must share the same correlation id.""" + # Build the widget first so its construction cascade is excluded + # from the assertion. + bw = _CascadeWidget(dirty_df, debug=False) + + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): + bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['needle']}} + + parsed = _parse_lines(caplog.records) + assert parsed, "expected cascade log lines for the state flip" + + cids = {p['cid'] for p in parsed} + # The buckaroo_state change can drive at most a single cascade per + # top-level set on the dataflow (post_processing / cleaning_method / + # quick_command_args). We're flipping only quick_command_args so the + # cascade is a single rooted tree — one correlation id. + assert len(cids) == 1, ( + f"one cascade should share one correlation id, got cids={sorted(cids)} " + f"across observers={sorted({p['observer'] for p in parsed})}" + ) + + +def test_correlation_id_is_monotonic(dirty_df, caplog): + """Successive cascades must mint strictly-increasing correlation + ids — otherwise observers from different cascades collide in the + logs.""" + bw = _CascadeWidget(dirty_df, debug=False) + + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): + bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['n1']}} + first_cycle = _parse_lines(caplog.records) + caplog.clear() + bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['n2']}} + second_cycle = _parse_lines(caplog.records) + + assert first_cycle and second_cycle, "both cycles must produce log lines" + max_first = max(p['cid'] for p in first_cycle) + min_second = min(p['cid'] for p in second_cycle) + assert min_second > max_first, ( + f"correlation ids must increase across cascades — " + f"first max={max_first}, second min={min_second}" + ) + + +def test_log_format_is_uniform(dirty_df, caplog): + """Every cache_timing line must match the documented format — + cascade cid= observer= trait= elapsed_ms=.""" + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): + _CascadeWidget(dirty_df, debug=False) + + records = [r for r in caplog.records if r.name == CACHE_TIMING_CHANNEL] + assert records, "expected cache_timing records" + for rec in records: + msg = rec.getMessage() + assert LINE_RE.match(msg), ( + f"cache_timing line does not match documented format: {msg!r}" + ) From 7cfac31c70076b6b64f75da7166c2c13ea332f11 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 21 May 2026 10:17:45 -0400 Subject: [PATCH 2/3] feat(dataflow): cascade observability with correlation id (#822) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Standardise the buckaroo.dataflow.cache_timing channel so the cascade is observable end to end. Each cascade observer wraps its body in a ``self._cascade_step(observer_name, change)`` context manager that: - mints a fresh correlation id (module-level monotonic counter) when it is the outermost observer in a cascade, reuses it for nested observers - times the call with time.perf_counter - emits one DEBUG line at exit, format: cascade cid= observer= trait= elapsed_ms= Observers instrumented: _sampled_df, _operation_result, _processed_result, _summary_sd, _merged_sd (base + CustomizableDataflow override), _populate_sd_cache, _widget_config, _handle_widget_change. The channel is at DEBUG so it does not flood normal operation; enable with ``logging.getLogger("buckaroo.dataflow.cache_timing").setLevel( DEBUG)``. No cascade behavior change — purely observability. The known _populate_sd_cache double-fire pattern is now visible in the log stream as two lines sharing one cid. --- buckaroo/dataflow/dataflow.py | 379 ++++++++++++++++++++-------------- 1 file changed, 225 insertions(+), 154 deletions(-) diff --git a/buckaroo/dataflow/dataflow.py b/buckaroo/dataflow/dataflow.py index 3c2790f89..a07de990d 100644 --- a/buckaroo/dataflow/dataflow.py +++ b/buckaroo/dataflow/dataflow.py @@ -1,3 +1,7 @@ +from contextlib import contextmanager +import itertools +import logging +import time from typing import List, Literal, Tuple, Type, TypedDict, Dict as TDict, Any as TAny, Union from typing_extensions import override import six @@ -24,6 +28,30 @@ from .sd_cache import hash_chain, split_chain_by_scope +# Cascade observability — issue #822. +# +# Every cascade observer in this module emits one DEBUG line on the +# ``buckaroo.dataflow.cache_timing`` channel per fire, with a stable +# per-cascade correlation id (``cid``) so the lines from one external +# trait change can be grouped — and so the known double-fire patterns +# (e.g. ``_populate_sd_cache`` running twice per state_change because it +# observes both ``summary_sd`` and ``operations``) become visible. +# +# A "cascade" is the synchronous tree of observer notifications rooted +# at one external trait set. traitlets dispatches observers +# synchronously from inside ``set()``, so the tree maps onto a single +# call stack — the outermost observer mints a new ``cid`` and every +# observer nested inside it (via the trait sets it performs) reuses +# the same id. When the outermost observer returns, the id is cleared. +# +# ``itertools.count`` is atomic for inc in CPython, so the counter is +# safe across threads; the per-instance ``_cascade_id`` attribute is +# only mutated from the cascade root, which is single-threaded with +# respect to the cascade it owns. +cache_timing_logger = logging.getLogger("buckaroo.dataflow.cache_timing") +_cascade_counter = itertools.count(1) + + class DfTrait(Any): """Any-trait for values that may contain a pandas/polars DataFrame. @@ -68,6 +96,7 @@ class DataFlow(ABCDataflow): """ def __init__(self, raw_df): self.exception = None + self._cascade_id = None super().__init__() self.summary_sd = {} self.existing_operations = [] @@ -78,6 +107,39 @@ def __init__(self, raw_df): except Exception: six.reraise(self.exception[0], self.exception[1], self.exception[2]) + @contextmanager + def _cascade_step(self, observer_name, change): + """Time one cascade observer and emit one log line on exit. + + First observer in a cascade mints a fresh correlation id from + the module-level monotonic counter; nested observers (the + synchronous tree of further trait sets) reuse the same id. + The cascade root clears the id on exit so the next external + trait change starts a fresh cascade. + + ``change`` is the traitlets observer payload — we only read + ``change['name']`` (the trait that fired this observer); the + helper tolerates any falsy/missing change so the same call + site works for synthetic invocations. + """ + started_cascade = self._cascade_id is None + if started_cascade: + self._cascade_id = next(_cascade_counter) + cid = self._cascade_id + trait_name = '' + if isinstance(change, dict): + trait_name = change.get('name', '') or '' + t0 = time.perf_counter() + try: + yield cid + finally: + elapsed_ms = (time.perf_counter() - t0) * 1000.0 + cache_timing_logger.debug( + "cascade cid=%d observer=%s trait=%s elapsed_ms=%.3f", + cid, observer_name, trait_name, elapsed_ms) + if started_cascade: + self._cascade_id = None + autocleaning_klass = SentinelAutocleaning autoclean_conf = tuple() @@ -142,7 +204,8 @@ def _compute_sampled_df(self, raw_df:pd.DataFrame, sample_method:str): @observe('raw_df', 'sample_method') @exception_protect('sampled_df-protector') def _sampled_df(self, _change:Any) -> None: - self.sampled_df = self._compute_sampled_df(self.raw_df, self.sample_method) + with self._cascade_step('_sampled_df', _change): + self.sampled_df = self._compute_sampled_df(self.raw_df, self.sample_method) @observe('sampled_df', 'cleaning_method', 'quick_command_args', 'operations') @exception_protect('operation_result-protector') @@ -178,16 +241,17 @@ def _operation_result(self, _change:Any) -> None: return self._in_operation_result = True try: - result = self.ac_obj.handle_ops_and_clean( - self.sampled_df, self.cleaning_method, self.quick_command_args, self.operations) - - if result is None: - return - else: - self.cleaned = result - self.operations = result[3] - self.operation_results = {'transformed_df':None, - 'generated_py_code': self.generated_code} + with self._cascade_step('_operation_result', _change): + result = self.ac_obj.handle_ops_and_clean( + self.sampled_df, self.cleaning_method, self.quick_command_args, self.operations) + + if result is None: + return + else: + self.cleaned = result + self.operations = result[3] + self.operation_results = {'transformed_df':None, + 'generated_py_code': self.generated_code} finally: self._in_operation_result = False @@ -221,9 +285,10 @@ def populate_df_meta(self): @observe('cleaned', 'post_processing_method') @exception_protect('processed_result-protector') def _processed_result(self, change): - #for now this is a no-op because I don't have a post_processing_function or mechanism - self.processed_result = self._compute_processed_result(self.cleaned_df, self.post_processing_method) - self.populate_df_meta() + with self._cascade_step('_processed_result', change): + #for now this is a no-op because I don't have a post_processing_function or mechanism + self.processed_result = self._compute_processed_result(self.cleaned_df, self.post_processing_method) + self.populate_df_meta() @property def processed_df(self): @@ -253,39 +318,42 @@ def _get_summary_sd(self, df:pd.DataFrame) -> Tuple[SDType, TAny]: @observe('processed_result', 'analysis_klasses') @exception_protect('summary_sd-protector') def _summary_sd(self, change): - # Dedupe: the autocleaning operations cascade re-fires - # _operation_result → cleaned → processed_result with a freshly-built - # tuple wrapper, which makes this observer fire twice per widget - # construction even when processed_df identity is unchanged. - # Skip when neither the dataframe nor analysis_klasses has actually - # changed since the last run. See issue #709. - df = self.processed_df - klasses = self.analysis_klasses - if (id(df), id(klasses)) == self._summary_sd_cache_key: - return - self._summary_sd_cache_key = (id(df), id(klasses)) - result_summary_sd, errs = self._get_summary_sd(df) - self.summary_sd = result_summary_sd - self.errs = errs + with self._cascade_step('_summary_sd', change): + # Dedupe: the autocleaning operations cascade re-fires + # _operation_result → cleaned → processed_result with a freshly-built + # tuple wrapper, which makes this observer fire twice per widget + # construction even when processed_df identity is unchanged. + # Skip when neither the dataframe nor analysis_klasses has actually + # changed since the last run. See issue #709. + df = self.processed_df + klasses = self.analysis_klasses + if (id(df), id(klasses)) == self._summary_sd_cache_key: + return + self._summary_sd_cache_key = (id(df), id(klasses)) + result_summary_sd, errs = self._get_summary_sd(df) + self.summary_sd = result_summary_sd + self.errs = errs @observe('summary_sd', 'processed_result') @exception_protect('merged_sd-protector') def _merged_sd(self, change): - #slightly inconsitent that processed_sd gets priority over - #summary_sd, given that processed_df is computed first. My - #thinking was that processed_sd has greater total knowledge - #and should supersede summary_sd. + with self._cascade_step('_merged_sd', change): + #slightly inconsitent that processed_sd gets priority over + #summary_sd, given that processed_df is computed first. My + #thinking was that processed_sd has greater total knowledge + #and should supersede summary_sd. + + self.merged_sd = merge_sds(self.cleaned_sd, self.summary_sd, self.processed_sd) - self.merged_sd = merge_sds(self.cleaned_sd, self.summary_sd, self.processed_sd) - @observe('merged_sd', 'style_method') @exception_protect('widget_config-protector') def _widget_config(self, change): - #how to control ordering of column_config??? - # dfviewer_config = self._get_dfviewer_config(self.merged_sd, self.style_method) - # self.widget_args_tuple = [self.processed_df, self.merged_sd, dfviewer_config] - self.widget_args_tuple = (id(self.processed_df), self.processed_df, self.merged_sd) + with self._cascade_step('_widget_config', change): + #how to control ordering of column_config??? + # dfviewer_config = self._get_dfviewer_config(self.merged_sd, self.style_method) + # self.widget_args_tuple = [self.processed_df, self.merged_sd, dfviewer_config] + self.widget_args_tuple = (id(self.processed_df), self.processed_df, self.merged_sd) BuckarooOptions = TypedDict('BuckarooOptions', { 'sampled': List[str], @@ -408,57 +476,58 @@ def setup_options_from_analysis(self): @observe('summary_sd', 'processed_result', 'filt_sd_key') @exception_protect('merged_sd-protector') def _merged_sd(self, change): - # Bare keys come from the raw scope's SD (computed on - # sampled_df). ``filtered_*`` keys are layered on top from the - # filt scope's SD when the filter is active. Scope SDs are read - # from the keyed cache (#783) — the cache observer - # ``_populate_sd_cache`` is what computes and stores them; this - # observer just assembles the wire shape #777's `?key` JS - # consumes. - # - # filt_sd_key is in the observed set so this fires after - # ``_populate_sd_cache`` has updated the pointer (which it - # always does, even on a pure cache hit) — guarantees the - # cache lookups below see the right keys for the current state. - - # Resolve scope SDs. Falls back to summary_sd / cleaned_sd - # for pre-cache-population states (initial startup, the brief - # window before _populate_sd_cache has fired). - cache = self.summary_stats_cache or {} - raw_sd = cache.get(self.raw_sd_key) if self.raw_sd_key else None - if raw_sd is None: - raw_sd = self.summary_sd or {} - filt_sd = cache.get(self.filt_sd_key) if self.filt_sd_key else None - if filt_sd is None: - filt_sd = self.summary_sd or {} - - # ``filtered_*`` keys reflect "search filter applied on top of - # cleaning", so the gate is "filt chain has ops the clean chain - # doesn't" — i.e. at least one quick-command op is present. Keying - # off ``filt_sd_key != raw_sd_key`` would also fire for - # cleaning-only states, mislabelling cleaned stats as filtered - # until the deferred ``cleaned_*`` scope lands. - chains = split_chain_by_scope(self.operations) - filter_active = chains['filt'] != chains['clean'] - - if self.processed_df is None: - #on initial startup - self.merged_sd = merge_sds(self.init_sd, self.cleaned_sd, raw_sd, self.processed_sd) - return + with self._cascade_step('_merged_sd', change): + # Bare keys come from the raw scope's SD (computed on + # sampled_df). ``filtered_*`` keys are layered on top from the + # filt scope's SD when the filter is active. Scope SDs are read + # from the keyed cache (#783) — the cache observer + # ``_populate_sd_cache`` is what computes and stores them; this + # observer just assembles the wire shape #777's `?key` JS + # consumes. + # + # filt_sd_key is in the observed set so this fires after + # ``_populate_sd_cache`` has updated the pointer (which it + # always does, even on a pure cache hit) — guarantees the + # cache lookups below see the right keys for the current state. + + # Resolve scope SDs. Falls back to summary_sd / cleaned_sd + # for pre-cache-population states (initial startup, the brief + # window before _populate_sd_cache has fired). + cache = self.summary_stats_cache or {} + raw_sd = cache.get(self.raw_sd_key) if self.raw_sd_key else None + if raw_sd is None: + raw_sd = self.summary_sd or {} + filt_sd = cache.get(self.filt_sd_key) if self.filt_sd_key else None + if filt_sd is None: + filt_sd = self.summary_sd or {} + + # ``filtered_*`` keys reflect "search filter applied on top of + # cleaning", so the gate is "filt chain has ops the clean chain + # doesn't" — i.e. at least one quick-command op is present. Keying + # off ``filt_sd_key != raw_sd_key`` would also fire for + # cleaning-only states, mislabelling cleaned stats as filtered + # until the deferred ``cleaned_*`` scope lands. + chains = split_chain_by_scope(self.operations) + filter_active = chains['filt'] != chains['clean'] + + if self.processed_df is None: + #on initial startup + self.merged_sd = merge_sds(self.init_sd, self.cleaned_sd, raw_sd, self.processed_sd) + return - #we do this to get rewrtten keys for init_sd - rewritten_init_sd = merge_sd_overrides({}, self.processed_df, self.init_sd) - intermediate_sd = merge_sds(rewritten_init_sd, self.cleaned_sd, raw_sd) - base = merge_sd_overrides(intermediate_sd, self.processed_df, self.processed_sd) + #we do this to get rewrtten keys for init_sd + rewritten_init_sd = merge_sd_overrides({}, self.processed_df, self.init_sd) + intermediate_sd = merge_sds(rewritten_init_sd, self.cleaned_sd, raw_sd) + base = merge_sd_overrides(intermediate_sd, self.processed_df, self.processed_sd) - # Layer ``filtered_*`` keys on top when a filter is active. - if filter_active and filt_sd: - for col, stats in filt_sd.items(): - col_dict = base.setdefault(col, {}) - for stat_name, val in stats.items(): - col_dict[f'filtered_{stat_name}'] = val + # Layer ``filtered_*`` keys on top when a filter is active. + if filter_active and filt_sd: + for col, stats in filt_sd.items(): + col_dict = base.setdefault(col, {}) + for stat_name, val in stats.items(): + col_dict[f'filtered_{stat_name}'] = val - self.merged_sd = base + self.merged_sd = base def _compute_scope_df(self, scope: str): """Return the df that scope's SD should be computed against. @@ -545,35 +614,36 @@ def _populate_sd_cache(self, _change): are un-synced — the frontend consumes only the merged prefixed-key ``merged_sd``. """ - if self.processed_df is None: - return - chains = split_chain_by_scope(self.operations) - keys = {scope: self._scope_cache_key(chain) - for scope, chain in chains.items()} - new_cache = dict(self.summary_stats_cache) - cache_grew = False - - # filt scope reuses the SD that _summary_sd just produced. - if keys['filt'] not in new_cache: - new_cache[keys['filt']] = dict(self.summary_sd or {}) - cache_grew = True - - # raw + clean: fresh compute, but only on cache miss. - for scope in ('raw', 'clean'): - if keys[scope] in new_cache: - continue - scope_df = self._compute_scope_df(scope) - if scope_df is None: - continue - sd, _errs = self._get_summary_sd(scope_df) - new_cache[keys[scope]] = sd - cache_grew = True - - if cache_grew: - self.summary_stats_cache = new_cache - self.raw_sd_key = keys['raw'] - self.clean_sd_key = keys['clean'] - self.filt_sd_key = keys['filt'] + with self._cascade_step('_populate_sd_cache', _change): + if self.processed_df is None: + return + chains = split_chain_by_scope(self.operations) + keys = {scope: self._scope_cache_key(chain) + for scope, chain in chains.items()} + new_cache = dict(self.summary_stats_cache) + cache_grew = False + + # filt scope reuses the SD that _summary_sd just produced. + if keys['filt'] not in new_cache: + new_cache[keys['filt']] = dict(self.summary_sd or {}) + cache_grew = True + + # raw + clean: fresh compute, but only on cache miss. + for scope in ('raw', 'clean'): + if keys[scope] in new_cache: + continue + scope_df = self._compute_scope_df(scope) + if scope_df is None: + continue + sd, _errs = self._get_summary_sd(scope_df) + new_cache[keys[scope]] = sd + cache_grew = True + + if cache_grew: + self.summary_stats_cache = new_cache + self.raw_sd_key = keys['raw'] + self.clean_sd_key = keys['clean'] + self.filt_sd_key = keys['filt'] ### start code interpreter block def add_command(self, incomingCommandKls): @@ -658,46 +728,47 @@ def _handle_widget_change(self, change): """ put together df_dict for consumption by the frontend """ - # Tuple[TAny, pd.DataFrame, SDType] - _unused, processed_df, merged_sd = self.widget_args_tuple - if processed_df is None: - return + with self._cascade_step('_handle_widget_change', change): + # Tuple[TAny, pd.DataFrame, SDType] + _unused, processed_df, merged_sd = self.widget_args_tuple + if processed_df is None: + return - # df_data_dict is still hardcoded for now - # eventually processed_df will be able to add or alter values of df_data_dict - # correlation would be added, filtered would probably be altered - - # to expedite processing maybe future provided dfs from - # postprcoessing could default to empty until that is - # selected, optionally - if self.skip_main_serial: - self.df_data_dict = {'main': [], - 'all_stats': self._sd_to_jsondf(merged_sd), - 'empty': []} - else: - self.df_data_dict = {'main': self._df_to_obj(processed_df), - 'all_stats': self._sd_to_jsondf(merged_sd), - 'empty': []} - - temp_display_args = {} - for display_name, A_Klass in self.df_display_klasses.items(): - df_viewer_config = A_Klass.get_dfviewer_config(merged_sd, processed_df) - base_column_config = df_viewer_config['column_config'] - df_viewer_config['column_config'] = merge_column_config( - base_column_config, self.processed_df, self.column_config_overrides) - disp_arg = {'data_key': A_Klass.data_key, - 'df_viewer_config': df_viewer_config, - 'summary_stats_key': A_Klass.summary_stats_key} - temp_display_args[display_name] = disp_arg - - if self.pinned_rows is not None: - temp_display_args['main']['df_viewer_config']['pinned_rows'] = self.pinned_rows - if self.extra_grid_config: - temp_display_args['main']['df_viewer_config']['extra_grid_config'] = self.extra_grid_config - if self.component_config: - temp_display_args['main']['df_viewer_config']['component_config'] = self.component_config - - self.df_display_args = temp_display_args + # df_data_dict is still hardcoded for now + # eventually processed_df will be able to add or alter values of df_data_dict + # correlation would be added, filtered would probably be altered + + # to expedite processing maybe future provided dfs from + # postprcoessing could default to empty until that is + # selected, optionally + if self.skip_main_serial: + self.df_data_dict = {'main': [], + 'all_stats': self._sd_to_jsondf(merged_sd), + 'empty': []} + else: + self.df_data_dict = {'main': self._df_to_obj(processed_df), + 'all_stats': self._sd_to_jsondf(merged_sd), + 'empty': []} + + temp_display_args = {} + for display_name, A_Klass in self.df_display_klasses.items(): + df_viewer_config = A_Klass.get_dfviewer_config(merged_sd, processed_df) + base_column_config = df_viewer_config['column_config'] + df_viewer_config['column_config'] = merge_column_config( + base_column_config, self.processed_df, self.column_config_overrides) + disp_arg = {'data_key': A_Klass.data_key, + 'df_viewer_config': df_viewer_config, + 'summary_stats_key': A_Klass.summary_stats_key} + temp_display_args[display_name] = disp_arg + + if self.pinned_rows is not None: + temp_display_args['main']['df_viewer_config']['pinned_rows'] = self.pinned_rows + if self.extra_grid_config: + temp_display_args['main']['df_viewer_config']['extra_grid_config'] = self.extra_grid_config + if self.component_config: + temp_display_args['main']['df_viewer_config']['component_config'] = self.component_config + + self.df_display_args = temp_display_args """ From 42c5e9f1969c093c809dcbb94824ff1772b016a3 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 21 May 2026 15:32:57 -0400 Subject: [PATCH 3/3] test(cascade-observability): scope coverage assertion to state_change only Codex P2 on #826: ``test_every_cascade_observer_logs_during_state_change`` started ``caplog.at_level`` before ``_CascadeWidget(...)`` ran, so construction-time emissions counted toward the coverage assertion. A regression that silenced an observer during the state_change cascade could still satisfy the test if the same observer fired during construction. - Build the widget outside the ``caplog.at_level`` block (mirrors the pattern in ``test_one_cascade_shares_one_correlation_id``) so only state_change cascade emissions count. - Drop ``_sampled_df`` from the expected-observer set for this test: it's keyed on ``raw_df`` / ``sample_method``, neither of which a ``quick_command_args`` flip touches. Keeping it in the set would encode a false-positive contract. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../dataflow/cascade_observability_test.py | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/tests/unit/dataflow/cascade_observability_test.py b/tests/unit/dataflow/cascade_observability_test.py index 06f70c1df..d704f537c 100644 --- a/tests/unit/dataflow/cascade_observability_test.py +++ b/tests/unit/dataflow/cascade_observability_test.py @@ -122,15 +122,30 @@ def test_cache_timing_channel_silent_at_info_level(dirty_df, caplog): def test_every_cascade_observer_logs_during_state_change(dirty_df, caplog): """Each cascade observer must emit at least one cache_timing line - during a state_change cycle.""" + during a state_change cycle. + + Construction is performed outside ``caplog.at_level`` so its + initialization-time emissions don't satisfy the assertion. The + coverage claim is about the state_change cascade specifically — a + regression that silences (e.g.) ``_summary_sd`` during a filter flip + would otherwise hide behind its construction-time emission. + """ + # Build outside caplog so only state_change emissions count. + bw = _CascadeWidget(dirty_df, debug=False) + with caplog.at_level(logging.DEBUG, logger=CACHE_TIMING_CHANNEL): - bw = _CascadeWidget(dirty_df, debug=False) # A filter flip exercises the full cascade. bw.buckaroo_state = {**bw.buckaroo_state, 'quick_command_args': {'search': ['needle']}} parsed = _parse_lines(caplog.records) observers_seen = {p['observer'] for p in parsed} - missing = CASCADE_OBSERVERS - observers_seen + # ``_sampled_df`` only fires when ``raw_df`` or ``sample_method`` + # changes — a ``quick_command_args`` flip leaves both alone, so it + # legitimately does not log during a state_change cascade. Drop it + # from the expected set so the assertion reflects what the + # state_change path is *supposed* to emit. + expected = CASCADE_OBSERVERS - {'_sampled_df'} + missing = expected - observers_seen assert not missing, ( f"cascade observers missing cache_timing coverage: {sorted(missing)}; " f"saw {sorted(observers_seen)}"