Introduce ErrorRecord dataclass for structured error files

src/tigerflow/cli/report.py

@@ -168,12 +168,12 @@ def fmt_duration(ms: float) -> str:
                 if shown >= 5:
                     break
                 err_detail = (
-                    f"{err.exception_type}: {err.message}"
-                    if err.exception_type
-                    else err.message or "Unknown error"
+                    f"{err.record.exception_type}: {err.record.message}"
+                    if err.record.exception_type
+                    else err.record.message or "Unknown error"
                 )
                 lines.append(
-                    f"  [dim]{task_name}[/dim]  {err.file}  [red]{err_detail}[/red]"
+                    f"  [dim]{task_name}[/dim]  {err.record.file}  [red]{err_detail}[/red]"
                 )
                 shown += 1
             if shown >= 5:
@@ -322,18 +322,18 @@ def report(
             }
         if "errors" in sections:
             result["errors"] = {
-                name: [
+                task: [
                     {
-                        "file": e.file,
-                        "path": e.path,
-                        "timestamp": e.timestamp.isoformat() if e.timestamp else None,
-                        "exception_type": e.exception_type,
-                        "message": e.message,
-                        "traceback": e.traceback,
+                        "file": err.record.file,
+                        "path": err.path,
+                        "timestamp": err.record.timestamp or None,
+                        "exception_type": err.record.exception_type,
+                        "message": err.record.message,
+                        "traceback": err.record.traceback,
                     }
-                    for e in errs
+                    for err in errors
                 ]
-                for name, errs in pipeline_report.errors.items()
+                for task, errors in pipeline_report.errors.items()
             }
 
         print(json.dumps(result, indent=2, default=str))

src/tigerflow/models.py

@@ -2,6 +2,7 @@
 import re
 import shlex
 import textwrap
+from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
 from pathlib import Path
@@ -14,6 +15,7 @@
 from tigerflow.staging import StagingPipeline
 from tigerflow.utils import (
     TEMP_FILE_PREFIX,
+    ErrorRecord,
     is_process_running,
     read_pid_file,
     validate_file_ext,
@@ -461,15 +463,16 @@ class TaskProgress(BaseModel):
     failed: int = 0
 
 
-class FileError(BaseModel):
-    """Error information for a failed file."""
+@dataclass(slots=True)
+class FileError:
+    """A failed file's error record paired with where it was found.
+
+    `path` is the location of the .err file itself, which the record does
+    not carry: it is known only to the reader that discovers the file.
+    """
 
-    file: str
     path: str
-    timestamp: datetime | None = None
-    exception_type: str = ""
-    message: str = ""
-    traceback: str = ""
+    record: ErrorRecord
 
 
 class TaskMeta(BaseModel):
@@ -622,19 +625,17 @@ def report(self) -> PipelineReport:
                     stem = file.name.removesuffix(".err")
                     failed_stems.add(stem)
                     try:
-                        data = json.loads(file.read_text())
-                        task_errors.append(
-                            FileError(
-                                file=data.get("file", stem),
-                                path=str(file),
-                                timestamp=datetime.fromisoformat(data["timestamp"]),
-                                exception_type=data.get("exception_type", ""),
-                                message=data.get("message", ""),
-                                traceback=data.get("traceback", ""),
-                            )
+                        record = ErrorRecord.read(file)
+                        record.file = record.file or stem
+                    except (OSError, ValueError):
+                        record = ErrorRecord(
+                            timestamp="",
+                            exception_type="",
+                            message="",
+                            traceback="",
+                            file=stem,
                         )
-                    except (OSError, json.JSONDecodeError, KeyError):
-                        task_errors.append(FileError(file=stem, path=str(file)))
+                    task_errors.append(FileError(path=str(file), record=record))
             if task_errors:
                 errors[task_dir.name] = task_errors
 

src/tigerflow/tasks/local.py

@@ -11,10 +11,10 @@
 from tigerflow.logconfig import logger
 from tigerflow.models import LocalTaskConfig
 from tigerflow.settings import settings
-from tigerflow.utils import SetupContext, atomic_write
+from tigerflow.utils import ErrorRecord, SetupContext, atomic_write
 
 from ._base import Task
-from .utils import log_metrics, write_error_file
+from .utils import log_metrics
 
 
 class LocalTask(Task):
@@ -52,7 +52,7 @@ def task(input_file: Path, output_file: Path):
                         output_file.name.removesuffix(self.config.output_ext) + ".err"
                     )
                     error_file = output_dir / error_fname
-                    write_error_file(error_file, input_file.name)
+                    ErrorRecord.from_exception(file=input_file.name).write(error_file)
                     logger.error("Failed processing: {}", input_file.name)
 
         # Clean up incomplete temporary files left behind by a prior process instance

src/tigerflow/tasks/local_async.py

@@ -10,10 +10,10 @@
 from tigerflow.logconfig import logger
 from tigerflow.models import LocalAsyncTaskConfig
 from tigerflow.settings import settings
-from tigerflow.utils import SetupContext, atomic_write
+from tigerflow.utils import ErrorRecord, SetupContext, atomic_write
 
 from ._base import Task
-from .utils import log_metrics, write_error_file
+from .utils import log_metrics
 
 
 class LocalAsyncTask(Task):
@@ -53,7 +53,7 @@ async def task(input_file: Path, output_file: Path):
                         output_file.name.removesuffix(self.config.output_ext) + ".err"
                     )
                     error_file = self.config.output_dir / error_fname
-                    write_error_file(error_file, input_file.name)
+                    ErrorRecord.from_exception(file=input_file.name).write(error_file)
                     logger.error("Failed processing: {}", input_file.name)
 
         async def worker():

src/tigerflow/tasks/slurm.py

@@ -22,13 +22,14 @@
 from tigerflow.settings import settings
 from tigerflow.utils import (
     TEMP_FILE_PREFIX,
+    ErrorRecord,
     SetupContext,
     atomic_write,
     submit_to_slurm,
 )
 
 from ._base import Task
-from .utils import get_slurm_task_status, log_metrics, write_error_file
+from .utils import get_slurm_task_status, log_metrics
 
 
 class SlurmTask(Task):
@@ -82,7 +83,8 @@ async def setup(self, worker: Worker):
                     logger.info("Task setup complete")
                 except Exception:
                     logger.exception("Task setup failed; aborting task")
-                    setup_failed_sentinel.touch()
+                    setup_failed_sentinel.touch()  # ensure sentinel exists even if write below fails
+                    ErrorRecord.from_exception().write(setup_failed_sentinel)
 
             async def teardown(self, worker: Worker):
                 logger.info("Tearing down task")
@@ -112,7 +114,7 @@ def task(input_file: Path, output_file: Path):
                     metrics["status"] = "error"
                     error_fname = output_file.name.removesuffix(output_ext) + ".err"
                     error_file = output_dir / error_fname
-                    write_error_file(error_file, input_file.name)
+                    ErrorRecord.from_exception(file=input_file.name).write(error_file)
                     logger.error("Failed processing: {}", input_file.name)
 
         # Define parameters for each Slurm job

src/tigerflow/tasks/utils.py

@@ -1,14 +1,10 @@
 import json
 import subprocess
-import sys
-import traceback
 from contextlib import contextmanager
 from datetime import datetime, timezone
-from pathlib import Path
 
 from tigerflow.logconfig import logger
 from tigerflow.models import TaskStatus, TaskStatusKind
-from tigerflow.utils import atomic_write
 
 
 @contextmanager
@@ -112,21 +108,3 @@ def get_pending_worker_ids(worker_job_name: str) -> list[int]:
     ).stdout
 
     return [int(job_id) for job_id in pending_ids.split() if job_id.isdigit()]
