diff --git a/evidence/driver-write-completion-20261007/README.md b/evidence/driver-write-completion-20261007/README.md new file mode 100644 index 0000000000..fba9a89558 --- /dev/null +++ b/evidence/driver-write-completion-20261007/README.md @@ -0,0 +1,40 @@ +# Driver-owned write completion + +Driver.call_after_flush schedules its synchronous callback on the original +Driver App loop after its output flush boundary. Synchronous drivers use their +actual flush; queued Linux and Windows writers use the original WriterThread +FIFO signal. The existing signal now carries and publishes onto that loop. +There is no Toad platform decision or ambient-loop wrapper in this capability. + +Headless has no output to wait for. Inline flushes its buffered file. Web waits +for complete framed-packet acceptance at its existing OS write boundary; that +boundary now handles partial os.write results. Neither terminal pixels nor +browser paint are acknowledged. Flush/write failures cannot publish success. + +Writer shutdown drains prior signals and clears the original driver's writer +custody; later queued-driver submission is refused outside application mode. +A retired App loop receives no completion. Callback exceptions execute under +the App loop's normal exception handling, not on the terminal writer thread. +WriterThread.call_after_flush now requires the original loop explicitly, and +all native consumers including its original control are migrated. + +Qualified native source checks: four focused controls passed in 0.35 seconds, +covering FIFO flush, callback thread/loop, headless scheduling, buffered inline +file flush, closed-loop retirement and joined writer shutdown. + +The original WebDriver accepted a 1,100,005-byte framed pipe packet across two +actual OS writes; exact bytes and owner-loop completion passed, and the reader +joined. This proves transport acceptance, not browser paint. + +The actual Linux source App rendered through its original writer and delivered +completion on its App loop. Writer and input thread joined. PTY input was used; +output was captured to stderr, not inspected as terminal pixels. The first run +failed only its cleanup oracle: Linux retains the joined input Thread object. +That negative is retained. The corrected Linux-only check passed without +repeating the Web control. Captured stderr includes normal terminal output. + +Windows console startup remains unexecuted on this Linux host. Its original +shared WriterThread path is source-reviewed and that writer is exercised above. +No latency improvement is claimed. Native verification used zero inputs and +providers; Parent separately verified and published the paired Linux saved-tab +App. Production remains byte-identical to the delivered 41e8a5319 checkpoint. diff --git a/evidence/driver-write-completion-20261007/app-first.stderr.log b/evidence/driver-write-completion-20261007/app-first.stderr.log new file mode 100644 index 0000000000..e16b41d8d0 --- /dev/null +++ b/evidence/driver-write-completion-20261007/app-first.stderr.log @@ -0,0 +1,26 @@ +[?1049h=[?25l[?1004h[>25u[?2026$p[?2048$p[?2004h[?7l]22;defaultNative driver write completion  +  +  +  +  +  +  +  +  + [?2004l[?7h>[ + asyncio.run(main(Path(sys.argv[1]))) + ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^ + File "/home/ts/.local/share/uv/python/cpython-3.14.2-linux-x86_64-gnu/lib/python3.14/asyncio/runners.py", line 204, in run + return runner.run(main) + ~~~~~~~~~~^^^^^^ + File "/home/ts/.local/share/uv/python/cpython-3.14.2-linux-x86_64-gnu/lib/python3.14/asyncio/runners.py", line 127, in run + return self._loop.run_until_complete(task) + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^ + File "/home/ts/.local/share/uv/python/cpython-3.14.2-linux-x86_64-gnu/lib/python3.14/asyncio/base_events.py", line 719, in run_until_complete + return future.result() + ~~~~~~~~~~~~~^^ + File "/home/ts/wt/textual-frame-publication-20261006/tests/driver_write_completion_app.py", line 117, in main + assert app._driver._key_thread is None + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +AssertionError diff --git a/evidence/driver-write-completion-20261007/app-first.stdout.log b/evidence/driver-write-completion-20261007/app-first.stdout.log new file mode 100644 index 0000000000..e69de29bb2 diff --git a/evidence/driver-write-completion-20261007/app-linux02.stderr.log b/evidence/driver-write-completion-20261007/app-linux02.stderr.log new file mode 100644 index 0000000000..e3663c37da --- /dev/null +++ b/evidence/driver-write-completion-20261007/app-linux02.stderr.log @@ -0,0 +1,10 @@ +[?1049h=[?25l[?1004h[>25u[?2026$p[?2048$p[?2004h[?7l]22;defaultNative driver write completion  +  +  +  +  +  +  +  +  + [?2004l[?7h>[ None: def flush(self) -> None: """Flush any buffered data.""" + def call_after_flush(self, callback: Callable[[], None]) -> None: + """Publish write completion on this driver's original application loop. + + Synchronous output completes at flush; queued output must override this + operation with its actual writer acknowledgment. Completion covers + earlier writes to the driver, not terminal or browser pixel presentation. + Callbacks are scheduled, never invoked inline. A closed application loop + has retired its receiver and receives no completion. + """ + self.flush() + _FlushSignal(callback, self._loop).deliver() + def _enable_application_keypad(self) -> None: """Enter the terminal's unambiguous application-keypad input mode.""" self.write("\x1b=") # DECKPAM; normal-mode Delete may collide with F1. diff --git a/src/textual/drivers/_writer_thread.py b/src/textual/drivers/_writer_thread.py index 59d666c83a..ce88e2b171 100644 --- a/src/textual/drivers/_writer_thread.py +++ b/src/textual/drivers/_writer_thread.py @@ -1,6 +1,7 @@ from __future__ import annotations import threading +from asyncio import AbstractEventLoop from dataclasses import dataclass from queue import Queue from typing import IO, Callable @@ -13,6 +14,15 @@ @dataclass(frozen=True) class _FlushSignal: callback: Callable[[], None] + loop: AbstractEventLoop + + def deliver(self) -> None: + """Publish completion to the original application loop, if still alive.""" + try: + self.loop.call_soon_threadsafe(self.callback) + except RuntimeError: + if not self.loop.is_closed(): + raise class WriterThread(threading.Thread): @@ -31,9 +41,9 @@ def write(self, text: str) -> None: """ self._queue.put(text) - def call_after_flush(self, callback: Callable[[], None]) -> None: - """Call back after all earlier queued terminal writes are flushed.""" - self._queue.put(_FlushSignal(callback)) + def call_after_flush(self, callback: Callable[[], None], loop: AbstractEventLoop) -> None: + """Publish on the supplied loop after all earlier queued writes flush.""" + self._queue.put(_FlushSignal(callback, loop)) def isatty(self) -> bool: """Pretend to be a terminal. @@ -69,7 +79,7 @@ def run(self) -> None: break if isinstance(text, _FlushSignal): flush() - text.callback() + text.deliver() continue write(text) if qsize() == 0: diff --git a/src/textual/drivers/linux_driver.py b/src/textual/drivers/linux_driver.py index d5e5b4538b..ed3f5ba40b 100644 --- a/src/textual/drivers/linux_driver.py +++ b/src/textual/drivers/linux_driver.py @@ -194,9 +194,9 @@ def write(self, data: str) -> None: self._writer_thread.write(data) def call_after_flush(self, callback: Callable[[], None]) -> None: - """Observe a flush of all terminal writes queued before this call.""" + """Publish on the App loop after all earlier terminal writes flush.""" assert self._writer_thread is not None, "Driver must be in application mode" - self._writer_thread.call_after_flush(callback) + self._writer_thread.call_after_flush(callback, self._loop) def start_application_mode(self): """Start application mode.""" @@ -406,6 +406,7 @@ def close(self) -> None: """Perform cleanup.""" if self._writer_thread is not None: self._writer_thread.stop() + self._writer_thread = None def _run_input_thread(self) -> None: """ diff --git a/src/textual/drivers/web_driver.py b/src/textual/drivers/web_driver.py index f21d19ed38..0d22d2ca36 100644 --- a/src/textual/drivers/web_driver.py +++ b/src/textual/drivers/web_driver.py @@ -17,7 +17,6 @@ import signal import sys from codecs import getincrementaldecoder -from functools import partial from pathlib import Path from threading import Event, Thread from typing import Any, BinaryIO, Literal, TextIO, cast @@ -60,7 +59,6 @@ def __init__( super().__init__(app, debug=debug, mouse=mouse, size=size) self.stdout = sys.__stdout__ self.fileno = sys.__stdout__.fileno() - self._write = partial(os.write, self.fileno) self.exit_event = Event() self._key_thread: Thread = Thread( target=self.run_input_thread, name="textual-input" @@ -75,6 +73,15 @@ def __init__( def is_web(self) -> bool: return True + def _write(self, data: bytes) -> None: + """Accept the complete framed packet at the original transport boundary.""" + remaining = memoryview(data) + while remaining: + written = os.write(self.fileno, remaining) + if not written: + raise OSError("Web output write made no progress") + remaining = remaining[written:] + def write(self, data: str) -> None: """Write string data to the output device, which may be piped to the parent process (i.e. textual-web/textual-serve). diff --git a/src/textual/drivers/windows_driver.py b/src/textual/drivers/windows_driver.py index 76fd965652..5eeae0a26c 100644 --- a/src/textual/drivers/windows_driver.py +++ b/src/textual/drivers/windows_driver.py @@ -53,6 +53,11 @@ def write(self, data: str) -> None: assert self._writer_thread is not None, "Driver must be in application mode" self._writer_thread.write(data) + def call_after_flush(self, callback: Callable[[], None]) -> None: + """Publish on the App loop after all earlier terminal writes flush.""" + assert self._writer_thread is not None, "Driver must be in application mode" + self._writer_thread.call_after_flush(callback, self._loop) + def _enable_mouse_support(self) -> None: """Enable reporting of mouse events.""" if not self._mouse: @@ -137,5 +142,6 @@ def close(self) -> None: """Perform cleanup.""" if self._writer_thread is not None: self._writer_thread.stop() + self._writer_thread = None if self._restore_console: self._restore_console() diff --git a/tests/driver_write_completion_app.py b/tests/driver_write_completion_app.py new file mode 100644 index 0000000000..bdf5ea7d97 --- /dev/null +++ b/tests/driver_write_completion_app.py @@ -0,0 +1,128 @@ +"""One source Linux App confirms a real terminal writer acknowledgment.""" + +import asyncio +import cProfile +import hashlib +import json +import os +from pathlib import Path +import pstats +import signal +from threading import Thread, get_ident +from time import sleep + +from textual.app import App +from textual.drivers.linux_driver import LinuxDriver +from textual.drivers.web_driver import WebDriver +from textual.widgets import Static + + +class CompletionApp(App): + def __init__(self): + self.requested = False + self.received = False + self.owner_thread = get_ident() + self.writer = None + super().__init__(driver_class=LinuxDriver) + + def compose(self): + yield Static("Native driver write completion") + + def _display(self, screen, renderable): + super()._display(screen, renderable) + if renderable is not None and not self._batch_count and not self.requested: + self.requested = True + self.writer = self._driver._writer_thread + self._driver.call_after_flush(self.written) + + def written(self): + assert get_ident() == self.owner_thread + assert asyncio.get_running_loop() is self._driver._loop + assert self.writer.is_alive() + self.received = True + self.exit() + + +async def web_transport_completion(): + """Interrupt a real large pipe write after progress, then recover its tail.""" + loop = asyncio.get_running_loop() + driver = WebDriver(App()) + read_fd, write_fd = os.pipe() + driver.fileno = write_fd + chunks = [] + + def read_output(): + try: + while data := os.read(read_fd, 4096): + chunks.append(data) + sleep(.0005) + finally: + os.close(read_fd) + + reader = Thread(target=read_output, name="native-web-output-reader") + text = "frame-data-" * 100000 + payload = text.encode() + expected = b"D" + len(payload).to_bytes(4, "big") + payload + previous_handler = signal.getsignal(signal.SIGALRM) + assert signal.getitimer(signal.ITIMER_REAL) == (0.0, 0.0) + done = loop.create_future() + profile = cProfile.Profile() + reader.start() + try: + signal.signal(signal.SIGALRM, lambda *_: None) + # This is a controlled kernel interruption in the verification program, + # not a renderer timer or runtime workaround. Earlier bytes already fit + # in the pipe, so os.write returns that partial progress on interruption. + signal.setitimer(signal.ITIMER_REAL, .03) + profile.enable() + try: + driver.write(text) + finally: + profile.disable() + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, previous_handler) + driver.call_after_flush(lambda: done.set_result((get_ident(), asyncio.get_running_loop()))) + assert not done.done() + assert await asyncio.wait_for(done, 2) == (get_ident(), loop) + finally: + os.close(write_fd) + reader.join(2) + driver._input_reader.close() + # No input thread was started for this output-only source control. + driver._input_reader._selector.close() + assert not reader.is_alive() + received = b"".join(chunks) + assert received == expected + writes = sum(value[1] for (_, _, function), value in pstats.Stats(profile).stats.items() + if function == "") + assert writes > 1 + return {"scope": "real WebDriver framed pipe write and App-loop callback, not browser pixels", + "packet_bytes": len(received), "packet_sha256": hashlib.sha256(received).hexdigest(), + "os_write_calls": writes, "owner_loop_completion": True, + "reader_joined": True, "inputs": 0, "providers": 0} + + +async def main(output: Path, *, linux_only: bool = False): + output.mkdir(parents=True, exist_ok=True) + os.environ.update(XDG_CONFIG_HOME=str(output / "config"), + XDG_STATE_HOME=str(output / "state"), + XDG_DATA_HOME=str(output / "data")) + if not linux_only: + web = await web_transport_completion() + (output / "web-output.json").write_text(json.dumps(web, indent=2) + "\n") + app = CompletionApp() + await asyncio.wait_for(app.run_async(mouse=False, size=(60, 10)), 15) + assert app.received and app._exception is None + assert app._driver._writer_thread is None + assert not app.writer.is_alive() + assert not app._driver._key_thread.is_alive() + result = {"scope": "real source Linux App/PTY writer completion, not pixel presentation", + "driver": type(app._driver).__name__, "callback_on_owner_loop": True, + "writer_joined": True, "input_thread_joined": True, + "inputs": 0, "providers": 0} + (output / "linux-app.json").write_text(json.dumps(result, indent=2) + "\n") + + +if __name__ == "__main__": + import sys + asyncio.run(main(Path(sys.argv[1]), linux_only="--linux-only" in sys.argv[2:])) diff --git a/tests/test_driver_write_completion.py b/tests/test_driver_write_completion.py new file mode 100644 index 0000000000..f0dc103efc --- /dev/null +++ b/tests/test_driver_write_completion.py @@ -0,0 +1,71 @@ +"""Original driver output boundaries publish on the original App loop.""" + +import asyncio +import os +from pathlib import Path +import sys +import tempfile +from threading import get_ident + +import pytest + +from textual.app import App +from textual.drivers._writer_thread import WriterThread + + +async def test_headless_completion_is_scheduled_on_its_original_loop(): + app = App() + async with app.run_test() as pilot: + await pilot.pause() + loop = asyncio.get_running_loop() + done = loop.create_future() + thread = get_ident() + app._driver.call_after_flush(lambda: done.set_result((get_ident(), asyncio.get_running_loop()))) + assert not done.done() + assert await asyncio.wait_for(done, 2) == (thread, loop) + + +@pytest.mark.skipif(sys.platform == "win32", reason="Original inline driver is POSIX-only") +async def test_inline_file_is_flushed_before_owner_loop_completion(): + from textual.drivers.linux_inline_driver import LinuxInlineDriver + + app = App() + async with app.run_test() as pilot: + await pilot.pause() + loop = asyncio.get_running_loop() + done = loop.create_future() + driver = LinuxInlineDriver(app) + with tempfile.TemporaryFile(mode="w+") as output: + # Original synchronous output owner, using a real buffered file. + driver._file = output + driver.write("inline output") + driver.call_after_flush(lambda: done.set_result(( + os.pread(output.fileno(), 13, 0), get_ident(), asyncio.get_running_loop()))) + assert not done.done() + assert await asyncio.wait_for(done, 2) == (b"inline output", get_ident(), loop) + driver.close() + + +async def test_writer_drains_signals_before_close_and_drops_retired_loop(tmp_path: Path): + loop = asyncio.get_running_loop() + retired = asyncio.new_event_loop() + retired.close() + observed = [] + done = loop.create_future() + with (tmp_path / "output.txt").open("w") as output: + writer = WriterThread(output) + writer.start() + try: + writer.write("first") + writer.call_after_flush(lambda: observed.append("retired"), retired) + writer.write(" second") + writer.call_after_flush(lambda: done.set_result((tmp_path / "output.txt").read_text()), loop) + # Joining output does not wait for the App loop callback, which is + # scheduled by the original writer and remains owned by that loop. + writer.stop() + assert not writer.is_alive() + assert await asyncio.wait_for(done, 2) == "first second" + assert observed == [] + finally: + if writer.is_alive(): + writer.stop() diff --git a/tests/test_writer_thread.py b/tests/test_writer_thread.py index 7d62444396..6f4d05be63 100644 --- a/tests/test_writer_thread.py +++ b/tests/test_writer_thread.py @@ -1,7 +1,8 @@ """Terminal-write callbacks acknowledge the write and flush, in order.""" +import asyncio from io import StringIO -from threading import Event +from threading import get_ident from textual.drivers._writer_thread import WriterThread @@ -20,16 +21,26 @@ def flush(self): return super().flush() -def test_callback_follows_all_earlier_writes_and_flush(): +async def test_callback_follows_all_earlier_writes_and_flush(): output = RecordingOutput() writer = WriterThread(output) - done = Event() + done = asyncio.Event() + callback_threads = [] + loop = asyncio.get_running_loop() + + def written(): + output.events.append(("callback", output.getvalue())) + callback_threads.append(get_ident()) + done.set() + writer.start() try: writer.write("opening") writer.write(" frame") - writer.call_after_flush(lambda: (output.events.append(("callback", output.getvalue())), done.set())) - assert done.wait(2), "Writer did not acknowledge its flush" + writer.call_after_flush(written, loop) + await asyncio.wait_for(done.wait(), 2) assert output.events.index(("flush", "opening frame")) < output.events.index(("callback", "opening frame")) + assert callback_threads == [get_ident()] + assert writer.ident != callback_threads[0] finally: writer.stop()