diff --git a/hermes_cli/logs.py b/hermes_cli/logs.py
index a214d52c8a5f..b4fa75a607b8 100644
--- a/hermes_cli/logs.py
+++ b/hermes_cli/logs.py
@@ -19,10 +19,15 @@
     hermes logs --since 30m -f     # follow, starting 30 min ago
 """
 
+import io
+import json
+import os
 import re
+import stat
 import sys
 import time
 from datetime import datetime, timedelta
+from collections import deque
 from pathlib import Path
 from typing import Optional, Sequence
 
@@ -35,8 +40,7 @@
     "gateway": "gateway.log",
     "gui": "gui.log",
     "desktop": "desktop.log",
-    # Every stdio MCP subprocess's stderr (tools/mcp_tool.py redirects it
-    # here, with per-server session markers) — the "MCP output channel".
+    # Legacy raw stderr and the index of profile-owned stdio attempt files.
     "mcp": "mcp-stderr.log",
 }
 
@@ -215,7 +219,8 @@ def tail_log(
 
     # Read and display the tail
     try:
-        lines = _read_tail(log_path, num_lines, has_filters=has_filters,
+        read_tail = _read_mcp_tail if log_name == "mcp" else _read_tail
+        lines = read_tail(log_path, num_lines, has_filters=has_filters,
                            min_level=min_level, session_filter=session,
                            since=since_dt, component_prefixes=component_prefixes)
     except PermissionError:
@@ -247,12 +252,257 @@ def tail_log(
 
     # Follow mode — poll for new content
     try:
-        _follow_log(log_path, min_level=min_level, session_filter=session,
+        follow_log = _follow_mcp_log if log_name == "mcp" else _follow_log
+        follow_log(log_path, min_level=min_level, session_filter=session,
                      since=since_dt, component_prefixes=component_prefixes)
     except KeyboardInterrupt:
         print("\n--- stopped ---")
 
 
+_MCP_MAX_ATTEMPTS = 128
+_MCP_INDEX_LINES = 2000
+_MCP_TAIL_BYTES = 1048576
+_MCP_MAX_PREVIEW_LINES = 10000
+_MCP_FOLLOW_BYTES = 65536
+_MCP_MAX_LINE_BYTES = 65536
+
+
+class _MCPPreviewNotice(str):
+    """Reader metadata, distinct from child content subject to log filters."""
+
+
+def _mcp_open_regular(path: Path):
+    return open(path, "rb", opener=lambda name, flags: os.open(
+        name, flags | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
+    ))
+
+
+def _mcp_oversized_line(index: bool) -> str:
+    source = "index" if index else "stderr"
+    return _MCPPreviewNotice(f"MCP {source} line exceeds MCP preview limit; full content remains in the log file.\n")
+
+
+def _mcp_tail_rows(path: Path, n: int, *, index: bool = False) -> list:
+    """A finite byte window; the generic log reader is unchanged."""
+    if n <= 0:
+        return []
+    rows = deque(maxlen=min(n, _MCP_MAX_PREVIEW_LINES))
+    try:
+        with _mcp_open_regular(path) as stream:
+            info = os.fstat(stream.fileno())
+            if not stat.S_ISREG(info.st_mode):
+                return []
+            start = max(0, info.st_size - _MCP_TAIL_BYTES)
+            stream.seek(start)
+            data = stream.read(_MCP_TAIL_BYTES)
+        if start:
+            fragment, separator, data = data.partition(b"\n")
+            if not separator or len(fragment) > _MCP_MAX_LINE_BYTES:
+                rows.append(_mcp_oversized_line(index))
+        for row in io.BytesIO(data):
+            payload = row.removesuffix(b"\n").removesuffix(b"\r")
+            if len(payload) > _MCP_MAX_LINE_BYTES:
+                rows.append(_mcp_oversized_line(index))
+            else:
+                decoded = row.decode("utf-8", errors="replace")
+                # An EOF preview is a display record, even before the child
+                # finishes its line. The stored bytes remain unchanged.
+                rows.append(decoded if decoded.endswith("\n") else decoded + "\n")
+    except OSError:
+        return []
+    return list(rows)
+
+
+def _mcp_control_record(line: str) -> Optional[dict]:
+    try:
+        record = json.loads(line)
+    except (ValueError, TypeError):
+        return None
+    return record if isinstance(record, dict) and record.get("kind") == "mcp.stdio.attempt" else None
+
+
+def _mcp_attempt(record: dict, index: Path) -> Optional[dict]:
+    """Resolve only this profile's UUID-named capture, never an arbitrary path."""
+    attempt = record.get("attempt_id")
+    if (not isinstance(attempt, str) or re.fullmatch(r"[0-9a-f]{32}", attempt) is None
+            or not isinstance(record.get("server"), str)
+            or type(record.get("parent_pid")) is not int or record["parent_pid"] <= 0
+            or record.get("destination") != "file"
+            or not isinstance(record.get("config_home"), str)
+            or not isinstance(record.get("stderr_path"), str)):
+        return None
+    try:
+        home = index.parent.parent.resolve()
+        expected = home / "logs" / "mcp-stderr" / f"{attempt}.log"
+        if (Path(record["config_home"]).resolve() != home
+                or Path(record["stderr_path"]).resolve() != expected
+                or not expected.is_file()):
+            return None
+    except (OSError, ValueError, RuntimeError):
+        return None
+    return {"attempt_id": attempt, "server": record["server"], "parent_pid": record["parent_pid"],
+            "config_home": str(home), "stderr_path": str(expected), "destination": "file"}
+
+
+def _mcp_child_line(line: str, record: dict, filters: dict) -> Optional[str]:
+    control = _mcp_control_record(line)
+    if control is not None and control.get("attempt_id") == record["attempt_id"]:
+        return None
+    owner = " ".join(f"{key}={json.dumps(value)}" for key, value in (
+        ("profile", record["config_home"]), ("server", record["server"]),
+        ("attempt", record["attempt_id"]),
+    ))
+    rendered = f"[mcp {owner}] {line}"
+    session = filters.get("session_filter")
+    if session is not None and session not in rendered:
+        return None
+    if isinstance(line, _MCPPreviewNotice):
+        return rendered
+    if not _matches_filters(line, **{**filters, "session_filter": None}):
+        return None
+    return rendered
+
+
+def _read_mcp_tail(path: Path, num_lines: int, *, has_filters: bool = False, **filters) -> list:
+    """Read actual stderr in attempt order; legacy rows keep their own format."""
+    if num_lines <= 0:
+        return []
+    requested = num_lines
+    num_lines = min(num_lines, _MCP_MAX_PREVIEW_LINES)
+    result = []
+    remaining = _MCP_TAIL_BYTES
+    exhausted = False
+    seen = set()
+
+    def append(line):
+        nonlocal remaining, exhausted
+        if len(line) > remaining:
+            exhausted = True
+            return
+        result.append(line)
+        remaining -= len(line)
+
+    for line in reversed(_mcp_tail_rows(path, max(num_lines * 20, _MCP_INDEX_LINES), index=True)):
+        control = _mcp_control_record(line)
+        if control is None:
+            if isinstance(line, _MCPPreviewNotice) or _matches_filters(line, **filters):
+                append(line)
+        else:
+            record = _mcp_attempt(control, path)
+            if record is None or record["attempt_id"] in seen or len(seen) >= _MCP_MAX_ATTEMPTS:
+                continue
+            seen.add(record["attempt_id"])
+            rows = _mcp_tail_rows(Path(record["stderr_path"]),
+                                  max(num_lines * 20, _MCP_INDEX_LINES) if has_filters else num_lines + 4)
+            for row in reversed(rows):
+                rendered = _mcp_child_line(row, record, filters)
+                if rendered is not None:
+                    append(rendered)
+                if exhausted or len(result) >= num_lines:
+                    break
+        if exhausted or len(result) >= num_lines:
+            break
+    lines = list(reversed(result))
+    if exhausted or requested > _MCP_MAX_PREVIEW_LINES:
+        lines.insert(0, "MCP preview output limit reached; full content remains in the log files.\n")
+    return lines
+
+
+def _mcp_cursor(path: Path):
+    try:
+        with _mcp_open_regular(path) as stream:
+            info = os.fstat(stream.fileno())
+            if not stat.S_ISREG(info.st_mode):
+                return None
+            # Seed only the bounded final fragment. If it later completes,
+            # display its complete new record; never replay complete history.
+            start = max(0, info.st_size - _MCP_MAX_LINE_BYTES - 1)
+            stream.seek(start)
+            recent = stream.read(min(info.st_size, _MCP_MAX_LINE_BYTES + 1))
+            pending = recent.rsplit(b"\n", 1)[-1]
+            dropping = len(pending) > _MCP_MAX_LINE_BYTES
+            return ((info.st_dev, info.st_ino), info.st_size, b"" if dropping else pending, dropping)
+    except OSError:
+        return None
+
+
+def _mcp_read_chunk(path: Path, cursor, *, index: bool = False):
+    """Frame bounded binary chunks; retain fragments, never file descriptors."""
+    try:
+        with _mcp_open_regular(path) as stream:
+            info = os.fstat(stream.fileno())
+            if not stat.S_ISREG(info.st_mode):
+                return [], cursor
+            identity = (info.st_dev, info.st_ino)
+            retained = cursor is not None and cursor[0] == identity and cursor[1] <= info.st_size
+            offset, pending, dropping = cursor[1:] if retained else (0, b"", False)
+            stream.seek(offset)
+            chunk = stream.read(_MCP_FOLLOW_BYTES)
+            offset = stream.tell()
+        data = pending + chunk
+        lines = []
+        if dropping:
+            _fragment, separator, data = data.partition(b"\n")
+            if not separator:
+                return [], (identity, offset, b"", True)
+        parts = data.split(b"\n")
+        for row in parts[:-1]:
+            lines.append(_mcp_oversized_line(index) if len(row) > _MCP_MAX_LINE_BYTES
+                         else row.decode("utf-8", errors="replace") + "\n")
+        pending = parts[-1]
+        dropping = len(pending) > _MCP_MAX_LINE_BYTES
+        if dropping:
+            lines.append(_mcp_oversized_line(index))
+            pending = b""
+        return lines, (identity, offset, pending, dropping)
+    except OSError:
+        return [], cursor
+
+
+def _follow_mcp_log(path: Path, **filters) -> None:
+    records = {}
+
+    def remember(line):
+        control = _mcp_control_record(line)
+        record = _mcp_attempt(control, path) if control is not None else None
+        if record is not None:
+            records[record["attempt_id"]] = record
+            if len(records) > _MCP_MAX_ATTEMPTS:
+                records.pop(next(iter(records)))
+                print("MCP follow preview retains the latest 128 attempts; older captures remain in the log files.")
+
+    for line in _mcp_tail_rows(path, _MCP_INDEX_LINES, index=True):
+        remember(line)
+    positions = {key: _mcp_cursor(Path(record["stderr_path"])) for key, record in records.items()}
+    for key, cursor in positions.items():
+        if cursor is not None and cursor[3]:
+            rendered = _mcp_child_line(_mcp_oversized_line(False), records[key], filters)
+            if rendered is not None:
+                print(rendered, end="")
+    index_cursor = _mcp_cursor(path)
+    if index_cursor is not None and index_cursor[3]:
+        print(_mcp_oversized_line(True), end="")
+    while True:
+        lines, index_cursor = _mcp_read_chunk(path, index_cursor, index=True)
+        for line in lines:
+            if _mcp_control_record(line) is not None:
+                remember(line)
+            elif isinstance(line, _MCPPreviewNotice) or _matches_filters(line, **filters):
+                print(line, end="")
+                sys.stdout.flush()
+        for key, record in records.items():
+            if _mcp_attempt(record, path) is None:
+                continue
+            lines, positions[key] = _mcp_read_chunk(Path(record["stderr_path"]), positions.get(key))
+            for line in lines:
+                rendered = _mcp_child_line(line, record, filters)
+                if rendered is not None:
+                    print(rendered, end="")
+                    sys.stdout.flush()
+        positions = {key: positions[key] for key in records if key in positions}
+        time.sleep(0.3)
+
+
 def _read_tail(
     path: Path,
     num_lines: int,
diff --git a/tests/hermes_cli/test_mcp_logs_readers.py b/tests/hermes_cli/test_mcp_logs_readers.py
new file mode 100644
index 000000000000..80d046870b4c
--- /dev/null
+++ b/tests/hermes_cli/test_mcp_logs_readers.py
@@ -0,0 +1,431 @@
+"""Actual MCP log-reader integration with temporary stderr files only."""
+
+import json
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def runtime(tmp_path, monkeypatch):
+    import hermes_cli.logs as logs
+    import tools.mcp_tool as mcp_tool
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    home = tmp_path / "profile"
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setattr(mcp_tool, "_mcp_stderr_log_files", {})
+    token = set_hermes_home_override(str(home))
+    try:
+        yield logs, mcp_tool, home
+    finally:
+        reset_hermes_home_override(token)
+        for fh in mcp_tool._mcp_stderr_log_files.values():
+            fh.close()
+
+
+def capture(runtime, text, server="cea_graph"):
+    _logs, mcp_tool, _home = runtime
+    record = mcp_tool._begin_stdio_diagnostic(server)
+    record["stream"].write(text)
+    record["stream"].flush()
+    record["status"] = "closed"
+    mcp_tool._finish_stdio_diagnostic(record)
+    return record
+
+
+def manual_record(home, attempt="a" * 32, server="cea_graph", text="child payload\n"):
+    path = home / "logs" / "mcp-stderr" / f"{attempt}.log"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text)
+    record = {
+        "kind": "mcp.stdio.attempt", "attempt_id": attempt, "server": server,
+        "config_home": str(home), "parent_pid": 123, "stderr_path": str(path),
+        "destination": "file", "phase": "transport", "status": "starting",
+    }
+    return record
+
+
+def write_index(home, records, legacy=""):
+    path = home / "logs" / "mcp-stderr.log"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(legacy + "".join(json.dumps(record) + "\n" for record in records))
+    return path
+
+
+def test_actual_writer_and_reader_show_stderr_for_only_the_selected_profile(
+    runtime, tmp_path, capsys
+):
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    logs, _mcp_tool, home = runtime
+    first = capture(runtime, "first alpha stderr\n")
+    token = set_hermes_home_override(str(tmp_path / "beta"))
+    try:
+        capture(runtime, "beta stderr must stay separate\n")
+    finally:
+        reset_hermes_home_override(token)
+    second = capture(runtime, "second alpha stderr\n", server="pilot_bridge")
+    logs.tail_log("mcp", num_lines=50)
+    output = capsys.readouterr().out
+    assert "first alpha stderr" in output
+    assert "second alpha stderr" in output
+    assert "beta stderr must stay separate" not in output
+    assert first["attempt_id"] in output and second["attempt_id"] in output
+    assert "cea_graph" in output and "pilot_bridge" in output
+    assert str(home) in output
+    # The reader retrieves actual content, without duplicating it in the index.
+    assert "first alpha stderr" not in (home / "logs" / "mcp-stderr.log").read_text()
+    assert '"kind": "mcp.stdio.attempt"' not in output
+
+
+def test_tail_limit_applies_to_child_content_and_preserves_raw_json(runtime, capsys):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home, text='older\n{"kind":"child.event","message":"newest"}\n')
+    write_index(home, [record])
+    logs.tail_log("mcp", num_lines=1)
+    output = capsys.readouterr().out
+    assert '"message":"newest"' in output
+    assert "older" not in output
+    assert record["attempt_id"] in output
+
+
+def test_filters_use_child_timestamp_level_component_and_attempt(runtime, capsys):
+    logs, _mcp_tool, home = runtime
+    now = datetime.now()
+    old = (now - timedelta(hours=3)).strftime("%Y-%m-%d %H:%M:%S")
+    recent = now.strftime("%Y-%m-%d %H:%M:%S")
+    record = manual_record(home, text=(
+        f"{old} ERROR tools.reader: stale payload\n"
+        f"{recent} INFO tools.reader: lower level\n"
+        f"{recent} ERROR gateway.reader: wrong component\n"
+        f"{recent} ERROR tools.reader: matching child payload\n"
+    ))
+    write_index(home, [record])
+    logs.tail_log("mcp", level="WARNING", since="1h", component="tools", session=record["attempt_id"])
+    output = capsys.readouterr().out
+    assert "matching child payload" in output
+    assert "stale payload" not in output
+    assert "lower level" not in output
+    assert "wrong component" not in output
+
+
+def test_legacy_stderr_and_new_attempts_are_both_readable(runtime, capsys):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home, text="attributed child payload\n")
+    write_index(home, [record], legacy='legacy raw stderr\n{"kind":"old.server","message":"legacy JSON"}\n')
+    logs.tail_log("mcp", num_lines=10)
+    output = capsys.readouterr().out
+    assert "legacy raw stderr" in output
+    assert '"message":"legacy JSON"' in output
+    assert "attributed child payload" in output
+
+
+@pytest.mark.parametrize("payload", ["last unterminated child payload", "progress\r"])
+def test_final_unterminated_child_line_stays_separate_from_metadata(runtime, capsys, payload):
+    logs, _mcp_tool, _home = runtime
+    capture(runtime, payload)
+    logs.tail_log("mcp", num_lines=1)
+    output = capsys.readouterr().out
+    assert payload in output
+    assert '"kind": "mcp.stdio.attempt"' not in output
+
+
+def test_tail_separates_partial_rows_from_two_active_attempts(runtime, capsys):
+    logs, mcp_tool, _home = runtime
+    first = mcp_tool._begin_stdio_diagnostic("cea_graph")
+    first["stream"].write("alpha partial")
+    first["stream"].flush()
+    second = mcp_tool._begin_stdio_diagnostic("pilot_bridge")
+    second["stream"].write("beta complete\n")
+    second["stream"].flush()
+    try:
+        logs.tail_log("mcp", num_lines=20)
+        assert Path(first["stderr_path"]).read_bytes().endswith(b"alpha partial")
+    finally:
+        for record in (second, first):
+            record["status"] = "closed"
+            mcp_tool._finish_stdio_diagnostic(record)
+    output = capsys.readouterr().out
+    assert "alpha partial\n" in output
+    assert "beta complete\n" in output
+
+
+def test_tail_to_follow_separates_the_preview_and_completes_its_partial_line(runtime, monkeypatch, capsys):
+    logs, mcp_tool, _home = runtime
+    record = mcp_tool._begin_stdio_diagnostic("cea_graph")
+    record["stream"].write("alpha partial")
+    record["stream"].flush()
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            record["stream"].write(" remainder\nfresh complete child\n")
+            record["stream"].flush()
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    try:
+        logs.tail_log("mcp", num_lines=20, follow=True)
+    finally:
+        record["status"] = "closed"
+        mcp_tool._finish_stdio_diagnostic(record)
+    output = capsys.readouterr().out
+    assert "alpha partial\n" in output
+    assert output.count("alpha partial remainder\n") == 1
+    assert output.count("fresh complete child\n") == 1
+
+
+@pytest.mark.parametrize("invalid", ["wrong_profile", "outside_path", "symlink", "bad_id", "discarded", "missing", "invalid_pid"])
+def test_index_cannot_read_an_unowned_or_unavailable_file(runtime, tmp_path, capsys, invalid):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home)
+    foreign = tmp_path / "other-profile" / "foreign.log"
+    foreign.parent.mkdir()
+    foreign.write_text("FOREIGN_CONTENT_MUST_NOT_BE_READ\n")
+    if invalid == "wrong_profile":
+        record["config_home"] = str(foreign.parent)
+    elif invalid == "outside_path":
+        record["stderr_path"] = str(foreign)
+    elif invalid == "symlink":
+        path = Path(record["stderr_path"])
+        path.unlink()
+        path.symlink_to(foreign)
+    elif invalid == "bad_id":
+        record["attempt_id"] = "../foreign"
+    elif invalid == "discarded":
+        record["destination"] = "discarded"
+    elif invalid == "missing":
+        Path(record["stderr_path"]).unlink()
+    elif invalid == "invalid_pid":
+        record["parent_pid"] = True
+    write_index(home, [record])
+    logs.tail_log("mcp", num_lines=10)
+    output = capsys.readouterr().out
+    assert "FOREIGN_CONTENT_MUST_NOT_BE_READ" not in output
+    assert "child payload" not in output
+
+
+def test_follow_reads_existing_and_new_attempts_without_replaying_history(
+    runtime, monkeypatch, capsys
+):
+    logs, mcp_tool, home = runtime
+    active = mcp_tool._begin_stdio_diagnostic("cea_graph")
+    active["stream"].write("historical child payload\n")
+    active["stream"].flush()
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            active["stream"].write("fresh active stderr\n")
+            active["stream"].flush()
+            capture(runtime, "fresh next-attempt stderr\n", server="pilot_bridge")
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    try:
+        logs.tail_log("mcp", num_lines=0, follow=True)
+    finally:
+        active["status"] = "closed"
+        mcp_tool._finish_stdio_diagnostic(active)
+    output = capsys.readouterr().out
+    assert "historical child payload" not in output
+    assert output.count("fresh active stderr") == 1
+    assert output.count("fresh next-attempt stderr") == 1
+    assert active["attempt_id"] in output
+    assert str(home) in output
+
+
+def test_follow_preserves_legacy_appends_and_handles_attempt_replacement(
+    runtime, monkeypatch, capsys
+):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home, text="existing child payload\n")
+    index = write_index(home, [record], legacy="legacy history\n")
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            with index.open("a") as stream:
+                stream.write("legacy fresh stderr\n")
+            path = Path(record["stderr_path"])
+            replacement = path.with_suffix(".new")
+            replacement.write_text("replacement child stderr\n")
+            replacement.replace(path)
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    logs.tail_log("mcp", num_lines=0, follow=True)
+    output = capsys.readouterr().out
+    assert output.count("legacy fresh stderr") == 1
+    assert output.count("replacement child stderr") == 1
+    assert "existing child payload" not in output
+
+
+def test_follow_preserves_an_index_record_across_the_read_budget(runtime, monkeypatch, capsys):
+    logs, _mcp_tool, home = runtime
+    index = write_index(home, [])
+    record = manual_record(home, text="boundary child stderr\n")
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            with index.open("a") as stream:
+                stream.write("x" * (65536 - 32) + "\n" + json.dumps(record) + "\n")
+        elif iterations > 2:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    logs.tail_log("mcp", num_lines=0, follow=True)
+    output = capsys.readouterr().out
+    assert output.count("boundary child stderr") == 1
+    assert '"kind": "mcp.stdio.attempt"' not in output
+
+
+@pytest.mark.parametrize("filtered", [False, True])
+def test_follow_frames_a_child_line_written_across_polls(runtime, monkeypatch, capsys, filtered):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home, text="")
+    write_index(home, [record])
+    path = Path(record["stderr_path"])
+    stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    first = f"{stamp} ERROR tools.reader: split child "
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            with path.open("a") as stream:
+                stream.write(first)
+        elif iterations == 2:
+            assert "split child" not in capsys.readouterr().out
+            with path.open("a") as stream:
+                stream.write("payload\n")
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    options = {"level": "WARNING", "component": "tools", "since": "1h"} if filtered else {}
+    logs.tail_log("mcp", num_lines=0, follow=True, **options)
+    output = capsys.readouterr().out
+    assert output.count(first + "payload\n") == 1
+    assert output.count("[mcp ") == 1
+
+
+def test_follow_frames_a_control_record_written_across_polls(runtime, monkeypatch, capsys):
+    logs, _mcp_tool, home = runtime
+    index = write_index(home, [])
+    record = manual_record(home, text="split-index child payload\n")
+    encoded = json.dumps(record)
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            with index.open("a") as stream:
+                stream.write(encoded[:40])
+        elif iterations == 2:
+            assert '"kind"' not in capsys.readouterr().out
+            with index.open("a") as stream:
+                stream.write(encoded[40:] + "\n")
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    logs.tail_log("mcp", num_lines=0, follow=True)
+    output = capsys.readouterr().out
+    assert output.count("split-index child payload") == 1
+    assert '"kind": "mcp.stdio.attempt"' not in output
+
+
+@pytest.mark.parametrize("filtered", [False, True])
+def test_follow_advances_past_an_oversized_line_with_a_visible_notice(runtime, monkeypatch, capsys, filtered):
+    logs, _mcp_tool, home = runtime
+    monkeypatch.setattr(logs, "_MCP_FOLLOW_BYTES", 512, raising=False)
+    monkeypatch.setattr(logs, "_MCP_MAX_LINE_BYTES", 2048, raising=False)
+    record = manual_record(home, text="")
+    write_index(home, [record])
+    path = Path(record["stderr_path"])
+    stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            with path.open("a") as stream:
+                stream.write("x" * 4097 + f"\n{stamp} ERROR tools.reader: after oversized child line\n")
+        elif iterations >= 12:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    options = {"level": "WARNING", "component": "tools", "since": "1h"} if filtered else {}
+    logs.tail_log("mcp", num_lines=0, follow=True, **options)
+    output = capsys.readouterr().out
+    assert output.count("exceeds MCP preview limit") == 1
+    assert output.count("after oversized child line") == 1
+
+
+@pytest.mark.parametrize("filtered", [False, True])
+def test_tail_has_a_finite_window_and_keeps_recent_content(runtime, monkeypatch, capsys, filtered):
+    logs, _mcp_tool, home = runtime
+    monkeypatch.setattr(logs, "_MCP_TAIL_BYTES", 2048, raising=False)
+    monkeypatch.setattr(logs, "_MCP_MAX_LINE_BYTES", 1024, raising=False)
+    stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    record = manual_record(home, text="x" * 4096 + f"\n{stamp} ERROR tools.reader: recent child payload\n")
+    write_index(home, [record])
+    options = {"level": "WARNING", "component": "tools", "since": "1h"} if filtered else {}
+    logs.tail_log("mcp", num_lines=10, **options)
+    output = capsys.readouterr().out
+    assert "recent child payload" in output
+    assert "exceeds MCP preview limit" in output
+    assert "x" * 2048 not in output
+
+
+def test_follow_discards_a_pending_fragment_on_observed_truncation(runtime, monkeypatch, capsys):
+    logs, _mcp_tool, home = runtime
+    record = manual_record(home, text="")
+    write_index(home, [record])
+    path = Path(record["stderr_path"])
+    iterations = 0
+
+    def advance(_seconds):
+        nonlocal iterations
+        iterations += 1
+        if iterations == 1:
+            path.write_text("stale fragment" * 100)
+        elif iterations == 2:
+            path.write_text("replacement complete\n")
+        else:
+            raise KeyboardInterrupt
+
+    monkeypatch.setattr(logs.time, "sleep", advance)
+    logs.tail_log("mcp", num_lines=0, follow=True)
+    output = capsys.readouterr().out
+    assert "stale fragment" not in output
+    assert output.count("replacement complete\n") == 1
+
+
+def test_non_mcp_reader_keeps_its_existing_behavior(runtime, capsys):
+    logs, _mcp_tool, home = runtime
+    path = home / "logs" / "agent.log"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("older agent row\nnewest agent row\n")
+    logs.tail_log("agent", num_lines=1)
+    output = capsys.readouterr().out
+    assert "newest agent row" in output and "older agent row" not in output
+    assert "[mcp " not in output
diff --git a/tests/tools/test_mcp_parked_self_probe.py b/tests/tools/test_mcp_parked_self_probe.py
index 61a1a6578e34..9883921efbca 100644
--- a/tests/tools/test_mcp_parked_self_probe.py
+++ b/tests/tools/test_mcp_parked_self_probe.py
@@ -68,6 +68,16 @@ async def _fast_sleep(_delay, *a, **kw):
         "revived_registration": 0,
     }
 
