feat: make structural analysis configurable

README.md

@@ -177,12 +177,19 @@ policy = AnalysisPolicy(
     high_threshold=4.0,
     critical_threshold=6.0,
     mapped_task_aggregation=MappedTaskAggregation.MAX,
+    change_point_minimum_segment_size=4,
+    change_point_score_threshold=4.0,
+    trend_minimum_observations=8,
+    trend_score_threshold=4.0,
+    trend_minimum_directional_consistency=0.75,
 )
 ```
 
 `baseline_window` limits the number of historical values used before the current
 run. Mapped task durations can be aggregated with `MAX`, `MEAN`, or `SUM`. The
 same policy options are available through the CLI and MCP tool.
+Change-point and trend detection can also be disabled independently with
+`change_point_detection_enabled=False` or `trend_detection_enabled=False`.
 
 Every `DAGAnalysis` exposes a derived `summary` with task-analysis coverage,
 severity distribution, anomalous task and handoff counts, uniquely affected

src/flowsense/application/analyzer.py

@@ -40,13 +40,28 @@ def analyze_dag(
     drift_results = {}
 
     for task_id, durations in duration_history.items():
-        change_point = detect_change_point(task_id, durations)
-        if change_point is not None:
-            change_point_results[task_id] = change_point
-
-        trend = detect_trend(task_id, durations)
-        if trend is not None:
-            trend_results[task_id] = trend
+        if policy.change_point_detection_enabled:
+            change_point = detect_change_point(
+                task_id,
+                durations,
+                minimum_segment_size=policy.change_point_minimum_segment_size,
+                score_threshold=policy.change_point_score_threshold,
+            )
+            if change_point is not None:
+                change_point_results[task_id] = change_point
+
+        if policy.trend_detection_enabled:
+            trend = detect_trend(
+                task_id,
+                durations,
+                minimum_observations=policy.trend_minimum_observations,
+                score_threshold=policy.trend_score_threshold,
+                minimum_directional_consistency=(
+                    policy.trend_minimum_directional_consistency
+                ),
+            )
+            if trend is not None:
+                trend_results[task_id] = trend
 
         try:
             drift_results[task_id] = calculate_drift(
@@ -86,16 +101,30 @@ def analyze_dag(
 
     for edge, delays in handoff_history.items():
         upstream_task, downstream_task = edge
-        change_point = detect_change_point(
-            f"{upstream_task}->{downstream_task}",
-            delays,
-        )
-        if change_point is not None:
-            handoff_change_point_results[edge] = change_point
-
-        trend = detect_trend(f"{upstream_task}->{downstream_task}", delays)
-        if trend is not None:
-            handoff_trend_results[edge] = trend
+        subject_id = f"{upstream_task}->{downstream_task}"
+
+        if policy.change_point_detection_enabled:
+            change_point = detect_change_point(
+                subject_id,
+                delays,
+                minimum_segment_size=policy.change_point_minimum_segment_size,
+                score_threshold=policy.change_point_score_threshold,
+            )
+            if change_point is not None:
+                handoff_change_point_results[edge] = change_point
+
+        if policy.trend_detection_enabled:
+            trend = detect_trend(
+                subject_id,
+                delays,
+                minimum_observations=policy.trend_minimum_observations,
+                score_threshold=policy.trend_score_threshold,
+                minimum_directional_consistency=(
+                    policy.trend_minimum_directional_consistency
+                ),
+            )
+            if trend is not None:
+                handoff_trend_results[edge] = trend
 
         try:
             handoff_drift_results[edge] = calculate_handoff_drift(

src/flowsense/cli/main.py

@@ -35,6 +35,23 @@ def analyze(
     medium_threshold: float = typer.Option(2.0, min=0.0),
     high_threshold: float = typer.Option(3.5, min=0.0),
     critical_threshold: float = typer.Option(5.0, min=0.0),
+    change_point_detection: bool = typer.Option(
+        True,
+        "--change-point-detection/--no-change-point-detection",
+    ),
+    change_point_minimum_segment_size: int = typer.Option(3, min=2),
+    change_point_score_threshold: float = typer.Option(3.5, min=0.0),
+    trend_detection: bool = typer.Option(
+        True,
+        "--trend-detection/--no-trend-detection",
+    ),
+    trend_minimum_observations: int = typer.Option(5, min=3),
+    trend_score_threshold: float = typer.Option(3.5, min=0.0),
+    trend_minimum_directional_consistency: float = typer.Option(
+        0.6,
+        min=0.0,
+        max=1.0,
+    ),
     mapped_task_aggregation: Annotated[
         MappedTaskAggregation,
         typer.Option(),
@@ -48,6 +65,15 @@ def analyze(
             high_threshold=high_threshold,
             critical_threshold=critical_threshold,
             mapped_task_aggregation=mapped_task_aggregation,
+            change_point_detection_enabled=change_point_detection,
+            change_point_minimum_segment_size=change_point_minimum_segment_size,
+            change_point_score_threshold=change_point_score_threshold,
+            trend_detection_enabled=trend_detection,
+            trend_minimum_observations=trend_minimum_observations,
+            trend_score_threshold=trend_score_threshold,
+            trend_minimum_directional_consistency=(
+                trend_minimum_directional_consistency
+            ),
         )
     except ValueError as exc:
         raise typer.BadParameter(str(exc)) from exc

src/flowsense/domain/policy.py

@@ -11,6 +11,13 @@ class AnalysisPolicy:
     high_threshold: float = 3.5
     critical_threshold: float = 5.0
     mapped_task_aggregation: MappedTaskAggregation = MappedTaskAggregation.MAX
+    change_point_detection_enabled: bool = True
+    change_point_minimum_segment_size: int = 3
+    change_point_score_threshold: float = 3.5
+    trend_detection_enabled: bool = True
+    trend_minimum_observations: int = 5
+    trend_score_threshold: float = 3.5
+    trend_minimum_directional_consistency: float = 0.6
 
     def __post_init__(self) -> None:
         if self.minimum_history < 2:
@@ -26,6 +33,16 @@ def __post_init__(self) -> None:
             0 < self.medium_threshold < self.high_threshold < self.critical_threshold
         ):
             raise ValueError("Severity thresholds must be positive and increasing.")
+        if self.change_point_minimum_segment_size < 2:
+            raise ValueError("change_point_minimum_segment_size must be at least 2.")
+        if self.change_point_score_threshold <= 0:
+            raise ValueError("change_point_score_threshold must be positive.")
+        if self.trend_minimum_observations < 3:
+            raise ValueError("trend_minimum_observations must be at least 3.")
+        if self.trend_score_threshold <= 0:
+            raise ValueError("trend_score_threshold must be positive.")
+        if not 0 < self.trend_minimum_directional_consistency <= 1:
+            raise ValueError("trend_minimum_directional_consistency must be in (0, 1].")
 
 
 DEFAULT_ANALYSIS_POLICY = AnalysisPolicy()

src/flowsense/mcp/server.py

@@ -42,6 +42,21 @@ def serialize_analysis(analysis: DAGAnalysis) -> dict:
             "high_threshold": analysis.policy.high_threshold,
             "critical_threshold": analysis.policy.critical_threshold,
             "mapped_task_aggregation": analysis.policy.mapped_task_aggregation,
+            "change_point_detection_enabled": (
+                analysis.policy.change_point_detection_enabled
+            ),
+            "change_point_minimum_segment_size": (
+                analysis.policy.change_point_minimum_segment_size
+            ),
+            "change_point_score_threshold": (
+                analysis.policy.change_point_score_threshold
+            ),
+            "trend_detection_enabled": analysis.policy.trend_detection_enabled,
+            "trend_minimum_observations": (analysis.policy.trend_minimum_observations),
+            "trend_score_threshold": analysis.policy.trend_score_threshold,
+            "trend_minimum_directional_consistency": (
+                analysis.policy.trend_minimum_directional_consistency
+            ),
         },
         "primary_origin": (
             {
@@ -164,6 +179,13 @@ def analyze_airflow_dag(
     medium_threshold: float = 2.0,
     high_threshold: float = 3.5,
     critical_threshold: float = 5.0,
+    change_point_detection_enabled: bool = True,
+    change_point_minimum_segment_size: int = 3,
+    change_point_score_threshold: float = 3.5,
+    trend_detection_enabled: bool = True,
+    trend_minimum_observations: int = 5,
+    trend_score_threshold: float = 3.5,
+    trend_minimum_directional_consistency: float = 0.6,
     mapped_task_aggregation: MappedTaskAggregation = MappedTaskAggregation.MAX,
 ) -> dict:
     """Analyze an Apache Airflow DAG for temporal drift and propagation."""
@@ -179,6 +201,17 @@ def analyze_airflow_dag(
                     high_threshold=high_threshold,
                     critical_threshold=critical_threshold,
                     mapped_task_aggregation=mapped_task_aggregation,
+                    change_point_detection_enabled=change_point_detection_enabled,
+                    change_point_minimum_segment_size=(
+                        change_point_minimum_segment_size
+                    ),
+                    change_point_score_threshold=change_point_score_threshold,
+                    trend_detection_enabled=trend_detection_enabled,
+                    trend_minimum_observations=trend_minimum_observations,
+                    trend_score_threshold=trend_score_threshold,
+                    trend_minimum_directional_consistency=(
+                        trend_minimum_directional_consistency
+                    ),
                 ),
             )
     except AirflowApiError as exc:

tests/test_analyzer.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from flowsense import AnalysisPolicy
 from flowsense.application import DAGDataSource, analyze_dag
 from flowsense.models import TaskRun
 
@@ -185,6 +186,34 @@ def test_analyze_dag_detects_task_and_handoff_trends() -> None:
     assert handoff_trend.slope_per_observation == 1.0
 
 
+def test_analyze_dag_uses_structural_analysis_policy() -> None:
+    client = MagicMock(spec=DAGDataSource)
+    client.collect_task_runs.return_value = [
+        TaskRun(
+            dag_id="demo",
+            dag_run_id=f"run_{index}",
+            task_id="transform",
+            state="success",
+            duration=duration,
+        )
+        for index, duration in enumerate(
+            [3.0, 4.0, 5.0, 6.0, 7.0, 8.0],
+            start=1,
+        )
+    ]
+    client.get_dag_dependencies.return_value = {"transform": []}
+    policy = AnalysisPolicy(
+        change_point_detection_enabled=False,
+        trend_detection_enabled=False,
+    )
+
+    analysis = analyze_dag("demo", client, policy=policy)
+
+    assert analysis.change_point_results == {}
+    assert analysis.trend_results == {}
+    assert analysis.policy is policy
+
+
 def test_analyze_dag_calculates_handoff_drift() -> None:
     client = MagicMock(spec=DAGDataSource)
 

tests/test_cli.py

@@ -1,4 +1,4 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from typer.testing import CliRunner
 
@@ -20,3 +20,40 @@ def test_analyze_reports_airflow_api_errors() -> None:
     assert result.exit_code == 1
     assert "Airflow request failed" in result.output
     assert "503" in result.output
+
+
+def test_analyze_builds_structural_analysis_policy_from_options() -> None:
+    with (
+        patch("flowsense.cli.main.AirflowClient"),
+        patch("flowsense.cli.main.analyze_dag", return_value=MagicMock()) as analyze,
+        patch("flowsense.cli.main.render_analysis"),
+    ):
+        result = CliRunner().invoke(
+            app,
+            [
+                "analyze",
+                "demo",
+                "--no-change-point-detection",
+                "--change-point-minimum-segment-size",
+                "4",
+                "--change-point-score-threshold",
+                "4.5",
+                "--no-trend-detection",
+                "--trend-minimum-observations",
+                "8",
+                "--trend-score-threshold",
+                "4.5",
+                "--trend-minimum-directional-consistency",
+                "0.75",
+            ],
+        )
+
+    assert result.exit_code == 0
+    policy = analyze.call_args.kwargs["policy"]
+    assert policy.change_point_detection_enabled is False
+    assert policy.change_point_minimum_segment_size == 4
+    assert policy.change_point_score_threshold == 4.5
+    assert policy.trend_detection_enabled is False
+    assert policy.trend_minimum_observations == 8
+    assert policy.trend_score_threshold == 4.5
+    assert policy.trend_minimum_directional_consistency == 0.75

tests/test_mcp_server.py

@@ -156,6 +156,21 @@ def test_serialize_analysis() -> None:
     assert result["dag_id"] == "demo"
     assert result["runs_analyzed"] == 5
     assert result["overall_severity"] == "CRITICAL"
+    assert result["policy"] == {
+        "minimum_history": 5,
+        "baseline_window": None,
+        "medium_threshold": 2.0,
+        "high_threshold": 3.5,
+        "critical_threshold": 5.0,
+        "mapped_task_aggregation": "MAX",
+        "change_point_detection_enabled": True,
+        "change_point_minimum_segment_size": 3,
+        "change_point_score_threshold": 3.5,
+        "trend_detection_enabled": True,
+        "trend_minimum_observations": 5,
+        "trend_score_threshold": 3.5,
+        "trend_minimum_directional_consistency": 0.6,
+    }
     assert result["summary"] == {
         "total_tasks": 2,
         "analyzed_tasks": 1,

tests/test_policy.py

@@ -20,6 +20,24 @@ def test_policy_validates_history_and_thresholds() -> None:
         AnalysisPolicy(minimum_history=5, baseline_window=3)
 
 
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"change_point_minimum_segment_size": 1},
+        {"change_point_score_threshold": 0.0},
+        {"trend_minimum_observations": 2},
+        {"trend_score_threshold": 0.0},
+        {"trend_minimum_directional_consistency": 0.0},
+        {"trend_minimum_directional_consistency": 1.1},
+    ],
+)
+def test_policy_validates_structural_analysis_settings(
+    overrides: dict[str, int | float],
+) -> None:
+    with pytest.raises(ValueError):
+        AnalysisPolicy(**overrides)
+
+
 def test_drift_uses_configured_minimum_history_and_thresholds() -> None:
     policy = AnalysisPolicy(
         minimum_history=3,