|
| 1 | +"""Diff two run traces — what changed between two executions of a flow. |
| 2 | +
|
| 3 | +A run history tells you a run *failed*, but not *what changed* from the run that |
| 4 | +passed: which step was added or dropped, which step flipped pass→fail, which step |
| 5 | +got slower. ``run_diff`` aligns the two step sequences with a longest-common- |
| 6 | +subsequence walk (so an inserted / removed step shifts the rest into place instead |
| 7 | +of mis-pairing everything) and classifies the differences: |
| 8 | +
|
| 9 | +* **added** / **removed** steps (present in only one run), |
| 10 | +* **status flips** (an aligned step whose status changed — with the new failure's |
| 11 | + :func:`failure_signature` when it carries an ``error``), |
| 12 | +* **timing regressions** (an aligned step that got ``regress_factor`` x slower). |
| 13 | +
|
| 14 | +A step is any dict with a name key (default ``"name"``) and optional ``status`` / |
| 15 | +``duration`` / ``error``. Pure standard library; no device, no ``PySide6``. |
| 16 | +""" |
| 17 | +from typing import Any, Dict, List, Sequence |
| 18 | + |
| 19 | +Step = Dict[str, Any] |
| 20 | + |
| 21 | + |
| 22 | +def _lcs_pairs(left: Sequence[str], right: Sequence[str]) -> List[tuple]: |
| 23 | + """Return aligned ``(i, j)`` index pairs of the longest common subsequence.""" |
| 24 | + n, m = len(left), len(right) |
| 25 | + table = [[0] * (m + 1) for _ in range(n + 1)] |
| 26 | + for i in range(n - 1, -1, -1): |
| 27 | + for j in range(m - 1, -1, -1): |
| 28 | + if left[i] == right[j]: |
| 29 | + table[i][j] = table[i + 1][j + 1] + 1 |
| 30 | + else: |
| 31 | + table[i][j] = max(table[i + 1][j], table[i][j + 1]) |
| 32 | + pairs, i, j = [], 0, 0 |
| 33 | + while i < n and j < m: |
| 34 | + if left[i] == right[j]: |
| 35 | + pairs.append((i, j)) |
| 36 | + i, j = i + 1, j + 1 |
| 37 | + elif table[i + 1][j] >= table[i][j + 1]: |
| 38 | + i += 1 |
| 39 | + else: |
| 40 | + j += 1 |
| 41 | + return pairs |
| 42 | + |
| 43 | + |
| 44 | +def _status_flip(before: Step, after: Step, name: str) -> Dict[str, Any]: |
| 45 | + """Build a status-flip record, attaching a signature for a new error.""" |
| 46 | + flip = {"name": name, "from": before.get("status"), |
| 47 | + "to": after.get("status")} |
| 48 | + error = after.get("error") |
| 49 | + if error: |
| 50 | + from je_auto_control.utils.failure_signature import failure_signature |
| 51 | + flip["signature"] = failure_signature(str(error)) |
| 52 | + return flip |
| 53 | + |
| 54 | + |
| 55 | +def _regression(before: Step, after: Step, name: str, |
| 56 | + factor: float) -> Dict[str, Any]: |
| 57 | + """Return a timing-regression record, or ``{}`` if not a regression.""" |
| 58 | + prev, curr = before.get("duration"), after.get("duration") |
| 59 | + if not isinstance(prev, (int, float)) or not isinstance(curr, (int, float)): |
| 60 | + return {} |
| 61 | + if prev > 0 and curr >= prev * float(factor): |
| 62 | + return {"name": name, "before": float(prev), "after": float(curr), |
| 63 | + "ratio": round(curr / prev, 3)} |
| 64 | + return {} |
| 65 | + |
| 66 | + |
| 67 | +def _aligned_changes(before: Sequence[Step], after: Sequence[Step], |
| 68 | + names: Sequence[str], pairs: List[tuple], |
| 69 | + factor: float) -> tuple: |
| 70 | + """Classify the LCS-aligned pairs into (status flips, timing regressions).""" |
| 71 | + flips, regressions = [], [] |
| 72 | + for i, j in pairs: |
| 73 | + if before[i].get("status") != after[j].get("status"): |
| 74 | + flips.append(_status_flip(before[i], after[j], names[i])) |
| 75 | + regression = _regression(before[i], after[j], names[i], factor) |
| 76 | + if regression: |
| 77 | + regressions.append(regression) |
| 78 | + return flips, regressions |
| 79 | + |
| 80 | + |
| 81 | +def _keys(steps: Sequence[Step], key: str) -> List[str]: |
| 82 | + return [str(step.get(key, "")) for step in steps] |
| 83 | + |
| 84 | + |
| 85 | +def _unmatched(steps: Sequence[Step], matched: set) -> List[Step]: |
| 86 | + return [steps[k] for k in range(len(steps)) if k not in matched] |
| 87 | + |
| 88 | + |
| 89 | +def diff_runs(before: Sequence[Step], after: Sequence[Step], *, |
| 90 | + key: str = "name", regress_factor: float = 1.5) -> Dict[str, Any]: |
| 91 | + """Diff two step sequences into ``{added, removed, status_flips, |
| 92 | + timing_regressions, aligned, identical}``. |
| 93 | +
|
| 94 | + Steps are aligned by their ``key`` value via LCS; ``regress_factor`` is the |
| 95 | + slowdown ratio that counts as a timing regression. |
| 96 | + """ |
| 97 | + left, right = _keys(before, key), _keys(after, key) |
| 98 | + pairs = _lcs_pairs(left, right) |
| 99 | + flips, regressions = _aligned_changes(before, after, left, pairs, |
| 100 | + regress_factor) |
| 101 | + added = _unmatched(after, {j for _, j in pairs}) |
| 102 | + removed = _unmatched(before, {i for i, _ in pairs}) |
| 103 | + return {"added": added, "removed": removed, "status_flips": flips, |
| 104 | + "timing_regressions": regressions, "aligned": len(pairs), |
| 105 | + "identical": not any((added, removed, flips, regressions))} |
| 106 | + |
| 107 | + |
| 108 | +def summarize_run_diff(diff: Dict[str, Any]) -> str: |
| 109 | + """Render a one-line human summary of a :func:`diff_runs` result.""" |
| 110 | + if diff.get("identical"): |
| 111 | + return "no change" |
| 112 | + parts = [] |
| 113 | + for label, field in (("+{} added", "added"), ("-{} removed", "removed"), |
| 114 | + ("{} status flip(s)", "status_flips"), |
| 115 | + ("{} regression(s)", "timing_regressions")): |
| 116 | + count = len(diff.get(field, [])) |
| 117 | + if count: |
| 118 | + parts.append(label.format(count)) |
| 119 | + return ", ".join(parts) |
0 commit comments