+    def _register(name, server, config):
+        # Production discovery registers an owned revival before readiness.
+        assert name == "srv"
+        assert mcp_tool._servers.get(name) is server
+        assert not server._ready.is_set()
+        state["revived_registration"] += 1
+        return ["srv__tool"]
+
+    monkeypatch.setattr(mcp_tool, "_register_server_tools", _register)
+
     async def _scenario():
         class _Task(MCPServerTask):
             def _is_http(self):
@@ -77,11 +87,6 @@ def _deregister_tools(self):
                 state["deregistered"] += 1
                 self._registered_tool_names = []
 
-            def _register_discovered_tools_if_needed(self):
-                if self._ready.is_set() and not self._registered_tool_names:
-                    state["revived_registration"] += 1
-                    self._registered_tool_names = ["srv__tool"]
-
             async def _run_stdio(self, config):
                 state["transport_calls"] += 1
                 if state["transport_calls"] == 1:
@@ -95,12 +100,21 @@ async def _run_stdio(self, config):
                     raise RuntimeError("backend still down")
                 # Backend recovered: establish a session and park in the
                 # lifecycle wait like the real transport does.
-                self.session = object()
-                self._register_discovered_tools_if_needed()
+                assert not self._ready.is_set()
+                self.session = SimpleNamespace(
+                    list_tools=AsyncMock(
+                        return_value=SimpleNamespace(tools=[SimpleNamespace(name="tool")]),
+                    )
+                )
+                # Match _run_stdio: discover/publish first, signal ready after.
+                await self._discover_tools()
+                assert not self._ready.is_set()
+                self._ready.set()
                 await self._wait_for_lifecycle_event()
 
         task = _Task("srv")
         task._registered_tool_names = ["srv__tool"]
