From 68475922bb36d8451335146488b40901dc4aa7a2 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 24 Aug 2026 19:28:49 +0200 Subject: [PATCH] fix: restore interactive PTY behavior Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- README.md | 20 ++- SCOPE.md | 20 +-- WORK-RECORDS.md | 2 +- src/tamq/cli.py | 5 +- src/tamq/control.py | 23 +++- src/tamq/ptytap.py | 123 +++++++++++++++--- src/tamq/service.py | 5 +- src/tamq/tmux.py | 16 ++- tests/test_cli.py | 9 +- tests/test_install_target.py | 57 ++++++++ tests/test_ptytap.py | 56 ++++++++ tests/test_ptytap_integration.py | 99 ++++++++++++++ tests/test_service_delivery.py | 17 +++ tests/test_tmux_manager.py | 25 ++++ .../TAMQ-WP-0003-delivery-reliability.md | 10 +- 15 files changed, 437 insertions(+), 50 deletions(-) create mode 100644 tests/test_ptytap.py create mode 100644 tests/test_ptytap_integration.py diff --git a/README.md b/README.md index f5bd23c..870e486 100644 --- a/README.md +++ b/README.md @@ -48,9 +48,20 @@ Repeated starts reuse existing windows and do not launch a second agent in them. `--cmd` remains an alias for `--command`. A quoted command may include arguments, for example `--command 'codex --profile local'`. -For degraded tmux-only use, `tamq start --no-service ...` skips socket endpoint -registration and message delivery. Stop the broker and remove the managed tmux -session explicitly when finished: +An already-running pane keeps the `tamq tap` code that launched it. After +upgrading from an earlier alpha, recreate the managed session once so terminal +mode and resize fixes take effect: + +```bash +tamq stop +tmux kill-session -t tamq +tamq start --command codex railiance-platform activity-core +``` + +For degraded tmux-only use, `tamq start --no-service ...` launches the selected +agent directly, without `tamq tap`, socket endpoint registration, or message +delivery. Stop the broker and remove the managed tmux session explicitly when +finished: ```bash tamq stop @@ -84,7 +95,8 @@ fallback. coordination-engine integration remains a later phase. 1. tmux control mode is the topology and output/control stream; 2. the local broker assigns endpoint/source identity and durably queues intent; 3. `tamq tap` is a full-duplex PTY proxy around the agent process. It forwards - bytes unchanged while observing complete input lines for `@repo:` routing. + bytes unchanged in raw terminal mode, propagates terminal resize and lifecycle + signals, and observes complete input lines for `@repo:` routing. This keeps tmux-specific topology concerns separate from reusable terminal I/O observation and message identity. diff --git a/SCOPE.md b/SCOPE.md index dd71a91..09aec29 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -43,7 +43,7 @@ transport. | Durable, inspectable local queue | Implemented | SQLite history, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, and purge are present. | | Local socket service | Implemented | Peer-credential checks and structured ping/register/send/history/ack/endpoints/disconnect operations are tested. | | Tmux endpoint lifecycle and delivery | Implemented for local alpha | Real isolated-tmux and installed-package tests prove exact repository windows, stable reuse, control-mode injection, durable history transition, service stop, and bounded cleanup. | -| Full-duplex input observation | Implemented, lightly proven | `tamq tap` preserves PTY traffic and observes complete address lines; PTY code coverage is 23%. | +| Full-duplex input observation | Implemented for local alpha | `tamq tap` uses raw outer-terminal mode, copies and propagates window dimensions, forwards lifecycle signals, preserves mouse/control bytes, and observes CR/LF address lines. A real tmux fixture covers geometry, resize, and raw mouse input. | | Bounded retry behavior | Not enforced | Failed injection remains pending and becomes claimable after lease expiry, but no attempt counter or terminal failure state applies the configured cap. | | Acknowledgement policy | Partially implemented | Explicit acknowledgement and the configuration field exist; delivery always marks a successful tmux injection as `injected`, irrespective of `delivery_ack_mode`. | | Coordination-engine interoperability | Not implemented | The adapter contract and implementation remain in `TAMQ-WP-0002`. | @@ -70,20 +70,22 @@ Not yet suitable: - Unattended or high-confidence delivery where bounded retries and positive recipient acknowledgement are required. -- Production-style operation without full PTY lifecycle, crash-recovery, and - stronger process-supervision evidence. +- Production-style operation without longer-running crash/terminal soak tests + and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 63 passing tests and 75% statement coverage. Coverage +The suite currently has 69 passing tests and 72% statement coverage. Coverage is strongest in durable storage and registry handling, and weakest in the PTY -tap and CLI orchestration. The Forgejo CI job installs tmux and uv, runs the -real-tmux and isolated installed-package session tests with a deterministic gita -fixture, and retains CLI help/version smoke checks on Python 3.11. +tap and CLI orchestration; PTY statement coverage increased from 23% to 33%, +while subprocess behavior is primarily proven by the real-tmux test. The +Forgejo CI job installs tmux and uv, runs the real-tmux and isolated +installed-package session tests with a deterministic gita fixture, and retains +CLI help/version smoke checks on Python 3.11. ## Next Usability Gates -- `TAMQ-WP-0003` owns bounded retry state, acknowledgement enforcement, crash - recovery, and deeper PTY lifecycle evidence. +- `TAMQ-WP-0003-T01` and `T02` still own bounded retry state and acknowledgement + enforcement. Its real tmux/PTY lifecycle gate (`T03`) is complete. - `TAMQ-WP-0002` owns the coordination-engine adapter after the local delivery contract is sufficiently reliable. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index d67c7ec..9dd2d31 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -22,7 +22,7 @@ | task | TAMQ-WP-0002-T03 | wait | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | | task | TAMQ-WP-0003-T01 | todo | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | task | TAMQ-WP-0003-T02 | wait | — | workplans/TAMQ-WP-0003-delivery-reliability.md | -| task | TAMQ-WP-0003-T03 | todo | — | workplans/TAMQ-WP-0003-delivery-reliability.md | +| task | TAMQ-WP-0003-T03 | done | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | task | TAMQ-WP-0003-T04 | wait | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | task | TAMQ-WP-0004-T01 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | task | TAMQ-WP-0004-T02 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | diff --git a/src/tamq/cli.py b/src/tamq/cli.py index a40ce7e..8476937 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -204,7 +204,8 @@ def main(argv: list[str] | None = None) -> int: if args.command == "start": manager = TmuxManager() try: - preflight_runtime_paths() + if not args.no_service: + preflight_runtime_paths() launch_plan = manager.preflight(args.repos, args.agent_command) except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) @@ -213,7 +214,7 @@ def main(argv: list[str] | None = None) -> int: print("tamq service failed to start; use --no-service to open repos without messaging", file=sys.stderr) return 1 try: - endpoint = manager.ensure_plan(launch_plan) + endpoint = manager.ensure_plan(launch_plan, tap=not args.no_service) except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) return 2 diff --git a/src/tamq/control.py b/src/tamq/control.py index 393c8a2..d2df3da 100644 --- a/src/tamq/control.py +++ b/src/tamq/control.py @@ -16,16 +16,29 @@ class ControlModeClient: tmux_command: Sequence[str] | None = None process: subprocess.Popen[str] | None = None - def start(self) -> None: - if self.process is not None: - return + def _command(self) -> tuple[str, ...]: socket_name = os.environ.get("TAMQ_TMUX_SOCKET") - command = tuple( + return tuple( self.tmux_command or (["tmux", "-L", socket_name] if socket_name else ["tmux"]) ) + + def session_exists(self, expected_pid: int | None = None) -> bool: + result = subprocess.run( + [*self._command(), "display-message", "-p", "-t", self.session, "#{pid}"], + text=True, + capture_output=True, + check=False, + ) + if result.returncode != 0: + return False + return expected_pid is None or result.stdout.strip() == str(expected_pid) + + def start(self) -> None: + if self.process is not None: + return self.process = subprocess.Popen( - [*command, "-C", "attach-session", "-t", self.session], + [*self._command(), "-C", "attach-session", "-t", self.session], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, diff --git a/src/tamq/ptytap.py b/src/tamq/ptytap.py index 50a9208..15ac5a3 100644 --- a/src/tamq/ptytap.py +++ b/src/tamq/ptytap.py @@ -1,10 +1,13 @@ from __future__ import annotations +import fcntl import os import pty import select import signal import sys +import termios +import tty from collections.abc import Callable, Sequence from .broker import InputBroker @@ -18,14 +21,70 @@ class PtyTap: self.broker = broker self.on_line = on_line + def _observe_input(self, buffer: bytearray, data: bytes) -> None: + buffer.extend(data) + while True: + delimiters = [index for marker in (b"\r", b"\n") if (index := buffer.find(marker)) >= 0] + if not delimiters: + return + index = min(delimiters) + raw = bytes(buffer[:index]) + del buffer[: index + 1] + line = raw.decode("utf-8", errors="replace") + routed = self.broker.inspect_line(line) + if routed and self.on_line: + self.on_line(line) + def run(self) -> int: - pid, master = pty.fork() + stdin_fd = sys.stdin.fileno() + stdout_fd = sys.stdout.fileno() + master, slave = pty.openpty() + copy_winsize(stdin_fd, slave) + pid = os.fork() if pid == 0: - os.execvp(self.command[0], self.command) + try: + os.close(master) + os.setsid() + fcntl.ioctl(slave, termios.TIOCSCTTY, 0) + for fd in (0, 1, 2): + os.dup2(slave, fd) + if slave > 2: + os.close(slave) + os.execvp(self.command[0], self.command) + except BaseException as exc: + os.write(2, f"tamq tap: cannot start {self.command[0]}: {exc}\n".encode()) + os._exit(127) + os.close(slave) input_buffer = bytearray() + saved_terminal = termios.tcgetattr(stdin_fd) if os.isatty(stdin_fd) else None + saved_handlers: dict[int, signal.Handlers] = {} + input_open = True + + def resize(_signum=None, _frame=None) -> None: + copy_winsize(stdin_fd, master) + + def forward(signum, _frame) -> None: + try: + os.killpg(pid, signum) + except ProcessLookupError: + pass + + status = 0 try: + if saved_terminal is not None: + tty.setraw(stdin_fd) + saved_handlers[signal.SIGWINCH] = signal.signal(signal.SIGWINCH, resize) + for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): + saved_handlers[signum] = signal.signal(signum, forward) + resize() while True: - readable, _, _ = select.select([master, sys.stdin], [], []) + sources = [master] + if input_open: + sources.append(stdin_fd) + try: + readable, _, _ = select.select(sources, [], []) + except InterruptedError: + continue if master in readable: try: data = os.read(master, 65536) @@ -33,23 +92,51 @@ class PtyTap: break if not data: break - os.write(sys.stdout.fileno(), data) - if sys.stdin in readable: - data = os.read(sys.stdin.fileno(), 65536) + write_all(stdout_fd, data) + if input_open and stdin_fd in readable: + data = os.read(stdin_fd, 65536) if not data: - break - os.write(master, data) - input_buffer.extend(data) - while b"\n" in input_buffer: - raw, _, rest = input_buffer.partition(b"\n") - input_buffer = bytearray(rest) - line = raw.decode("utf-8", errors="replace") - routed = self.broker.inspect_line(line) - if routed and self.on_line: - self.on_line(line) + input_open = False + try: + os.killpg(pid, signal.SIGHUP) + except ProcessLookupError: + pass + continue + write_all(master, data) + self._observe_input(input_buffer, data) + except BaseException: + try: + os.killpg(pid, signal.SIGHUP) + except ProcessLookupError: + pass + raise finally: + for signum, handler in saved_handlers.items(): + signal.signal(signum, handler) + if saved_terminal is not None: + termios.tcsetattr(stdin_fd, termios.TCSAFLUSH, saved_terminal) + os.close(master) try: _, status = os.waitpid(pid, 0) - return os.waitstatus_to_exitcode(status) except ChildProcessError: - return 0 + status = 0 + return os.waitstatus_to_exitcode(status) + + +def copy_winsize(source_fd: int, target_fd: int) -> bool: + """Copy terminal rows/columns and pixel dimensions when a source tty exists.""" + try: + size = fcntl.ioctl(source_fd, termios.TIOCGWINSZ, b"\0" * 8) + fcntl.ioctl(target_fd, termios.TIOCSWINSZ, size) + return True + except OSError: + return False + + +def write_all(fd: int, data: bytes) -> None: + view = memoryview(data) + while view: + written = os.write(fd, view) + if written == 0: + raise OSError("terminal write returned zero bytes") + view = view[written:] diff --git a/src/tamq/service.py b/src/tamq/service.py index 8005d5f..bb3e339 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -74,10 +74,13 @@ class Service: def _deliver_once(self) -> None: for endpoint in self.store.endpoints(): repos = json.loads(endpoint["repos"]) + control = ControlModeClient(endpoint["session"]) + if not control.session_exists(expected_pid=int(endpoint["pid"])): + self.store.disconnect_endpoint(endpoint["endpoint_id"]) + continue pending = [row for row in self.store.list(state="pending") if row["endpoint_id"] in (None, endpoint["endpoint_id"]) and row["target_repo"] in repos] if not pending: continue - control = ControlModeClient(endpoint["session"]) try: control.start() for row in pending: diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py index f299deb..6987ada 100644 --- a/src/tamq/tmux.py +++ b/src/tamq/tmux.py @@ -120,7 +120,7 @@ class TmuxManager: def ensure(self, repos: list[str], command: str = "codex") -> Endpoint: return self.ensure_plan(self.preflight(repos, command)) - def ensure_plan(self, plan: LaunchPlan) -> Endpoint: + def ensure_plan(self, plan: LaunchPlan, *, tap: bool = True) -> Endpoint: created_session = False created_windows: list[str] = [] try: @@ -158,11 +158,15 @@ class TmuxManager: self._run("new-window", "-d", "-t", self.session, "-n", repo, "-c", plan.paths[repo]) windows.append(repo) created_windows.append(repo) - tap = [ - *self.tamq_command, - "tap", "--repo", repo, "--endpoint", endpoint_id, "--", *plan.command, - ] - self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(tap), "C-m") + command = ( + [ + *self.tamq_command, + "tap", "--repo", repo, "--endpoint", endpoint_id, "--", *plan.command, + ] + if tap + else list(plan.command) + ) + self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(command), "C-m") self._run("select-window", "-t", f"{self.session}:{plan.repos[0]}") return Endpoint( self.session, diff --git a/tests/test_cli.py b/tests/test_cli.py index 1a7ea13..9b9ff4c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -57,8 +57,9 @@ def test_detached_no_service_start_skips_socket_registration(monkeypatch, capsys assert command == "codex" return "plan" - def ensure_plan(self, plan): + def ensure_plan(self, plan, *, tap=True): assert plan == "plan" + assert tap is False return endpoint monkeypatch.setattr("tamq.cli.TmuxManager", Manager) @@ -88,7 +89,8 @@ def test_interactive_no_service_start_attaches(monkeypatch, capsys): def preflight(self, repos, command): return "plan" - def ensure_plan(self, plan): + def ensure_plan(self, plan, *, tap=True): + assert tap is False return endpoint attached = [] @@ -118,7 +120,8 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys): def preflight(self, repos, command): return "plan" - def ensure_plan(self, plan): + def ensure_plan(self, plan, *, tap=True): + assert tap is True return endpoint def rollback(self, value): diff --git a/tests/test_install_target.py b/tests/test_install_target.py index 5eed6a4..801f89f 100644 --- a/tests/test_install_target.py +++ b/tests/test_install_target.py @@ -161,6 +161,63 @@ def test_isolated_installed_tool_session_smoke(tmp_path): assert status["service"] is True assert [item["endpoint_id"] for item in status["endpoints"]] == [started["instance_id"]] assert run(str(tamq), "stop").returncode == 0 + assert json.loads(run(str(tamq), "status").stdout)["service"] is False + + restarted = json.loads( + run( + str(tamq), + "start", + "--detach", + "--command", + "alpha-agent", + "railiance-platform", + "activity-core", + ).stdout + ) + assert restarted["instance_id"] == started["instance_id"] + assert run( + *tmux, + "list-windows", + "-t", + "tamq", + "-F", + "#{window_name}|#{pane_current_path}|#{pane_pid}", + ).stdout.splitlines() == rows + + run(*tmux, "kill-session", "-t", "tamq") + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + status = json.loads(run(str(tamq), "status").stdout) + if status["endpoints"] == []: + break + time.sleep(0.05) + assert status["endpoints"] == [] + + recovered = json.loads( + run( + str(tamq), + "start", + "--detach", + "--command", + "alpha-agent", + "railiance-platform", + "activity-core", + ).stdout + ) + assert recovered["instance_id"] != started["instance_id"] + recovered_rows = run( + *tmux, + "list-windows", + "-t", + "tamq", + "-F", + "#{window_name}|#{pane_current_path}", + ).stdout.splitlines() + assert recovered_rows == [ + f"railiance-platform|{repo_a}", + f"activity-core|{repo_b}", + ] + assert run(str(tamq), "stop").returncode == 0 finally: subprocess.run( [str(tamq), "stop"], diff --git a/tests/test_ptytap.py b/tests/test_ptytap.py new file mode 100644 index 0000000..28742b3 --- /dev/null +++ b/tests/test_ptytap.py @@ -0,0 +1,56 @@ +import fcntl +import os +import pty +import struct +import termios + +from tamq.ptytap import PtyTap, copy_winsize, write_all + + +class RecordingBroker: + def __init__(self): + self.lines = [] + + def inspect_line(self, line): + self.lines.append(line) + return line.startswith("@") + + +def test_input_observer_accepts_raw_terminal_carriage_returns(): + broker = RecordingBroker() + observed = [] + tap = PtyTap(["true"], broker, on_line=observed.append) + buffer = bytearray() + + tap._observe_input(buffer, b"@activity-core: hel") + tap._observe_input(buffer, b"lo\rplain line\n") + + assert broker.lines == ["@activity-core: hello", "plain line"] + assert observed == ["@activity-core: hello"] + + +def test_copy_winsize_preserves_rows_columns_and_pixels(): + source_master, source_slave = pty.openpty() + target_master, target_slave = pty.openpty() + expected = struct.pack("HHHH", 41, 103, 900, 1600) + try: + fcntl.ioctl(source_slave, termios.TIOCSWINSZ, expected) + assert copy_winsize(source_slave, target_master) is True + actual = fcntl.ioctl(target_slave, termios.TIOCGWINSZ, b"\0" * 8) + assert actual == expected + finally: + for fd in (source_master, source_slave, target_master, target_slave): + os.close(fd) + + +def test_write_all_retries_partial_writes(monkeypatch): + writes = [] + + def partial_write(fd, data): + chunk = bytes(data[:2]) + writes.append((fd, chunk)) + return len(chunk) + + monkeypatch.setattr(os, "write", partial_write) + write_all(9, b"abcde") + assert writes == [(9, b"ab"), (9, b"cd"), (9, b"e")] diff --git a/tests/test_ptytap_integration.py b/tests/test_ptytap_integration.py new file mode 100644 index 0000000..7a374ea --- /dev/null +++ b/tests/test_ptytap_integration.py @@ -0,0 +1,99 @@ +import os +import shutil +import sys +import time +from pathlib import Path +from uuid import uuid4 + +import pytest + +from tamq.tmux import LaunchPlan, TmuxManager + + +@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") +def test_tap_propagates_terminal_size_resize_and_raw_mouse_input(tmp_path, monkeypatch): + repo = tmp_path / "activity-core" + repo.mkdir() + fixture = tmp_path / "terminal_fixture.py" + fixture.write_text( + "import os, signal, tty\n" + "def size(prefix):\n" + " value = os.get_terminal_size(0)\n" + " print(f'{prefix} {value.lines}x{value.columns}', flush=True)\n" + "def resized(signum, frame): size('RESIZE')\n" + "signal.signal(signal.SIGWINCH, resized)\n" + "size('INITIAL')\n" + "tty.setraw(0)\n" + "print('READY', flush=True)\n" + "data = b''\n" + "while not data.endswith(b'M'):\n" + " data += os.read(0, 1024)\n" + "print(f'INPUT {data.hex()}', flush=True)\n" + "while True: signal.pause()\n", + encoding="utf-8", + ) + project_src = str(Path(__file__).resolve().parents[1] / "src") + existing_pythonpath = os.environ.get("PYTHONPATH") + monkeypatch.setenv( + "PYTHONPATH", + project_src if not existing_pythonpath else f"{project_src}:{existing_pythonpath}", + ) + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + + socket_name = f"tamq-pty-{os.getpid()}-{uuid4().hex[:8]}" + session = f"tamq-pty-{uuid4().hex[:8]}" + manager = TmuxManager( + session, + tmux_command=("tmux", "-L", socket_name), + tamq_command=(sys.executable, "-m", "tamq.cli"), + ) + plan = LaunchPlan( + ("activity-core",), + {"activity-core": str(repo)}, + (sys.executable, str(fixture)), + ) + + def capture() -> str: + return manager._run("capture-pane", "-p", "-t", f"{session}:activity-core") + + def wait_for(marker: str) -> str: + deadline = time.monotonic() + 5 + output = "" + while time.monotonic() < deadline: + output = capture() + if marker in output: + return output + time.sleep(0.05) + pytest.fail(f"timed out waiting for {marker!r}; pane contained:\n{output}") + + try: + manager.ensure_plan(plan) + output = wait_for("READY") + dimensions = manager._run( + "display-message", + "-p", + "-t", + f"{session}:activity-core", + "#{pane_height}x#{pane_width}", + ) + assert f"INITIAL {dimensions}" in output + + mouse_sequence = "\x1b[<64;10;5M" + manager._run( + "send-keys", "-t", f"{session}:activity-core", "-l", "--", mouse_sequence + ) + output = wait_for(f"INPUT {mouse_sequence.encode().hex()}") + assert f"INPUT {mouse_sequence.encode().hex()}" in output + + manager._run("resize-window", "-t", session, "-x", "100", "-y", "40") + resized = manager._run( + "display-message", + "-p", + "-t", + f"{session}:activity-core", + "#{pane_height}x#{pane_width}", + ) + output = wait_for(f"RESIZE {resized}") + assert f"RESIZE {resized}" in output + finally: + manager._run("kill-server", check=False) diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py index 5ec4ea7..c90baa0 100644 --- a/tests/test_service_delivery.py +++ b/tests/test_service_delivery.py @@ -13,6 +13,9 @@ class FakeControl: def start(self): return None + def session_exists(self, expected_pid=None): + return True + def inject(self, window, text): self.injected.append((window, text)) @@ -29,3 +32,17 @@ def test_service_delivery_injects_pending_messages(tmp_path, monkeypatch): assert FakeControl.instances[-1].injected == [("tamq:repo-b", "#repo-a: continue")] assert store.list()[0]["message_id"] == message_id assert store.list()[0]["state"] == "injected" + + +def test_service_disconnects_disappeared_tmux_endpoint(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"]) + + class MissingControl(FakeControl): + def session_exists(self, expected_pid=None): + assert expected_pid == 9 + return False + + monkeypatch.setattr("tamq.service.ControlModeClient", MissingControl) + Service(store=store)._deliver_once() + assert store.endpoints() == [] diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py index bf7b2ee..e56782c 100644 --- a/tests/test_tmux_manager.py +++ b/tests/test_tmux_manager.py @@ -93,6 +93,31 @@ def test_preflight_rejects_foreign_session(tmp_path, monkeypatch): manager.preflight(["a"]) +def test_tmux_only_plan_starts_agent_without_tap(tmp_path, monkeypatch): + manager = tmux.TmuxManager("tamq-test") + calls = [] + plan = tmux.LaunchPlan(("a",), {"a": str(tmp_path)}, ("codex", "--quiet")) + + def run(*args, check=True): + calls.append(args) + if args[:2] == ("display-message", "-p"): + return "42" + if args[:2] == ("list-windows", "-t"): + return "__tamq_boot" + return "" + + monkeypatch.setattr(manager, "_run", run) + monkeypatch.setattr( + tmux.subprocess, + "run", + lambda *args, **kwargs: type("R", (), {"returncode": 1})(), + ) + manager.ensure_plan(plan, tap=False) + send = next(call for call in calls if call and call[0] == "send-keys") + assert "codex --quiet" in send + assert "tamq tap" not in send + + def test_preflight_rejects_missing_agent(tmp_path, monkeypatch): repo = tmp_path / "a" repo.mkdir() diff --git a/workplans/TAMQ-WP-0003-delivery-reliability.md b/workplans/TAMQ-WP-0003-delivery-reliability.md index 17d7787..13cc3ec 100644 --- a/workplans/TAMQ-WP-0003-delivery-reliability.md +++ b/workplans/TAMQ-WP-0003-delivery-reliability.md @@ -48,7 +48,7 @@ and late-acknowledgement behavior. This task follows the state model from T01. ```task id: TAMQ-WP-0003-T03 -status: todo +status: done priority: high state_hub_task_id: "9a3d84b4-55a5-5d18-b4ee-99ac4f111875" ``` @@ -58,6 +58,14 @@ service shutdown, endpoint disappearance, and restart recovery without starting an external coding agent. Keep fast unit tests while adding a bounded integration suite. +Completed with a real isolated-tmux PTY fixture that proves initial dimensions, +`SIGWINCH` resize propagation, raw mouse-sequence forwarding without newline +buffering, and bounded cleanup. The installed-package lifecycle smoke now also +proves service shutdown/restart, stable pane reuse, disappeared-session endpoint +retirement, and clean recovery into a new tmux instance. The PTY proxy restores +terminal state and forwards termination signals; `--no-service` bypasses the tap +entirely as documented. + ## Align CI and operator documentation ```task