Fix four latent pipeline bugs found in a full-pass review

prox/analytics.py

@@ -10,6 +10,20 @@
 logger = logging.getLogger(__name__)
 
 
+def _contains_any(series: pd.Series, values: list) -> pd.Series:
+    """Case-insensitive substring match against any of `values`.
+
+    An empty `values` list returns an all-False mask - '|'.join([]) would
+    otherwise produce the empty-string pattern '', and str.contains('')
+    matches every row, silently classifying everything as a match instead
+    of nothing.
+    """
+    if not values:
+        return pd.Series(False, index=series.index)
+    pattern = '|'.join(str(v).lower() for v in values)
+    return series.astype(str).str.lower().str.contains(pattern, na=False)
+
+
 def get_event_log_summary(event_log_df: pd.DataFrame) -> Tuple[Dict[str, Any] | None, list]:
     """
     Computes high-level statistics for an event log.
@@ -523,8 +537,7 @@ def analyze_repeat_purchases(
     # GA4-style logs commonly attach event_value/price to browsing events too
     # (view_item, add_to_cart), not just completed transactions, so "revenue > 0
     # somewhere in this case" previously misclassified cart-abandoners as buyers.
-    pattern = '|'.join([p.lower() for p in purchase_values])
-    mask_text = df[real_activity_col].astype(str).str.lower().str.contains(pattern, na=False)
+    mask_text = _contains_any(df[real_activity_col], purchase_values)
 
     mask_col = pd.Series(False, index=df.index)
     for flag_col in [c for c in df.columns if 'purchase' in c.lower() or 'conversion' in c.lower()]:
@@ -543,8 +556,7 @@ def analyze_repeat_purchases(
     # --- Cart detection (activity name or an explicit flag column) ---
     # Computed independently of purchase detection so abandonment can still be
     # reported even when zero purchases exist (100% abandonment).
-    cart_pattern = '|'.join([c.lower() for c in cart_values])
-    mask_cart_text = df[real_activity_col].astype(str).str.lower().str.contains(cart_pattern, na=False)
+    mask_cart_text = _contains_any(df[real_activity_col], cart_values)
 
     mask_cart_col = pd.Series(False, index=df.index)
     for flag_col in [c for c in df.columns if 'cart' in c.lower() or 'basket' in c.lower()]:
@@ -1096,15 +1108,10 @@ def classify_sessions(
         return pd.DataFrame(columns=empty_cols)
 
     df = df.copy()
-    activities_lower = df[real_activity_col].astype(str).str.lower()
-
-    purchase_pattern = '|'.join(p.lower() for p in purchase_values)
-    cart_pattern = '|'.join(c.lower() for c in cart_values)
-    research_pattern = '|'.join(r.lower() for r in research_keywords)
 
-    df['_is_purchase'] = activities_lower.str.contains(purchase_pattern, na=False)
-    df['_is_cart'] = activities_lower.str.contains(cart_pattern, na=False)
-    df['_is_research'] = activities_lower.str.contains(research_pattern, na=False)
+    df['_is_purchase'] = _contains_any(df[real_activity_col], purchase_values)
+    df['_is_cart'] = _contains_any(df[real_activity_col], cart_values)
+    df['_is_research'] = _contains_any(df[real_activity_col], research_keywords)
 
     grouped = df.groupby(real_session_col).agg(
         user_id=(real_user_col, 'first'),

prox/conformance.py

@@ -297,18 +297,29 @@ def _fitness_state_equation_alignments(
             nt.append(Event({'concept:name': str(event['concept:name'])}))
         clean_log.append(nt)
 
-    # Rebuild markings from net structure
-    rim = Marking()
-    rfm = Marking()
-    for p in process_model.places:
-        if not p.in_arcs:
-            rim[p] = 1
-        if not p.out_arcs:
-            rfm[p] = 1
+    # Use the real markings discovery/import already produced (same pattern
+    # as _fitness_token_replay above) rather than guessing from net topology.
+    # A "place with no in-arcs is the initial marking" heuristic happens to
+    # match for most discovered nets, but silently diverges from the actual
+    # im/fm for reference models built via build_structured_reference_model()
+    # or import_reference_model_bpmn() - e.g. any net with loops or multiple
+    # branches, where sourceless/sinkless places don't coincide with the real
+    # start/end. Only fall back to the topology guess if no markings were
+    # actually supplied.
+    rim = initial_marking if initial_marking else Marking()
+    rfm = final_marking if final_marking else Marking()
     if not rim:
-        rim[list(process_model.places)[0]] = 1
+        for p in process_model.places:
+            if not p.in_arcs:
+                rim[p] = 1
+        if not rim:
+            rim[list(process_model.places)[0]] = 1
     if not rfm:
-        rfm[list(process_model.places)[-1]] = 1
+        for p in process_model.places:
+            if not p.out_arcs:
+                rfm[p] = 1
+        if not rfm:
+            rfm[list(process_model.places)[-1]] = 1
 
     max_cores = max(1, os.cpu_count() - 1) if cores == 0 else cores
     params = {'cores': max_cores, 'ret_tuple_as_trans_desc': True}

prox/data_manager.py

@@ -319,12 +319,16 @@ def refine_activity_labels(
 
     context_values = df.loc[mask, context_column].fillna('unknown').astype(str)
 
-    if not context_values.empty:
-        first_val = str(context_values.iloc[0])
-        if 'http' in first_val or '/' in first_val:
-            context_values = context_values.str.split('?').str[0]
-            context_values = context_values.str.strip('/')
-            context_values = context_values.apply(lambda x: x.split('/')[-1] if '/' in x else x)
+    # URL-cleaning applied per row, not gated on whether the column "looks
+    # like" a URL overall - a single whole-column decision (previously based
+    # on just the first matched row's value) missed URL-like values in any
+    # later row once the first row happened to be plain text, and vice
+    # versa. These steps are no-ops for genuinely plain values (no '?' to
+    # split on, no leading/trailing '/' to strip, no '/' to take the last
+    # segment of), so applying them unconditionally is safe either way.
+    context_values = context_values.str.split('?').str[0]
+    context_values = context_values.str.strip('/')
+    context_values = context_values.apply(lambda x: x.split('/')[-1] if '/' in x else x)
 
     df.loc[mask, 'concept:name'] = target_activity + '_' + context_values.str.upper()
     return df

prox/discovery.py

@@ -12,7 +12,13 @@
 
 
 def _discover_inductive_miner(log, *, noise_threshold=0.2, **_ignored):
-    tree = inductive_miner.apply(log, parameters={'noise_threshold': noise_threshold})
+    # noise_threshold only has an effect under the IMf variant - the default
+    # IM variant silently ignores it (verified: IM produces byte-identical
+    # trees regardless of noise_threshold). IMf at noise_threshold=0.0
+    # matches plain IM exactly, so this is safe at the default too.
+    tree = inductive_miner.apply(
+        log, parameters={'noise_threshold': noise_threshold}, variant=inductive_miner.Variants.IMf
+    )
     net, im, fm = pt_converter.apply(tree, variant=pt_converter.Variants.TO_PETRI_NET)
     message = f"Petri net discovered via Inductive Miner (noise={noise_threshold})"
     return net, im, fm, message

tests/test_analytics.py

@@ -576,3 +576,20 @@ def test_summarize_user_journeys_empty_input_returns_empty():
     result = summarize_user_journeys(pd.DataFrame())
     assert result.empty
     assert list(result.columns) == ['user_id', 'session_count', 'journey']
+
+
+# --- _contains_any / empty keyword-list guard (regression) ---
+
+def test_analyze_repeat_purchases_empty_purchase_values_matches_nothing(tmp_path):
+    """Regression test: '|'.join([]) produces the empty-string pattern '',
+    and str.contains('') matches every row - purchase_values=[] used to
+    silently classify every case as a purchase instead of none."""
+    df = make_purchase_log()
+    result = analyze_repeat_purchases(df, purchase_values=[], output_folder=str(tmp_path))
+    assert result['metrics']['total_buyers'] == 0
+
+
+def test_classify_sessions_empty_keyword_lists_labels_everything_browsing():
+    df = make_session_intent_log()
+    labels = classify_sessions(df, purchase_values=[], cart_values=[], research_keywords=[])
+    assert set(labels['label']) == {'Browsing'}

tests/test_conformance.py

@@ -8,6 +8,7 @@
     build_structured_reference_model,
     import_reference_model_bpmn,
     diff_reference_model_coverage,
+    _fitness_state_equation_alignments,
 )
 
 from conftest import make_event_log, make_simple_variant_log
@@ -341,3 +342,51 @@ def test_diff_reference_model_coverage_empty_diff_when_log_and_reference_match()
     diff = diff_reference_model_coverage(net, df)
     assert diff['unexpected_in_data'] == []
     assert diff['never_observed'] == []
+
+
+# --- _fitness_state_equation_alignments uses the real markings, not a guess ---
+
+def test_state_equation_alignments_uses_supplied_markings_not_topology_guess():
+    """Regression test: the function used to ignore its initial_marking/
+    final_marking arguments and rebuild markings by guessing from net
+    topology ('no in-arcs' = initial, 'no out-arcs' = final). A place with
+    no arcs at all satisfies both conditions, so it used to get folded into
+    both the guessed initial and final marking even though it's neither -
+    corrupting the marking used for alignment. With the real markings
+    passed through, that decoy place is correctly ignored and a perfectly
+    matching trace gets fitness 1.0."""
+    from pm4py.objects.petri_net.obj import PetriNet, Marking
+    from pm4py.objects.petri_net.utils import petri_utils
+    from pm4py.objects.log.obj import EventLog, Trace, Event
+
+    net = PetriNet('test')
+    real_start = PetriNet.Place('real_start')
+    mid = PetriNet.Place('mid')
+    real_end = PetriNet.Place('real_end')
+    decoy = PetriNet.Place('decoy')  # no arcs at all: "no in-arcs" AND "no out-arcs"
+    for p in (real_start, mid, real_end, decoy):
+        net.places.add(p)
+
+    t_a = PetriNet.Transition('t_a', 'a')
+    t_b = PetriNet.Transition('t_b', 'b')
+    net.transitions.add(t_a)
+    net.transitions.add(t_b)
+    petri_utils.add_arc_from_to(real_start, t_a, net)
+    petri_utils.add_arc_from_to(t_a, mid, net)
+    petri_utils.add_arc_from_to(mid, t_b, net)
+    petri_utils.add_arc_from_to(t_b, real_end, net)
+
+    im = Marking({real_start: 1})
+    fm = Marking({real_end: 1})
+
+    log = EventLog()
+    tr = Trace()
+    tr.attributes['concept:name'] = 'case1'
+    tr.append(Event({'concept:name': 'a'}))
+    tr.append(Event({'concept:name': 'b'}))
+    log.append(tr)
+
+    result = _fitness_state_equation_alignments(
+        log, net, im, fm, max_align=10, cores=1, optimize_variants=True
+    )
+    assert result['fitness']['log_fitness'] == pytest.approx(1.0)

tests/test_data_manager.py

@@ -387,3 +387,22 @@ def test_winsorize_series_all_nan_returns_zero_bounds():
     assert clipped.isna().all()
     assert lower == 0.0
     assert upper == 0.0
+
+
+# --- refine_activity_labels ---
+
+def test_refine_activity_labels_cleans_urls_per_row_not_by_first_row_only():
+    """Regression test: the URL-cleaning step (strip query string, keep last
+    path segment) used to decide whether to apply at all based only on the
+    first matched row's value, then applied that single decision to the
+    whole column. A column mixing plain and URL-like values had every
+    non-first-style row leak raw slashes/query strings into the activity
+    name."""
+    df = pd.DataFrame({
+        'concept:name': ['page_view', 'page_view'],
+        'page_type': ['product', '/category/product?ref=x'],
+        'time:timestamp': [pd.Timestamp('2024-01-01'), pd.Timestamp('2024-01-01 00:01')],
+        'case:concept:name': ['c1', 'c1'],
+    })
+    out = refine_activity_labels(df, target_activity='page_view', context_column='page_type')
+    assert out['concept:name'].tolist() == ['page_view_PRODUCT', 'page_view_PRODUCT']

tests/test_discovery.py

@@ -1,6 +1,11 @@
+import io
+
+import pm4py
 import pytest
 
-from prox.discovery import perform_process_discovery
+from prox.discovery import perform_process_discovery, _discover_inductive_miner
+from prox.data_manager import load_and_validate_csv
+from prox.mock_data import generate_mock_event_log
 
 from conftest import make_simple_variant_log
 
@@ -38,3 +43,20 @@ def test_perform_process_discovery_missing_columns():
     model, errors, messages = perform_process_discovery(pd.DataFrame({'foo': [1]}))
     assert model is None
     assert any('missing' in e.lower() for e in errors)
+
+
+def test_inductive_miner_noise_threshold_actually_affects_the_model():
+    """Regression test: inductive_miner.apply() defaults to the plain IM
+    variant, which silently ignores noise_threshold entirely - only the IMf
+    variant applies it. Without variant=Variants.IMf, a noisy log discovered
+    at noise_threshold=0.0 and noise_threshold=0.9 produced byte-identical
+    models despite the UI's Noise Threshold slider suggesting otherwise."""
+    raw = generate_mock_event_log(n_sessions=60, seed=1)
+    csv_bytes = raw.to_csv(index=False).encode()
+    log_df, messages, has_category = load_and_validate_csv(io.BytesIO(csv_bytes), case_grouping='user')
+    log = pm4py.convert_to_event_log(log_df)
+
+    net_low, im_low, fm_low, _ = _discover_inductive_miner(log, noise_threshold=0.0)
+    net_high, im_high, fm_high, _ = _discover_inductive_miner(log, noise_threshold=0.9)
+
+    assert (len(net_low.places), len(net_low.transitions)) != (len(net_high.places), len(net_high.transitions))