+        monkeypatch.setitem(mcp_tool._servers, task.name, task)
 
         run_task = asyncio.ensure_future(task.run({"command": "x"}))
 
@@ -117,7 +131,7 @@ async def _run_stdio(self, config):
         state["backend_up"] = True
         for _ in range(200):
             await _real_sleep(0.01)
-            if task.session is not None:
+            if task.session is not None and task._ready.is_set():
                 break
 
         assert task.session is not None, (
@@ -127,6 +141,9 @@ async def _run_stdio(self, config):
         assert state["revived_registration"] >= 1, (
             "revived server did not re-register its tools"
         )
+        assert task._ready.is_set(), "revived server never completed discovery"
+        assert task._registered_tool_names == ["srv__tool"]
+        assert [tool.name for tool in task._tools] == ["tool"]
 
         task._shutdown_event.set()
         task._reconnect_event.set()
diff --git a/tests/tools/test_mcp_readiness_status.py b/tests/tools/test_mcp_readiness_status.py
new file mode 100644
index 000000000000..a2b478ecc264
--- /dev/null
+++ b/tests/tools/test_mcp_readiness_status.py
@@ -0,0 +1,83 @@
+"""Status boundaries only; no MCP processes, RPCs or provider calls."""
+
+import pytest
+
+
+@pytest.fixture
+def status_runtime(monkeypatch):
+    import tools.mcp_tool as mcp_tool
+
+    monkeypatch.setattr(mcp_tool, "_servers", {})
+    monkeypatch.setattr(mcp_tool, "_server_connecting", set())
+    monkeypatch.setattr(mcp_tool, "_server_connect_errors", {})
+    monkeypatch.setattr(
+        mcp_tool, "_load_mcp_config", lambda: {"cea_graph": {"command": "inert"}}
+    )
+    server = mcp_tool.MCPServerTask("cea_graph")
+    mcp_tool._servers["cea_graph"] = server
+    return mcp_tool, server
+
+
+def test_session_before_discovery_is_connecting(status_runtime):
+    mcp_tool, server = status_runtime
+    # The actual stdio path assigns this before awaiting discovery. A retained
+    # task can be reconnecting without appearing in _server_connecting.
+    server.session = object()
+    status = mcp_tool.get_mcp_status()[0]
+    assert status["status"] == "connecting"
+    assert status["connected"] is False
+    assert status["tools"] == 0
+
+
+def test_ready_session_with_no_tools_is_still_connected(status_runtime):
+    mcp_tool, server = status_runtime
+    server.session = object()
+    server._ready.set()
+    # Resource/prompt-only servers legitimately have zero tool definitions.
+    status = mcp_tool.get_mcp_status()[0]
+    assert status["status"] == "connected"
+    assert status["connected"] is True
+    assert status["tools"] == 0
+
+
+def test_ready_failure_is_not_connection_success(status_runtime):
+    mcp_tool, server = status_runtime
+    server.session = object()
+    server._ready.set()  # start() also wakes its waiter on failure.
+    server._error = RuntimeError("Connection closed")
+    status = mcp_tool.get_mcp_status()[0]
+    assert status["status"] == "failed"
+    assert status["connected"] is False
+    assert "Connection closed" in status["error"]
+
+
+def test_retained_failure_reports_task_error_without_discovery_map(status_runtime):
+    mcp_tool, server = status_runtime
+    server._error = RuntimeError("Connection closed")
+    status = mcp_tool.get_mcp_status()[0]
+    assert status["status"] == "failed"
+    assert status["connected"] is False
+
+
+def test_recovered_ready_session_overrides_old_discovery_error(status_runtime):
+    mcp_tool, server = status_runtime
+    server.session = object()
+    server._ready.set()
+    mcp_tool._server_connect_errors["cea_graph"] = "older attempt failed"
+    assert mcp_tool.get_mcp_status()[0]["status"] == "connected"
+
+
+def test_ready_event_without_session_is_not_connected(status_runtime):
+    mcp_tool, server = status_runtime
+    server._ready.set()
+    assert mcp_tool.get_mcp_status()[0]["connected"] is False
+
+
+def test_disabled_unstarted_server_keeps_disabled_status(status_runtime, monkeypatch):
+    mcp_tool, _server = status_runtime
+    mcp_tool._servers.clear()
+    monkeypatch.setattr(
+        mcp_tool, "_load_mcp_config",
+        lambda: {"cea_graph": {"command": "inert", "enabled": False}},
+    )
+    assert mcp_tool.get_mcp_status()[0]["status"] == "disabled"
diff --git a/tests/tools/test_mcp_stdio_diagnostics.py b/tests/tools/test_mcp_stdio_diagnostics.py
new file mode 100644
index 000000000000..1243d7077068
--- /dev/null
+++ b/tests/tools/test_mcp_stdio_diagnostics.py
@@ -0,0 +1,239 @@
+"""Actual diagnostic/stdio orchestration with inert transport seams only."""
+
+import asyncio
+import json
+import os
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+
+@pytest.fixture
+def diagnostics(tmp_path, monkeypatch):
+    import tools.mcp_tool as mcp_tool
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setattr(mcp_tool, "_mcp_stderr_log_files", {}, raising=False)
+    if hasattr(mcp_tool, "_mcp_stderr_log_fh"):
+        monkeypatch.setattr(mcp_tool, "_mcp_stderr_log_fh", None)
+    token = set_hermes_home_override(str(tmp_path / "profile"))
+    try:
+        yield mcp_tool
+    finally:
+        reset_hermes_home_override(token)
+        for fh in mcp_tool._mcp_stderr_log_files.values():
+            fh.close()
+        legacy = getattr(mcp_tool, "_mcp_stderr_log_fh", None)
+        if legacy is not None:
+            legacy.close()
+
+
+def test_existing_index_seam_follows_the_current_config_home(diagnostics, tmp_path):
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    for name in ("alpha", "beta"):
+        token = set_hermes_home_override(str(tmp_path / name))
+        try:
+            fh = diagnostics._get_mcp_stderr_log()
+            fh.write(name + "\n")
+            fh.flush()
+        finally:
+            reset_hermes_home_override(token)
+    assert (tmp_path / "alpha" / "logs" / "mcp-stderr.log").read_text() == "alpha\n"
+    assert (tmp_path / "beta" / "logs" / "mcp-stderr.log").read_text() == "beta\n"
+
+
+def test_profile_and_attempt_output_remain_distinct(diagnostics, tmp_path):
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    mcp_tool = diagnostics
+    records = []
+    for home, text in [("alpha", "first"), ("beta", "second"), ("alpha", "third")]:
+        token = set_hermes_home_override(str(tmp_path / home))
+        try:
+            capture = mcp_tool._begin_stdio_diagnostic("cea_graph")
+            capture["stream"].write(text + "\n")
+            capture["status"] = "closed"
+            mcp_tool._finish_stdio_diagnostic(capture)
+            assert capture["stream"].closed
+            records.append(capture)
+        finally:
+            reset_hermes_home_override(token)
+    assert len({record["stderr_path"] for record in records}) == 3
+    assert len({record["attempt_id"] for record in records}) == 3
+    for record, text in zip(records, ["first", "second", "third"]):
+        path = Path(record["stderr_path"])
+        assert path.parent.parent.parent == Path(record["config_home"])
+        content = path.read_text()
+        assert text + "\n" in content
+        assert all(other + "\n" not in content for other in {"first", "second", "third"} - {text})
+        if os.name == "posix":
+            assert path.stat().st_mode & 0o777 == 0o600
+    for home, expected in [("alpha", 2), ("beta", 1)]:
+        index = (tmp_path / home / "logs" / "mcp-stderr.log").read_text().splitlines()
+        entries = [json.loads(line) for line in index]
+        assert len(entries) == expected
+        assert all(entry["config_home"] == str(tmp_path / home) for entry in entries)
+    assert mcp_tool._mcp_stdio_diagnostic.get() is None
+
+
+def test_nested_attempts_restore_context_and_close_only_their_stream(diagnostics):
+    mcp_tool = diagnostics
+    outer = mcp_tool._begin_stdio_diagnostic("cea_graph")
+    inner = mcp_tool._begin_stdio_diagnostic("pilot_bridge")
+    inner["status"] = "closed"
+    mcp_tool._finish_stdio_diagnostic(inner)
+    assert inner["stream"].closed
+    assert not outer["stream"].closed
+    assert mcp_tool._mcp_stdio_diagnostic.get() is outer
+    outer["status"] = "closed"
+    mcp_tool._finish_stdio_diagnostic(outer)
+    assert mcp_tool._mcp_stdio_diagnostic.get() is None
+
+
+def test_missing_capture_reports_discarded_without_retaining_error_text(
+    diagnostics, monkeypatch, caplog
+):
+    import builtins
+
+    real_open = builtins.open
+
+    def unavailable(path, *args, **kwargs):
+        if str(path) != os.devnull:
+            raise OSError("MUST_NOT_RETAIN_SECRET")
+        return real_open(path, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "open", unavailable)
+    capture = diagnostics._begin_stdio_diagnostic("cea_graph")
+    capture["status"] = "closed"
+    diagnostics._finish_stdio_diagnostic(capture)
+    assert capture["destination"] == "discarded"
+    assert "discarded" in caplog.text
+    assert "MUST_NOT_RETAIN_SECRET" not in caplog.text
+    assert capture["stream"].closed
+    assert diagnostics._mcp_stdio_diagnostic.get() is None
+
+
+@pytest.mark.parametrize("phase", ["negotiate", "discover_tools"])
+@pytest.mark.parametrize("error_type", [RuntimeError, asyncio.CancelledError])
+def test_actual_stdio_failure_preserves_error_and_phase_without_live_io(
+    diagnostics, monkeypatch, caplog, phase, error_type
+):
+    import tools.osv_check
+
+    mcp_tool = diagnostics
+    monkeypatch.setattr(mcp_tool, "_ensure_mcp_sdk", lambda: True)
+    monkeypatch.setattr(mcp_tool, "StdioServerParameters", SimpleNamespace, raising=False)
+    monkeypatch.setattr(mcp_tool, "_resolve_stdio_command", lambda command, env: (command, env))
+    monkeypatch.setattr(mcp_tool, "_wrap_command_with_watchdog", lambda command, args: (command, args))
+    monkeypatch.setattr(mcp_tool, "_kill_orphaned_mcp_children", lambda: None)
+    monkeypatch.setattr(mcp_tool, "_snapshot_child_pids", lambda: set())
+    monkeypatch.setattr(mcp_tool, "_MCP_NOTIFICATION_TYPES", False)
+    monkeypatch.setattr(mcp_tool, "_MCP_LOGGING_CALLBACK_SUPPORTED", False)
+    monkeypatch.setattr(tools.osv_check, "check_package_for_malware", lambda *_args: None)
+
+    @asynccontextmanager
+    async def inert_stdio(_params, *, errlog):
+        errlog.write("inert child stderr\n")
+        yield None, None
+
+    @asynccontextmanager
+    async def inert_session(*_args, **_kwargs):
+        yield object()
+
+    monkeypatch.setattr(mcp_tool, "stdio_client", inert_stdio, raising=False)
+    monkeypatch.setattr(mcp_tool, "ClientSession", inert_session, raising=False)
+    server = mcp_tool.MCPServerTask("cea_graph")
+    monkeypatch.setattr(mcp_tool, "_servers", {"cea_graph": server})
+    monkeypatch.setattr(mcp_tool, "_server_connecting", set())
+    monkeypatch.setattr(mcp_tool, "_server_connect_errors", {})
+    monkeypatch.setattr(mcp_tool, "_load_mcp_config", lambda: {"cea_graph": {"command": "inert"}})
+    original = error_type("MUST_NOT_RETAIN_SECRET")
+
+    async def discover(_server):
+        # Exercise get_mcp_status during the actual retained-task window:
+        # session assigned, tools not discovered, readiness not yet signalled.
+        assert mcp_tool.get_mcp_status()[0]["status"] == "connecting"
+        raise original
+
+    monkeypatch.setattr(
+        mcp_tool.MCPServerTask,
+        "_negotiate_session",
+        AsyncMock(
+            side_effect=original if phase == "negotiate" else None,
+            return_value=SimpleNamespace(),
+        ),
+    )
+    monkeypatch.setattr(mcp_tool.MCPServerTask, "_discover_tools", discover)
+    with pytest.raises(error_type) as caught:
+        asyncio.run(server._run_stdio({"command": "inert-never-executed"}))
+    assert caught.value is original
+    assert not server._ready.is_set()
+    assert mcp_tool._mcp_stdio_diagnostic.get() is None
+    records = [json.loads(record.getMessage().split(": ", 1)[1]) for record in caplog.records
+               if record.getMessage().startswith("MCP stdio attempt ended: ")]
+    assert len(records) == 1
+    record = records[0]
+    assert record["phase"] == phase
+    assert record["status"] == "failed"
+    assert record["exception_type"] == error_type.__name__
+    assert record["destination"] == "file"
+    assert "MUST_NOT_RETAIN_SECRET" not in caplog.text
+    assert "MUST_NOT_RETAIN_SECRET" not in Path(record["stderr_path"]).read_text()
+    assert "inert child stderr" in Path(record["stderr_path"]).read_text()
+
+
+def test_actual_run_retry_clears_readiness_before_second_stdio_discovery(diagnostics, monkeypatch):
+    import tools.osv_check
+
+    mcp_tool = diagnostics
+    monkeypatch.setattr(mcp_tool, "_ensure_mcp_sdk", lambda: True)
+    monkeypatch.setattr(mcp_tool, "StdioServerParameters", SimpleNamespace, raising=False)
+    monkeypatch.setattr(mcp_tool, "_resolve_stdio_command", lambda command, env: (command, env))
+    monkeypatch.setattr(mcp_tool, "_wrap_command_with_watchdog", lambda command, args: (command, args))
+    monkeypatch.setattr(mcp_tool, "_kill_orphaned_mcp_children", lambda: None)
+    monkeypatch.setattr(mcp_tool, "_snapshot_child_pids", lambda: set())
+    monkeypatch.setattr(mcp_tool, "_MCP_NOTIFICATION_TYPES", False)
+    monkeypatch.setattr(mcp_tool, "_MCP_LOGGING_CALLBACK_SUPPORTED", False)
+    monkeypatch.setattr(tools.osv_check, "check_package_for_malware", lambda *_args: None)
+    monkeypatch.setattr(mcp_tool.asyncio, "sleep", AsyncMock())
+
+    @asynccontextmanager
+    async def inert_stdio(_params, *, errlog):
+        errlog.write("inert retry stderr\n")
+        yield None, None
+
+    @asynccontextmanager
+    async def inert_session(*_args, **_kwargs):
+        yield object()
+
+    monkeypatch.setattr(mcp_tool, "stdio_client", inert_stdio, raising=False)
+    monkeypatch.setattr(mcp_tool, "ClientSession", inert_session, raising=False)
+    monkeypatch.setattr(mcp_tool.MCPServerTask, "_negotiate_session", AsyncMock(return_value=SimpleNamespace()))
+    server = mcp_tool.MCPServerTask("cea_graph")
+    monkeypatch.setattr(mcp_tool, "_servers", {"cea_graph": server})
+    monkeypatch.setattr(mcp_tool, "_server_connecting", set())
+    monkeypatch.setattr(mcp_tool, "_server_connect_errors", {})
+    monkeypatch.setattr(mcp_tool, "_load_mcp_config", lambda: {"cea_graph": {"command": "inert"}})
+    statuses = []
+
+    async def discover(_server):
+        statuses.append(mcp_tool.get_mcp_status()[0]["status"])
+        if len(statuses) == 2:
+            server._shutdown_event.set()
+
+    async def lifecycle(_server):
+        if len(statuses) == 1:
+            raise BrokenPipeError("inert transient failure after first readiness")
+        return "shutdown"
+
+    monkeypatch.setattr(mcp_tool.MCPServerTask, "_discover_tools", discover)
+    monkeypatch.setattr(mcp_tool.MCPServerTask, "_wait_for_lifecycle_event", lifecycle)
+    asyncio.run(server.run({"command": "inert-never-executed", "sampling": {"enabled": False},
+                           "elicitation": {"enabled": False}}))
+    assert statuses == ["connecting", "connecting"]
+    assert server._ever_connected
diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py
index a33710a5fe0e..5ea82914c987 100644
--- a/tools/mcp_tool.py
+++ b/tools/mcp_tool.py
@@ -114,8 +114,10 @@
 from types import SimpleNamespace
 from typing import Callable
 from datetime import datetime
+from pathlib import Path
 from typing import Any, Coroutine, Dict, List, Optional, Set, Tuple
 from urllib.parse import urlparse
+from uuid import uuid4
 
 from tools.registry import tool_error
 from tools.ansi_strip import strip_unicode_tags
@@ -179,63 +181,157 @@ def _truncate_mcp_text_result(text: str, max_chars: int = _MCP_HARD_RESULT_CAP_C
 # the terminal while prompt_toolkit / Rich is rendering the TUI — which
 # corrupts the display and can hang the session.
 #
-# Instead we redirect every stdio MCP subprocess's stderr into a shared
-# per-profile log file (~/.hermes/logs/mcp-stderr.log), tagged with the
-# server name so individual servers remain debuggable.
+# Each stdio attempt owns a separate file under its config home's logs.
+# The existing mcp-stderr.log remains an index, without interleaving child
+# output from concurrent servers or borrowing another profile's handle.
 #
 # Fallback is os.devnull if opening the log file fails for any reason.
 
-_mcp_stderr_log_fh: Optional[Any] = None
+_mcp_stderr_log_files: Dict[str, Any] = {}
 _mcp_stderr_log_lock = threading.Lock()
+_mcp_stdio_diagnostic: contextvars.ContextVar[Optional[dict]] = contextvars.ContextVar(
+    "mcp_stdio_diagnostic", default=None
+)
 
 
 def _get_mcp_stderr_log() -> Any:
-    """Return a shared append-mode file handle for MCP subprocess stderr.
+    """Return this attempt's stderr file, or this config home's index.
 
-    Opened once per process and reused for every stdio server.  Must have a
-    real OS-level file descriptor (``fileno()``) because asyncio's subprocess
-    machinery wires the child's stderr directly to that fd.  Falls back to
-    ``/dev/null`` if opening the log file fails.
+    The no-argument seam is retained for existing stdio adapters. Attempt
+    handles are closed with the transport; only per-home index handles are
+    cached. Diagnostics must never prevent a connection attempt.
     """
-    global _mcp_stderr_log_fh
+    capture = _mcp_stdio_diagnostic.get()
+    if capture is not None and "stream" in capture:
+        return capture["stream"]
     with _mcp_stderr_log_lock:
-        if _mcp_stderr_log_fh is not None:
-            return _mcp_stderr_log_fh
+        log_path = None
+        fh = None
         try:
-            from hermes_constants import get_hermes_home
-            log_dir = get_hermes_home() / "logs"
-            log_dir.mkdir(parents=True, exist_ok=True)
-            log_path = log_dir / "mcp-stderr.log"
-            # Line-buffered so server output lands on disk promptly; errors=
-            # "replace" tolerates garbled binary output from misbehaving
-            # servers.
-            fh = open(log_path, "a", encoding="utf-8", errors="replace", buffering=1)
-            # Sanity-check: confirm a real fd is available before we commit.
+            if capture is not None:
+                log_path = Path(capture["stderr_path"])
+            else:
+                from hermes_constants import get_hermes_home
+
+                log_path = get_hermes_home() / "logs" / "mcp-stderr.log"
+                cached = _mcp_stderr_log_files.get(str(log_path))
+                if cached is not None and not cached.closed:
+                    return cached
+            log_path.parent.mkdir(parents=True, exist_ok=True)
+            fh = open(
+                log_path, "x+" if capture is not None else "a", encoding="utf-8",
+                errors="replace", buffering=1, newline="" if capture is not None else None,
+                opener=lambda path, flags: os.open(path, flags, 0o600),
+            )
             fh.fileno()
-            _mcp_stderr_log_fh = fh
+            destination = "file"
         except Exception as exc:  # pragma: no cover — best-effort fallback
-            logger.debug("Failed to open MCP stderr log, using devnull: %s", exc)
+            if fh is not None:
+                try:
+                    fh.close()
+                except Exception:
+                    pass
+            logger.debug("Failed to open MCP stderr log: %s", type(exc).__name__)
             try:
-                _mcp_stderr_log_fh = open(os.devnull, "w", encoding="utf-8")
+                fh = open(os.devnull, "w", encoding="utf-8")
+                destination = "discarded"
             except Exception:
-                # Last resort: the real stderr.  Not ideal for TUI users but
-                # it matches pre-fix behavior.
-                _mcp_stderr_log_fh = sys.stderr
-        return _mcp_stderr_log_fh
+                fh = sys.stderr
+                destination = "parent_stderr"
+        if capture is not None:
+            capture["stream"] = fh
+            capture["destination"] = destination
+        elif log_path is not None:
+            _mcp_stderr_log_files[str(log_path)] = fh
+        return fh
+
+
+def _stdio_diagnostic_fields(capture: dict) -> dict:
+    """Only ownership and lifecycle metadata; no argv, env or error text."""
+    return {"kind": "mcp.stdio.attempt", **{key: capture.get(key) for key in (
+        "attempt_id", "server", "config_home", "parent_pid", "stderr_path",
+        "destination", "phase", "status", "exception_type",
+    )}}
+
+
+def _begin_stdio_diagnostic(server_name: str) -> dict:
+    capture = {
+        "attempt_id": uuid4().hex, "server": server_name,
+        "config_home": None, "parent_pid": os.getpid(), "stderr_path": None,
+        "phase": "transport", "status": "starting",
+    }
+    try:
+        from hermes_constants import get_hermes_home
+
+        home = get_hermes_home()
+        capture["config_home"] = str(home)
+        capture["stderr_path"] = str(home / "logs" / "mcp-stderr" / f"{capture['attempt_id']}.log")
+    except Exception:
+        pass  # Unavailable ownership is explicit; capture still fails softly.
+    capture["token"] = _mcp_stdio_diagnostic.set(capture)
+    capture["stream"] = _get_mcp_stderr_log()
+    _write_stderr_log_header(server_name)
+    logger.debug("MCP stdio attempt: %s", json.dumps(_stdio_diagnostic_fields(capture)))
+    return capture
+
+
+def _finish_stdio_diagnostic(capture: dict) -> None:
+    try:
+        record = json.dumps(_stdio_diagnostic_fields(capture))
+        fh = capture.get("stream")
+        try:
+            if fh is not None:
+                boundary = "\n"
+                if capture.get("destination") == "file":
+                    # The transport has unwound. Inspect its last byte through
+                    # the same owned handle; child stderr may omit a newline.
+                    fh.flush()
+                    fh.seek(0, os.SEEK_END)
+                    end = fh.tell()
+                    if end:
+                        fh.seek(end - 1)
+                        boundary = "" if fh.read(1) == "\n" else "\n"
+                    else:
+                        boundary = ""
+                    fh.seek(0, os.SEEK_END)
+                fh.write(boundary + record + "\n")
+                fh.flush()
+        except Exception:
+            pass
+        if capture["status"] != "closed" or capture.get("destination") != "file":
+            logger.warning("MCP stdio attempt ended: %s", record)
+    finally:
+        _mcp_stdio_diagnostic.reset(capture["token"])
+        fh = capture.get("stream")
+        if fh is not None and fh is not sys.stderr:
+            try:
+                fh.close()
+            except Exception:
+                pass
 
 
 def _write_stderr_log_header(server_name: str) -> None:
     """Write a human-readable session marker before launching a server.
 
-    Gives operators a way to find each server's output in the shared
-    ``mcp-stderr.log`` file without needing per-line prefixes (which would
-    require a pipe + reader thread and complicate shutdown).
+    The per-home index points at an attempt-owned file, so concurrent child
+    output can be attributed without changing the MCP protocol streams.
     """
     fh = _get_mcp_stderr_log()
     try:
         ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         fh.write(f"\n===== [{ts}] starting MCP server '{server_name}' =====\n")
         fh.flush()
+        capture = _mcp_stdio_diagnostic.get()
+        if capture is not None:
+            record = json.dumps(_stdio_diagnostic_fields(capture))
+            fh.write(record + "\n")
+            token = _mcp_stdio_diagnostic.set(None)
+            try:
+                index = _get_mcp_stderr_log()
+            finally:
+                _mcp_stdio_diagnostic.reset(token)
+            index.write(record + "\n")
+            index.flush()
     except Exception:
         pass
 
@@ -3280,12 +3376,10 @@ async def _run_stdio(self, config: dict):
         # Snapshot child PIDs before spawning so we can track the new one.
         pids_before = _snapshot_child_pids()
         new_pids: set = set()
-        # Redirect subprocess stderr into a shared log file so MCP servers
-        # (FastMCP banners, slack-mcp startup JSON, etc.) don't dump onto
-        # the user's TTY and corrupt the TUI.  Preserves debuggability via
-        # ~/.hermes/logs/mcp-stderr.log.
-        _write_stderr_log_header(self.name)
-        _errlog = _get_mcp_stderr_log()
+        # Each attempt keeps its child stderr and config-home ownership.
+        # The per-home mcp-stderr.log is an index of those files.
+        _diagnostic = _begin_stdio_diagnostic(self.name)
+        _errlog = _diagnostic["stream"]
         try:
             async with stdio_client(server_params, errlog=_errlog) as (
                 read_stream,
@@ -3356,13 +3450,16 @@ async def _run_stdio(self, config: dict):
                     connect_timeout = float(
                         config.get("connect_timeout", _DEFAULT_CONNECT_TIMEOUT)
                     )
+                    _diagnostic["phase"] = "negotiate"
                     self.initialize_result = await self._negotiate_session(
                         session, connect_timeout
                     )
                     self.session = session
                     self._mark_lifecycle_started()
+                    _diagnostic["phase"] = "discover_tools"
                     await self._discover_tools()
                     self._ready.set()
+                    _diagnostic["phase"] = "active"
                     self._ever_connected = True
                     # Session is live again: clear any breaker state from a
                     # prior outage so the first call after recovery isn't
@@ -3378,7 +3475,17 @@ async def _run_stdio(self, config: dict):
                     # _reconnect_event (e.g. future manual /mcp refresh) for
                     # consistency with _run_http.
                     return await self._wait_for_lifecycle_event()
+        except BaseException as exc:
+            _diagnostic["status"] = "failed"
+            _diagnostic["exception_type"] = type(exc).__name__
+            raise
         finally:
+            if _diagnostic["status"] == "starting":
+                _diagnostic["status"] = "closed"
+            try:
+                _finish_stdio_diagnostic(_diagnostic)
+            except Exception:
+                logger.debug("Failed to finalize MCP stdio diagnostics", exc_info=False)
             # Runs on clean exit, exceptions, AND asyncio cancellation.
             # If any of the spawned PIDs are still alive, the SDK's
             # teardown failed (common when the task is cancelled mid-way
@@ -4020,6 +4127,9 @@ async def run(self, config: dict):
 
         while True:
             try:
+                # Readiness belongs to this transport attempt. Exception
+                # retries and lazy stdio revival also rebuild the session.
+                self._ready.clear()
                 if self._is_http():
                     lifecycle_reason = await self._run_http(config)
                 else:
@@ -8001,7 +8111,11 @@ def get_mcp_status() -> List[dict]:
         transport = cfg.get("transport", "http") if "url" in cfg else "stdio"
         enabled = _parse_boolish(cfg.get("enabled", True), default=True)
         server = active_servers.get(name)
-        if server and server.session is not None:
+        # A session is assigned after initialize, before tools/list completes.
+        # The task's ready event and error state own readiness, including on
+        # reconnect; a transport object alone must not advertise success.
+        if (server and server.session is not None and server._ready.is_set()
+                and server._error is None):
             entry = {
                 "name": name,
                 "transport": transport,
@@ -8025,7 +8139,8 @@ def get_mcp_status() -> List[dict]:
                 "disabled": True,
                 "status": "disabled",
             })
-        elif name in connecting:
+        elif name in connecting or (server and not server._ready.is_set()
+                                    and server._error is None):
             result.append({
                 "name": name,
                 "transport": transport,
@@ -8034,7 +8149,7 @@ def get_mcp_status() -> List[dict]:
                 "disabled": False,
                 "status": "connecting",
             })
-        elif name in connect_errors:
+        elif name in connect_errors or (server and server._error is not None):
             result.append({
                 "name": name,
                 "transport": transport,
@@ -8042,7 +8157,7 @@ def get_mcp_status() -> List[dict]:
                 "connected": False,
                 "disabled": False,
                 "status": "failed",
-                "error": connect_errors[name],
+                "error": connect_errors.get(name) or _format_connect_error(server._error),
             })
         else:
             result.append({