From 9fcfba3c1703e7345eff09a8282a0ad9e236615e Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 24 Aug 2026 22:23:57 +0200 Subject: [PATCH] feat: display messages as terminal output Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- README.md | 37 +++++++--- SCOPE.md | 15 +++-- WORK-RECORDS.md | 12 ++-- src/tamq/cli.py | 58 ++++++++-------- src/tamq/control.py | 12 ++++ src/tamq/service.py | 35 ++++++++-- src/tamq/store.py | 30 +++++++-- src/tamq/terminal.py | 67 +++++++++++++++++++ tests/test_cli.py | 13 +++- tests/test_cli_helpers.py | 22 +++++- tests/test_control.py | 19 +++++- tests/test_install_target.py | 21 +++--- tests/test_service_delivery.py | 55 +++++++++++++++ tests/test_store.py | 18 ++++- tests/test_terminal_output.py | 45 +++++++++++++ ...Q-WP-0007-terminal-output-notifications.md | 36 ++++++++-- 16 files changed, 415 insertions(+), 80 deletions(-) create mode 100644 src/tamq/terminal.py create mode 100644 tests/test_terminal_output.py diff --git a/README.md b/README.md index 222a3fd..2c9c303 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,13 @@ tamq start --command 'htop --tree' flex-auth audit-core `--cmd` remains an alias for `--command`. Repeated starts reuse existing windows and never run another initial command in them. +By default, a routed message is also written as sanitized output to its target +pane. Use `--no-display` when you want durable inbox-only delivery: + +```bash +tamq --no-display flex-auth audit-core +``` + ## Exchange messages manually Each managed shell exports its own repository slug as `TAMQ_REPO`. From the @@ -80,6 +87,18 @@ send operation. The long form remains available: tamq send '@audit-core: please review the auth boundary' ``` +With normal `output` delivery, the target pane visibly receives: + +```text +#flex-auth: please review the auth boundary [m-...] +``` + +This uses the pane's tmux-reported `/dev/pts/` device—the same Unix +terminal-output mechanism underlying tools such as `write(1)`. It does not use +`send-keys`, send Enter, or place bytes on the foreground process's stdin. +Output can visually interleave with a prompt and a full-screen program may +redraw over it, so the durable inbox remains authoritative. + In the `audit-core` window, inspect and acknowledge it: ```bash @@ -107,13 +126,13 @@ it and all later messages pending. Filters never run in the background and cannot be combined with `--all` or `--json`. Outside a managed window, use `tamq inbox --repo audit-core` and optionally -`--json`. Manual messages remain durable and pending until acknowledged. They -are never injected as terminal keystrokes, so they cannot corrupt a command -being typed in the target pane. +`--json`. Displayed messages remain durable and pending until acknowledged. +Neither normal output delivery nor `--no-display` injects terminal keystrokes, +so they cannot execute or alter a command being typed in the target pane. After upgrading from an earlier alpha, recreate the managed session once so existing panes inherit the neutral shell contract, repository command `PATH`, -and manual broker registration: +and terminal-output broker registration: ```bash tamq stop @@ -161,10 +180,12 @@ later phase. This keeps tmux-specific topology concerns separate from reusable terminal I/O observation and message identity. -Neutral endpoints use manual delivery: messages stay in the durable inbox and -never become pane input. The legacy pane-delivery path is available only with -the explicit `tamq start --tap --command ...` opt-in; it remains subject to the -retry and acknowledgement limitations tracked by `TAMQ-WP-0003`. +Normal endpoints use terminal-output delivery: each message is written once to +the target pane's PTY output and stays pending in the durable inbox until +acknowledged. `--no-display` selects inbox-only manual mode. Neither becomes +pane input. The legacy pane-input path is available only with the explicit +`tamq start --tap --command ...` opt-in; it remains subject to the retry and +acknowledgement limitations tracked by `TAMQ-WP-0003`. The visible endpoint label is `tmux-amq-`; each boot also receives an instance nonce so PID reuse cannot collide with prior leases or receipts. diff --git a/SCOPE.md b/SCOPE.md index 10e7841..d8c6282 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -19,6 +19,8 @@ tamq does not choose or infer them. - Durable manual send/inbox/acknowledgement with per-window repository identity, shell-native address commands, comment-safe display, and explicit pull-time filters. +- Sanitized one-time output notifications through target tmux pane PTYs, with + inbox-only delivery as an explicit option and no foreground-process input. - Explicit opt-in control-mode pane delivery and the full-duplex `tamq tap` PTY broker for integration experiments. - Exact `gita` repository validation and direct `@repo: message` routing. @@ -50,10 +52,10 @@ transport. | Durable, inspectable local queue | Implemented | SQLite history, manual inbox, 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. | | Neutral tmux session lifecycle | Implemented for local alpha | Repository-first startup opens untouched shells at exact gita paths, exports per-window identity, and runs no initial command unless `--command` is explicit. Stable reuse, service restart, and cleanup are covered by the installed-package test. | -| Safe manual messaging | Implemented for local alpha | Manual endpoints never inject pending messages into panes. Installed testing proves shell-native send, comment-safe inbox, explicit filter/ack exchange, and an unchanged target pane; legacy rows migrate to manual mode. | +| Safe manual messaging | Implemented for local alpha | Normal endpoints write one sanitized comment to the target PTY output without injecting stdin; messages remain pending until acknowledgement. Inbox-only manual mode is explicit with `--no-display`, and legacy rows migrate to manual mode. | | Full-duplex input observation | Explicit opt-in | `--tap --command ...` enables the PTY integration path. It is absent from neutral startup and remains covered for geometry, resize, raw mouse input, and lifecycle behavior. | -| 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`. | +| Bounded retry behavior | Not enforced | Failed output or 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 | Terminal output remains pending until explicit acknowledgement, while legacy pane injection becomes `injected`; the configured `delivery_ack_mode` does not yet govern both paths. | | Coordination-engine interoperability | Not implemented | The adapter contract and implementation remain in `TAMQ-WP-0002`. | ## Practical Usability @@ -62,8 +64,9 @@ The terminal-neutral alpha path was exercised successfully on 2026-08-24 with an isolated installed tool. Repository-first startup created two ordinary shells at exact gita paths without sending initial keystrokes. The test proved per-window repository identity, stable reuse, shell-native addressing, -comment-safe inbox/filter/ack exchange, zero target-pane mutation, service -restart, endpoint disappearance, explicit initial-command startup, and cleanup. +comment-safe target output and inbox/filter/ack exchange, zero target-input +mutation, service restart, endpoint disappearance, explicit initial-command +startup, and cleanup. Suitable today: @@ -83,7 +86,7 @@ Not yet suitable: and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 85 passing tests and 75% statement coverage. Coverage +The suite currently has 93 passing tests and 76% statement coverage. Coverage is strongest in durable storage and registry handling, and weakest in the PTY tap and CLI orchestration; PTY statement coverage increased from 23% to 33%, while subprocess behavior is primarily proven by the real-tmux test. The diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 3a99fc8..2a6edae 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -15,7 +15,7 @@ | workplan | TAMQ-WP-0004 | finished | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | workplan | TAMQ-WP-0005 | finished | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | | workplan | TAMQ-WP-0006 | finished | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | -| workplan | TAMQ-WP-0007 | active | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| workplan | TAMQ-WP-0007 | finished | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | | task | TAMQ-WP-ADHOC-2026-08-24-T01 | done | — | workplans/ADHOC-2026-08-24.md | | task | TAMQ-WP-0001-T01 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | | task | TAMQ-WP-0001-T02 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | @@ -41,8 +41,8 @@ | task | TAMQ-WP-0006-T03 | done | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | | task | TAMQ-WP-0006-T04 | done | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | | task | TAMQ-WP-0006-T05 | done | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | -| task | TAMQ-WP-0007-T01 | progress | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | -| task | TAMQ-WP-0007-T02 | todo | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | -| task | TAMQ-WP-0007-T03 | todo | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | -| task | TAMQ-WP-0007-T04 | todo | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | -| task | TAMQ-WP-0007-T05 | todo | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| task | TAMQ-WP-0007-T01 | done | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| task | TAMQ-WP-0007-T02 | done | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| task | TAMQ-WP-0007-T03 | done | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| task | TAMQ-WP-0007-T04 | done | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | +| task | TAMQ-WP-0007-T05 | done | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | diff --git a/src/tamq/cli.py b/src/tamq/cli.py index c4a7430..05ecec4 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -27,12 +27,13 @@ from .control import ControlModeClient from .diagnostics import configure from .policy import load_profile from .registry import RegistryError, validate_targets +from .terminal import format_comment SUBCOMMANDS = frozenset( "start attach serve stop status ping inbox history inspect ack send export replay purge completion db-version config tap".split() ) -START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service"}) +START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service", "--no-display"}) GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"}) @@ -99,15 +100,15 @@ def stop_service_process() -> int | None: return pid -def ensure_manual_service() -> bool: - """Ensure the broker understands neutral endpoints before registration.""" +def ensure_service_capabilities(required: set[str]) -> bool: + """Restart an older broker once before registering a newer endpoint mode.""" if not ensure_service(): return False try: capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", []) except (OSError, json.JSONDecodeError): return False - if "manual_delivery" in capabilities: + if required.issubset(capabilities): return True stop_service_process() if not ensure_service(): @@ -116,7 +117,15 @@ def ensure_manual_service() -> bool: capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", []) except (OSError, json.JSONDecodeError): return False - return "manual_delivery" in capabilities + return required.issubset(capabilities) + + +def ensure_manual_service() -> bool: + return ensure_service_capabilities({"manual_delivery"}) + + +def ensure_output_service() -> bool: + return ensure_service_capabilities({"manual_delivery", "terminal_output"}) def preflight_runtime_paths() -> None: @@ -133,24 +142,9 @@ def attach_session(session: str) -> int: return subprocess.run(command, check=False).returncode -def _terminal_safe(text: str) -> str: - return "".join( - character - if character.isprintable() - else f"\\x{ord(character):02x}" - for character in text - ) - - def format_comment_message(row: sqlite3.Row) -> str: """Render a durable message so every displayed line remains a shell comment.""" - sender = _terminal_safe(str(row["sender_repo"])) - body = str(row["body"]) - lines = body.splitlines() or [""] - rendered = [f"#{sender}: {_terminal_safe(lines[0])}"] - rendered.extend(f"# {_terminal_safe(line)}" for line in lines[1:]) - rendered[-1] += f" [{_terminal_safe(str(row['message_id']))}]" - return "\n".join(rendered) + return format_comment(row["sender_repo"], row["body"], row["message_id"]) def build_parser() -> argparse.ArgumentParser: @@ -159,11 +153,12 @@ def build_parser() -> argparse.ArgumentParser: description="Repository-aware tmux sessions with durable local messaging.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""session shorthand: - tamq [--detach] [--command COMMAND] REPO [REPO ...] + tamq [--detach] [--command COMMAND] [--no-display] REPO [REPO ...] With no --command, tamq opens ordinary repository shells. --command is an - explicit initial command for newly created windows; it does not opt in to - terminal observation or message injection. + explicit initial command for newly created windows. Messages appear as + sanitized terminal output by default; --no-display keeps them inbox-only. + Neither mode injects message bytes into pane input. manual messaging from a managed shell: @TARGET: MESSAGE... @@ -180,6 +175,7 @@ manual messaging from a managed shell: start.add_argument("repos", nargs="*", help="gita-registered repository slugs") start.add_argument("--command", "--cmd", dest="initial_command", default=None, help="explicit initial command for newly created windows (default: ordinary shell)") start.add_argument("--tap", action="store_true", help="explicitly opt in to PTY input observation and pane message injection") + start.add_argument("--no-display", action="store_true", help="keep messages in the durable inbox without writing target terminal output") start.add_argument("--detach", action="store_true", help="detach after startup") start.add_argument("--no-service", action="store_true", help="open tmux windows without service registration or messaging") attach = subparsers.add_parser("attach", help="attach to the managed tmux session") @@ -312,6 +308,9 @@ def main(argv: list[str] | None = None) -> int: if args.tap and args.no_service: print("tamq: --tap cannot be combined with --no-service", file=sys.stderr) return 2 + if args.tap and args.no_display: + print("tamq: --tap cannot be combined with --no-display", file=sys.stderr) + return 2 manager = TmuxManager() try: if not args.no_service: @@ -320,8 +319,11 @@ def main(argv: list[str] | None = None) -> int: except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) return 2 - if not args.no_service and not ensure_manual_service(): - print("tamq service failed to start with manual-delivery safety; use --no-service to open repos without messaging", file=sys.stderr) + service_ready = ( + ensure_manual_service() if args.no_display else ensure_output_service() + ) if not args.no_service else True + if not service_ready: + print("tamq service failed to start with the required delivery capability; use --no-service to open repos without messaging", file=sys.stderr) return 1 try: endpoint = manager.ensure_plan(launch_plan, tap=args.tap) @@ -332,7 +334,7 @@ def main(argv: list[str] | None = None) -> int: registered = False if not args.no_service: try: - delivery_mode = "pane" if args.tap else "manual" + delivery_mode = "pane" if args.tap else ("manual" if args.no_display else "output") registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos, "delivery_mode": delivery_mode})) if not registration.get("ok"): raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}") @@ -360,7 +362,7 @@ def main(argv: list[str] | None = None) -> int: "delivered": delivered, "service": not args.no_service, "messaging": registered, - "delivery_mode": "none" if args.no_service else ("pane" if args.tap else "manual"), + "delivery_mode": "none" if args.no_service else ("pane" if args.tap else ("manual" if args.no_display else "output")), } print(json.dumps(summary), flush=True) if args.detach: diff --git a/src/tamq/control.py b/src/tamq/control.py index d2df3da..cd33dc2 100644 --- a/src/tamq/control.py +++ b/src/tamq/control.py @@ -34,6 +34,18 @@ class ControlModeClient: return False return expected_pid is None or result.stdout.strip() == str(expected_pid) + def pane_tty(self, window: str) -> str: + result = subprocess.run( + [*self._command(), "display-message", "-p", "-t", window, "#{pane_tty}"], + text=True, + capture_output=True, + check=False, + ) + tty_path = result.stdout.strip() + if result.returncode != 0 or not tty_path: + raise ControlModeError(result.stderr.strip() or f"cannot resolve pane tty: {window}") + return tty_path + def start(self) -> None: if self.process is not None: return diff --git a/src/tamq/service.py b/src/tamq/service.py index 28afcf3..f57d032 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -13,6 +13,7 @@ from .config import db_path from .store import Store from .registry import RegistryError, validate_targets from .control import ControlModeClient +from .terminal import terminal_frame, write_terminal_output PROTOCOL_VERSION = "0.1" @@ -59,7 +60,7 @@ class Service: self.store.close() async def _delivery_loop(self, stopped: asyncio.Event) -> None: - """Continuously inject pending messages into registered live endpoints.""" + """Continuously deliver pending messages to registered live endpoints.""" while not stopped.is_set(): try: self._deliver_once() @@ -78,22 +79,42 @@ class Service: if not control.session_exists(expected_pid=int(endpoint["pid"])): self.store.disconnect_endpoint(endpoint["endpoint_id"]) continue - if endpoint["delivery_mode"] != "pane": + delivery_mode = endpoint["delivery_mode"] + if delivery_mode == "manual": 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] + 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 + and (delivery_mode == "pane" or row["displayed_at"] is None) + ] if not pending: continue try: - control.start() + if delivery_mode == "pane": + control.start() for row in pending: lease = self.store.claim(row["message_id"], endpoint["endpoint_id"]) if lease is None: continue try: - control.inject(f'{endpoint["session"]}:{row["target_repo"]}', f'#{row["sender_repo"]}: {row["body"]}') + target = f'{endpoint["session"]}:{row["target_repo"]}' + if delivery_mode == "output": + tty_path = control.pane_tty(target) + write_terminal_output( + tty_path, + terminal_frame( + row["sender_repo"], row["body"], row["message_id"] + ), + ) + else: + control.inject(target, f'#{row["sender_repo"]}: {row["body"]}') except Exception: continue - self.store.release(row["message_id"], lease, "injected") + if delivery_mode == "output": + self.store.mark_displayed(row["message_id"], lease) + else: + self.store.release(row["message_id"], lease, "injected") finally: control.close() @@ -109,7 +130,7 @@ class Service: if protocol.split(".")[0] != PROTOCOL_VERSION.split(".")[0]: response = {"ok": False, "error": f"incompatible protocol: {protocol}", "protocol": PROTOCOL_VERSION} elif op == "ping": - response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery"]} + response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery", "terminal_output"]} elif op == "register": required = ("endpoint_id", "pid", "session", "repos") if any(key not in request for key in required): diff --git a/src/tamq/store.py b/src/tamq/store.py index f3ac634..59e4a2a 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -7,7 +7,7 @@ from pathlib import Path from typing import Iterable from uuid import uuid4 -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 class Store: @@ -30,7 +30,8 @@ class Store: endpoint_id TEXT, provenance TEXT, injected_at REAL, - acknowledged_at REAL + acknowledged_at REAL, + displayed_at REAL ); CREATE INDEX IF NOT EXISTS messages_target_state ON messages(target_repo, state); CREATE TABLE IF NOT EXISTS endpoints ( @@ -57,6 +58,11 @@ class Store: self.db.execute( "ALTER TABLE endpoints ADD COLUMN delivery_mode TEXT NOT NULL DEFAULT 'manual'" ) + message_columns = { + row["name"] for row in self.db.execute("PRAGMA table_info(messages)") + } + if "displayed_at" not in message_columns: + self.db.execute("ALTER TABLE messages ADD COLUMN displayed_at REAL") self.db.execute( "INSERT INTO metadata(key,value) VALUES('schema_version',?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", @@ -76,7 +82,7 @@ class Store: delivery_mode: str = "manual", ) -> None: import json - if delivery_mode not in {"manual", "pane"}: + if delivery_mode not in {"manual", "output", "pane"}: raise ValueError(f"invalid delivery mode: {delivery_mode}") self.db.execute( "UPDATE endpoints SET disconnected_at=strftime('%s','now') " @@ -131,8 +137,9 @@ class Store: raise ValueError("message body exceeds 8 KiB limit") message_id = f"m-{uuid4()}" self.db.execute( - "INSERT INTO messages VALUES(?,?,?,?,?,?,?,?,?,?)", - (message_id, sender, target, body, time.time(), "pending", endpoint, provenance, None, None), + "INSERT INTO messages(message_id,sender_repo,target_repo,body,created_at,state,endpoint_id,provenance,injected_at,acknowledged_at,displayed_at) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?)", + (message_id, sender, target, body, time.time(), "pending", endpoint, provenance, None, None, None), ) self.db.commit() return message_id @@ -161,6 +168,19 @@ class Store: self.set_state(message_id, "acknowledged") return True + def mark_displayed(self, message_id: str, lease_id: str) -> bool: + with self.db: + result = self.db.execute( + "DELETE FROM leases WHERE message_id=? AND lease_id=?", + (message_id, lease_id), + ) + if result.rowcount: + self.db.execute( + "UPDATE messages SET displayed_at=? WHERE message_id=?", + (time.time(), message_id), + ) + return result.rowcount == 1 + def claim(self, message_id: str, endpoint_id: str, ttl: float = 30.0) -> str | None: now = time.time() lease_id = f"lease-{uuid4()}" diff --git a/src/tamq/terminal.py b/src/tamq/terminal.py new file mode 100644 index 0000000..ea190ed --- /dev/null +++ b/src/tamq/terminal.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import os +import re +import stat +from pathlib import Path + + +class TerminalOutputError(RuntimeError): + pass + + +def terminal_safe(text: str) -> str: + """Escape terminal controls while retaining printable Unicode verbatim.""" + return "".join( + character + if character.isprintable() + else f"\\x{ord(character):02x}" + for character in text + ) + + +def format_comment(sender: str, body: str, message_id: str) -> str: + """Render a message so every line is visibly comment-prefixed.""" + lines = body.splitlines() or [""] + rendered = [f"#{terminal_safe(sender)}: {terminal_safe(lines[0])}"] + rendered.extend(f"# {terminal_safe(line)}" for line in lines[1:]) + rendered[-1] += f" [{terminal_safe(message_id)}]" + return "\n".join(rendered) + + +def terminal_frame(sender: str, body: str, message_id: str) -> str: + """Frame asynchronous output away from the current visual input line.""" + content = format_comment(sender, body, message_id).replace("\n", "\r\n") + return f"\r\n{content}\r\n" + + +def write_terminal_output(tty_path: str | Path, text: str) -> None: + """Write output to a tmux pane PTY without placing bytes on its stdin.""" + try: + path = Path(tty_path).resolve(strict=True) + except OSError as exc: + raise TerminalOutputError(f"terminal device is unavailable: {tty_path}") from exc + if re.fullmatch(r"/dev/pts/[0-9]+", str(path)) is None: + raise TerminalOutputError(f"refusing non-PTY terminal path: {path}") + flags = os.O_WRONLY | os.O_NOCTTY | os.O_NONBLOCK | os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise TerminalOutputError(f"cannot open terminal output: {path}") from exc + try: + details = os.fstat(descriptor) + if not stat.S_ISCHR(details.st_mode) or details.st_uid != os.getuid(): + raise TerminalOutputError(f"terminal device is not owned by this user: {path}") + payload = text.encode("utf-8") + offset = 0 + while offset < len(payload): + written = os.write(descriptor, payload[offset:]) + if written <= 0: + raise TerminalOutputError(f"terminal output made no progress: {path}") + offset += written + except OSError as exc: + raise TerminalOutputError(f"cannot write terminal output: {path}") from exc + finally: + os.close(descriptor) diff --git a/tests/test_cli.py b/tests/test_cli.py index e3ef84b..d981dd3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -18,7 +18,7 @@ def test_root_help_exposes_repository_shorthand_and_command(capsys): main(["--help"]) assert exc.value.code == 0 output = capsys.readouterr().out - assert "tamq [--detach] [--command COMMAND] REPO [REPO ...]" in output + assert "tamq [--detach] [--command COMMAND] [--no-display] REPO" in output assert "With no --command" in output assert "@TARGET: MESSAGE" in output @@ -44,10 +44,12 @@ def test_start_parser_supports_command_and_cmd_alias(): canonical = parser.parse_args(["start", "--command", "codex --quiet", "a", "b"]) compatibility = parser.parse_args(["start", "--cmd", "claude", "a"]) neutral = parser.parse_args(["start", "a", "b"]) + inbox_only = parser.parse_args(["start", "--no-display", "a"]) assert canonical.initial_command == "codex --quiet" assert canonical.repos == ["a", "b"] assert compatibility.initial_command == "claude" assert neutral.initial_command is None + assert inbox_only.no_display is True def test_repository_first_and_command_first_start_shorthand(): @@ -62,6 +64,11 @@ def test_repository_first_and_command_first_start_shorthand(): "codex", "flex-auth", ] + assert normalize_argv(["--no-display", "flex-auth"]) == [ + "start", + "--no-display", + "flex-auth", + ] assert normalize_argv(["status"]) == ["status"] assert normalize_argv(["--verbose", "flex-auth", "audit-core"]) == [ "--verbose", @@ -165,7 +172,7 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys): monkeypatch.setattr("tamq.cli.TmuxManager", Manager) monkeypatch.setattr("tamq.cli.preflight_runtime_paths", lambda: None) - monkeypatch.setattr("tamq.cli.ensure_manual_service", lambda: True) + monkeypatch.setattr("tamq.cli.ensure_output_service", lambda: True) monkeypatch.setattr("tamq.cli.request", failed_registration) assert main(["start", "--detach", "a"]) == 1 assert rolled_back == [endpoint] @@ -177,3 +184,5 @@ def test_tap_requires_explicit_command_and_service(capsys): assert "--tap requires an explicit --command" in capsys.readouterr().err assert main(["start", "--tap", "--no-service", "--command", "sh", "a"]) == 2 assert "--tap cannot be combined with --no-service" in capsys.readouterr().err + assert main(["start", "--tap", "--no-display", "--command", "sh", "a"]) == 2 + assert "--tap cannot be combined with --no-display" in capsys.readouterr().err diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index 46f0337..2f30ac4 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -1,4 +1,4 @@ -from tamq.cli import ensure_manual_service, parse_size +from tamq.cli import ensure_manual_service, ensure_output_service, parse_size def test_parse_size(): @@ -22,3 +22,23 @@ def test_manual_service_restarts_legacy_broker(monkeypatch): assert ensure_manual_service() is True assert len(starts) == 2 assert stops == [True] + + +def test_output_service_requires_terminal_output_capability(monkeypatch): + capabilities = iter([ + ["manual_delivery"], + ["manual_delivery", "terminal_output"], + ]) + starts = [] + stops = [] + + async def request(payload): + return {"capabilities": next(capabilities)} + + monkeypatch.setattr("tamq.cli.ensure_service", lambda: starts.append(True) or True) + monkeypatch.setattr("tamq.cli.stop_service_process", lambda: stops.append(True) or 42) + monkeypatch.setattr("tamq.cli.request", request) + + assert ensure_output_service() is True + assert len(starts) == 2 + assert stops == [True] diff --git a/tests/test_control.py b/tests/test_control.py index e0b75ac..cac6947 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -1,4 +1,5 @@ -from tamq.control import shell_quote +from tamq import control +from tamq.control import ControlModeClient, shell_quote from tamq.tmux import shell_join @@ -10,3 +11,19 @@ def test_shell_quote(): def test_tmux_shell_join_quotes_arguments(): assert shell_join(["tamq", "tap", "--repo", "net-kingdom", "--", "codex"]) == "tamq tap --repo net-kingdom -- codex" + + +def test_pane_tty_is_resolved_through_tmux(monkeypatch): + calls = [] + + def run(command, **kwargs): + calls.append(command) + return type("Result", (), {"returncode": 0, "stdout": "/dev/pts/7\n", "stderr": ""})() + + monkeypatch.setattr(control.subprocess, "run", run) + client = ControlModeClient("tamq", tmux_command=("tmux", "-L", "test")) + assert client.pane_tty("tamq:audit-core") == "/dev/pts/7" + assert calls == [[ + "tmux", "-L", "test", "display-message", "-p", "-t", + "tamq:audit-core", "#{pane_tty}", + ]] diff --git a/tests/test_install_target.py b/tests/test_install_target.py index 3991e22..429bee5 100644 --- a/tests/test_install_target.py +++ b/tests/test_install_target.py @@ -92,7 +92,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ) assert run(str(tamq), "--version").stdout.strip() == "0.1.0" root_help = run(str(tamq), "--help").stdout - assert "tamq [--detach] [--command COMMAND] REPO [REPO ...]" in root_help + assert "tamq [--detach] [--command COMMAND] [--no-display] REPO" in root_help started = json.loads( run( str(tamq), @@ -102,7 +102,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ).stdout ) assert started["repos"] == ["railiance-platform", "activity-core"] - assert started["delivery_mode"] == "manual" + assert started["delivery_mode"] == "output" rows = run( *tmux, "list-windows", @@ -168,7 +168,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): assert [item["endpoint_id"] for item in pre_delivery_status["endpoints"]] == [ started["instance_id"] ] - assert pre_delivery_status["endpoints"][0]["delivery_mode"] == "manual" + assert pre_delivery_status["endpoints"][0]["delivery_mode"] == "output" target_before = run( *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" @@ -183,6 +183,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ) deadline = time.monotonic() + 5 inbox = [] + target_after = target_before while time.monotonic() < deadline: inbox = [ json.loads(line) @@ -190,7 +191,10 @@ def test_isolated_installed_tool_session_smoke(tmp_path): str(tamq), "inbox", "--repo", "activity-core", "--json" ).stdout.splitlines() ] - if inbox: + target_after = run( + *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" + ).stdout + if inbox and inbox[-1]["displayed_at"] is not None and "#railiance-platform: installed-message" in target_after: break time.sleep(0.05) assert inbox, run( @@ -199,14 +203,13 @@ def test_isolated_installed_tool_session_smoke(tmp_path): assert inbox[-1]["sender_repo"] == "railiance-platform" assert inbox[-1]["body"] == "installed-message" assert inbox[-1]["state"] == "pending" + assert inbox[-1]["displayed_at"] is not None message_id = inbox[-1]["message_id"] assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == ( f"#railiance-platform: installed-message [{message_id}]\n" ) - target_after = run( - *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" - ).stdout - assert target_after == target_before + assert target_after != target_before + assert f"#railiance-platform: installed-message [{message_id}]" in target_after assert run(str(tamq), "ack", message_id).stdout.strip() == message_id assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == "" @@ -289,7 +292,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ).stdout ) assert recovered["instance_id"] != started["instance_id"] - assert recovered["delivery_mode"] == "manual" + assert recovered["delivery_mode"] == "output" recovered_rows = run( *tmux, "list-windows", diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py index a5bfc19..c59333d 100644 --- a/tests/test_service_delivery.py +++ b/tests/test_service_delivery.py @@ -19,6 +19,9 @@ class FakeControl: def inject(self, window, text): self.injected.append((window, text)) + def pane_tty(self, window): + return f"/dev/pts/{window.rsplit(':', 1)[-1]}" + def close(self): return None @@ -46,6 +49,58 @@ def test_service_never_injects_manual_endpoint_messages(tmp_path, monkeypatch): assert store.list()[0]["state"] == "pending" +def test_service_writes_output_once_and_retains_pending_ack(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "output") + message_id = store.add("repo-a", "repo-b", "continue") + writes = [] + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + monkeypatch.setattr( + "tamq.service.write_terminal_output", + lambda path, text: writes.append((path, text)), + ) + service = Service(store=store) + + service._deliver_once() + service._deliver_once() + + assert writes == [ + ( + "/dev/pts/repo-b", + f"\r\n#repo-a: continue [{message_id}]\r\n", + ) + ] + row = store.list()[0] + assert row["state"] == "pending" + assert row["displayed_at"] is not None + assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0 + + +def test_failed_terminal_output_remains_undisplayed_and_retryable(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "output") + store.add("repo-a", "repo-b", "continue") + attempts = [] + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + + def fail_once(path, text): + attempts.append((path, text)) + if len(attempts) == 1: + raise OSError("temporary failure") + + monkeypatch.setattr("tamq.service.write_terminal_output", fail_once) + service = Service(store=store) + + service._deliver_once() + assert store.list()[0]["displayed_at"] is None + store.db.execute("UPDATE leases SET expires_at=0") + store.db.commit() + service._deliver_once() + + assert len(attempts) == 2 + assert store.list()[0]["displayed_at"] is not None + + 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"]) diff --git a/tests/test_store.py b/tests/test_store.py index de43c98..c4cd819 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -41,5 +41,21 @@ def test_store_migrates_legacy_endpoint_rows_to_manual_delivery(tmp_path): assert store.endpoints()[0]["delivery_mode"] == "manual" assert store.db.execute( "SELECT value FROM metadata WHERE key='schema_version'" - ).fetchone()[0] == "2" + ).fetchone()[0] == "3" + assert "displayed_at" in { + row["name"] for row in store.db.execute("PRAGMA table_info(messages)") + } + + +def test_mark_displayed_releases_lease_without_acknowledging(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add("a", "b", "hello") + lease_id = store.claim(message_id, "endpoint") + assert lease_id is not None + + assert store.mark_displayed(message_id, lease_id) is True + row = store.list()[0] + assert row["state"] == "pending" + assert row["displayed_at"] is not None + assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0 store.close() diff --git a/tests/test_terminal_output.py b/tests/test_terminal_output.py new file mode 100644 index 0000000..4501ccd --- /dev/null +++ b/tests/test_terminal_output.py @@ -0,0 +1,45 @@ +import os +import pty +import select + +import pytest + +from tamq.terminal import ( + TerminalOutputError, + format_comment, + terminal_frame, + write_terminal_output, +) + + +def test_comment_format_escapes_controls_and_prefixes_every_line(): + assert format_comment("repo-a", "first\nsecond\x1b[31m", "m-1") == ( + "#repo-a: first\n# second\\x1b[31m [m-1]" + ) + assert terminal_frame("repo-a", "hello", "m-1") == ( + "\r\n#repo-a: hello [m-1]\r\n" + ) + + +def test_terminal_output_reaches_pty_output_but_not_input(): + master, slave = pty.openpty() + try: + tty_path = os.ttyname(slave) + write_terminal_output(tty_path, terminal_frame("repo-a", "hello", "m-1")) + + readable, _, _ = select.select([master], [], [], 1) + assert readable == [master] + assert b"#repo-a: hello [m-1]" in os.read(master, 4096) + readable_input, _, _ = select.select([slave], [], [], 0) + assert readable_input == [] + finally: + os.close(master) + os.close(slave) + + +def test_terminal_output_rejects_non_pty_paths(tmp_path): + target = tmp_path / "ordinary-file" + target.write_text("untouched", encoding="utf-8") + with pytest.raises(TerminalOutputError, match="non-PTY"): + write_terminal_output(target, "message") + assert target.read_text(encoding="utf-8") == "untouched" diff --git a/workplans/TAMQ-WP-0007-terminal-output-notifications.md b/workplans/TAMQ-WP-0007-terminal-output-notifications.md index 5b4cd43..4f4c47d 100644 --- a/workplans/TAMQ-WP-0007-terminal-output-notifications.md +++ b/workplans/TAMQ-WP-0007-terminal-output-notifications.md @@ -4,7 +4,7 @@ type: workplan title: "Target-pane terminal output notifications" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social planning_priority: P0 @@ -47,7 +47,7 @@ from being emitted repeatedly after broker polls or restarts. ```task id: TAMQ-WP-0007-T01 -status: progress +status: done priority: high state_hub_task_id: "00972410-bc94-50f7-b7be-8c3c8228a43a" ``` @@ -58,11 +58,15 @@ current user, sanitize terminal control bytes, use bounded non-blocking writes, and document visual interleaving as a display limitation rather than an input safety failure. +Completed with a dedicated terminal-output module, strict tmux PTY path and +ownership validation, non-blocking output-only writes, printable Unicode +retention, terminal-control escaping, and comment-prefixed CRLF framing. + ## Persist one-time display state ```task id: TAMQ-WP-0007-T02 -status: todo +status: done priority: high state_hub_task_id: "e9bc0a58-3488-5122-a899-26dc1b4b2f88" ``` @@ -72,11 +76,16 @@ output, record successful display and release its lease atomically, retain its pending acknowledgement state, and make broker restart/poll behavior idempotent after a successful write. +Completed with SQLite schema v3 and `displayed_at`. Successful output records +the timestamp and releases the lease without changing the pending state; +subsequent polls skip it. Failed writes retain the claim until expiry and then +retry without falsely recording display. + ## Make terminal output the normal managed-session mode ```task id: TAMQ-WP-0007-T03 -status: todo +status: done priority: high state_hub_task_id: "e59a59b0-af2d-570b-9e35-963216dedfa4" ``` @@ -86,11 +95,15 @@ broker capability. Preserve `--tap --command` as explicit pane-input mode, provide `--no-display` for durable inbox-only operation, and restart an older broker before registering the new mode. +Completed with normal `output` registration, `terminal_output` capability +negotiation and legacy-broker restart, explicit `--no-display` manual mode, and +the unchanged explicit `--tap --command` pane-input boundary. + ## Prove output without input ```task id: TAMQ-WP-0007-T04 -status: todo +status: done priority: high state_hub_task_id: "e51146f4-0e44-5de9-87dd-725fe78c792a" ``` @@ -101,11 +114,16 @@ for acknowledgement, emits only once across polls, handles multiline/control content safely, retries failed output, and retains manual and explicit tap modes. +Completed with real PTY proof that output reaches the master while the slave +stdin remains unreadable, invalid-device rejection, control-character and +multiline formatting, successful one-time display, failure/lease retry, schema +migration, and real installed tmux capture evidence. + ## Install and document the operator workflow ```task id: TAMQ-WP-0007-T05 -status: todo +status: done priority: high state_hub_task_id: "af6100d5-ee2c-5bf4-9086-1578dec770ec" ``` @@ -113,3 +131,9 @@ state_hub_task_id: "af6100d5-ee2c-5bf4-9086-1578dec770ec" Update help, README, SCOPE, and installed-package acceptance. Install the verified build, recreate or re-register a live two-repository session, exchange and acknowledge one visible smoke message, and record State Hub evidence. + +Completed with updated CLI help, README, and SCOPE; 93 passing tests at 76% +statement coverage; and an installed live `flex-auth`/`audit-core` session in +`output` mode. Both previously pending user messages and a fresh smoke appeared +in the target pane. The smoke alone was acknowledged; the two user messages +remain pending with persisted display timestamps.