#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ABEND Intelligence V2 Advanced standalone decoder and analyzer for z/OS / MVS ABEND codes, return codes, condition codes, reason codes, and diagnostic messages found in JES2/JES3 job logs, JESMSGLG, JESYSMSG, SYSOUT, SYSPRINT, SYSABEND, SYSUDUMP, CEEDUMP extracts, and scheduler logs. Design goals: - No external dependency. - Direct code decoding and full text file analysis. - Built-in ABEND knowledge catalog with custom JSON override support. - Message family classification and evidence context windows. - Root-cause candidates with confidence and action recommendations. - HTML, JSON, CSV findings, CSV timeline, catalog export. - Batch-friendly exit codes with severity gate. """ from __future__ import annotations import argparse import csv import hashlib import html import json import re import sys from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple VERSION = "2.0.0" SEVERITY_ORDER = { "INFO": 10, "WARNING": 30, "ERROR": 60, "CRITICAL": 90, } ORDER_TO_SEVERITY = sorted(SEVERITY_ORDER.items(), key=lambda kv: kv[1]) MESSAGE_FAMILY_HINTS = { "IEF": "MVS job management / step termination", "IEC": "Data set / I/O / allocation / catalog / DCB", "IGD": "SMS allocation / storage management", "ICH": "RACF security / authorization", "IRR": "RACF security / identity / access control", "CEE": "Language Environment runtime", "IDC": "IDCAMS / VSAM utility", "ICE": "DFSORT / ICETOOL", "WER": "SYNCSORT / sort processing", "IEW": "Binder / link-edit", "IGY": "COBOL compiler", "DFH": "CICS", "DSN": "Db2 for z/OS", "DFS": "IMS", "HASP": "JES2", "$HASP": "JES2", } DATASET_RE = re.compile(r"\b[A-Z0-9@#$][A-Z0-9@#$-]{0,7}(?:\.[A-Z0-9@#$][A-Z0-9@#$-]{0,7}){1,20}(?:\([A-Z0-9@#$-]+\))?\b") JOB_ID_RE = re.compile(r"\b(JOB\d{4,8}|STC\d{4,8}|TSU\d{4,8})\b", re.IGNORECASE) TIME_RE = re.compile(r"\b(?:\d{2}:\d{2}:\d{2}(?:\.\d+)?|\d{2}\.\d{2}\.\d{2})\b") MSG_ID_RE = re.compile(r"\b(\$?HASP\d{3,5}|[A-Z]{3}\d{3,5}[A-Z]?)\b") @dataclass class CatalogEntry: code: str aliases: List[str] kind: str category: str default_severity: str title: str explanation: str probable_causes: List[str] = field(default_factory=list) recommended_checks: List[str] = field(default_factory=list) recommended_actions: List[str] = field(default_factory=list) confidence: str = "MEDIUM" tags: List[str] = field(default_factory=list) reason_notes: Dict[str, str] = field(default_factory=dict) @dataclass class DecodeEvent: code: str normalized_code: str kind: str category: str severity: str title: str explanation: str confidence: str source: str line_number: Optional[int] column: Optional[int] message_id: Optional[str] step_name: Optional[str] program_name: Optional[str] job_name: Optional[str] job_id: Optional[str] reason_code: Optional[str] return_code: Optional[int] evidence: str context_before: List[str] = field(default_factory=list) context_after: List[str] = field(default_factory=list) probable_causes: List[str] = field(default_factory=list) recommended_checks: List[str] = field(default_factory=list) recommended_actions: List[str] = field(default_factory=list) tags: List[str] = field(default_factory=list) @dataclass class TimelineEvent: line_number: int event_type: str severity: str message_id: Optional[str] code: Optional[str] step_name: Optional[str] program_name: Optional[str] job_name: Optional[str] job_id: Optional[str] evidence: str @dataclass class RootCauseCandidate: rank: int category: str severity: str confidence: str score: int title: str evidence_count: int explanation: str recommended_actions: List[str] = field(default_factory=list) supporting_codes: List[str] = field(default_factory=list) supporting_lines: List[int] = field(default_factory=list) @dataclass class AnalysisResult: tool: str version: str generated_at: str source_name: str source_hash: Optional[str] status: str highest_severity: str risk_score: int total_lines: int total_events: int total_timeline_events: int system_abends: int user_abends: int return_codes: int reason_codes: int message_families: Dict[str, int] codes_seen: List[str] job_names: List[str] job_ids: List[str] steps_seen: List[str] programs_seen: List[str] datasets_seen: List[str] root_cause_candidates: List[RootCauseCandidate] events: List[DecodeEvent] timeline: List[TimelineEvent] def entry( code: str, kind: str, category: str, severity: str, title: str, explanation: str, probable_causes: Sequence[str], checks: Sequence[str], actions: Sequence[str], confidence: str = "MEDIUM", tags: Sequence[str] = (), aliases: Sequence[str] = (), reason_notes: Optional[Dict[str, str]] = None, ) -> CatalogEntry: return CatalogEntry( code=code.upper(), aliases=[x.upper() for x in aliases], kind=kind, category=category, default_severity=severity, title=title, explanation=explanation, probable_causes=list(probable_causes), recommended_checks=list(checks), recommended_actions=list(actions), confidence=confidence, tags=list(tags), reason_notes=reason_notes or {}, ) BASE_CATALOG: Dict[str, CatalogEntry] = { "S0C1": entry( "S0C1", "SYSTEM_ABEND", "PROGRAM_EXCEPTION", "CRITICAL", "Operation exception", "The program attempted to execute an invalid operation code. This often indicates corrupted executable storage, branching into data, or a wrong entry point.", ["Branch to invalid address", "Corrupted load module", "Storage overlay", "Wrong entry point or AMODE/RMODE mismatch"], ["Review PSW and failing instruction", "Check recent binder/link-edit changes", "Inspect caller/callee entry definitions"], ["Rebuild the load module", "Validate branch logic and entry points", "Use SYSUDUMP/SYSMDUMP and IPCS for deep diagnosis"], tags=["program", "psw", "load-module"], ), "S0C4": entry( "S0C4", "SYSTEM_ABEND", "PROGRAM_STORAGE", "CRITICAL", "Protection or addressing exception", "The program referenced storage it was not allowed to access, or used an invalid address.", ["Bad pointer or address", "Subscript out of range", "Wrong linkage section", "Storage overlay", "Use after free"], ["Check dump registers and PSW", "Review COBOL table bounds", "Validate linkage parameter list"], ["Correct pointer/subscript logic", "Align caller and callee definitions", "Enable runtime checks in test"], confidence="HIGH", tags=["program", "storage", "addressing"], ), "S0C5": entry( "S0C5", "SYSTEM_ABEND", "PROGRAM_STORAGE", "CRITICAL", "Addressing exception", "The program used an address that is not valid for the current addressing mode or storage context.", ["Invalid address calculation", "Wrong AMODE", "Corrupted save area", "Branch through bad pointer"], ["Check AMODE/RMODE", "Inspect save area chain", "Review called module linkage"], ["Correct addressing assumptions", "Relink with correct options", "Validate program call interface"], tags=["program", "addressing"], ), "S0C6": entry( "S0C6", "SYSTEM_ABEND", "PROGRAM_EXCEPTION", "CRITICAL", "Specification exception", "An instruction or operand was not properly aligned or not valid for the instruction specification.", ["Invalid operand alignment", "Assembler instruction misuse", "Corrupted operand address"], ["Inspect failing instruction", "Review assembler macros or generated code", "Check register contents"], ["Correct operand alignment", "Fix generated code or assembler logic"], tags=["assembler", "program"], ), "S0C7": entry( "S0C7", "SYSTEM_ABEND", "DATA_EXCEPTION", "CRITICAL", "Data exception", "A numeric instruction encountered invalid data, commonly invalid packed decimal or zoned decimal content.", ["Non-numeric data in numeric field", "Copybook does not match physical layout", "Upstream file format changed", "Invalid packed decimal sign nibble"], ["Identify failing record", "Compare copybook with physical file", "Validate packed/zoned decimal fields", "Review recent upstream changes"], ["Reject or repair corrupt input records", "Synchronize copybook and file layout", "Add pre-validation before arithmetic"], confidence="HIGH", tags=["cobol", "numeric", "data-quality", "packed-decimal"], ), "S0CB": entry( "S0CB", "SYSTEM_ABEND", "DATA_EXCEPTION", "CRITICAL", "Decimal divide exception", "A decimal divide operation failed, often due to divide by zero, overflow, or invalid decimal operands.", ["Division by zero", "Invalid decimal operand", "Scale/precision mismatch"], ["Review arithmetic fields", "Validate divisor values", "Check copybook precision and scale"], ["Add defensive divisor checks", "Correct data layout or invalid input values"], tags=["cobol", "numeric", "decimal"], ), "S0CE": entry( "S0CE", "SYSTEM_ABEND", "DATA_EXCEPTION", "CRITICAL", "Floating-point divide exception", "A floating-point operation failed, commonly due to division by zero or invalid floating-point value.", ["Division by zero", "Invalid floating-point data", "Overflow/underflow scenario"], ["Review arithmetic around failing instruction", "Validate input ranges"], ["Add input validation", "Handle exceptional arithmetic conditions"], tags=["numeric", "floating-point"], ), "S013": entry( "S013", "SYSTEM_ABEND", "DATASET_DCB", "ERROR", "Open or DCB error", "A data set could not be opened correctly, often due to DCB, data set organization, label, member, or disposition problems.", ["Incorrect DCB attributes", "Wrong DSORG/RECFM/LRECL", "Missing PDS member", "Program expects a different file shape"], ["Check DD statement and DSN", "Inspect data set attributes", "Compare DCB with program definition"], ["Correct DCB or allocation", "Correct member name or DSORG", "Recreate data set with expected attributes"], tags=["dataset", "open", "dcb"], ), "S106": entry( "S106", "SYSTEM_ABEND", "LOAD_MODULE", "ERROR", "Load module error", "The system could not load a required module or found an invalid module condition.", ["Module not found", "Invalid load module", "STEPLIB/JOBLIB/APF issue", "Architecture or authorization mismatch"], ["Check STEPLIB/JOBLIB", "Verify module exists", "Review APF authorization requirements"], ["Correct library concatenation", "Rebuild module", "Fix APF or authorization setup"], tags=["load", "module", "steplib"], ), "S122": entry( "S122", "SYSTEM_ABEND", "OPERATOR_CANCEL", "ERROR", "Operator or system cancel", "A task was canceled by operator, system policy, or automation.", ["Operator cancel", "Automation rule", "System policy"], ["Check SYSLOG/OPERLOG", "Check scheduler and automation logs"], ["Confirm expected cancellation", "Tune automation policy if needed"], tags=["cancel", "operations"], ), "S222": entry( "S222", "SYSTEM_ABEND", "OPERATOR_CANCEL", "ERROR", "Job or task canceled", "The job, step, or task was canceled, commonly by operator action or automation.", ["Operator canceled the job", "Automation canceled after policy rule", "Job exceeded local limits"], ["Check SYSLOG/OPERLOG for cancel command", "Check scheduler or automation rules"], ["Document the cancel", "Tune rules if cancellation was unexpected"], tags=["cancel", "operations"], ), "S322": entry( "S322", "SYSTEM_ABEND", "TIME_LIMIT", "ERROR", "CPU time limit exceeded", "The job or step exceeded its CPU time limit.", ["Unexpected loop", "Input volume spike", "TIME parameter too low", "Inefficient SQL or sort"], ["Check step TIME parameter", "Review last successful runtime", "Inspect loop or volume changes"], ["Fix loop or inefficient logic", "Adjust TIME only after validation", "Split workload if necessary"], confidence="HIGH", tags=["timeout", "cpu", "batch"], ), "S422": entry( "S422", "SYSTEM_ABEND", "WAIT_LIMIT", "ERROR", "Wait time limit exceeded", "The job exceeded wait-time or inactive time limits.", ["Waiting for resource", "Deadlock or enqueue wait", "External dependency not available"], ["Check ENQ/resource waits", "Check database or dataset locks", "Review dependency status"], ["Resolve resource wait", "Tune timeout policy if legitimate"], tags=["wait", "enqueue", "resource"], ), "S522": entry( "S522", "SYSTEM_ABEND", "WAIT_LIMIT", "ERROR", "Wait limit exceeded", "The job or step exceeded the allowed wait time.", ["Waiting on tape/device", "Dataset enqueue", "DB wait", "External subsystem wait"], ["Check SYSLOG/OPERLOG", "Check enqueue/resource contention", "Look for preceding IEC/IGD/DSN messages"], ["Resolve blocked resource", "Re-run after resource availability", "Adjust limit only when justified"], tags=["wait", "resource", "batch"], ), "S722": entry( "S722", "SYSTEM_ABEND", "OUTPUT_LIMIT", "WARNING", "SYSOUT line limit exceeded", "The job exceeded a SYSOUT or output line limit.", ["Excessive logging", "Loop producing output", "Incorrect report selection"], ["Check SYSOUT class and line limits", "Inspect repeated messages"], ["Reduce logging", "Correct loop or output selection", "Increase limit only if valid"], tags=["sysout", "output"], ), "S804": entry( "S804", "SYSTEM_ABEND", "REGION_STORAGE", "ERROR", "Region unavailable", "Insufficient virtual storage was available for the requested region.", ["REGION too small", "Storage leak", "Large in-memory sort or table"], ["Check REGION parameter", "Review storage usage", "Check recent data volume growth"], ["Tune REGION", "Reduce memory usage", "Use external sort/work datasets"], tags=["region", "memory"], ), "S806": entry( "S806", "SYSTEM_ABEND", "LOAD_MODULE", "ERROR", "Program not found or load failure", "A program could not be loaded. The usual cause is a missing module or incorrect STEPLIB/JOBLIB/LNKLST setup.", ["Program not found", "Incorrect STEPLIB/JOBLIB", "Wrong program name", "Module not in APF-authorized library when required"], ["Check EXEC PGM name", "Check STEPLIB/JOBLIB concatenation", "Verify module exists"], ["Correct library concatenation", "Deploy the load module", "Fix EXEC PGM spelling"], confidence="HIGH", tags=["load", "module", "steplib"], ), "S80A": entry( "S80A", "SYSTEM_ABEND", "REGION_STORAGE", "ERROR", "GETMAIN or storage request failed", "The job could not obtain requested virtual storage.", ["REGION too small", "Storage fragmentation", "Application memory growth", "Sort or buffer allocation too large"], ["Check REGION", "Review storage usage and recent data volume", "Check LE storage options"], ["Tune memory usage", "Increase REGION if appropriate", "Fix storage leak"], tags=["memory", "region", "getmain"], ), "S878": entry( "S878", "SYSTEM_ABEND", "REGION_STORAGE", "ERROR", "Insufficient virtual storage", "The program was unable to obtain enough virtual storage for a request.", ["REGION too small", "Large sort/work area", "Storage leak", "Unexpected volume growth"], ["Review REGION and MEMLIMIT", "Inspect application storage use", "Check DFSORT/utility options"], ["Increase REGION/MEMLIMIT if justified", "Reduce in-memory structures", "Investigate leak"], confidence="HIGH", tags=["memory", "region", "storage"], ), "S913": entry( "S913", "SYSTEM_ABEND", "SECURITY", "CRITICAL", "Security authorization failure", "A data set, resource, or function could not be accessed because the user or job did not have sufficient authority.", ["RACF/ACF2/Top Secret denial", "Protected dataset", "Missing facility or program authorization", "Wrong submitter or surrogate"], ["Check ICH/IRR security messages", "Verify job user ID and group", "Check dataset and facility profiles"], ["Request proper access", "Correct job credentials or surrogate setup", "Avoid bypassing security controls"], confidence="HIGH", tags=["security", "racf", "authorization"], reason_notes={"04": "Commonly associated with insufficient authority for the requested resource."}, ), "SB37": entry( "SB37", "SYSTEM_ABEND", "DATASET_SPACE", "ERROR", "End-of-volume or space exhausted", "A data set ran out of available space on the current volume or could not extend as needed.", ["Primary/secondary SPACE too small", "Volume full", "No secondary extents available", "Large unexpected output"], ["Check IEC messages", "Check DD SPACE parameter", "Check volume free space", "Review output growth"], ["Increase SPACE", "Move to larger storage class or volume", "Fix runaway output"], confidence="HIGH", tags=["dataset", "space", "allocation"], ), "SD37": entry( "SD37", "SYSTEM_ABEND", "DATASET_SPACE", "ERROR", "No secondary space specified", "A data set needed to extend but no secondary allocation was available or usable.", ["SPACE lacks secondary quantity", "Output larger than expected", "Allocation policy unsuitable"], ["Inspect DD SPACE parameter", "Check allocation class", "Review record counts"], ["Add secondary allocation", "Increase primary allocation", "Use SMS storage class policy"], confidence="HIGH", tags=["dataset", "space", "allocation"], ), "SE37": entry( "SE37", "SYSTEM_ABEND", "DATASET_SPACE", "ERROR", "Maximum extents or space exhausted", "The data set could not extend further, often due to extent limit or volume constraints.", ["Maximum extents reached", "Volume full", "Poor allocation sizing", "Output growth"], ["Check extents", "Check volume free space", "Review allocation and SMS policy"], ["Reallocate with larger extents", "Move to suitable volume/storage class", "Tune output volume"], confidence="HIGH", tags=["dataset", "space", "extents"], ), "U4038": entry( "U4038", "USER_ABEND", "LANGUAGE_ENVIRONMENT", "ERROR", "Language Environment user abend", "Language Environment terminated the application after an unhandled condition. The true root cause is often a preceding CEE message or system ABEND.", ["Unhandled LE condition", "Application called CEETERM/abort", "Preceding COBOL or PL/I runtime error"], ["Search preceding CEE messages", "Check CEEDUMP", "Look for S0C7/S0C4 before U4038"], ["Fix preceding runtime condition", "Review LE options", "Analyze CEEDUMP"], confidence="MEDIUM", tags=["le", "runtime", "cobol"], ), "U0004": entry( "U0004", "USER_ABEND", "APPLICATION", "ERROR", "Application user abend 0004", "The application deliberately issued user abend 0004 or propagated an application-defined failure.", ["Application validation failure", "Business rule error", "Utility or product-defined abend"], ["Check application messages immediately before the abend", "Review vendor/product documentation"], ["Fix the triggering business or configuration condition"], tags=["application", "user-abend"], aliases=["U4", "U004"], ), "U0999": entry( "U0999", "USER_ABEND", "APPLICATION", "ERROR", "Generic application user abend", "A program or product issued a user abend 0999. The detailed meaning is application-specific.", ["Application-defined failure", "Controlled termination", "Product-specific error"], ["Read messages before the abend", "Check program/product documentation", "Identify failing step and module"], ["Fix product/application condition", "Escalate with full joblog and dumps if needed"], tags=["application", "user-abend"], aliases=["U999"], ), "RC=0": entry( "RC=0", "RETURN_CODE", "RETURN_CODE", "INFO", "Successful condition code", "Condition code 0 normally means successful completion.", [], ["Confirm downstream scheduler expectations"], [], tags=["return-code"], aliases=["CC=0", "MAXCC=0", "LASTCC=0"], ), "RC=4": entry( "RC=4", "RETURN_CODE", "RETURN_CODE", "WARNING", "Warning condition code", "Condition code 4 usually indicates a warning or minor condition. It may be acceptable depending on local standards.", ["Utility warning", "Non-blocking data condition", "Compiler warning"], ["Check step messages", "Confirm whether RC=4 is allowed for this job"], ["Document or resolve warnings according to standards"], tags=["return-code", "warning"], aliases=["CC=4", "MAXCC=4", "LASTCC=4"], ), "RC=8": entry( "RC=8", "RETURN_CODE", "RETURN_CODE", "ERROR", "Error condition code", "Condition code 8 usually indicates an error requiring attention.", ["Utility failed a validation", "Compiler error", "Application detected an error"], ["Inspect failing step messages", "Check scheduler threshold", "Review related dataset/program messages"], ["Correct the triggering error and rerun if needed"], tags=["return-code", "error"], aliases=["CC=8", "MAXCC=8", "LASTCC=8"], ), "RC=12": entry( "RC=12", "RETURN_CODE", "RETURN_CODE", "ERROR", "Severe condition code", "Condition code 12 usually indicates a severe error in a utility, compiler, or application step.", ["Compiler severe error", "Utility rejected operation", "Application fatal validation error"], ["Read diagnostic messages around the step", "Check input and parameters", "Check generated reports"], ["Fix the severe error before rerun", "Do not bypass without owner approval"], tags=["return-code", "severe"], aliases=["CC=12", "MAXCC=12", "LASTCC=12"], ), "RC=16": entry( "RC=16", "RETURN_CODE", "RETURN_CODE", "CRITICAL", "Critical condition code", "Condition code 16 usually indicates a critical failure and should be treated as a failed step unless local standards say otherwise.", ["Utility critical failure", "Compiler unrecoverable error", "Application fatal condition"], ["Inspect all step messages", "Check input and environment", "Review scheduler thresholds"], ["Correct root cause and rerun", "Escalate if production batch chain is blocked"], tags=["return-code", "critical"], aliases=["CC=16", "MAXCC=16", "LASTCC=16"], ), } def normalize_code(raw: str) -> Optional[str]: if not raw: return None token = raw.strip().upper().replace(" ", "") token = token.rstrip(".,;:)") token = token.lstrip("(") if re.fullmatch(r"(?:MAXCC|LASTCC|RC|CC)[:=]?\d{1,4}", token): label, value = re.match(r"(MAXCC|LASTCC|RC|CC)[:=]?(\d{1,4})", token).groups() try: n = int(value) except ValueError: return None if n <= 0: return "RC=0" if n <= 4: return "RC=4" if n <= 8: return "RC=8" if n <= 12: return "RC=12" return "RC=16" if re.fullmatch(r"[US][0-9A-F]{1,4}", token): prefix = token[0] rest = token[1:] if prefix == "U": if rest.zfill(4) == "0000": return None return "U" + rest.zfill(4) if prefix == "S": # Keep common B37/D37/E37 family without zero padding. if rest in {"B37", "D37", "E37"}: return prefix + rest return "S" + rest.zfill(3) if re.fullmatch(r"ABEND[=:-]?[US][0-9A-F]{1,4}", token): return normalize_code(re.sub(r"^ABEND[=:-]?", "", token)) if re.fullmatch(r"COMP(?:LETION)?CODE[=:-]?[US][0-9A-F]{1,4}", token): return normalize_code(re.sub(r"^COMP(?:LETION)?CODE[=:-]?", "", token)) return None def severity_name(score: int) -> str: result = "INFO" for name, threshold in ORDER_TO_SEVERITY: if score >= threshold: result = name return result def max_severity(values: Iterable[str]) -> str: best = "INFO" for value in values: if SEVERITY_ORDER.get(value, 0) > SEVERITY_ORDER.get(best, 0): best = value return best def severity_at_least(value: str, minimum: str) -> bool: return SEVERITY_ORDER.get(value, 0) >= SEVERITY_ORDER.get(minimum, 0) def redact_text(text: str) -> str: text = DATASET_RE.sub("", text) text = re.sub(r"\b[A-Z0-9]{1,8}\.(JOB|STC|TSU)\d{4,8}\b", "", text, flags=re.IGNORECASE) return text def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as fh: for block in iter(lambda: fh.read(65536), b""): h.update(block) return h.hexdigest() def catalog_with_aliases(catalog: Dict[str, CatalogEntry]) -> Dict[str, CatalogEntry]: result: Dict[str, CatalogEntry] = {} for key, item in catalog.items(): result[item.code] = item result[key.upper()] = item for alias in item.aliases: norm = normalize_code(alias) or alias.upper() result[norm] = item result[alias.upper()] = item return result def load_custom_catalog(path: Optional[str]) -> Dict[str, CatalogEntry]: catalog = dict(BASE_CATALOG) if not path: return catalog payload = json.loads(Path(path).read_text(encoding="utf-8")) if isinstance(payload, dict): items = payload.get("entries", payload.values()) else: items = payload for raw in items: item = CatalogEntry( code=str(raw["code"]).upper(), aliases=[str(x).upper() for x in raw.get("aliases", [])], kind=str(raw.get("kind", "CUSTOM")), category=str(raw.get("category", "CUSTOM")), default_severity=str(raw.get("default_severity", raw.get("severity", "WARNING"))).upper(), title=str(raw.get("title", raw["code"])), explanation=str(raw.get("explanation", "Custom catalog entry.")), probable_causes=list(raw.get("probable_causes", [])), recommended_checks=list(raw.get("recommended_checks", [])), recommended_actions=list(raw.get("recommended_actions", [])), confidence=str(raw.get("confidence", "MEDIUM")), tags=list(raw.get("tags", [])), reason_notes=dict(raw.get("reason_notes", {})), ) catalog[item.code] = item return catalog class AbendAnalyzer: def __init__(self, catalog: Dict[str, CatalogEntry], context_lines: int = 2, redact: bool = False): self.catalog = catalog self.catalog_index = catalog_with_aliases(catalog) self.context_lines = max(0, context_lines) self.redact = redact def analyze_direct_codes(self, codes: Sequence[str], source_name: str = "direct") -> AnalysisResult: lines = [" ".join(codes)] if codes else [] events: List[DecodeEvent] = [] timeline: List[TimelineEvent] = [] for idx, raw in enumerate(codes, start=1): event = self.decode_token(raw, source="direct", line_number=None, line_text=raw, column=None) if event: events.append(event) timeline.append(self.timeline_from_event(event, len(timeline) + 1)) return self.build_result(source_name, None, lines, events, timeline) def analyze_text(self, text: str, source_name: str, source_hash: Optional[str] = None) -> AnalysisResult: lines = text.splitlines() events: List[DecodeEvent] = [] timeline: List[TimelineEvent] = [] state: Dict[str, Optional[str]] = { "job_name": None, "job_id": None, "step_name": None, "program_name": None, } for idx, raw_line in enumerate(lines, start=1): line = raw_line.rstrip("\n") if self.redact: line = redact_text(line) self.update_context_state(line, state) msg_id = self.extract_message_id(line) family = self.message_family(msg_id) if msg_id or self.is_timeline_line(line): timeline.append(TimelineEvent( line_number=idx, event_type=self.classify_timeline_event(line, msg_id), severity=self.infer_line_severity(line, msg_id), message_id=msg_id, code=None, step_name=state.get("step_name"), program_name=state.get("program_name"), job_name=state.get("job_name"), job_id=state.get("job_id"), evidence=line.strip(), )) for raw_token, col in self.extract_codes_from_line(line): event = self.decode_token( raw_token, source="file", line_number=idx, line_text=line, column=col, message_id=msg_id, step_name=state.get("step_name"), program_name=state.get("program_name"), job_name=state.get("job_name"), job_id=state.get("job_id"), reason_code=self.extract_reason_code(line), context_before=self.context_slice(lines, idx, before=True), context_after=self.context_slice(lines, idx, before=False), ) if event: events.append(event) timeline.append(self.timeline_from_event(event, idx)) inferred = self.infer_from_message_line(line, idx, msg_id, state, lines) events.extend(inferred) for event in inferred: timeline.append(self.timeline_from_event(event, idx)) # Remove duplicate decode events caused by ABEND= and bare token on the same line. events = self.deduplicate_events(events) timeline = self.deduplicate_timeline(timeline) return self.build_result(source_name, source_hash, lines, events, timeline) def context_slice(self, lines: Sequence[str], line_number: int, before: bool) -> List[str]: if self.context_lines <= 0: return [] idx = line_number - 1 if before: start = max(0, idx - self.context_lines) result = list(lines[start:idx]) else: end = min(len(lines), idx + 1 + self.context_lines) result = list(lines[idx + 1:end]) if self.redact: return [redact_text(x) for x in result] return result def extract_message_id(self, line: str) -> Optional[str]: match = MSG_ID_RE.search(line) return match.group(1).upper() if match else None def message_family(self, msg_id: Optional[str]) -> Optional[str]: if not msg_id: return None if msg_id.startswith("$HASP"): return "$HASP" return msg_id[:3] def update_context_state(self, line: str, state: Dict[str, Optional[str]]) -> None: # JES2 start/end examples: $HASP373 PAYROLL1 STARTED - INIT 1 - CLASS A - SYS MVSA m = re.search(r"\$HASP\d+\s+([A-Z0-9@$#]{1,8})\b", line, re.IGNORECASE) if m: state["job_name"] = m.group(1).upper() j = JOB_ID_RE.search(line) if j: state["job_id"] = j.group(1).upper() # EXEC statement examples: //STEP030 EXEC PGM=PAYCALC s = re.search(r"//([A-Z0-9@$#]{1,8})\s+EXEC\s+(?:PGM=([A-Z0-9@$#]+)|PROC=([A-Z0-9@$#]+))", line, re.IGNORECASE) if s: state["step_name"] = s.group(1).upper() state["program_name"] = (s.group(2) or s.group(3) or "").upper() or None # IEF messages often include job step/program tokens. m2 = re.search(r"\bIEF\d+[A-Z]?\s+([A-Z0-9@$#]{1,8})\s+([A-Z0-9@$#]{1,8})\s+([A-Z0-9@$#]{1,8})", line, re.IGNORECASE) if m2: state["job_name"] = m2.group(1).upper() state["step_name"] = m2.group(2).upper() state["program_name"] = m2.group(3).upper() def extract_codes_from_line(self, line: str) -> List[Tuple[str, int]]: found: List[Tuple[str, int]] = [] patterns = [ r"\bABEND[=:-]?\s*([US][0-9A-F]{1,4})\b", r"\bCOMP(?:LETION)?\s*CODE[=:-]?\s*([US][0-9A-F]{1,4})\b", r"\b(?:SYSTEM|USER)?\s*ABEND\s+([US][0-9A-F]{1,4})\b", r"\b([US][0-9A-F]{2,4})\b", r"\b(MAXCC|LASTCC|RC|CC)\s*[=:]?\s*(\d{1,4})\b", r"\bCOND\s+CODE\s+(\d{1,4})\b", r"\bCONDITION\s+CODE\s+(\d{1,4})\b", r"\bRETURN\s+CODE\s+(\d{1,4})\b", ] for pattern in patterns: for m in re.finditer(pattern, line, re.IGNORECASE): if m.lastindex == 2 and m.group(1).upper() in {"MAXCC", "LASTCC", "RC", "CC"}: token = f"{m.group(1).upper()}={m.group(2)}" elif "COND" in pattern or "RETURN" in pattern: token = f"RC={m.group(1)}" else: token = m.group(1) norm = normalize_code(token) if norm: found.append((token, m.start())) return found def extract_reason_code(self, line: str) -> Optional[str]: m = re.search(r"\b(?:REASON|RSN|REAS)\s*[=:]?\s*(?:X'([0-9A-F]+)'|([0-9A-F]{1,8}))\b", line, re.IGNORECASE) if not m: return None value = (m.group(1) or m.group(2) or "").upper() return value.zfill(2) if len(value) <= 2 else value def decode_token( self, raw_token: str, source: str, line_number: Optional[int], line_text: str, column: Optional[int], message_id: Optional[str] = None, step_name: Optional[str] = None, program_name: Optional[str] = None, job_name: Optional[str] = None, job_id: Optional[str] = None, reason_code: Optional[str] = None, context_before: Optional[List[str]] = None, context_after: Optional[List[str]] = None, ) -> Optional[DecodeEvent]: norm = normalize_code(raw_token) if not norm: return None catalog_item = self.catalog_index.get(norm) if not catalog_item: catalog_item = self.unknown_entry(norm) rc_value = self.return_code_value(norm) explanation = catalog_item.explanation if reason_code: reason_hint = self.reason_hint(catalog_item, reason_code) if reason_hint: explanation = f"{explanation} Reason {reason_code}: {reason_hint}" return DecodeEvent( code=raw_token.upper(), normalized_code=catalog_item.code, kind=catalog_item.kind, category=catalog_item.category, severity=self.adjust_severity(catalog_item.default_severity, line_text, norm), title=catalog_item.title, explanation=explanation, confidence=catalog_item.confidence, source=source, line_number=line_number, column=column, message_id=message_id, step_name=step_name, program_name=program_name, job_name=job_name, job_id=job_id, reason_code=reason_code, return_code=rc_value, evidence=line_text.strip(), context_before=context_before or [], context_after=context_after or [], probable_causes=list(catalog_item.probable_causes), recommended_checks=list(catalog_item.recommended_checks), recommended_actions=list(catalog_item.recommended_actions), tags=list(catalog_item.tags), ) def unknown_entry(self, code: str) -> CatalogEntry: if code.startswith("U"): return entry( code, "USER_ABEND", "APPLICATION", "ERROR", f"User ABEND {code}", "A program or product issued a user ABEND. The precise meaning is application or product-specific.", ["Application-controlled failure", "Product-specific failure", "Deliberate termination after detected error"], ["Read messages immediately before the ABEND", "Identify failing program or product", "Check application documentation"], ["Resolve the application-specific cause", "Escalate with joblog and dump if unclear"], tags=["user-abend", "application"], ) if code.startswith("S"): return entry( code, "SYSTEM_ABEND", "SYSTEM", "ERROR", f"System ABEND {code}", "A system completion code was detected, but it is not in the built-in catalog.", ["System or subsystem failure condition", "Program/environment issue"], ["Check official system code documentation", "Review surrounding messages and dump"], ["Add this code to a custom catalog once diagnosed"], tags=["system-abend", "unknown"], ) return entry( code, "RETURN_CODE", "RETURN_CODE", "WARNING", f"Return code {code}", "A return or condition code was detected.", ["Utility or application return condition"], ["Check step documentation and scheduler threshold"], ["Resolve according to application standards"], tags=["return-code"], ) def return_code_value(self, norm: str) -> Optional[int]: m = re.match(r"RC=(\d+)", norm) return int(m.group(1)) if m else None def reason_hint(self, item: CatalogEntry, reason_code: str) -> Optional[str]: candidates = [reason_code.upper(), reason_code.upper().lstrip("0"), reason_code.upper().zfill(2)] for c in candidates: if c in item.reason_notes: return item.reason_notes[c] return None def adjust_severity(self, severity: str, line: str, norm: str) -> str: uline = line.upper() if norm.startswith("RC=") and any(x in uline for x in ["WARNING", "WARN"]): if severity == "ERROR": return "WARNING" if any(x in uline for x in ["FATAL", "ABEND", "FAILED", "NOT AUTHORIZED", "DENIED"]): if SEVERITY_ORDER.get(severity, 0) < SEVERITY_ORDER["ERROR"]: return "ERROR" return severity def infer_from_message_line( self, line: str, line_number: int, msg_id: Optional[str], state: Dict[str, Optional[str]], lines: Sequence[str], ) -> List[DecodeEvent]: results: List[DecodeEvent] = [] uline = line.upper() custom: Optional[CatalogEntry] = None code = None if msg_id and msg_id.startswith("ICH") or msg_id and msg_id.startswith("IRR"): custom = self.catalog_index.get("S913") code = "S913" elif "NOT AUTHORIZED" in uline or "RACF" in uline or "ACCESS DENIED" in uline: custom = self.catalog_index.get("S913") code = "S913" elif msg_id and msg_id.startswith("IEC") and any(x in uline for x in ["B37", "D37", "E37", "SPACE", "EXTENT"]): if "D37" in uline: custom = self.catalog_index.get("SD37") code = "SD37" elif "E37" in uline: custom = self.catalog_index.get("SE37") code = "SE37" else: custom = self.catalog_index.get("SB37") code = "SB37" elif msg_id and msg_id.startswith("CEE") and "U4038" in uline: custom = self.catalog_index.get("U4038") code = "U4038" elif msg_id and msg_id.startswith("IEW") and any(x in uline for x in ["UNRESOLVED", "NOT FOUND", "ERROR"]): custom = entry( "IEW-ERROR", "MESSAGE", "BINDER", "ERROR", "Binder/link-edit error", "A binder or link-edit diagnostic was detected.", ["Unresolved external reference", "Missing object module", "Invalid binder control statements"], ["Review IEW messages", "Check object libraries and binder cards"], ["Correct binder input or library concatenation"], tags=["binder", "link-edit"], ) code = "IEW-ERROR" elif msg_id and msg_id.startswith("IGY") and any(x in uline for x in ["ERROR", "SEVERE", "RC="]): custom = entry( "COBOL-COMPILE", "MESSAGE", "COMPILER", "ERROR", "COBOL compiler error", "A COBOL compiler diagnostic was detected.", ["Syntax error", "Copybook missing", "Invalid compiler option"], ["Review IGY messages", "Check copybooks and compile options"], ["Correct source or copybook issue and recompile"], tags=["cobol", "compiler"], ) code = "COBOL-COMPILE" elif msg_id and (msg_id.startswith("ICE") or msg_id.startswith("WER")) and any(x in uline for x in ["ERROR", "SEVERE", "INSUFFICIENT", "EXCEEDED", "FAILED", "RC=8", "RC=12", "RC=16"]): custom = entry( "SORT-ERROR", "MESSAGE", "SORT", "ERROR", "Sort utility error", "A DFSORT/SYNCSORT diagnostic was detected.", ["SORTWK shortage", "Invalid control card", "Input/output dataset issue"], ["Review sort messages", "Check SORTIN/SORTOUT/SORTWK DDs", "Check control statements"], ["Correct sort control cards or work allocation"], tags=["sort", "utility"], ) code = "SORT-ERROR" if custom and code: event = DecodeEvent( code=code, normalized_code=custom.code, kind=custom.kind, category=custom.category, severity=custom.default_severity, title=custom.title, explanation=custom.explanation, confidence=custom.confidence, source="inference", line_number=line_number, column=None, message_id=msg_id, step_name=state.get("step_name"), program_name=state.get("program_name"), job_name=state.get("job_name"), job_id=state.get("job_id"), reason_code=self.extract_reason_code(line), return_code=None, evidence=line.strip(), context_before=self.context_slice(lines, line_number, before=True), context_after=self.context_slice(lines, line_number, before=False), probable_causes=list(custom.probable_causes), recommended_checks=list(custom.recommended_checks), recommended_actions=list(custom.recommended_actions), tags=list(custom.tags), ) results.append(event) return results def is_timeline_line(self, line: str) -> bool: u = line.upper() return bool(TIME_RE.search(line) or "$HASP" in u or "IEF" in u or "ABEND" in u or "MAXCC" in u) def classify_timeline_event(self, line: str, msg_id: Optional[str]) -> str: u = line.upper() if "ABEND" in u: return "ABEND" if "MAXCC" in u or "COND CODE" in u or "RC=" in u: return "RETURN_CODE" if msg_id: family = self.message_family(msg_id) or "MESSAGE" return f"{family}_MESSAGE" return "LOG" def infer_line_severity(self, line: str, msg_id: Optional[str]) -> str: u = line.upper() if any(x in u for x in ["ABEND", "FATAL", "NOT AUTHORIZED", "DENIED"]): return "CRITICAL" if any(x in u for x in ["ERROR", "FAILED", "SEVERE"]): return "ERROR" if any(x in u for x in ["WARNING", "WARN"]): return "WARNING" return "INFO" def timeline_from_event(self, event: DecodeEvent, fallback_line: int) -> TimelineEvent: return TimelineEvent( line_number=event.line_number or fallback_line, event_type=event.kind, severity=event.severity, message_id=event.message_id, code=event.normalized_code, step_name=event.step_name, program_name=event.program_name, job_name=event.job_name, job_id=event.job_id, evidence=event.evidence, ) def deduplicate_events(self, events: List[DecodeEvent]) -> List[DecodeEvent]: seen = set() result = [] for e in events: key = (e.normalized_code, e.line_number, e.evidence[:160], e.source) if key in seen: continue # Suppress inference duplicate if explicit event same code same line already exists. explicit_key = (e.normalized_code, e.line_number) if e.source == "inference" and any((x.normalized_code, x.line_number) == explicit_key and x.source != "inference" for x in result): continue seen.add(key) result.append(e) return result def deduplicate_timeline(self, events: List[TimelineEvent]) -> List[TimelineEvent]: seen = set() result = [] for e in events: key = (e.line_number, e.event_type, e.code, e.message_id, e.evidence[:160]) if key in seen: continue seen.add(key) result.append(e) return sorted(result, key=lambda x: x.line_number) def build_result( self, source_name: str, source_hash: Optional[str], lines: Sequence[str], events: List[DecodeEvent], timeline: List[TimelineEvent], ) -> AnalysisResult: highest = max_severity([e.severity for e in events] or ["INFO"]) risk = self.compute_risk_score(events) status = self.compute_status(events, risk) families: Dict[str, int] = {} for t in timeline: fam = self.message_family(t.message_id) if t.message_id else None if fam: families[fam] = families.get(fam, 0) + 1 root_causes = self.root_cause_candidates(events) return AnalysisResult( tool="ABEND Intelligence", version=VERSION, generated_at=datetime.now(timezone.utc).isoformat(), source_name=source_name, source_hash=source_hash, status=status, highest_severity=highest, risk_score=risk, total_lines=len(lines), total_events=len(events), total_timeline_events=len(timeline), system_abends=sum(1 for e in events if e.kind == "SYSTEM_ABEND"), user_abends=sum(1 for e in events if e.kind == "USER_ABEND"), return_codes=sum(1 for e in events if e.kind == "RETURN_CODE"), reason_codes=sum(1 for e in events if e.reason_code), message_families=dict(sorted(families.items())), codes_seen=sorted({e.normalized_code for e in events}), job_names=sorted({e.job_name for e in events if e.job_name}), job_ids=sorted({e.job_id for e in events if e.job_id}), steps_seen=sorted({e.step_name for e in events if e.step_name}), programs_seen=sorted({e.program_name for e in events if e.program_name}), datasets_seen=sorted({d for line in lines for d in DATASET_RE.findall(redact_text(line) if self.redact else line)})[:300], root_cause_candidates=root_causes, events=events, timeline=timeline, ) def compute_risk_score(self, events: Sequence[DecodeEvent]) -> int: if not events: return 0 score = 0 for e in events: base = SEVERITY_ORDER.get(e.severity, 0) weight = 1 if e.kind in {"SYSTEM_ABEND", "USER_ABEND"}: weight = 2 if e.category in {"SECURITY", "DATA_EXCEPTION", "PROGRAM_STORAGE", "DATASET_SPACE"}: weight += 1 score += int(base * weight / 4) if any(e.severity == "CRITICAL" for e in events): score += 25 if len({e.category for e in events if e.severity in {"ERROR", "CRITICAL"}}) >= 3: score += 15 return min(100, score) def compute_status(self, events: Sequence[DecodeEvent], risk: int) -> str: if any(e.severity == "CRITICAL" for e in events): return "FAILED" if any(e.severity == "ERROR" for e in events): return "ERROR" if any(e.severity == "WARNING" for e in events): return "WARNING" if risk > 0: return "INFO" return "OK" def root_cause_candidates(self, events: Sequence[DecodeEvent]) -> List[RootCauseCandidate]: grouped: Dict[str, List[DecodeEvent]] = {} for event in events: if event.severity == "INFO" and event.kind == "RETURN_CODE": continue grouped.setdefault(event.category, []).append(event) candidates: List[RootCauseCandidate] = [] templates = { "DATA_EXCEPTION": ("Input data or copybook mismatch", "Numeric data exception evidence suggests invalid packed/zoned decimal data or layout mismatch."), "PROGRAM_STORAGE": ("Program storage/addressing failure", "Addressing/protection evidence suggests bad pointer, linkage mismatch, storage overlay, or table overrun."), "DATASET_SPACE": ("Dataset allocation or space exhaustion", "Dataset space messages or B37/D37/E37 ABENDs indicate output allocation failure or volume/extent exhaustion."), "SECURITY": ("Security authorization failure", "Security messages or S913 evidence indicate insufficient authority or protected resource access."), "LOAD_MODULE": ("Load module or library concatenation issue", "Load failure evidence points to missing program, wrong STEPLIB/JOBLIB, or invalid module deployment."), "TIME_LIMIT": ("CPU time limit exceeded", "Timeout evidence suggests loop, volume spike, or inadequate TIME policy."), "WAIT_LIMIT": ("Resource wait timeout", "Wait-limit evidence suggests blocked resource, enqueue, external device, or subsystem wait."), "REGION_STORAGE": ("Region or virtual storage shortage", "Virtual storage evidence suggests REGION/MEMLIMIT shortage, storage leak, or large in-memory allocation."), "DATASET_DCB": ("Dataset open or DCB mismatch", "Open/DCB evidence suggests data set attributes or DD definitions do not match program expectations."), "LANGUAGE_ENVIRONMENT": ("Language Environment termination", "LE evidence indicates an unhandled runtime condition; the preceding message is usually decisive."), "RETURN_CODE": ("Non-zero return code", "One or more non-zero condition codes were found and may require job-owner validation."), } for category, evs in grouped.items(): sev = max_severity([e.severity for e in evs]) score = min(100, sum(SEVERITY_ORDER.get(e.severity, 0) for e in evs) // max(1, len(evs)) + min(30, len(evs) * 5)) title, explanation = templates.get(category, (category.replace("_", " ").title(), "Evidence from this category was detected.")) actions: List[str] = [] for e in evs: for action in e.recommended_actions: if action not in actions: actions.append(action) if len(actions) >= 5: break candidates.append(RootCauseCandidate( rank=0, category=category, severity=sev, confidence="HIGH" if len(evs) >= 2 or sev == "CRITICAL" else "MEDIUM", score=score, title=title, evidence_count=len(evs), explanation=explanation, recommended_actions=actions, supporting_codes=sorted({e.normalized_code for e in evs}), supporting_lines=sorted({e.line_number for e in evs if e.line_number})[:20], )) candidates.sort(key=lambda c: (SEVERITY_ORDER.get(c.severity, 0), c.score, c.evidence_count), reverse=True) for idx, c in enumerate(candidates, start=1): c.rank = idx return candidates[:10] def result_to_dict(result: AnalysisResult) -> Dict[str, Any]: return asdict(result) def write_json(result: AnalysisResult, path: str) -> None: Path(path).write_text(json.dumps(result_to_dict(result), indent=2, ensure_ascii=False), encoding="utf-8") def write_csv_findings(result: AnalysisResult, path: str, min_severity: str = "INFO") -> None: fields = [ "severity", "kind", "category", "normalized_code", "title", "line_number", "message_id", "step_name", "program_name", "job_name", "job_id", "reason_code", "return_code", "confidence", "evidence", ] with Path(path).open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for e in result.events: if not severity_at_least(e.severity, min_severity): continue writer.writerow({field: getattr(e, field, None) for field in fields}) def write_csv_timeline(result: AnalysisResult, path: str) -> None: fields = ["line_number", "event_type", "severity", "message_id", "code", "step_name", "program_name", "job_name", "job_id", "evidence"] with Path(path).open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for e in result.timeline: writer.writerow({field: getattr(e, field, None) for field in fields}) def write_catalog(catalog: Dict[str, CatalogEntry], path: str) -> None: payload = {"tool": "ABEND Intelligence", "version": VERSION, "entries": [asdict(v) for v in sorted(catalog.values(), key=lambda x: x.code)]} Path(path).write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") def html_list(items: Sequence[str]) -> str: if not items: return "None" return "" def severity_class(sev: str) -> str: return { "CRITICAL": "sev-critical", "ERROR": "sev-error", "WARNING": "sev-warning", "INFO": "sev-info", }.get(sev, "sev-info") def write_html(result: AnalysisResult, path: str, min_severity: str = "INFO") -> None: top_events = [e for e in result.events if severity_at_least(e.severity, min_severity)] top_events.sort(key=lambda e: (SEVERITY_ORDER.get(e.severity, 0), e.line_number or 0), reverse=True) rows = [] for e in top_events[:300]: rows.append(f""" {html.escape(e.severity)} {html.escape(e.normalized_code)}
{html.escape(e.kind)} {html.escape(e.category)} {html.escape(e.title)}
{html.escape(e.explanation)} {html.escape(str(e.line_number or ''))} {html.escape(e.evidence)} """) rc_rows = [] for c in result.root_cause_candidates: rc_rows.append(f"""
#{c.rank}{html.escape(c.severity)}{html.escape(c.title)}

