diff --git a/docs/mcp.md b/docs/mcp.md
index 933b6ab..eed0e3e 100644
--- a/docs/mcp.md
+++ b/docs/mcp.md
@@ -17,8 +17,9 @@ MCP support is an optional extra:
 pip install kci-dev[mcp]
 ```
 
-Read-only dashboard query tools (trees, builds, boots, tests, hardware,
-known issues) are always available and need no configuration. Maestro
+Read-only dashboard query tools (trees, builds, boots, tests, logs,
+hardware, known issues) are always available and need no configuration.
+Maestro
 node lookup tools are enabled when the configured instance has an `api`
 URL, and job retry/checkout trigger tools when it also has a `pipeline`
 URL and a `token`. See the [config file](../config_file.md) documentation.
diff --git a/kcidev/api.py b/kcidev/api.py
index d7d6192..8a264f8 100644
--- a/kcidev/api.py
+++ b/kcidev/api.py
@@ -8,11 +8,18 @@
 without invoking Click commands or shelling out to ``kci-dev``.
 """
 
+import ipaddress
+import socket
+import zlib
 from datetime import datetime, timezone
+from time import monotonic
+from urllib.parse import urljoin, urlparse
 
 import click
+import requests
 from click.testing import CliRunner
 
+from kcidev.libs.common import kcidev_session
 from kcidev.libs.dashboard import (
     dashboard_api_url,
     dashboard_fetch_boot_issues,
@@ -102,6 +109,101 @@ def _as_library_error(action, func, *args, **kwargs):
         raise KciDevError(action) from exc
 
 
+MAX_LOG_BYTES = 1 << 20
+LOG_SCAN_LIMIT = 64 << 20
+LOG_DEADLINE_SECONDS = 60
+_LOG_CHUNK = 1 << 16
+_MAX_LOG_REDIRECTS = 5
+
+
+def _require_public_url(url):
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise KciDevError(f"Refusing to fetch non-http(s) log URL: {url}")
+    host = parsed.hostname
+    if not host:
+        raise KciDevError(f"Log URL has no host: {url}")
+    default_port = 443 if parsed.scheme == "https" else 80
+    try:
+        port = parsed.port or default_port
+    except ValueError as exc:
+        raise KciDevError(f"Invalid port in log URL {url}: {exc}") from exc
+    try:
+        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
+    except (OSError, UnicodeError) as exc:
+        raise KciDevError(f"Could not resolve log host {host}: {exc}") from exc
+    for info in infos:
+        ip = ipaddress.ip_address(info[4][0])
+        if not ip.is_global:
+            raise KciDevError(
+                f"Refusing to fetch log from non-public address {ip} ({host})"
+            )
+
+
+def _stream_public_get(url):
+    for _ in range(_MAX_LOG_REDIRECTS + 1):
+        _require_public_url(url)
+        response = kcidev_session.get(
+            url, stream=True, timeout=30, allow_redirects=False
+        )
+        if 300 <= response.status_code < 400:
+            location = response.headers.get("Location")
+            response.close()
+            if not location:
+                raise KciDevError(f"Redirect without Location header: {url}")
+            url = urljoin(url, location)
+            continue
+        return response
+    raise KciDevError("Too many redirects while fetching log")
+
+
+def _gunzip_iter(raw_iter):
+    decomp = zlib.decompressobj(zlib.MAX_WBITS | 16)
+    carry = b""
+    trailing_garbage = False
+    for chunk in raw_iter:
+        to_feed = carry + chunk
+        carry = b""
+        while to_feed:
+            if decomp.eof:
+                if to_feed[:2] == b"\x1f\x8b":
+                    decomp = zlib.decompressobj(zlib.MAX_WBITS | 16)
+                elif len(to_feed) < 2:
+                    carry = to_feed
+                    break
+                else:
+                    trailing_garbage = True
+                    break
+            yield decomp.decompress(to_feed, _LOG_CHUNK)
+            to_feed = decomp.unused_data if decomp.eof else decomp.unconsumed_tail
+        if trailing_garbage:
+            break
+    if trailing_garbage:
+        return
+    rest = decomp.flush()
+    if rest:
+        yield rest
+    if not decomp.eof:
+        raise KciDevError("Log gzip stream is incomplete or malformed")
+
+
+def _pick_log_url(test):
+    if not isinstance(test, dict):
+        return None
+    if test.get("log_url"):
+        return test["log_url"]
+    logs = [
+        f
+        for f in (test.get("output_files") or [])
+        if isinstance(f, dict)
+        and f.get("url")
+        and isinstance(f.get("name"), str)
+        and "log" in f["name"].lower()
+    ]
+    logs.sort(key=lambda f: ("stderr" in f["name"].lower(), f["name"].lower()))
+    return logs[0]["url"] if logs else None
+
+
 class KernelCIClient:
     """Client for interacting with KernelCI services from Python code.
 
@@ -367,6 +469,116 @@ def get_test(self, test_id):
             "Dashboard test request failed", dashboard_fetch_test, test_id, True
         )
 
+    def get_log(self, test_id, max_bytes=16384, tail=True):
+        """Fetch the raw log for a test, decompressing gzip, size-bounded.
+
+        Resolves the test's log URL (``log_url`` or, when that is empty, a
+        log entry from ``output_files``), downloads it with a bounded head or
+        tail buffer, decompressing gzip (including multi-member streams)
+        incrementally, and returns the text with the decompressed
+        ``total_bytes`` and a ``truncated`` flag. At most ``max_bytes`` bytes
+        are returned (capped at ``MAX_LOG_BYTES``), taken from the end when
+        ``tail`` is true (where failures usually are) or the start otherwise.
+        Reading stops once ``LOG_SCAN_LIMIT`` compressed or decompressed
+        bytes are seen, to bound memory and download against oversized or
+        malicious logs; ``scan_limited`` is then set and ``total_bytes`` is a
+        floor rather than the exact size. Reading also stops after
+        ``LOG_DEADLINE_SECONDS``, since the request timeout is per read
+        rather than total and a slow server would otherwise hold the
+        caller indefinitely; ``deadline_exceeded`` is then set and
+        ``total_bytes`` is likewise a floor. The log URL is validated (scheme
+        and resolved address) before fetching, though a DNS rebind between
+        that check and the request remains a residual gap.
+        """
+        if isinstance(max_bytes, bool) or not isinstance(max_bytes, int):
+            raise KciDevError("max_bytes must be a positive integer")
+        if max_bytes <= 0:
+            raise KciDevError("max_bytes must be a positive integer")
+        max_bytes = min(max_bytes, MAX_LOG_BYTES)
+
+        test = self.get_test(test_id)
+        log_url = _pick_log_url(test)
+        if not log_url:
+            raise KciDevError(f"No log available for test {test_id}")
+
+        buf = bytearray()
+        total = 0
+        raw_total = 0
+        scan_limited = False
+        deadline_exceeded = False
+        deadline = monotonic() + LOG_DEADLINE_SECONDS
+        response = None
+        try:
+            response = _stream_public_get(log_url)
+            response.raise_for_status()
+            chunks = response.iter_content(_LOG_CHUNK)
+
+            prefix = b""
+            for chunk in chunks:
+                if not chunk:
+                    continue
+                prefix += chunk
+                if len(prefix) >= 2:
+                    break
+
+            def raw_iter():
+                nonlocal raw_total, scan_limited, deadline_exceeded
+                if prefix:
+                    raw_total += len(prefix)
+                    yield prefix
+                for chunk in chunks:
+                    if not chunk:
+                        continue
+                    if monotonic() > deadline:
+                        deadline_exceeded = True
+                        return
+                    raw_total += len(chunk)
+                    if raw_total > LOG_SCAN_LIMIT:
+                        scan_limited = True
+                        return
+                    yield chunk
+
+            source = (
+                _gunzip_iter(raw_iter()) if prefix[:2] == b"\x1f\x8b" else raw_iter()
+            )
+            try:
+                for out in source:
+                    if not out:
+                        continue
+                    total += len(out)
+                    if tail:
+                        buf += out
+                        if len(buf) > max_bytes:
+                            del buf[:-max_bytes]
+                    elif len(buf) < max_bytes:
+                        buf += out[: max_bytes - len(buf)]
+                    if total >= LOG_SCAN_LIMIT:
+                        scan_limited = True
+                        break
+            except KciDevError:
+                if not (scan_limited or deadline_exceeded):
+                    raise
+        except KciDevError:
+            raise
+        except (requests.exceptions.RequestException, zlib.error, OSError) as exc:
+            raise KciDevError(f"Log download failed for test {test_id}: {exc}") from exc
+        finally:
+            if response is not None:
+                response.close()
+
+        returned = bytes(buf)
+        return {
+            "test_id": test_id,
+            "log_url": log_url,
+            "total_bytes": total,
+            "returned_bytes": len(returned),
+            "truncated": scan_limited or deadline_exceeded or total > len(returned),
+            "scan_limited": scan_limited,
+            "deadline_exceeded": deadline_exceeded,
+            "tail": tail,
+            "text": returned.decode("utf-8", errors="replace"),
+        }
+
     def get_tree_list(self, origin, days=7):
         return self._dashboard_request(
             "Dashboard tree list request failed",
@@ -420,21 +632,33 @@ def get_hardware_tests(self, name, origin):
             True,
         )
 
+    def _issues_or_empty(self, action, func, item_id, error_verbose):
+        """Fetch issues for one artifact, treating "none tracked" as empty.
+
+        The dashboard reports an artifact with no known issues as an
+        error rather than an empty list, which callers that ask "is this
+        failure already known" should read as a clean answer.
+        """
+        try:
+            return self._dashboard_request(action, func, item_id, True, error_verbose)
+        except KciDevError as exc:
+            if "No issues" in str(exc):
+                return []
+            raise
+
     def get_build_issues(self, build_id, error_verbose=True):
-        return self._dashboard_request(
+        return self._issues_or_empty(
             "Dashboard build issues request failed",
             dashboard_fetch_build_issues,
             build_id,
-            True,
             error_verbose,
         )
 
     def get_boot_issues(self, test_id, error_verbose=True):
-        return self._dashboard_request(
+        return self._issues_or_empty(
             "Dashboard boot issues request failed",
             dashboard_fetch_boot_issues,
             test_id,
-            True,
             error_verbose,
         )
 
diff --git a/kcidev/mcp/tools_dashboard.py b/kcidev/mcp/tools_dashboard.py
index 44fa05f..57a4696 100644
--- a/kcidev/mcp/tools_dashboard.py
+++ b/kcidev/mcp/tools_dashboard.py
@@ -236,6 +236,46 @@ def get_test(test_id: str):
     return _current_client().get_test(test_id)
 
 
+@tool_errors
+def get_log(test_id: str, max_bytes: int = 16384, tail: bool = True):
+    """Fetch the raw log for a test or job by dashboard test id.
+
+    Downloads and decompresses the log, resolving it from the test's
+    log_url or, when that is empty (common for failures), a log entry in
+    output_files. Returns it size-bounded: by default the last max_bytes,
+    where failures usually are (set tail=false for the start). The
+    response reports total_bytes and truncated so you can widen max_bytes
+    if needed, up to a 1 MiB ceiling: asking for more returns that
+    ceiling rather than the whole log, so compare returned_bytes with
+    total_bytes rather than retrying the same call. A download that runs
+    past about a minute stops early and sets deadline_exceeded. Use
+    get_test first for the shorter log_excerpt.
+    """
+    return _current_client().get_log(test_id, max_bytes=max_bytes, tail=tail)
+
+
+@tool_errors
+def get_test_issues(test_id: str):
+    """List known issues detected on a specific test or boot.
+
+    Use this to check a failing test against issues KernelCI already
+    tracks before treating the failure as new. Test ids look like
+    'maestro:<hex>'.
+    """
+    return _current_client().get_boot_issues(test_id)
+
+
+@tool_errors
+def get_build_issues(build_id: str):
+    """List known issues detected on a specific build.
+
+    Use this to check a failing build against issues KernelCI already
+    tracks before treating the failure as new. Build ids look like
+    'maestro:<hex>'.
+    """
+    return _current_client().get_build_issues(build_id)
+
+
 @tool_errors
 def list_hardware(origin: str = "maestro"):
     """List hardware platforms with results over the last 7 days.
@@ -321,6 +361,9 @@ def get_issue_tests(
     list_tests,
     get_build,
     get_test,
+    get_log,
+    get_test_issues,
+    get_build_issues,
     list_hardware,
     get_hardware_summary,
     list_issues,
diff --git a/tests/test_api.py b/tests/test_api.py
index 000451e..87940cb 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -4,7 +4,7 @@
 import pytest
 import requests
 
-from kcidev import KciDevError, KernelCIClient
+from kcidev import KciDevError, KernelCIClient, api
 from kcidev.libs import maestro_common
 
 CFG = {
@@ -301,3 +301,443 @@ def test_compare_results_only_fetches_issues_when_requested(monkeypatch):
 
     assert report["items"][0]["known_issues"] == ["issue-1"]
     get_issues.assert_called_once_with("head-test", error_verbose=False)
+
+
+class _FakeStream:
+    def __init__(self, chunks, status_code=200, headers=None, raise_after=None):
+        self._chunks = list(chunks)
+        self.status_code = status_code
+        self.headers = headers or {}
+        self.closed = False
+        self._raise_after = raise_after
+
+    def raise_for_status(self):
+        pass
+
+    def iter_content(self, size):
+        for i, chunk in enumerate(self._chunks):
+            if self._raise_after is not None and i == self._raise_after:
+                raise requests.exceptions.ChunkedEncodingError("conn reset")
+            yield chunk
+
+    def close(self):
+        self.closed = True
+
+
+def _log_test(url="https://files.kernelci.org/x.log", output_files=None):
+    return {"log_url": url, "output_files": output_files or []}
+
+
+def _mock_stream(monkeypatch, chunks):
+    monkeypatch.setattr(api, "_stream_public_get", lambda url: _FakeStream(chunks))
+
+
+def _addrinfo(ip, port=443):
+    return [(2, 1, 6, "", (ip, port))]
+
+
+def test_get_log_decompresses_gzip(monkeypatch):
+    import gzip
+
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/log.gz")
+    )
+    _mock_stream(monkeypatch, [gzip.compress(b"boot ok\nTEST FAIL: oops\n")])
+    out = _client().get_log("maestro:abc")
+    assert out["truncated"] is False
+    assert "TEST FAIL: oops" in out["text"]
+    assert out["total_bytes"] == len(b"boot ok\nTEST FAIL: oops\n")
+
+
+def test_get_log_tail_truncates(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"A" * 100 + b"TAILEND"])
+    out = _client().get_log("t", max_bytes=7, tail=True)
+    assert out["truncated"] is True
+    assert out["text"] == "TAILEND"
+    assert out["returned_bytes"] == 7
+
+
+def test_get_log_head_truncates(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"HEADSTART" + b"Z" * 100])
+    out = _client().get_log("t", max_bytes=9, tail=False)
+    assert out["text"] == "HEADSTART"
+    assert out["truncated"] is True
+
+
+def test_get_log_tail_spans_chunk_boundaries(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"1234567890", b"abcde", b"XYZ"])
+    out = _client().get_log("t", max_bytes=5, tail=True)
+    assert out["text"] == "deXYZ"
+    assert out["total_bytes"] == 18
+
+
+def test_get_log_no_url_raises(monkeypatch):
+    monkeypatch.setattr(
+        KernelCIClient,
+        "get_test",
+        lambda self, tid: {"log_url": None, "output_files": []},
+    )
+    with pytest.raises(KciDevError, match="No log available"):
+        _client().get_log("t")
+
+
+def test_get_log_falls_back_to_output_files(monkeypatch):
+    test = {
+        "log_url": None,
+        "output_files": [
+            {"name": "build_kselftest_stderr_log", "url": "https://f/stderr.log.gz"},
+            {"name": "test_log", "url": "https://f/test.log.gz"},
+            {"name": "job_definition", "url": "https://f/def.json"},
+        ],
+    }
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: test)
+    _mock_stream(monkeypatch, [b"from output_files"])
+    out = _client().get_log("t")
+    assert out["log_url"] == "https://f/test.log.gz"
+    assert out["text"] == "from output_files"
+
+
+@pytest.mark.parametrize("bad", [0, -1, -1000, 1.5, True, "100"])
+def test_get_log_rejects_bad_max_bytes(bad):
+    with pytest.raises(KciDevError, match="positive integer"):
+        _client().get_log("t", max_bytes=bad)
+
+
+def test_get_log_clamps_oversized_max_bytes(monkeypatch):
+    monkeypatch.setattr(api, "MAX_LOG_BYTES", 10)
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"B" * 50])
+    out = _client().get_log("t", max_bytes=10_000, tail=False)
+    assert out["returned_bytes"] == 10
+    assert out["truncated"] is True
+
+
+def test_get_log_scan_limit_bounds_download(monkeypatch):
+    monkeypatch.setattr(api, "LOG_SCAN_LIMIT", 20)
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"C" * 15, b"D" * 15, b"E" * 15])
+    out = _client().get_log("t", max_bytes=100, tail=False)
+    assert out["scan_limited"] is True
+    assert out["truncated"] is True
+    assert out["total_bytes"] == 15
+
+
+def test_get_log_truncated_gzip_raises(monkeypatch):
+    import gzip
+
+    good = gzip.compress(b"hello world" * 50)
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/x.gz")
+    )
+    _mock_stream(monkeypatch, [good[:20]])
+    with pytest.raises(KciDevError, match="incomplete or malformed"):
+        _client().get_log("t")
+
+
+def test_get_log_corrupt_gzip_raises(monkeypatch):
+    import gzip
+
+    good = gzip.compress(b"hello world" * 50)
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/x.gz")
+    )
+    _mock_stream(monkeypatch, [good[:10] + b"\x00" * 40])
+    with pytest.raises(KciDevError, match="Log download failed"):
+        _client().get_log("t")
+
+
+def test_get_log_download_failure_raises(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+
+    def boom(url):
+        raise requests.exceptions.ConnectionError("no route")
+
+    monkeypatch.setattr(api, "_stream_public_get", boom)
+    with pytest.raises(KciDevError, match="Log download failed"):
+        _client().get_log("t")
+
+
+def test_require_public_url_rejects_non_http():
+    with pytest.raises(KciDevError, match="non-http"):
+        api._require_public_url("ftp://files.kernelci.org/x")
+    with pytest.raises(KciDevError, match="non-http"):
+        api._require_public_url("file:///etc/passwd")
+
+
+@pytest.mark.parametrize(
+    "ip", ["127.0.0.1", "10.0.0.5", "192.168.1.1", "169.254.169.254", "::1"]
+)
+def test_require_public_url_rejects_private(monkeypatch, ip):
+    monkeypatch.setattr(api.socket, "getaddrinfo", lambda *a, **k: _addrinfo(ip))
+    with pytest.raises(KciDevError, match="non-public"):
+        api._require_public_url("https://evil.example/x")
+
+
+def test_require_public_url_allows_public(monkeypatch):
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("93.184.216.34")
+    )
+    api._require_public_url("https://files.kernelci.org/x")
+
+
+def test_require_public_url_unresolvable(monkeypatch):
+    def boom(*a, **k):
+        raise api.socket.gaierror("nope")
+
+    monkeypatch.setattr(api.socket, "getaddrinfo", boom)
+    with pytest.raises(KciDevError, match="resolve"):
+        api._require_public_url("https://nope.invalid/x")
+
+
+def test_stream_public_get_follows_validated_redirect(monkeypatch):
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("93.184.216.34")
+    )
+    r1 = Mock(status_code=302, headers={"Location": "https://cdn.example/final"})
+    r1.close = Mock()
+    r2 = _FakeStream([b"ok"])
+    calls = []
+
+    def fake_get(url, **k):
+        calls.append(url)
+        return r1 if len(calls) == 1 else r2
+
+    monkeypatch.setattr(api.kcidev_session, "get", fake_get)
+    assert api._stream_public_get("https://files.kernelci.org/x") is r2
+    assert calls == ["https://files.kernelci.org/x", "https://cdn.example/final"]
+
+
+def test_stream_public_get_rejects_redirect_to_private(monkeypatch):
+    def ai(host, *a, **k):
+        good = host == "files.kernelci.org"
+        return _addrinfo("93.184.216.34" if good else "169.254.169.254")
+
+    monkeypatch.setattr(api.socket, "getaddrinfo", ai)
+    r1 = Mock(
+        status_code=302, headers={"Location": "http://169.254.169.254/latest/meta"}
+    )
+    r1.close = Mock()
+    monkeypatch.setattr(api.kcidev_session, "get", Mock(return_value=r1))
+    with pytest.raises(KciDevError, match="non-public"):
+        api._stream_public_get("https://files.kernelci.org/x")
+
+
+def test_stream_public_get_too_many_redirects(monkeypatch):
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("93.184.216.34")
+    )
+    rr = Mock(status_code=302, headers={"Location": "https://a.example/loop"})
+    rr.close = Mock()
+    monkeypatch.setattr(api.kcidev_session, "get", Mock(return_value=rr))
+    with pytest.raises(KciDevError, match="Too many redirects"):
+        api._stream_public_get("https://a.example/loop")
+
+
+def test_gunzip_iter_roundtrip():
+    import gzip
+
+    assert b"".join(api._gunzip_iter([gzip.compress(b"hello")])) == b"hello"
+
+
+def test_gunzip_iter_truncated_raises():
+    import gzip
+
+    good = gzip.compress(b"data" * 100)
+    with pytest.raises(KciDevError, match="incomplete or malformed"):
+        list(api._gunzip_iter([good[:15]]))
+
+
+def test_gunzip_iter_multi_member():
+    import gzip
+
+    stream = gzip.compress(b"first\n") + gzip.compress(b"SECOND\n")
+    assert b"".join(api._gunzip_iter([stream])) == b"first\nSECOND\n"
+
+
+def test_gunzip_iter_multi_member_split_and_boundary():
+    import gzip
+
+    stream = gzip.compress(b"AAA") + gzip.compress(b"BBB")
+    split = [stream[:4], stream[4:]]
+    assert b"".join(api._gunzip_iter(split)) == b"AAABBB"
+    boundary = [gzip.compress(b"AAA"), gzip.compress(b"BBB")]
+    assert b"".join(api._gunzip_iter(boundary)) == b"AAABBB"
+
+
+def test_gunzip_iter_tolerates_trailing_garbage():
+    import gzip
+
+    assert b"".join(api._gunzip_iter([gzip.compress(b"log") + b"junk"])) == b"log"
+
+
+def test_get_log_reads_multi_member_gzip(monkeypatch):
+    import gzip
+
+    stream = gzip.compress(b"member one\n") + gzip.compress(b"MEMBER TWO FAIL\n")
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/x.gz")
+    )
+    _mock_stream(monkeypatch, [stream])
+    out = _client().get_log("t")
+    assert "MEMBER TWO FAIL" in out["text"]
+    assert out["truncated"] is False
+    assert out["total_bytes"] == len(b"member one\nMEMBER TWO FAIL\n")
+
+
+def test_get_log_closes_response_on_success(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    stream = _FakeStream([b"hello"])
+    monkeypatch.setattr(api, "_stream_public_get", lambda url: stream)
+    _client().get_log("t")
+    assert stream.closed is True
+
+
+def test_get_log_closes_response_on_gzip_error(monkeypatch):
+    import gzip
+
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/x.gz")
+    )
+    stream = _FakeStream([gzip.compress(b"data" * 50)[:20]])
+    monkeypatch.setattr(api, "_stream_public_get", lambda url: stream)
+    with pytest.raises(KciDevError):
+        _client().get_log("t")
+    assert stream.closed is True
+
+
+def test_get_log_closes_response_on_stream_error(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    stream = _FakeStream([b"a", b"b"], raise_after=1)
+    monkeypatch.setattr(api, "_stream_public_get", lambda url: stream)
+    with pytest.raises(KciDevError, match="Log download failed"):
+        _client().get_log("t")
+    assert stream.closed is True
+
+
+def test_get_log_raw_scan_limit_bounds_compressed(monkeypatch):
+    import gzip
+
+    monkeypatch.setattr(api, "LOG_SCAN_LIMIT", 15)
+    monkeypatch.setattr(
+        KernelCIClient, "get_test", lambda self, tid: _log_test("https://f/x.gz")
+    )
+    gz = gzip.compress(b"hi there friend")
+    _mock_stream(monkeypatch, [gz[:2], gz[2:10], gz[10:20], gz[20:]])
+    out = _client().get_log("t")
+    assert out["scan_limited"] is True
+    assert out["truncated"] is True
+
+
+def test_get_log_output_files_url_goes_through_guard(monkeypatch):
+    test = {
+        "log_url": None,
+        "output_files": [{"name": "test_log", "url": "https://internal.evil/x.log"}],
+    }
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: test)
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("169.254.169.254")
+    )
+    with pytest.raises(KciDevError, match="non-public"):
+        _client().get_log("t")
+
+
+def test_get_log_ignores_non_string_output_file_name(monkeypatch):
+    test = {
+        "log_url": None,
+        "output_files": [
+            {"name": 7, "url": "https://f/weird"},
+            {"name": "test_log", "url": "https://f/ok.log"},
+        ],
+    }
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: test)
+    _mock_stream(monkeypatch, [b"ok"])
+    out = _client().get_log("t")
+    assert out["log_url"] == "https://f/ok.log"
+
+
+def test_require_public_url_rejects_bad_port():
+    with pytest.raises(KciDevError, match="[Pp]ort"):
+        api._require_public_url("https://files.kernelci.org:99999/x")
+
+
+@pytest.mark.parametrize(
+    "ip", ["::ffff:127.0.0.1", "::ffff:169.254.169.254", "0.0.0.0"]
+)
+def test_require_public_url_rejects_mapped_and_unspecified(monkeypatch, ip):
+    monkeypatch.setattr(api.socket, "getaddrinfo", lambda *a, **k: _addrinfo(ip))
+    with pytest.raises(KciDevError, match="non-public"):
+        api._require_public_url("https://evil.example/x")
+
+
+def test_require_public_url_idna_error_is_clean(monkeypatch):
+    def boom(*a, **k):
+        raise UnicodeError("label too long")
+
+    monkeypatch.setattr(api.socket, "getaddrinfo", boom)
+    with pytest.raises(KciDevError, match="resolve"):
+        api._require_public_url("https://" + "a" * 70 + ".example/x")
+
+
+def test_stream_public_get_redirect_without_location(monkeypatch):
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("93.184.216.34")
+    )
+    r = Mock(status_code=302, headers={})
+    r.close = Mock()
+    monkeypatch.setattr(api.kcidev_session, "get", Mock(return_value=r))
+    with pytest.raises(KciDevError, match="without Location"):
+        api._stream_public_get("https://files.kernelci.org/x")
+
+
+def test_stream_public_get_rejects_redirect_to_file_scheme(monkeypatch):
+    monkeypatch.setattr(
+        api.socket, "getaddrinfo", lambda *a, **k: _addrinfo("93.184.216.34")
+    )
+    r = Mock(status_code=302, headers={"Location": "file:///etc/passwd"})
+    r.close = Mock()
+    monkeypatch.setattr(api.kcidev_session, "get", Mock(return_value=r))
+    with pytest.raises(KciDevError, match="non-http"):
+        api._stream_public_get("https://files.kernelci.org/x")
+
+
+def test_get_log_stops_at_the_total_deadline(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"X" * 1000 for _ in range(50)])
+    ticks = iter([0.0] + [api.LOG_DEADLINE_SECONDS + 1] * 200)
+    monkeypatch.setattr(api, "monotonic", lambda: next(ticks))
+
+    out = _client().get_log("t", max_bytes=100000)
+
+    assert out["deadline_exceeded"] is True
+    assert out["truncated"] is True
+    assert out["total_bytes"] < 50 * 1000
+
+
+def test_get_log_normal_download_is_not_deadline_limited(monkeypatch):
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, [b"Y" * 100])
+
+    out = _client().get_log("t", max_bytes=100000)
+
+    assert out["deadline_exceeded"] is False
+    assert out["truncated"] is False
+
+
+def test_get_log_deadline_returns_partial_gzip(monkeypatch):
+    import gzip
+
+    raw = bytes(range(256)) * 40
+    payload = gzip.compress(raw)
+    chunks = [payload[i : i + 64] for i in range(0, len(payload), 64)]
+    monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test())
+    _mock_stream(monkeypatch, chunks)
+    ticks = iter([0.0] + [api.LOG_DEADLINE_SECONDS + 1] * 500)
+    monkeypatch.setattr(api, "monotonic", lambda: next(ticks))
+
+    out = _client().get_log("t", max_bytes=100000)
+
+    assert out["deadline_exceeded"] is True
+    assert out["truncated"] is True
diff --git a/tests/test_mcp_tools_dashboard.py b/tests/test_mcp_tools_dashboard.py
index c24d189..70b3aba 100644
--- a/tests/test_mcp_tools_dashboard.py
+++ b/tests/test_mcp_tools_dashboard.py
@@ -214,3 +214,50 @@ def test_get_summary_detail_returns_full_payload(monkeypatch):
         detail=True,
     )
     assert result == SUMMARY_PAYLOAD
+
+
+def test_get_test_issues_fetches_dashboard(monkeypatch):
+    get = _mock_get(monkeypatch, [{"id": "issue1"}])
+    result = tools_dashboard.get_test_issues("maestro:t1")
+    assert result == [{"id": "issue1"}]
+    assert "test/maestro:t1/issues" in get.call_args[0][0]
+
+
+def test_get_build_issues_fetches_dashboard(monkeypatch):
+    get = _mock_get(monkeypatch, [{"id": "issue2"}])
+    result = tools_dashboard.get_build_issues("maestro:b1")
+    assert result == [{"id": "issue2"}]
+    assert "build/maestro:b1/issues" in get.call_args[0][0]
+
+
+def test_get_log_returns_client_payload(monkeypatch):
+    from kcidev.api import KernelCIClient
+
+    monkeypatch.setattr(
+        KernelCIClient,
+        "get_log",
+        lambda self, tid, max_bytes=16384, tail=True: {
+            "test_id": tid,
+            "truncated": False,
+            "text": "log body",
+        },
+    )
+    result = tools_dashboard.get_log("maestro:t1")
+    assert result["text"] == "log body"
+    assert result["test_id"] == "maestro:t1"
+
+
+def test_get_test_issues_returns_empty_when_none_are_tracked(monkeypatch):
+    _mock_get(monkeypatch, {"error": "No issues were found for this test"})
+    assert tools_dashboard.get_test_issues("maestro:t1") == []
+
+
+def test_get_build_issues_returns_empty_when_none_are_tracked(monkeypatch):
+    _mock_get(monkeypatch, {"error": "No issues found for this build"})
+    assert tools_dashboard.get_build_issues("maestro:b1") == []
+
+
+def test_get_test_issues_still_reports_other_errors(monkeypatch):
+    _mock_get(monkeypatch, {"error": "Test not found"})
+    with pytest.raises(ToolExecutionError):
+        tools_dashboard.get_test_issues("maestro:nope")