post_execute fires during comm_msg handling on a subshell, so output published from the hook is parented to the comm_msg and dropped by frontends

Anything that publishes output from a post_execute hook -- ipympl's
flush_figures, matplotlib_inline's -- therefore publishes it with
parent_header.msg_id pointing at the comm_msg rather than at the cell's
execute_request. A frontend routes iopub output to a cell by that id, so the
output is silently dropped. Worse, the hook has now drained its pending queue,
so the cell's own post_execute has nothing left to display.

This is, I believe, the cause of matplotlib/ipympl#609: an ipympl figure that is
already on screen keeps a comm busy, so the next figure is the one that
vanishes -- intermittently, more often with slower cells, and never with
%matplotlib inline (no comms).

No matplotlib / ipympl / ipywidgets required.

python repro_parent_header.py

Exit code 0 = correct behaviour, 1 = bug reproduced.
"""

from future import annotations

import queue
import sys
import time

import ipykernel
from jupyter_client.manager import start_new_kernel

SETUP = r'''
import time, threading
from IPython.display import display
from ipykernel.comm import Comm

T0 = time.time()
EVENTS = []
_pending = []

def _post_execute():
# This is what ipympl's flush_figures does: display whatever the cell queued.
EVENTS.append(("post_execute", round(time.time() - T0, 2), threading.current_thread().name))
while _pending:
display(_pending.pop())

get_ipython().events.register("post_execute", _post_execute)

_comm = Comm(target_name="repro")
_comm.on_msg(lambda msg: None)
print("COMM_ID", _comm.comm_id)
'''

CELL = r'''
_pending.append("PAYLOAD-PUBLISHED-FROM-POST-EXECUTE")
time.sleep(3) # window for a comm_msg to land mid-cell
'''

REPORT = "for e in EVENTS: print(*e)"

def drain(kc, stop_msg_id, log=None):
deadline = time.time() + 30
while time.time() < deadline:
try:
msg = kc.get_iopub_msg(timeout=0.5)
except queue.Empty:
continue
parent = msg.get("parent_header") or {}
item = (msg["msg_type"], parent.get("msg_type"), parent.get("msg_id"), msg["content"])
if log is not None:
log.append(item)
yield item
if (
msg["msg_type"] == "status"
and msg["content"]["execution_state"] == "idle"
and parent.get("msg_id") == stop_msg_id
):
return

def create_subshell(kc) -> str | None:
"""Ask for a subshell, as JupyterLab does for kernel comms. None on ipykernel 6."""
msg = kc.session.msg("create_subshell_request", {})
kc.control_channel.send(msg)
deadline = time.time() + 10
while time.time() < deadline:
try:
reply = kc.get_control_msg(timeout=2)
except queue.Empty:
return None
if reply["parent_header"].get("msg_id") == msg["header"]["msg_id"]:
return reply["content"].get("subshell_id")
return None

def main() -> int:
print(f"ipykernel {ipykernel.version}, python {sys.version.split()[0]}")
km, kc = start_new_kernel(kernel_name="python3")
try:
setup_id = kc.execute(SETUP)
comm_id = None
for mtype, _, _, content in drain(kc, setup_id):
if mtype == "stream" and content["text"].startswith("COMM_ID"):
comm_id = content["text"].split()[1]
assert comm_id, "no comm id"

    subshell_id = create_subshell(kc)
    print(f"subshell: {subshell_id or 'not supported (ipykernel 6)'}")

    cell_id = kc.execute(CELL)
    time.sleep(1.0)  # let the cell get going
    comm_msg = kc.session.msg(
        "comm_msg", {"comm_id": comm_id, "data": {"method": "custom", "content": {}}}
    )
    if subshell_id:
        comm_msg["header"]["subshell_id"] = subshell_id
    kc.shell_channel.send(comm_msg)
    comm_msg_id = comm_msg["header"]["msg_id"]

    trace: list = []
    display_parent = None
    for mtype, ptype, pid, _ in drain(kc, cell_id, log=trace):
        if mtype == "display_data":
            display_parent = (ptype, pid)

    print("\niopub during the cell")
    for mtype, ptype, pid, content in trace:
        who = "CELL" if pid == cell_id else "COMM" if pid == comm_msg_id else "-"
        note = ""
        if mtype == "status":
            note = content["execution_state"]
        elif mtype == "display_data":
            note = str(list(content.get("data", {}).keys()))
        print(f"  {mtype:14s} parent={ptype or '-':22s} [{who}]  {note}")

    print("\nhook invocations (elapsed, thread)")
    report_id = kc.execute(REPORT)
    for mtype, _, _, content in drain(kc, report_id):
        if mtype == "stream":
            for line in content["text"].splitlines():
                print(f"  {line}")

    print(f"\n  execute_request      {cell_id}")
    print(f"  comm_msg             {comm_msg_id}")
    print(f"  display_data parent  {display_parent}\n")

    if display_parent is None:
        print("INCONCLUSIVE: no display_data observed")
        return 2
    if display_parent[1] == cell_id:
        print("OK   display_data is parented to the cell's execute_request")
        return 0
    print("BUG  display_data is parented to the comm_msg, not to the cell.")
    print("     A frontend has no cell for that parent, so the output is dropped.")
    return 1
finally:
    kc.stop_channels()
    km.shutdown_kernel()

if name == "main":
raise SystemExit(main())