feat: improve propagation scoring

README.md

@@ -233,6 +233,11 @@ The latest execution is compared against that baseline using a robust Z-score.
 
 This makes the detector less sensitive to historical outliers than approaches based only on mean and standard deviation.
 
+Propagation scores are normalized to the `0.0–1.0` range. Downstream task
+severity is weighted by graph distance with a `0.8` decay per hop, so anomalies
+closer to the origin contribute more strongly than anomalies farther along the
+same path.
+
 ## Project Status
 
 FlowSense is currently in early development.
@@ -245,7 +250,6 @@ Planned areas include:
 
 - DAG-level analysis models
 - configurable historical baseline windows
-- improved propagation scoring
 - change-point detection
 - trend detection
 - richer CLI reporting

src/flowsense/engine/propagation.py

@@ -3,6 +3,9 @@
 from flowsense.domain import DriftResult, PropagationResult, Severity
 from flowsense.domain.enums import SEVERITY_SCORE
 
+_HOP_DECAY = 0.8
+_MAX_SEVERITY_SCORE = max(SEVERITY_SCORE.values())
+
 
 def _build_reverse_dependencies(
     dependencies: dict[str, list[str]],
@@ -97,6 +100,28 @@ def _find_propagation_paths(
     return paths
 
 
+def _calculate_propagation_score(
+    affected_tasks: list[str],
+    drift_results: dict[str, DriftResult],
+) -> float:
+    """Return a bounded, distance-weighted score for a propagation path."""
+    weighted_severity = 0.0
+    total_weight = 0.0
+
+    for hop, task_id in enumerate(affected_tasks):
+        weight = _HOP_DECAY**hop
+        normalized_severity = (
+            SEVERITY_SCORE[drift_results[task_id].severity] / _MAX_SEVERITY_SCORE
+        )
+        weighted_severity += normalized_severity * weight
+        total_weight += weight
+
+    if total_weight == 0.0:
+        return 0.0
+
+    return weighted_severity / total_weight
+
+
 def analyze_propagation(
     drift_results: dict[str, DriftResult],
     dependencies: dict[str, list[str]],
@@ -131,14 +156,9 @@ def analyze_propagation(
             if not affected_tasks:
                 continue
 
-            origin_score = SEVERITY_SCORE[drift.severity]
-
-            downstream_scores = [
-                SEVERITY_SCORE[drift_results[task].severity] for task in affected_tasks
-            ]
-
-            propagation_score = sum(downstream_scores) / (
-                len(downstream_scores) * origin_score
+            propagation_score = _calculate_propagation_score(
+                affected_tasks=affected_tasks,
+                drift_results=drift_results,
             )
 
             results.append(

tests/test_propagation.py

@@ -2,6 +2,65 @@
 from flowsense.engine.propagation import analyze_propagation
 
 
+def _drift(task_id: str, severity: str) -> DriftResult:
+    return DriftResult(
+        task_id=task_id,
+        baseline=1.0,
+        current=2.0,
+        mad=0.1,
+        robust_z_score=3.0,
+        deviation_percent=100.0,
+        severity=severity,
+    )
+
+
+def test_propagation_score_is_bounded_when_downstream_is_more_severe() -> None:
+    drift_results = {
+        "origin": _drift("origin", "HIGH"),
+        "critical": _drift("critical", "CRITICAL"),
+    }
+
+    results = analyze_propagation(
+        drift_results,
+        {"origin": ["critical"], "critical": []},
+    )
+
+    assert results[0].propagation_score == 1.0
+
+
+def test_propagation_score_gives_closer_tasks_more_weight() -> None:
+    closer_critical_results = analyze_propagation(
+        {
+            "origin": _drift("origin", "CRITICAL"),
+            "critical": _drift("critical", "CRITICAL"),
+            "medium": _drift("medium", "MEDIUM"),
+        },
+        {
+            "origin": ["critical"],
+            "critical": ["medium"],
+            "medium": [],
+        },
+    )
+    farther_critical_results = analyze_propagation(
+        {
+            "origin": _drift("origin", "CRITICAL"),
+            "medium": _drift("medium", "MEDIUM"),
+            "critical": _drift("critical", "CRITICAL"),
+        },
+        {
+            "origin": ["medium"],
+            "medium": ["critical"],
+            "critical": [],
+        },
+    )
+
+    closer_critical = closer_critical_results[0].propagation_score
+    farther_critical = farther_critical_results[0].propagation_score
+
+    assert closer_critical > farther_critical
+    assert 0.0 <= farther_critical <= closer_critical <= 1.0
+
+
 def test_detects_downstream_propagation() -> None:
     drift_results = {
         "extract": DriftResult(