-
-
-def write_error_file(error_path: Path, input_file: str) -> None:
-    """Write structured error JSON for a failed file.
-
-    Must be called from within an exception handler.
-    """
-    exc_type, exc_value, _ = sys.exc_info()
-    error_data = {
-        "file": input_file,
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "exception_type": exc_type.__name__ if exc_type else "Unknown",
-        "message": str(exc_value) if exc_value else "",
-        "traceback": traceback.format_exc(),
-    }
-    with atomic_write(error_path) as temp_path:
-        with open(temp_path, "w") as f:
-            json.dump(error_data, f, indent=2)

src/tigerflow/utils.py

@@ -1,11 +1,15 @@
+import dataclasses
 import importlib
+import json
 import os
 import re
 import subprocess
 import sys
 import tempfile
+import traceback
 from collections.abc import Callable
 from contextlib import contextmanager
+from datetime import datetime, timezone
 from pathlib import Path
 from subprocess import TimeoutExpired
 from types import SimpleNamespace
@@ -226,3 +230,52 @@ def atomic_write(filepath: str | os.PathLike[str]):
         raise
     else:
         temp_path.replace(filepath)
+
+
+@dataclasses.dataclass(slots=True)
+class ErrorRecord:
+    """Structured error record for JSON serialization.
+
+    Represents the on-disk schema used by .err and .setup-failed files.
+    """
+
+    timestamp: str
+    exception_type: str
+    message: str
+    traceback: str
+    file: str | None = None
+
+    @classmethod
+    def from_exception(cls, file: str | None = None) -> "ErrorRecord":
+        """Capture error details from the current exception context.
+
+        Must be called from within an exception handler. *file* is the
+        name of the input file being processed, if any; omit for errors
+        not associated with a specific file (e.g. task setup failures).
+        """
+        exc_type, exc_value, _ = sys.exc_info()
+        return cls(
+            timestamp=datetime.now(timezone.utc).isoformat(),
+            exception_type=exc_type.__name__ if exc_type else "Unknown",
+            message=str(exc_value) if exc_value else "",
+            traceback=traceback.format_exc(),
+            file=file,
+        )
+
+    def write(self, path: Path) -> None:
+        """Write error record as JSON to *path* using atomic write."""
+        with atomic_write(path) as temp_path:
+            with open(temp_path, "w") as f:
+                json.dump(dataclasses.asdict(self), f, indent=2)
+
+    @classmethod
+    def read(cls, path: Path) -> "ErrorRecord":
+        """Read error record from a JSON file.
+
+        Raises `ValueError` if the file content is malformed or incomplete.
+        """
+        try:
+            data = json.loads(path.read_text())
+            return cls(**data)
+        except (json.JSONDecodeError, TypeError) as exc:
+            raise ValueError(f"invalid error record: {path}") from exc

tests/integration/test_slurm_task.py

@@ -9,6 +9,7 @@
 Set SLURM_TEST_DIR to a shared filesystem path accessible by compute nodes.
 """
 
+import json
 import os
 import shutil
 import signal
@@ -270,6 +271,13 @@ def test_setup_failure_aborts(self, task_dirs, input_files, tasks_dir):
         sentinel_files = list(log_base.rglob(".setup-failed"))
         assert len(sentinel_files) > 0, "Expected .setup-failed sentinel file"
 
+        # Sentinel should contain expected error details
+        error_data = json.loads(sentinel_files[0].read_text())
+        assert error_data["exception_type"] == "RuntimeError"
+        assert "Intentional setup failure" in error_data["message"]
+        assert "timestamp" in error_data
+        assert "traceback" in error_data
+
         # No output files should have been produced
         output_files = [
             f