master to dev

.github/workflows/ci.yaml

@@ -23,7 +23,7 @@ jobs:
         uses: astral-sh/setup-uv@v6
 
       - name: Install dependencies
-        run: uv pip install --system -e ".[dev]"
+        run: uv pip install --system -e ".[dev,mcp]"
 
       - name: Run Ruff
         run: ruff check .

README.md

@@ -26,8 +26,12 @@ FlowSense is designed to answer questions such as:
 - Median-based historical baselines
 - MAD-based robust Z-score drift detection
 - Severity classification
-- Dependency-aware propagation analysis
+- Task handoff delay analysis
+- Task impact classification (`OWN_DRIFT`, `INHERITED_DELAY`, and `COMBINED`)
+- Multi-hop and branching propagation analysis
+- Primary root-cause selection
 - CLI-based DAG analysis
+- MCP server integration
 
 ## Example
 
@@ -40,13 +44,19 @@ Example output:
 ```text
 FlowSense Analysis — flowsense_demo
 
-┏━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━┓
-┃ Task      ┃ Baseline ┃ Current ┃ Deviation ┃ Z-Score ┃ Severity ┃
-┡━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━┩
-│ extract   │ 1.56s    │ 1.61s   │ +3.4%     │ 0.17    │ NORMAL   │
-│ transform │ 3.34s    │ 9.61s   │ +187.6%   │ 7.61    │ CRITICAL │
-│ load      │ 1.40s    │ 2.11s   │ +50.2%    │ 3.17    │ MEDIUM   │
-└───────────┴──────────┴─────────┴───────────┴─────────┴──────────┘
+┏━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━┓
+┃ Task      ┃ Baseline ┃ Current ┃ Deviation ┃ Z-Score ┃ Severity ┃ Impact    ┃
+┡━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━┩
+│ extract   │ 1.56s    │ 1.61s   │ +3.4%     │ 0.17    │ NORMAL   │ NORMAL    │
+│ transform │ 3.34s    │ 9.61s   │ +187.6%   │ 7.61    │ CRITICAL │ OWN_DRIFT │
+│ load      │ 1.40s    │ 2.11s   │ +50.2%    │ 3.17    │ MEDIUM   │ COMBINED  │
+└───────────┴──────────┴─────────┴───────────┴─────────┴──────────┴───────────┘
+
+Overall Severity: CRITICAL
+Primary Origin: transform
+Reason: OWN_DRIFT
+Severity: CRITICAL
+Propagation Score: 0.33
 
 Propagation Analysis
 
@@ -64,17 +74,16 @@ Apache Airflow
   Collector
       │
       ▼
-Task Run History
+Task Run and Handoff History
       │
       ▼
- Drift Engine
+Drift and Impact Analysis
       │
       ▼
-Propagation Analysis
+Propagation and Root-Cause Analysis
       │
       ├── CLI
-      │
-      └── MCP Server (planned)
+      └── MCP Server
 ```
 
 ## Installation
@@ -93,7 +102,7 @@ Create a virtual environment and install the project:
 ```bash
 uv venv --python 3.12
 source .venv/bin/activate
-uv pip install -e ".[dev]"
+uv pip install -e ".[dev,mcp]"
 ```
 
 ## Configuration
@@ -126,6 +135,18 @@ Then run:
 flowsense analyze <dag_id>
 ```
 
+## MCP Server
+
+Start the FlowSense MCP server over stdio:
+
+```bash
+flowsense-mcp
+```
+
+The server exposes the `analyze_airflow_dag` tool, which returns task drift,
+handoff drift, impact classification, propagation paths, and primary root-cause
+information for a DAG.
+
 ## Development
 
 Run unit tests:
@@ -167,7 +188,11 @@ src/flowsense/
 ├── engine/
 │   ├── drift.py
 │   ├── history.py
-│   └── propagation.py
+│   ├── impact.py
+│   ├── propagation.py
+│   ├── root_cause.py
+│   └── timing.py
+├── mcp/
 └── models/
 ```
 
