feat: add new JobStatusInfo model

This is intended to mirror and eventually replace the `ErrorMessage`
model in `src/dvsim/launcher/base.py`. Notably, it extends it to include
the ability to provide no context (`None`) rather than requiring an
empty list, and it allows multiple ranges of lines to be provided (in
the form (start, end)) rather than just a single line number.

The intention is that:
 - This extended functionality can be used to provide richer error
   context where possible in the future.
 - This `JobStatusInfo` will be used by the new async scheduler and
   async backends, with the `CompletedJobStatus` eventually returning
   this as its `fail_msg`. It supersedes `ErrorMessage`, and removes the
   dependency of `job/data` on `launcher/base`.
 - While we could change all the `ErrorMessage`s to equivalent
   `JobStatusInfo` objects now, we instead retain the old type to reduce
   code churn for launchers that will be rewritten.

Signed-off-by: Alex Jones <alex.jones@lowrisc.org>

Author: AlexJones0

src/dvsim/job/data.py

@@ -15,12 +15,12 @@
 from pydantic import BaseModel, ConfigDict
 
 from dvsim.job.status import JobStatus
-from dvsim.launcher.base import ErrorMessage
 from dvsim.report.data import IPMeta, ToolMeta
 
 __all__ = (
     "CompletedJobStatus",
     "JobSpec",
+    "JobStatusInfo",
     "WorkspaceConfig",
 )
 
@@ -123,6 +123,19 @@ def timeout_secs(self) -> int | None:
         return None if self.timeout_mins is None else self.timeout_mins * 60
 
 
+class JobStatusInfo(BaseModel):
+    """Context about some sort of failure / error within a job."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    message: str
+    """Human readable error message."""
+    lines: Sequence[int | tuple[int, int]] | None = None
+    """Relevant line information (in the job script or the job itself)."""
+    context: Sequence[str] | None = None
+    """Arbitrary context strings."""
+
+
 class CompletedJobStatus(BaseModel):
     """Job status."""
 
@@ -166,5 +179,5 @@ class CompletedJobStatus(BaseModel):
 
     status: JobStatus
     """Status of the job."""
-    fail_msg: ErrorMessage
+    fail_msg: JobStatusInfo | None
     """Error message."""

src/dvsim/scheduler/core.py

@@ -22,7 +22,7 @@
 
 from dvsim import instrumentation
 from dvsim.instrumentation import NoOpInstrumentation
-from dvsim.job.data import CompletedJobStatus, JobSpec
+from dvsim.job.data import CompletedJobStatus, JobSpec, JobStatusInfo
 from dvsim.job.status import JobStatus
 from dvsim.launcher.base import Launcher, LauncherBusyError, LauncherError
 from dvsim.logging import log
@@ -264,6 +264,14 @@ def on_signal(signal_received: int, _: FrameType | None) -> None:
             launcher = self._launchers[name]
             job_spec = self._jobs[name]
 
+            fail_msg = None
+            if launcher.fail_msg is not None:
+                launcher_fail = launcher.fail_msg
+                lines = None if launcher_fail.line_number is None else [launcher_fail.line_number]
+                fail_msg = JobStatusInfo(
+                    message=launcher_fail.message, lines=lines, context=launcher_fail.context
+                )
+
             results.append(
                 CompletedJobStatus(
                     name=job_spec.name,
@@ -279,7 +287,7 @@ def on_signal(signal_received: int, _: FrameType | None) -> None:
                     job_runtime=launcher.job_runtime.with_unit("s").get()[0],
                     simulated_time=launcher.simulated_time.with_unit("us").get()[0],
                     status=status,
-                    fail_msg=launcher.fail_msg,
+                    fail_msg=fail_msg,
                 )
             )
 

src/dvsim/sim_results.py

@@ -133,19 +133,29 @@ def from_job_status(results: Sequence["CompletedJobStatus"]) -> "BucketedFailure
 
         for job_status in results:
             if job_status.status in (JobStatus.FAILED, JobStatus.KILLED):
+                if job_status.fail_msg is None:
+                    continue
+
                 bucket = _bucketize(job_status.fail_msg.message)
 
                 if bucket not in buckets:
                     buckets[bucket] = []
 
+                # TODO: expose all relevant line numbers through the bucket, not just the first.
+                # We only expose the first one for now to keep changes to the scheduler minimal.
+                first_line_num = None
+                if job_status.fail_msg.lines:
+                    lines = job_status.fail_msg.lines[0]
+                    first_line_num = lines if isinstance(lines, int) else lines[0]
+
                 buckets[bucket].append(
                     JobFailureOverview(
                         name=job_status.name,
                         qual_name=job_status.qual_name,
                         seed=job_status.seed,
-                        line=job_status.fail_msg.line_number,
+                        line=first_line_num,
                         log_path=job_status.log_path,
-                        log_context=job_status.fail_msg.context,
+                        log_context=job_status.fail_msg.context or [],
                     ),
                 )