diff --git a/scripts/README.md b/scripts/README.md
index 97fd769ee..1e00b8ea2 100644
--- a/scripts/README.md
+++ b/scripts/README.md
@@ -92,6 +92,18 @@ They cannot be combined with a diff base. Both a path and its resolved symlink
 destination must be in scope. Results are sorted and deduplicated.
 An empty selection means there are no eligible inputs, not a completed scan.
 
+## Shared process and log helpers
+
+`lint_common.py` handles process-group cleanup, exit statuses and Make's
+parallel jobserver. Drivers must check command results or use `check=True`.
+The entry-point helper supports command tracing with `V=1` or `--verbose`.
+Progress is separate from captured analyzer output.
+
+Drivers save complete stdout/stderr under `.cache/lint/`. The helpers preserve
+raw bytes without filtering. Reusing a log name replaces the previous report
+by default; callers can preserve it with `truncate=False`. There is no automatic
+log archive.
+
 ## Dependencies and tooling tests
 
 Shared analyzer tooling requires Linux, Git and Python 3.10+, using only the
diff --git a/scripts/lint_common.py b/scripts/lint_common.py
new file mode 100644
index 000000000..b8a5b8ac4
--- /dev/null
+++ b/scripts/lint_common.py
@@ -0,0 +1,172 @@
+"""Process and report helpers for the analyzer drivers."""
+
+import contextlib
+import os
+import re
+import shlex
+import shutil
+import signal
+import stat
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+PROGRESS_INTERVAL = 5
+SPINNER_INTERVAL = 0.1
+SPINNER_FRAMES = "|/-\\"
+VERBOSE = os.environ.get("V", "0") not in {"", "0"}
+
+
+def show_log(path, label="Full log"):
+    print(f"  {label}: {Path(path).resolve()}", file=sys.stderr, flush=True)
+
+
+class Progress:
+    """Animate only a quiet child; redirected output gets periodic plain lines."""
+
+    def __init__(self, label, *, quiet=True):
+        self.label = label
+        self.started = time.monotonic()
+        self.animated = (bool(label) and quiet and sys.stderr.isatty()
+                         and os.environ.get("TERM") != "dumb")
+        self.interval = SPINNER_INTERVAL if self.animated else PROGRESS_INTERVAL
+        self.frame = 0
+        self.visible = False
+
+    def update(self):
+        elapsed = int(time.monotonic() - self.started)
+        if self.animated:
+            elapsed -= elapsed % PROGRESS_INTERVAL
+            frame = SPINNER_FRAMES[self.frame % len(SPINNER_FRAMES)]
+            text = f"  {frame} {self.label}  {elapsed // 60:02}:{elapsed % 60:02}"
+            width = shutil.get_terminal_size().columns
+            print("\r\033[2K" + text[:max(1, width - 1)], end="", file=sys.stderr, flush=True)
+            self.frame += 1
+        else:
+            print(f"  {self.label}: running ({elapsed}s elapsed)", file=sys.stderr, flush=True)
+        self.visible = True
+
+    def clear(self):
+        if self.animated and self.visible:
+            print("\r\033[2K", end="", file=sys.stderr, flush=True)
+
+    def finish(self, status):
+        self.clear()
+        if self.visible:
+            elapsed = int(time.monotonic() - self.started)
+            print(f"  {self.label}: finished ({elapsed}s, exit {status})",
+                  file=sys.stderr, flush=True)
+
+
+def jobserver_fds(env):
+    """Forward an inherited pipe jobserver without acquiring its tokens."""
+    matches = re.findall(r"--jobserver-(?:auth|fds)=(\S+)", env.get("MAKEFLAGS", ""))
+    if not matches or not re.fullmatch(r"\d+,\d+", matches[-1]):
+        return ()
+    descriptors = tuple(map(int, matches[-1].split(",")))
+    try:
+        for descriptor in descriptors:
+            if (not os.get_inheritable(descriptor)
+                    or not stat.S_ISFIFO(os.fstat(descriptor).st_mode)):
+                return ()
+    except OSError:
+        return ()
+    return descriptors
+
+
+def start(command, *, env=None, **kwargs):
+    environment = os.environ if env is None else env
+    if VERBOSE:
+        print("  $ " + shlex.join(map(str, command)), file=sys.stderr, flush=True)
+    return subprocess.Popen(command, env=environment, start_new_session=True,
+                            pass_fds=jobserver_fds(environment), **kwargs)
+
+
+def stop(process):
+    """Stop the owned process group, including children of an exited driver."""
+    with contextlib.suppress(ProcessLookupError):
+        os.killpg(process.pid, signal.SIGTERM)
+    try:
+        process.wait(timeout=2)
+    except subprocess.TimeoutExpired:
+        pass
+    with contextlib.suppress(ProcessLookupError):
+        os.killpg(process.pid, signal.SIGKILL)
+    process.wait()
+
+
+def exit_status(returncode):
+    return returncode if returncode >= 0 else 128 - returncode
+
+
+def run(command, *, input=None, check=False, progress=None, **kwargs):
+    if input is not None:
+        kwargs["stdin"] = subprocess.PIPE
+    process = start(command, **kwargs)
+    indicator = Progress(progress, quiet=kwargs.get("stdout") is not None
+                         and kwargs.get("stderr") is not None)
+    try:
+        if indicator.animated:
+            indicator.update()
+        while True:
+            try:
+                stdout, stderr = process.communicate(
+                    input, timeout=indicator.interval if progress else None)
+                break
+            except subprocess.TimeoutExpired:
+                # communicate retains pending input and output across retries.
+                input = None
+                indicator.update()
+    except BaseException:
+        indicator.clear()
+        stop(process)
+        raise
+    result = subprocess.CompletedProcess(command, exit_status(process.returncode), stdout, stderr)
+    indicator.finish(result.returncode)
+    if check:
+        result.check_returncode()
+    return result
+
+
+def log_path(root, name, *, truncate=True):
+    path = root / ".cache/lint" / name
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if truncate:
+        path.write_bytes(b"")
+    return path
+
+
+def copy_output(path, output=None):
+    destination = output if output is not None else sys.stderr.buffer
+    with Path(path).open("rb") as stream:
+        shutil.copyfileobj(stream, destination)
+    destination.flush()
+
+
+class Interrupted(Exception):
+    def __init__(self, signum):
+        self.signum = signum
+
+
+def entrypoint(main):
+    global VERBOSE
+    VERBOSE = VERBOSE or "--verbose" in sys.argv[1:]
+    sys.argv[1:] = [arg for arg in sys.argv[1:] if arg != "--verbose"]
+
+    def interrupt(signum, _frame):
+        raise Interrupted(signum)
+
+    signal.signal(signal.SIGTERM, interrupt)
+    try:
+        return main()
+    except Interrupted as error:
+        return 128 + error.signum
+    except KeyboardInterrupt:
+        return 130
+    except subprocess.CalledProcessError as error:
+        print(f"Command failed ({error.returncode}): {error.cmd}", file=sys.stderr)
+        return error.returncode
+    except (OSError, ValueError) as error:
+        print(str(error), file=sys.stderr)
+        return 2
diff --git a/scripts/tests/test_lint_common.py b/scripts/tests/test_lint_common.py
new file mode 100644
index 000000000..334b0dae7
--- /dev/null
+++ b/scripts/tests/test_lint_common.py
@@ -0,0 +1,279 @@
+#!/usr/bin/env python3
+"""Check process, progress and raw-log contracts shared by analyzer drivers."""
+
+import contextlib
+import io
+import os
+import pty
+import selectors
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.dont_write_bytecode = True
+SCRIPTS = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(SCRIPTS))
+from lint_common import Progress, copy_output, jobserver_fds, log_path, run  # noqa: E402
+
+
+class LogTest(unittest.TestCase):
+    def test_log_creation_append_and_reset_preserve_raw_bytes(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            log = log_path(root, "fixture.log")
+            self.assertEqual(log, root / ".cache/lint/fixture.log")
+            self.assertEqual(log.read_bytes(), b"")
+            raw = b"diagnostic\x00\xff\r\n"
+            log.write_bytes(raw)
+            self.assertEqual(log_path(root, "fixture.log", truncate=False), log)
+            output = io.BytesIO()
+            copy_output(log, output)
+            self.assertEqual(output.getvalue(), raw)
+            self.assertEqual(log.read_bytes(), raw)
+            self.assertEqual(log_path(root, "fixture.log"), log)
+            self.assertEqual(log.read_bytes(), b"")
+
+
+class ProcessTest(unittest.TestCase):
+    def test_progress_is_live_and_preserves_piped_data_and_failure(self):
+        for check in (False, True):
+            with self.subTest(check=check), tempfile.TemporaryDirectory() as directory:
+                release = Path(directory) / "release"
+
+                class Progress(io.StringIO):
+                    def write(self, text):
+                        size = super().write(text)
+                        if self.getvalue().count("running (") >= 2:
+                            release.touch()
+                        return size
+
+                script = """
+import os, sys, time
+from pathlib import Path
+payload = sys.stdin.buffer.read()
+deadline = time.monotonic() + 5
+while not Path(sys.argv[1]).exists():
+    if time.monotonic() > deadline:
+        sys.exit(99)
+    time.sleep(0.01)
+os.write(1, payload)
+os.write(2, b'diagnostic\\x00\\xff')
+sys.exit(7)
+"""
+                payload = b"input\x00\xff" * 20000
+                progress = Progress()
+                with patch("lint_common.PROGRESS_INTERVAL", 0.02), \
+                        contextlib.redirect_stderr(progress):
+                    try:
+                        result = run([sys.executable, "-c", script, str(release)],
+                                     input=payload, stdout=subprocess.PIPE,
+                                     stderr=subprocess.PIPE, progress="Fixture", check=check)
+                    except subprocess.CalledProcessError as error:
+                        self.assertTrue(check)
+                        result = error
+                    else:
+                        self.assertFalse(check)
+                self.assertEqual(result.returncode, 7)
+                self.assertEqual(result.stdout, payload)
+                self.assertEqual(result.stderr, b"diagnostic\x00\xff")
+                self.assertGreaterEqual(progress.getvalue().count("Fixture: running ("), 2)
+
+    def test_fast_process_does_not_print_periodic_progress(self):
+        progress = io.StringIO()
+        with contextlib.redirect_stderr(progress):
+            result = run([sys.executable, "-c", "pass"], progress="Fixture")
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(progress.getvalue(), "")
+
+    def test_terminal_spinner_is_live_and_separate_from_child_output(self):
+        with tempfile.TemporaryDirectory() as directory:
+            release = Path(directory) / "release"
+            child = """
+import os, sys, time
+from pathlib import Path
+deadline = time.monotonic() + 5
+while not Path(sys.argv[1]).exists():
+    if time.monotonic() > deadline:
+        sys.exit(99)
+    time.sleep(0.01)
+os.write(1, b'raw\\x00\\xff')
+os.write(2, b'diagnostic\\n')
+sys.exit(7)
+"""
+            script = """
+import subprocess, sys
+sys.path.insert(0, sys.argv[1])
+from lint_common import entrypoint, run
+def main():
+    result = run([sys.executable, '-c', sys.argv[2], sys.argv[3]],
+                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, progress='Fixture')
+    sys.stdout.buffer.write(result.stdout + result.stderr)
+    return result.returncode
+sys.exit(entrypoint(main))
+"""
+            master, slave = pty.openpty()
+            process = subprocess.Popen(
+                [sys.executable, "-B", "-c", script, str(SCRIPTS), child, str(release)],
+                stdout=subprocess.PIPE, stderr=slave, env=dict(os.environ, TERM="xterm"))
+            os.close(slave)
+            display = b""
+            try:
+                with selectors.DefaultSelector() as selector:
+                    selector.register(master, selectors.EVENT_READ)
+                    deadline = time.monotonic() + 5
+                    while display.count(b"\r\x1b[2K") < 2:
+                        self.assertGreater(deadline - time.monotonic(), 0, display)
+                        if selector.select(timeout=0.1):
+                            display += os.read(master, 4096)
+                    self.assertIsNone(process.poll())
+                    release.touch()
+                    stdout, _ = process.communicate(timeout=5)
+                    while selector.select(timeout=0.1):
+                        try:
+                            chunk = os.read(master, 4096)
+                        except OSError:
+                            break  # A closed PTY reports EIO on Linux.
+                        if not chunk:
+                            break
+                        display += chunk
+                self.assertEqual(process.returncode, 7, display)
+                self.assertEqual(stdout, b"raw\x00\xffdiagnostic\n")
+                self.assertIn(b"| Fixture", display)
+                self.assertIn(b"/ Fixture", display)
+                self.assertIn(b"exit 7)", display)
+                self.assertNotIn(b"running (", display)
+            finally:
+                release.touch()
+                if process.poll() is None:
+                    process.terminate()
+                process.communicate(timeout=5)
+                os.close(master)
+
+    def test_plain_progress_updates_every_five_seconds(self):
+        stream = io.StringIO()
+        with contextlib.redirect_stderr(stream), \
+                patch("lint_common.time.monotonic", side_effect=(10, 15)):
+            progress = Progress("Fixture")
+            self.assertEqual(progress.interval, 5)
+            progress.update()
+        self.assertEqual(stream.getvalue(), "  Fixture: running (5s elapsed)\n")
+
+    def test_dumb_terminal_does_not_receive_escape_sequences(self):
+        stream = io.StringIO()
+        with patch.object(stream, "isatty", return_value=True), \
+                patch.dict(os.environ, TERM="dumb"), contextlib.redirect_stderr(stream):
+            progress = Progress("Fixture")
+            progress.update()
+            progress.finish(0)
+        self.assertNotIn("\r", stream.getvalue())
+        self.assertNotIn("\033", stream.getvalue())
+
+    def test_interrupted_spinner_clears_its_line_and_stops_child(self):
+        stream = io.StringIO()
+        with patch.object(stream, "isatty", return_value=True), \
+                patch.dict(os.environ, TERM="xterm"), contextlib.redirect_stderr(stream), \
+                patch("lint_common.start") as start, patch("lint_common.stop") as stop:
+            start.return_value.communicate.side_effect = KeyboardInterrupt
+            with self.assertRaises(KeyboardInterrupt):
+                run(["fixture"], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                    progress="Fixture")
+            stop.assert_called_once_with(start.return_value)
+        self.assertTrue(stream.getvalue().endswith("\r\033[2K"))
+
+    def test_inherited_pipe_jobserver_survives_python_driver(self):
+        reader, writer = os.pipe()
+        try:
+            os.set_inheritable(reader, True)
+            os.set_inheritable(writer, True)
+            env = dict(os.environ, MAKEFLAGS=f"-j --jobserver-auth={reader},{writer}")
+            script = ("import os,sys; r,w=map(int,sys.argv[1:]); "
+                      "os.write(w,b'token'); assert os.read(r,5)==b'token'")
+            result = run([sys.executable, "-c", script, str(reader), str(writer)],
+                         env=env, stderr=subprocess.PIPE)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            # FIFO jobservers must not forward a superseded pipe specification.
+            env["MAKEFLAGS"] += " --jobserver-auth=fifo:/tmp/make-test"
+            self.assertEqual(jobserver_fds(env), ())
+        finally:
+            os.close(reader)
+            os.close(writer)
+
+    def test_stale_jobserver_numbers_do_not_forward_reused_log_descriptors(self):
+        with tempfile.TemporaryFile() as first, tempfile.TemporaryFile() as second:
+            env = {"MAKEFLAGS": f"--jobserver-auth={first.fileno()},{second.fileno()}"}
+            self.assertEqual(jobserver_fds(env), ())
+
+    def test_child_signal_keeps_shell_exit_status_and_raw_log(self):
+        with tempfile.TemporaryFile() as output:
+            result = run([sys.executable, "-c",
+                          "import os,signal; os.write(1,b'out\\xff\\n'); "
+                          "os.write(2,b'err\\r\\n'); os.kill(os.getpid(),signal.SIGTERM)"],
+                         stdout=output, stderr=subprocess.STDOUT, progress="Fixture")
+            self.assertEqual(result.returncode, 143)
+            output.seek(0)
+            self.assertEqual(output.read(), b"out\xff\nerr\r\n")
+
+    def test_terminating_driver_also_stops_child_and_grandchild(self):
+        with tempfile.TemporaryDirectory(prefix="scst-lint-signal.") as directory:
+            root = Path(directory)
+            grandchild = root / "grandchild.py"
+            grandchild.write_text("""
+import signal, sys, time
+from pathlib import Path
+root = Path(sys.argv[1])
+def stop(signum, frame):
+    (root / 'grandchild-stopped').touch()
+    sys.exit(0)
+signal.signal(signal.SIGTERM, stop)
+(root / 'grandchild-ready').touch()
+while True:
+    time.sleep(1)
+""")
+            child = root / "child.py"
+            child.write_text("""
+import signal, subprocess, sys
+from pathlib import Path
+root = Path(sys.argv[1])
+process = subprocess.Popen([sys.executable, str(root / 'grandchild.py'), str(root)])
+stopping = False
+def stop(signum, frame):
+    global stopping
+    stopping = True
+signal.signal(signal.SIGTERM, stop)
+(root / 'child-ready').touch()
+process.wait()
+if stopping:
+    (root / 'child-stopped').touch()
+""")
+            script = ("import sys; sys.path.insert(0, sys.argv[1]); "
+                      "from lint_common import entrypoint, run; "
+                      "sys.exit(entrypoint(lambda: run([sys.executable,sys.argv[2],"
+                      "sys.argv[3]], progress='Fixture').returncode))")
+            process = subprocess.Popen([sys.executable, "-B", "-c", script,
+                                        str(SCRIPTS), str(child), str(root)],
+                                       stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            try:
+                deadline = time.monotonic() + 5
+                while (not all((root / name).exists() for name in
+                               ("child-ready", "grandchild-ready")) and process.poll() is None):
+                    if time.monotonic() > deadline:
+                        self.fail("grandchild did not start")
+                    time.sleep(0.01)
+                process.send_signal(signal.SIGTERM)
+                stdout, stderr = process.communicate(timeout=5)
+                self.assertEqual(process.returncode, 143, stdout + stderr)
+                self.assertTrue((root / "child-stopped").exists())
+                self.assertTrue((root / "grandchild-stopped").exists())
+            finally:
+                if process.poll() is None:
+                    process.send_signal(signal.SIGTERM)
+                    process.communicate(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()