Skip to content

job_monitor_helper

Helper class for Job Monitor MCP tools.

This module contains the non-MCP business logic used by JobMonitorServer so the server can delegate execution through BaseMCPServer.run_tool() while keeping the tool-registration layer thin.

JobMonitorHelper

Encapsulate Job Monitor tool implementations behind the run_tool contract.

Attributes:

Name Type Description
tail_bytes

Maximum number of bytes to read from the end of each log file.
status_depth

Optional future-facing status-history depth setting stored alongside the helper configuration.

Methods:

Name Description
read_logs

Read tailed stdout and stderr content for a job run.
summarize_status

Build a structured status payload with failure classification data.
Source code in src/mada_tools/monitor/job_monitor/job_monitor_helper.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class JobMonitorHelper:
    """Encapsulate Job Monitor tool implementations behind the run_tool contract.

    Attributes:
        tail_bytes: Maximum number of bytes to read from the end of each log
            file.
        status_depth: Optional future-facing status-history depth setting stored
            alongside the helper configuration.

    Methods:
        read_logs: Read tailed stdout and stderr content for a job run.
        summarize_status: Build a structured status payload with failure
            classification data.
    """

    def __init__(self, tail_bytes: int = 50000, status_depth: Optional[int] = None):
        """Initialize the helper with log-tail configuration.

        Args:
            tail_bytes: Maximum number of bytes to read from the end of each
                log file.
            status_depth: Optional status-history depth value retained for
                compatibility with server-level configuration.
        """
        self.tail_bytes = tail_bytes
        self.status_depth = status_depth

    def read_logs(self, run_location: str, stdout_file: str, stderr_file: str) -> tuple[bool, str]:
        """Return tailed stdout and stderr logs for a job.

        Args:
            run_location: Path to the directory containing the log files.
            stdout_file: Name of the stdout log file.
            stderr_file: Name of the stderr log file.

        Returns:
            A `(success, payload)` tuple where `success` is always `True` and
            `payload` is a JSON string containing `stdout` and `stderr` log
            tails.
        """
        out_path = os.path.join(run_location, stdout_file)
        err_path = os.path.join(run_location, stderr_file)

        result = {
            "stdout": tail_log(out_path, self.tail_bytes),
            "stderr": tail_log(err_path, self.tail_bytes),
        }
        return True, json.dumps(result, indent=2)

    def summarize_status(
        self,
        run_location: str,
        stdout_file: str,
        stderr_file: str,
        exit_code: Optional[int] = None,
    ) -> tuple[bool, str]:
        """Return a structured status summary including failure classification.

        Args:
            run_location: Path to the directory containing the log files.
            stdout_file: Name of the stdout log file.
            stderr_file: Name of the stderr log file.
            exit_code: Optional scheduler or application exit code to include in
                the returned summary.

        Returns:
            A `(success, payload)` tuple where `success` is always `True` and
            `payload` is a JSON string containing the exit code, tailed logs,
            and detected failure classifications.
        """
        out_path = os.path.join(run_location, stdout_file)
        err_path = os.path.join(run_location, stderr_file)

        out_text = tail_log(out_path, self.tail_bytes)
        err_text = tail_log(err_path, self.tail_bytes)
        combined = out_text + "\n" + err_text

        findings = classify_failure(combined, self.tail_bytes)

        summary = {
            "exit_code": exit_code,
            "stdout_tail": out_text,
            "stderr_tail": err_text,
            "detected_failures": findings,
        }

        return True, json.dumps(summary, indent=2)

__init__(tail_bytes=50000, status_depth=None)

Initialize the helper with log-tail configuration.

Parameters:

Name Type Description Default
tail_bytes int

Maximum number of bytes to read from the end of each log file.

 50000
status_depth Optional[int]

Optional status-history depth value retained for compatibility with server-level configuration.

 None
Source code in src/mada_tools/monitor/job_monitor/job_monitor_helper.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, tail_bytes: int = 50000, status_depth: Optional[int] = None):
    """Initialize the helper with log-tail configuration.

    Args:
        tail_bytes: Maximum number of bytes to read from the end of each
            log file.
        status_depth: Optional status-history depth value retained for
            compatibility with server-level configuration.
    """
    self.tail_bytes = tail_bytes
    self.status_depth = status_depth

read_logs(run_location, stdout_file, stderr_file)

Return tailed stdout and stderr logs for a job.

Parameters:

Name Type Description Default
run_location str

Path to the directory containing the log files.

 required
stdout_file str

Name of the stdout log file.

 required
stderr_file str

Name of the stderr log file.

 required

Returns:

Type Description
bool

A (success, payload) tuple where success is always True and
str

payload is a JSON string containing stdout and stderr log
tuple[bool, str]

tails.
Source code in src/mada_tools/monitor/job_monitor/job_monitor_helper.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def read_logs(self, run_location: str, stdout_file: str, stderr_file: str) -> tuple[bool, str]:
    """Return tailed stdout and stderr logs for a job.

    Args:
        run_location: Path to the directory containing the log files.
        stdout_file: Name of the stdout log file.
        stderr_file: Name of the stderr log file.

    Returns:
        A `(success, payload)` tuple where `success` is always `True` and
        `payload` is a JSON string containing `stdout` and `stderr` log
        tails.
    """
    out_path = os.path.join(run_location, stdout_file)
    err_path = os.path.join(run_location, stderr_file)

    result = {
        "stdout": tail_log(out_path, self.tail_bytes),
        "stderr": tail_log(err_path, self.tail_bytes),
    }
    return True, json.dumps(result, indent=2)

summarize_status(run_location, stdout_file, stderr_file, exit_code=None)

Return a structured status summary including failure classification.

Parameters:

Name Type Description Default
run_location str

Path to the directory containing the log files.

 required
stdout_file str

Name of the stdout log file.

 required
stderr_file str

Name of the stderr log file.

 required
exit_code Optional[int]

Optional scheduler or application exit code to include in the returned summary.

 None

Returns:

Type Description
bool

A (success, payload) tuple where success is always True and
str

payload is a JSON string containing the exit code, tailed logs,
tuple[bool, str]

and detected failure classifications.
Source code in src/mada_tools/monitor/job_monitor/job_monitor_helper.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def summarize_status(
    self,
    run_location: str,
    stdout_file: str,
    stderr_file: str,
    exit_code: Optional[int] = None,
) -> tuple[bool, str]:
    """Return a structured status summary including failure classification.

    Args:
        run_location: Path to the directory containing the log files.
        stdout_file: Name of the stdout log file.
        stderr_file: Name of the stderr log file.
        exit_code: Optional scheduler or application exit code to include in
            the returned summary.

    Returns:
        A `(success, payload)` tuple where `success` is always `True` and
        `payload` is a JSON string containing the exit code, tailed logs,
        and detected failure classifications.
    """
    out_path = os.path.join(run_location, stdout_file)
    err_path = os.path.join(run_location, stderr_file)

    out_text = tail_log(out_path, self.tail_bytes)
    err_text = tail_log(err_path, self.tail_bytes)
    combined = out_text + "\n" + err_text

    findings = classify_failure(combined, self.tail_bytes)

    summary = {
        "exit_code": exit_code,
        "stdout_tail": out_text,
        "stderr_tail": err_text,
        "detected_failures": findings,
    }

    return True, json.dumps(summary, indent=2)