surface warnings in the report dashboard

src/tigerflow/cli/report.py

@@ -182,6 +182,18 @@ def fmt_duration(ms: float) -> str:
             lines.append(f"  [dim]... +{total_errors - 5} more[/dim]")
         lines.append("")
 
+    # Warnings summary (for this run)
+    total_warnings = sum(
+        task_warnings["num_warnings"] for task_warnings in report.num_warnings.values()
+    )
+    if total_warnings > 0:
+        lines.append(f"[bold]Warnings:[/bold] {total_warnings}")
+        for task_name, warning_data in report.num_warnings.items():
+            if warning_data["num_warnings"] > 0:
+                lines.append(
+                    f"  [dim]{task_name}[/dim]  [yellow]{warning_data['num_warnings']} warnings across {warning_data['num_files_processed']} file(s)[/yellow]   See logs in .tigerflow/{task_name} for more details"
+                )
+
     content = "\n".join(lines)
     return Panel(content, title="[bold]tigerflow report[/bold]", title_align="left")
 

src/tigerflow/models.py

@@ -491,6 +491,7 @@ class PipelineReport(BaseModel):
     staged: int | None = None  # None if stopped
     tasks: list[TaskProgress] = []
     metrics: dict[str, list[FileMetrics]] = {}
+    num_warnings: dict[str, dict[str, int]] = {}
     errors: dict[str, list[FileError]] = {}
 
 
@@ -595,6 +596,46 @@ def _parse_all_metrics(self) -> list[FileMetrics]:
 
         return metrics
 
+    def _parse_warnings(self) -> dict[str, dict[str, int]]:
+        """Parse WARNINGS from task log files.
+
+        Reads from:
+        - {task}/logs/{pid}/task-{pid}.log (local/local_async tasks)
+        - {task}/logs/{pid}/task-worker-{job_id}.log (Slurm worker logs)
+        """
+        all_warnings = {}
+
+        for task_dir in self._get_task_dirs():
+            log_files = list(task_dir.glob("logs/**/task*.log"))
+
+            for log_file in log_files:
+                if task_dir.name not in all_warnings:
+                    all_warnings[task_dir.name] = {
+                        "num_files_processed": 0,
+                        "num_warnings": 0,
+                    }
+                try:
+                    with open(log_file) as f:
+                        for line in f:
+                            if "METRICS" in line:
+                                all_warnings[task_dir.name]["num_files_processed"] = (
+                                    all_warnings[task_dir.name]["num_files_processed"]
+                                    + 1
+                                )
+                            if "WARNING" in line:
+                                if re.search(
+                                    r"Received signal \d+, initiating shutdown",
+                                    line,
+                                ):
+                                    continue
+                                all_warnings[task_dir.name]["num_warnings"] = (
+                                    all_warnings[task_dir.name]["num_warnings"] + 1
+                                )
+
+                except OSError:
+                    continue
+        return all_warnings
+
     def report(self) -> PipelineReport:
         """Generate a complete pipeline status report."""
         self.validate()
@@ -667,6 +708,7 @@ def report(self) -> PipelineReport:
         # === Per-Task Progress (from METRICS logs, all runs) ===
 
         all_metrics = self._parse_all_metrics()
+        warnings = self._parse_warnings()
         task_meta = self._get_task_meta()
 
         # Group metrics by task
@@ -721,5 +763,6 @@ def report(self) -> PipelineReport:
             staged=len(staged_stems) if is_running else None,
             tasks=tasks,
             metrics=metrics_by_task,
+            num_warnings=warnings,
             errors=errors,
         )

tests/unit/cli/test_report.py

@@ -248,7 +248,7 @@ def test_parse_all_metrics(self, tmp_path: Path):
         all_lines = []
         for i in range(5):
             all_lines.append(
-                f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
+                f"2026-03-10 10:00:01 | METRICS  | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
             )
         (log_dir / "task-12345.log").write_text("\n".join(all_lines))
 
@@ -302,7 +302,7 @@ def test_task_progress_with_symlinks(self, tmp_path: Path):
         for i in range(8):
             status = "success" if i < 6 else "error"
             task_lines.append(
-                f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}"
+                f"2026-03-10 12:00:01 | METRICS  | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}"
             )
         (log_dir / "task-100.log").write_text("\n".join(task_lines))
 
@@ -351,7 +351,7 @@ def test_multi_task_dependency_chain(self, tmp_path: Path):
         for i in range(10):
             status = "success" if i < 8 else "error"
             lines1.append(
-                f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}"
+                f"2026-03-10 12:00:01 | METRICS  | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}"
             )
         (log_dir1 / "task-100.log").write_text("\n".join(lines1))
 
@@ -364,7 +364,7 @@ def test_multi_task_dependency_chain(self, tmp_path: Path):
         lines2 = []
         for i in range(5):
             lines2.append(
-                f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
+                f"2026-03-10 12:00:02 | METRICS  | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
             )
         (log_dir2 / "task-100.log").write_text("\n".join(lines2))
 
@@ -442,7 +442,7 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path):
         run1_lines = []
         for i in range(5):
             run1_lines.append(
-                f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'run1_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
+                f"2026-03-10 10:00:01 | METRICS  | {json.dumps({'file': f'run1_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
             )
             (finished / f"run1_file{i}.txt").touch()
         (log_dir1 / "task-100.log").write_text("\n".join(run1_lines))
@@ -453,7 +453,7 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path):
         run2_lines = []
         for i in range(7):
             run2_lines.append(
-                f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'run2_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
+                f"2026-03-10 12:00:01 | METRICS  | {json.dumps({'file': f'run2_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}"
             )
             (finished / f"run2_file{i}.txt").touch()
         (log_dir2 / "task-200.log").write_text("\n".join(run2_lines))
@@ -488,14 +488,14 @@ def test_parse_metrics_from_multiple_worker_logs(self, tmp_path: Path):
         # Worker 1 processed file1
         worker1_log = log_dir / "task-worker-12345.log"
         worker1_log.write_text(
-            f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': 'file1.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n"
+            f"2026-03-10 12:00:01 | METRICS  | {json.dumps({'file': 'file1.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n"
         )
 
         # Worker 2 processed file2 and file3
         worker2_log = log_dir / "task-worker-12346.log"
         worker2_log.write_text(
-            f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': 'file2.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n"
-            f"2026-03-10 12:00:03 | METRICS | {json.dumps({'file': 'file3.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'error'})}\n"
+            f"2026-03-10 12:00:02 | METRICS  | {json.dumps({'file': 'file2.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n"
+            f"2026-03-10 12:00:03 | METRICS  | {json.dumps({'file': 'file3.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'error'})}\n"
         )
 
         output = PipelineOutput(tmp_path)