{html.escape(c.explanation)}

Confidence: {html.escape(c.confidence)} | Score: {c.score} | Evidence: {c.evidence_count}
{' '.join('' + html.escape(x) + '' for x in c.supporting_codes)}

Recommended actions

{html_list(c.recommended_actions)}
""") timeline_rows = [] for t in result.timeline[:300]: timeline_rows.append(f""" {t.line_number}{html.escape(t.event_type)}{html.escape(t.severity)} {html.escape(t.message_id or '')}{html.escape(t.code or '')}{html.escape(t.step_name or '')}{html.escape(t.evidence)} """) families = "".join(f"{html.escape(k)}: {v}" for k, v in sorted(result.message_families.items())) or "None" codes = "".join(f"{html.escape(x)}" for x in result.codes_seen) or "None" datasets = "".join(f"
  • {html.escape(x)}
    • " for x in result.datasets_seen[:80]) or "
  • None
    • " document = f""" ABEND Intelligence V2 Report

    ABEND Intelligence V2 Report

    Source: {html.escape(result.source_name)} | Generated: {html.escape(result.generated_at)} | Hash: {html.escape(result.source_hash or 'n/a')}

    {html.escape(result.status)}
    Status
    {result.risk_score}/100
    Risk score
    {result.total_events}
    Findings
    {html.escape(result.highest_severity)}
    Highest severity

    Detected codes

    {codes}

    Message families

    {families}

    Root-cause candidates

    {''.join(rc_rows) if rc_rows else '

    No root-cause candidate.

    '}

    Findings

    {''.join(rows)}
    SeverityCodeCategoryDiagnosticLineEvidence

    Timeline

    {''.join(timeline_rows)}
    LineTypeSeverityMessageCodeStepEvidence

    Datasets seen

      {datasets}
    """ Path(path).write_text(document, encoding="utf-8") def print_console(result: AnalysisResult, min_severity: str = "INFO", summary_only: bool = False) -> None: print("ABEND Intelligence V2") print("=" * 80) print(f"Source : {result.source_name}") print(f"Status : {result.status}") print(f"Highest severity : {result.highest_severity}") print(f"Risk score : {result.risk_score}/100") print(f"Lines : {result.total_lines}") print(f"Findings : {result.total_events}") print(f"System ABENDs : {result.system_abends}") print(f"User ABENDs : {result.user_abends}") print(f"Return codes : {result.return_codes}") print(f"Codes seen : {', '.join(result.codes_seen) if result.codes_seen else '-'}") if result.job_names: print(f"Jobs : {', '.join(result.job_names)}") if result.steps_seen: print(f"Steps : {', '.join(result.steps_seen)}") print() print("ROOT CAUSE CANDIDATES") print("-" * 80) if not result.root_cause_candidates: print("No root-cause candidate.") for c in result.root_cause_candidates[:8]: print(f"#{c.rank} [{c.severity}] {c.title} | confidence={c.confidence} | evidence={c.evidence_count}") print(f" {c.explanation}") if c.supporting_codes: print(f" Codes: {', '.join(c.supporting_codes)}") if c.recommended_actions: print(f" Action: {c.recommended_actions[0]}") if summary_only: return print() print("FINDINGS") print("-" * 80) shown = 0 for e in sorted(result.events, key=lambda x: (SEVERITY_ORDER.get(x.severity, 0), x.line_number or 0), reverse=True): if not severity_at_least(e.severity, min_severity): continue shown += 1 where = f"line {e.line_number}" if e.line_number else "direct" print(f"[{e.severity}] {e.normalized_code} {e.title} ({where})") print(f" {e.evidence}") if e.recommended_actions: print(f" Recommended: {e.recommended_actions[0]}") if shown == 0: print("No finding at selected severity.") def demo_text(name: str) -> str: demos = { "s0c7": """//PAYROLL1 JOB (ACCT),'PAYROLL',CLASS=A,MSGCLASS=X,NOTIFY=&SYSUID $HASP373 PAYROLL1 STARTED - INIT 1 - CLASS A - SYS MVSA //STEP010 EXEC PGM=IDCAMS IEF142I PAYROLL1 STEP010 IDCAMS - STEP WAS EXECUTED - COND CODE 0000 //STEP020 EXEC PGM=SORT ICE000I 0 - DFSORT REL 3.1 IEF142I PAYROLL1 STEP020 SORT - STEP WAS EXECUTED - COND CODE 0004 //STEP030 EXEC PGM=PAYCALC //PAYIN DD DSN=CORP.PAYROLL.INPUT.G0007V00,DISP=SHR CEE3207S The system detected a data exception (System Completion Code=0C7). PAYERR01E Invalid packed decimal amount in employee payment input record. IEF450I PAYROLL1 STEP030 PAYCALC - ABEND=S0C7 U0000 REASON=00000000 CEE3250C The system or user abend S0C7 was issued. $HASP395 PAYROLL1 ENDED - ABEND=S0C7 """, "sb37": """//INVOICE JOB (ACCT),'BILLING',CLASS=A,MSGCLASS=X $HASP373 INVOICE STARTED - INIT 2 - CLASS A - SYS MVSA //STEP010 EXEC PGM=IEBGENER //SYSUT2 DD DSN=CORP.BILLING.EXTRACT.G0001V00,DISP=(NEW,CATLG,DELETE),SPACE=(CYL,(10,1)) IEC030I B37-04,IFG0554A,INVOICE,STEP010,SYSUT2,3390,VOL001,CORP.BILLING.EXTRACT.G0001V00 IEF450I INVOICE STEP010 IEBGENER - ABEND=SB37 U0000 REASON=00000004 $HASP395 INVOICE ENDED - ABEND=SB37 """, "s913": """//SECJOB JOB (ACCT),'SECURITY',CLASS=A,MSGCLASS=X $HASP373 SECJOB STARTED - INIT 3 - CLASS A - SYS MVSA //STEP010 EXEC PGM=IDCAMS //INFILE DD DSN=PROD.SECRET.PAYROLL.MASTER,DISP=SHR ICH408I USER(BATCH01 ) GROUP(BATCHGRP ) NAME(BATCH USER) PROD.SECRET.PAYROLL.MASTER CL(DATASET ) INSUFFICIENT ACCESS AUTHORITY IRR012I VERIFICATION FAILED. USER NOT AUTHORIZED TO RESOURCE. IEF450I SECJOB STEP010 IDCAMS - ABEND=S913 U0000 REASON=00000004 $HASP395 SECJOB ENDED - ABEND=S913 """, "mixed": """//BADLOAD JOB (ACCT),'MIXED',CLASS=A,MSGCLASS=X $HASP373 BADLOAD STARTED - INIT 4 - CLASS A - SYS MVSA //STEP010 EXEC PGM=DOESNOTX CSV003I REQUESTED MODULE DOESNOTX NOT FOUND IEF450I BADLOAD STEP010 DOESNOTX - ABEND=S806 U0000 REASON=00000004 //STEP020 EXEC PGM=BIGSORT ICE046A SORT CAPACITY EXCEEDED - INSUFFICIENT SORTWK SPACE IEF142I BADLOAD STEP020 BIGSORT - STEP WAS EXECUTED - COND CODE 0016 $HASP395 BADLOAD ENDED - MAXCC=16 """, } return demos.get(name, demos["s0c7"]) def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="ABEND Intelligence V2 - z/OS/MVS ABEND, return code, reason code and message analyzer" ) parser.add_argument("codes", nargs="*", help="Codes to decode directly, for example S0C7 SB37 S913 U4038 RC=8 MAXCC=16") parser.add_argument("--input", "-i", help="Input text file: JESMSGLG, JESYSMSG, SYSOUT, joblog, dump extract") parser.add_argument("--stdin", action="store_true", help="Read input text from stdin") parser.add_argument("--demo", choices=["s0c7", "sb37", "s913", "mixed"], help="Run a built-in demo scenario") parser.add_argument("--context-lines", type=int, default=2, help="Number of context lines before/after each finding") parser.add_argument("--redact", action="store_true", help="Redact dataset-like names in output") parser.add_argument("--custom-catalog", help="JSON catalog extension/override file") parser.add_argument("--export-catalog", help="Export the effective catalog to JSON") parser.add_argument("--catalog", action="store_true", help="Print the built-in catalog summary and exit") parser.add_argument("--explain", help="Explain a single code and exit") parser.add_argument("--json", dest="json_output", help="Write full analysis JSON") parser.add_argument("--csv-findings", help="Write findings CSV") parser.add_argument("--csv-timeline", help="Write timeline CSV") parser.add_argument("--html", dest="html_output", help="Write HTML report") parser.add_argument("--min-severity", choices=list(SEVERITY_ORDER.keys()), default="INFO", help="Minimum severity for console/CSV/HTML findings") parser.add_argument("--summary-only", action="store_true", help="Print only summary and root-cause candidates") parser.add_argument("--fail-on", choices=list(SEVERITY_ORDER.keys()), help="Exit with code 2 if analysis has at least this severity") return parser.parse_args(argv) def print_catalog(catalog: Dict[str, CatalogEntry]) -> None: print("ABEND Intelligence catalog") print("=" * 80) for item in sorted(catalog.values(), key=lambda x: x.code): print(f"{item.code:10} {item.kind:14} {item.default_severity:9} {item.category:20} {item.title}") def print_explain(catalog: Dict[str, CatalogEntry], code: str) -> None: idx = catalog_with_aliases(catalog) norm = normalize_code(code) or code.upper() item = idx.get(norm) if not item: print(f"No built-in catalog entry for {code}. Normalized: {norm}") return print(f"Code : {item.code}") print(f"Kind : {item.kind}") print(f"Category : {item.category}") print(f"Severity : {item.default_severity}") print(f"Title : {item.title}") print(f"Explanation: {item.explanation}") if item.probable_causes: print("Probable causes:") for x in item.probable_causes: print(f" - {x}") if item.recommended_actions: print("Recommended actions:") for x in item.recommended_actions: print(f" - {x}") def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) catalog = load_custom_catalog(args.custom_catalog) if args.export_catalog: write_catalog(catalog, args.export_catalog) print(f"Catalog exported to: {args.export_catalog}") return 0 if args.catalog: print_catalog(catalog) return 0 if args.explain: print_explain(catalog, args.explain) return 0 analyzer = AbendAnalyzer(catalog, context_lines=args.context_lines, redact=args.redact) source_hash = None if args.demo: text = demo_text(args.demo) result = analyzer.analyze_text(text, source_name=f"demo:{args.demo}", source_hash=None) elif args.stdin: text = sys.stdin.read() result = analyzer.analyze_text(text, source_name="stdin", source_hash=None) elif args.input: path = Path(args.input) text = path.read_text(encoding="utf-8", errors="ignore") source_hash = sha256_file(path) result = analyzer.analyze_text(text, source_name=str(path), source_hash=source_hash) elif args.codes: result = analyzer.analyze_direct_codes(args.codes, source_name="direct") else: print("No input. Use direct codes, --input, --stdin, or --demo.", file=sys.stderr) return 1 print_console(result, min_severity=args.min_severity, summary_only=args.summary_only) if args.json_output: write_json(result, args.json_output) print(f"JSON written to: {args.json_output}") if args.csv_findings: write_csv_findings(result, args.csv_findings, min_severity=args.min_severity) print(f"Findings CSV written to: {args.csv_findings}") if args.csv_timeline: write_csv_timeline(result, args.csv_timeline) print(f"Timeline CSV written to: {args.csv_timeline}") if args.html_output: write_html(result, args.html_output, min_severity=args.min_severity) print(f"HTML report written to: {args.html_output}") if args.fail_on and severity_at_least(result.highest_severity, args.fail_on): return 2 return 0 if __name__ == "__main__": raise SystemExit(main())