@@ -194,13 +219,9 @@ Planned areas include:
 - DAG-level analysis models
 - configurable historical baseline windows
 - improved propagation scoring
-- temporal delay analysis
-- upstream/downstream impact separation
-- root-cause analysis
 - change-point detection
 - trend detection
 - richer CLI reporting
-- MCP server integration
 - broader Airflow compatibility testing
 
 ## License

pyproject.toml

@@ -21,11 +21,12 @@ dev = [
 ]
 
 mcp = [
-    "mcp",
+    "mcp[cli]>=2.0",
 ]
 
 [project.scripts]
 flowsense = "flowsense.cli.main:app"
+flowsense-mcp = "flowsense.mcp.server:main"
 
 [build-system]
 requires = ["setuptools>=75"]

src/flowsense/cli/main.py

@@ -4,10 +4,7 @@
 from rich.console import Console
 from rich.table import Table
 
-from flowsense.collector.airflow_client import AirflowClient
-from flowsense.engine.drift import calculate_drift
-from flowsense.engine.history import build_duration_history
-from flowsense.engine.propagation import analyze_propagation
+from flowsense.engine.analyzer import analyze_dag
 
 app = typer.Typer(
     name="flowsense",
@@ -27,39 +24,12 @@ def main() -> None:
 def analyze(
     dag_id: str = typer.Argument(
         ...,
-        help="Airflow DAG ID to analyze.",
+        help="Airflow DAG id to analyze.",
     ),
 ) -> None:
-    client = AirflowClient()
+    analysis = analyze_dag(dag_id)
 
-    task_runs = client.collect_task_runs(dag_id)
-
-    if not task_runs:
-        console.print(f"[red]No successful task runs found for DAG: {dag_id}[/red]")
-        raise typer.Exit(code=1)
-
-    history = build_duration_history(task_runs)
-
-    drift_results = {}
-
-    for task_id, durations in history.items():
-        result = calculate_drift(
-            task_id,
-            durations,
-        )
-
-        drift_results[task_id] = result
-
-    dependencies = client.get_dag_dependencies(dag_id)
-
-    propagation_results = analyze_propagation(
-        drift_results,
-        dependencies,
-    )
-
-    console.print()
-    console.print(f"[bold]FlowSense Analysis[/bold] — {dag_id}")
-    console.print()
+    console.print(f"\n[bold]FlowSense Analysis — {analysis.dag_id}[/bold]\n")
 
     table = Table()
 
@@ -69,34 +39,38 @@ def analyze(
     table.add_column("Deviation")
     table.add_column("Z-Score")
     table.add_column("Severity")
+    table.add_column("Impact")
+
+    for task_id, result in analysis.drift_results.items():
+        impact = analysis.task_impacts.get(task_id)
+        impact_label = impact.classification if impact else "-"
 
-    for result in drift_results.values():
         table.add_row(
-            result.task_id,
+            task_id,
             f"{result.baseline:.2f}s",
             f"{result.current:.2f}s",
             f"{result.deviation_percent:+.1f}%",
             f"{result.robust_z_score:.2f}",
             result.severity,
+            impact_label,
         )
 
     console.print(table)
 
-    console.print()
-    console.print("[bold]Propagation Analysis[/bold]")
+    console.print(f"\nOverall Severity: [bold]{analysis.overall_severity}[/bold]")
 
-    if not propagation_results:
-        console.print("No propagation detected.")
-        return
-
-    for propagation in propagation_results:
-        console.print()
-        console.print(f"Origin: [bold]{propagation.origin_task}[/bold]")
-
-        console.print("Path: " + " -> ".join(propagation.path))
-
-        console.print(f"Propagation Score: {propagation.propagation_score:.2f}")
+    if analysis.primary_origin:
+        console.print(f"Primary Origin: [bold]{analysis.primary_origin.task_id}[/bold]")
+        console.print(f"Reason: [bold]{analysis.primary_origin.classification}[/bold]")
+        console.print(f"Severity: [bold]{analysis.primary_origin.severity}[/bold]")
+        console.print(
+            f"Propagation Score: {analysis.primary_origin.propagation_score:.2f}"
+        )
 
+    if analysis.propagation_results:
+        console.print("\n[bold]Propagation Analysis[/bold]\n")
 
-if __name__ == "__main__":
-    app()
+        for result in analysis.propagation_results:
+            console.print(f"Origin: {result.origin_task}")
+            console.print(f"Path: {' -> '.join(result.path)}")
+            console.print(f"Propagation Score: {result.propagation_score:.2f}")

src/flowsense/engine/analyzer.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+from flowsense.collector.airflow_client import AirflowClient
+from flowsense.engine.drift import calculate_drift
+from flowsense.engine.history import build_duration_history
+from flowsense.engine.impact import classify_task_impact
+from flowsense.engine.propagation import analyze_propagation
+from flowsense.engine.root_cause import select_primary_origin
+from flowsense.engine.timing import (
+    build_handoff_history,
+    calculate_handoff_drift,
+)
+from flowsense.models import DAGAnalysis
+
+
+def analyze_dag(dag_id: str) -> DAGAnalysis:
+    client = AirflowClient()
+
+    task_runs = client.collect_task_runs(dag_id)
+    duration_history = build_duration_history(task_runs)
+
+    drift_results = {}
+
+    for task_id, durations in duration_history.items():
+        try:
+            drift_results[task_id] = calculate_drift(
+                task_id=task_id,
+                durations=durations,
+            )
+        except ValueError:
+            continue
+
+    dependencies = client.get_dag_dependencies(dag_id)
+
+    handoff_history = build_handoff_history(
+        task_runs=task_runs,
+        dependencies=dependencies,
+    )
+
+    handoff_drift_results = {}
+
+    for edge, delays in handoff_history.items():
+        _upstream_task, downstream_task = edge
+
+        try:
+            handoff_drift_results[edge] = calculate_handoff_drift(
+                upstream_task=_upstream_task,
+                downstream_task=downstream_task,
+                handoff_delays=delays,
+            )
+        except ValueError:
+            continue
+
+    task_impacts = {}
+
+    for task_id, task_drift in drift_results.items():
+        upstream_handoff_drifts = [
+            drift
+            for (
+                _upstream_task,
+                downstream_task,
+            ), drift in handoff_drift_results.items()
+            if downstream_task == task_id
+        ]
+
+        task_impacts[task_id] = classify_task_impact(
+            task_id=task_id,
+            task_drift=task_drift,
+            upstream_handoff_drifts=upstream_handoff_drifts,
+        )
+
+    propagation_results = analyze_propagation(
+        drift_results=drift_results,
+        dependencies=dependencies,
+    )
+
+    severity_order = {
+        "NORMAL": 0,
+        "MEDIUM": 1,
+        "HIGH": 2,
+        "CRITICAL": 3,
+    }
+
+    overall_severity = "NORMAL"
+
+    all_drift_results = [
+        *drift_results.values(),
+        *handoff_drift_results.values(),
+    ]
+
+    if all_drift_results:
+        overall_severity = max(
+            all_drift_results,
+            key=lambda result: severity_order[result.severity],
+        ).severity
+
+    primary_origin = select_primary_origin(
+        drift_results=drift_results,
+        task_impacts=task_impacts,
+        dependencies=dependencies,
+        propagation_results=propagation_results,
+    )
+
+    return DAGAnalysis(
+        dag_id=dag_id,
+        runs_analyzed=len({run.dag_run_id for run in task_runs}),
+        overall_severity=overall_severity,
+        primary_origin=primary_origin,
+        drift_results=drift_results,
+        handoff_drift_results=handoff_drift_results,
+        task_impacts=task_impacts,
+        propagation_results=propagation_results,
+        dependencies=dependencies,
+    )