Run a line at a terminal's prompt through pty/run and read how it ended

/pane/<n>/pty/run takes factotum's rpc shape: write one line on an open,
read the answer on that same open. The answer is exit N once the command
ended and the shell is back at a prompt; busy at once when a command runs,
text is typed at the prompt, or the shell has not drawn its first prompt;
error not run when the shell refused the line without running it (fish on
a syntax error keeps it on the prompt, so it is taken back with Ctrl-U);
error shell gone when the pane closes or its shell is replaced; error no
prompt marks for a shell pardes could not instrument.

The end comes from the shell's OSC 133 marks. ghostty parses D's exit
status and drops it, so the stream now runs through a handler that wraps
ghostty's and follows prompt -> input -> running -> done. The marks pardes
injects into bash and fish carry aid=pardes and only those count, so fish
4's own marks (which doubled ours), a nested shell's, and a stray 133;D in
printed output are ignored.

Checked end to end against real bash and fish: false, exit 7, a syntax
error, sleep, busy while running, and the pane closing mid-command.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

1 files changed, 39 insertions, 0 deletions

diff --git a/test/fs.py b/test/fs.py
index 1332c319..ff085a55 100644
--- a/test/fs.py
+++ b/test/fs.py
@@ -290,8 +290,47 @@ def discovery(binary, embedded=False):
     print('9P discovery: listing/stat/find are inert; new, remove, look, exec, name, sel and log behave')
 
 
+def run_file(binary):
+    """pty/run: a line in, how it ended out, on the same open."""
+    def run(c, serial, line):
+        fid = c.open(f'/pane/{serial}/pty/run', 2)
+        try:
+            c.rpc(118, struct.pack('<IQI', fid, 0, len(line)) + line)
+            return c.read_fid(fid)
+        finally:
+            c.close(fid)
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        with session(binary, root, 'run') as (client, address):
+            client.write('/exec', b'Shell bash\n')
+            client.write('/exec', b'Tty\n')
+            term = int(max(r.split()[0] for r in client.read('/index').splitlines() if r.split()[1] == b'term'))
+            deadline = time.monotonic() + 10
+            while run(client, term, b'true\n') == b'busy\n':  # until its first prompt
+                assert time.monotonic() < deadline, 'the shell never reached a prompt'
+                time.sleep(.05)
+            assert run(client, term, b'false\n') == b'exit 1\n'
+            assert run(client, term, b'sh -c "exit 7"\n') == b'exit 7\n'
+            with Client(address) as other:
+                slow = []
+                waiter = threading.Thread(target=lambda: slow.append(run(other, term, b'sleep 1\n')), daemon=True)
+                waiter.start()
+                time.sleep(.3)
+                assert run(client, term, b'true\n') == b'busy\n'
+                waiter.join(5)
+                assert slow == [b'exit 0\n'], slow
+                closing = []
+                waiter = threading.Thread(target=lambda: closing.append(run(other, term, b'sleep 30\n')), daemon=True)
+                waiter.start()
+                time.sleep(.3)
+                client.write(f'/pane/{term}/exec', b'Del\n')
+                waiter.join(5)
+                assert closing == [b'error shell gone\n'], closing
+
+
 def test(binary, quic=False):
     started = time.monotonic()
+    run_file(binary)
     for options, message in [
         (['--fs'], b'no such option'),
         (['--fs9'], b'no such option'),