From 8186550c9a8049fbe1cb63feedb8b6f499ed78b9 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 24 Aug 2026 23:24:18 +0200 Subject: [PATCH] feat: add interactive recipient composer Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- README.md | 26 ++-- SCOPE.md | 6 +- WORK-RECORDS.md | 8 +- src/tamq/cli.py | 38 ++++- src/tamq/composer.py | 131 ++++++++++++++++++ src/tamq/tmux.py | 17 ++- tests/test_composer.py | 101 ++++++++++++++ tests/test_manual_messaging.py | 60 ++++++++ tests/test_tmux_manager.py | 3 + ...-WP-0009-interactive-recipient-composer.md | 22 ++- 10 files changed, 379 insertions(+), 33 deletions(-) create mode 100644 src/tamq/composer.py create mode 100644 tests/test_composer.py diff --git a/README.md b/README.md index 1239f2f..c5be7f8 100644 --- a/README.md +++ b/README.md @@ -77,17 +77,27 @@ The spelling without the trailing colon is equivalent: @audit-core please review the auth boundary ``` -After receiving a message, use the bare `@` command to answer its sender: +Run bare `@` to open the inline composer. It shows the latest sender as the +recipient, and ordinary prose is read directly rather than parsed by the shell: -```bash -@ Thanks, I will take a look. +```text +$ @ +@audit-core: What's up? ``` -Bare `@` selects the sender of the latest durable inbound message for the -current repository, including an already acknowledged message. Self-addressed -messages are ignored. If this window has no prior counterparty, the command -reports an error and queues nothing. The explicit equivalent is `tamq reply -MESSAGE...`. +Press Tab to cycle through the other repository windows while preserving the +draft, Enter to send, or Ctrl-C to cancel without queuing. The initial recipient +is the sender of the latest durable inbound message, including an already +acknowledged message; without history, it is the first peer window. +Self-addressed messages are ignored. + +The one-line fast path remains available, but its text is parsed by the shell +and therefore follows ordinary shell quoting rules: + +```bash +@ "What's up?" +tamq reply "What's up?" +``` These are tamq-owned executable commands beside the installed `tamq` command, not shell-specific aliases. Set `TAMQ_COMMAND_DIR` before startup to select a diff --git a/SCOPE.md b/SCOPE.md index cfceef1..f9d4308 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -18,7 +18,7 @@ tamq does not choose or infer them. - Managed neutral-shell tmux lifecycle and explicit initial commands. - Durable manual send/inbox/acknowledgement with per-window repository identity, shell-native direct and latest-counterparty reply commands, comment-safe - display, and explicit pull-time filters. + display, a recipient-aware inline composer, 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 @@ -48,7 +48,7 @@ transport. | Intent capability | State | Evidence and remaining gap | | --- | --- | --- | -| Direct repository addressing | Implemented for local alpha | Exact `gita` validation, per-session `@repo`/`@repo:` commands, bare `@` latest-counterparty replies, and long-form parsing are covered without modifying shell configuration. | +| Direct repository addressing | Implemented for local alpha | Exact `gita` validation, per-session `@repo`/`@repo:` commands, a bare `@` composer with latest-counterparty default and Tab recipient cycling, and long-form parsing are covered without modifying shell configuration. | | 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. | @@ -87,7 +87,7 @@ Not yet suitable: and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 100 passing tests and 76% statement coverage. Coverage +The suite currently has 109 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 4d95ef4..a10f48a 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -17,7 +17,7 @@ | workplan | TAMQ-WP-0006 | finished | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | | workplan | TAMQ-WP-0007 | finished | — | workplans/TAMQ-WP-0007-terminal-output-notifications.md | | workplan | TAMQ-WP-0008 | finished | — | workplans/TAMQ-WP-0008-reply-shorthand-stable-output.md | -| workplan | TAMQ-WP-0009 | active | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | +| workplan | TAMQ-WP-0009 | finished | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.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 | @@ -52,6 +52,6 @@ | task | TAMQ-WP-0008-T02 | done | — | workplans/TAMQ-WP-0008-reply-shorthand-stable-output.md | | task | TAMQ-WP-0008-T03 | done | — | workplans/TAMQ-WP-0008-reply-shorthand-stable-output.md | | task | TAMQ-WP-0008-T04 | done | — | workplans/TAMQ-WP-0008-reply-shorthand-stable-output.md | -| task | TAMQ-WP-0009-T01 | progress | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | -| task | TAMQ-WP-0009-T02 | todo | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | -| task | TAMQ-WP-0009-T03 | todo | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | +| task | TAMQ-WP-0009-T01 | done | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | +| task | TAMQ-WP-0009-T02 | done | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | +| task | TAMQ-WP-0009-T03 | done | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | diff --git a/src/tamq/cli.py b/src/tamq/cli.py index aaafa4e..b0a6307 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -24,6 +24,7 @@ from .tmux import TmuxError, TmuxManager from .broker import BrokerIdentity, InputBroker from .ptytap import PtyTap from .control import ControlModeClient +from .composer import ComposerError, compose_message, recipient_order from .diagnostics import configure from .policy import load_profile from .registry import RegistryError, validate_targets @@ -162,7 +163,8 @@ def build_parser() -> argparse.ArgumentParser: manual messaging from a managed shell: @TARGET: MESSAGE... - @ MESSAGE... reply to the latest sender for this window + @ compose with a visible, Tab-selectable recipient + @ MESSAGE... fast reply (ordinary shell quoting applies) tamq inbox [--filter COMMAND] """, ) @@ -208,7 +210,7 @@ manual messaging from a managed shell: send.add_argument("--endpoint-id") send.add_argument("--from", dest="sender_repo", help="sender repository (default: TAMQ_REPO or local)") reply = subparsers.add_parser("reply", help="reply to the latest sender for the current repository") - reply.add_argument("body", nargs="+", help="message body") + reply.add_argument("body", nargs="*", help="message body; omit to open the interactive composer") export = subparsers.add_parser("export", help="export history as JSONL") export.add_argument("--output", required=True) export.add_argument("--repo", dest="target_repo") @@ -383,11 +385,33 @@ def main(argv: list[str] | None = None) -> int: if not sender: print("tamq reply requires a managed window with TAMQ_REPO", file=sys.stderr) return 2 - body = " ".join(args.body).strip() - target = store.latest_counterparty(sender) - if target is None: - print(f"tamq: no counterparty has sent a message to {sender}", file=sys.stderr) - return 2 + latest = store.latest_counterparty(sender) + if args.body: + body = " ".join(args.body).strip() + target = latest + if target is None: + print(f"tamq: no counterparty has sent a message to {sender}", file=sys.stderr) + return 2 + else: + try: + encoded_recipients = json.loads(os.environ.get("TAMQ_RECIPIENTS", "[]")) + except json.JSONDecodeError: + encoded_recipients = [] + session_repos = ( + encoded_recipients + if isinstance(encoded_recipients, list) + and all(isinstance(repo, str) for repo in encoded_recipients) + else [] + ) + recipients = recipient_order(sender, latest, session_repos) + try: + composed = compose_message(recipients) + except ComposerError as exc: + print(f"tamq: {exc}", file=sys.stderr) + return 2 + if composed is None: + return 130 + target, body = composed else: text = " ".join([args.address, *args.body]).strip() if not text.startswith("@") or ":" not in text: diff --git a/src/tamq/composer.py b/src/tamq/composer.py new file mode 100644 index 0000000..0c0c07f --- /dev/null +++ b/src/tamq/composer.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import os +import re +import sys +import termios +import tty +from collections.abc import Sequence + + +class ComposerError(RuntimeError): + pass + + +def recipient_order( + current_repo: str, + latest_counterparty: str | None, + session_repos: Sequence[str], +) -> tuple[str, ...]: + """Put the latest sender first, then the remaining peer repositories.""" + ordered: list[str] = [] + for repo in ([latest_counterparty] if latest_counterparty else []): + if ( + repo != current_repo + and repo not in ordered + and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is not None + ): + ordered.append(repo) + for repo in session_repos: + if ( + repo != current_repo + and repo not in ordered + and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is not None + ): + ordered.append(repo) + return tuple(ordered) + + +def _remove_last_character(value: bytearray) -> None: + if not value: + return + removed = value.pop() + if removed & 0xC0 == 0x80: + while value and value[-1] & 0xC0 == 0x80: + value.pop() + if value and value[-1] & 0xC0 == 0xC0: + value.pop() + + +def compose_message( + recipients: Sequence[str], + *, + input_fd: int | None = None, + output_fd: int | None = None, +) -> tuple[str, str] | None: + """Read one literal terminal line while Tab cycles the visible recipient.""" + if not recipients: + raise ComposerError("no recipient is available") + try: + source = sys.stdin.fileno() if input_fd is None else input_fd + destination = sys.stderr.fileno() if output_fd is None else output_fd + except (AttributeError, OSError) as exc: + raise ComposerError("interactive composition requires a terminal") from exc + if not os.isatty(source) or not os.isatty(destination): + raise ComposerError("interactive composition requires a terminal") + + saved = termios.tcgetattr(source) + draft = bytearray() + recipient_index = 0 + escape_state = 0 + + def write(value: bytes) -> None: + offset = 0 + while offset < len(value): + offset += os.write(destination, value[offset:]) + + def render() -> None: + prompt = f"@{recipients[recipient_index]}: ".encode("utf-8") + write(b"\r\x1b[2K" + prompt + draft) + + try: + tty.setraw(source) + render() + while True: + value = os.read(source, 1) + if not value: + write(b"\r\n") + return None + byte = value[0] + if escape_state == 1: + escape_state = 2 if byte in (ord("["), ord("O")) else 0 + continue + if escape_state == 2: + if 0x40 <= byte <= 0x7E: + escape_state = 0 + continue + if byte == 0x1B: + escape_state = 1 + elif byte == 0x09: + recipient_index = (recipient_index + 1) % len(recipients) + render() + elif byte in (0x0A, 0x0D): + try: + body = draft.decode("utf-8") + except UnicodeDecodeError: + write(b"\a") + continue + if not body.strip(): + write(b"\a") + continue + write(b"\r\n") + return recipients[recipient_index], body + elif byte == 0x03: + write(b"^C\r\n") + return None + elif byte == 0x04 and not draft: + write(b"\r\n") + return None + elif byte in (0x08, 0x7F): + _remove_last_character(draft) + render() + elif byte == 0x15: + draft.clear() + render() + elif byte >= 0x20: + draft.extend(value) + write(value) + else: + write(b"\a") + finally: + termios.tcsetattr(source, termios.TCSADRAIN, saved) diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py index 16591ec..72d3359 100644 --- a/src/tamq/tmux.py +++ b/src/tamq/tmux.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import os import re import shlex @@ -78,13 +79,7 @@ class TmuxManager: self.command_dir.mkdir(parents=True, exist_ok=True, mode=0o700) if not os.access(self.command_dir, os.W_OK | os.X_OK): raise TmuxError(f"address command directory is not writable: {self.command_dir}") - scripts = { - "@": ( - "#!/bin/sh\n" - "# tamq-address-command v1\n" - f"exec {shell_join(list(self.tamq_command))} reply -- \"$@\"\n" - ) - } + scripts: dict[str, str] = {} for repo in repos: if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is None: raise TmuxError( @@ -98,6 +93,14 @@ class TmuxManager: ) for name in (f"@{repo}", target): scripts[name] = script + recipient_data = shlex.quote(json.dumps(list(repos), separators=(",", ":"))) + scripts["@"] = ( + "#!/bin/sh\n" + "# tamq-address-command v1\n" + f"TAMQ_RECIPIENTS={recipient_data}\n" + "export TAMQ_RECIPIENTS\n" + f"exec {shell_join(list(self.tamq_command))} reply -- \"$@\"\n" + ) for name in scripts: destination = self.command_dir / name if destination.exists() or destination.is_symlink(): diff --git a/tests/test_composer.py b/tests/test_composer.py new file mode 100644 index 0000000..80a8d62 --- /dev/null +++ b/tests/test_composer.py @@ -0,0 +1,101 @@ +import os +import pty +import select +import termios +import threading +import time + +import pytest + +from tamq.composer import ComposerError, compose_message, recipient_order + + +def _read_until(master: int, marker: bytes, timeout: float = 2) -> bytes: + output = bytearray() + deadline = time.monotonic() + timeout + while marker not in output and time.monotonic() < deadline: + readable, _, _ = select.select([master], [], [], 0.05) + if readable: + output.extend(os.read(master, 4096)) + return bytes(output) + + +def _compose_in_pty(recipients, typed): + master, slave = pty.openpty() + original = termios.tcgetattr(slave) + result = {} + + def run(): + try: + result["value"] = compose_message( + recipients, input_fd=slave, output_fd=slave + ) + except Exception as exc: # surfaced in the test thread + result["error"] = exc + + thread = threading.Thread(target=run) + try: + thread.start() + output = bytearray(_read_until(master, b": ")) + os.write(master, typed) + thread.join(timeout=2) + assert not thread.is_alive() + output.extend(_read_until(master, b"\r\n", timeout=0.2)) + assert "error" not in result + assert termios.tcgetattr(slave) == original + return result["value"], bytes(output) + finally: + os.close(master) + os.close(slave) + + +def test_recipient_order_prefers_latest_then_session_peers(): + assert recipient_order( + "audit-core", + "flex-auth", + ["audit-core", "railiance-platform", "flex-auth", "railiance-platform"], + ) == ("flex-auth", "railiance-platform") + assert recipient_order( + "audit-core", None, ["audit-core", "railiance-platform"] + ) == ("railiance-platform",) + assert recipient_order("audit-core", "bad\x1btarget", ["../bad"]) == () + + +def test_composer_accepts_literal_prose_and_cycles_recipient_with_tab(): + result, output = _compose_in_pty( + ("flex-auth", "railiance-platform"), + b"What's up?\t Really?\r", + ) + + assert result == ("railiance-platform", "What's up? Really?") + assert b"@flex-auth: " in output + assert b"@railiance-platform: What's up?" in output + + +def test_composer_supports_unicode_backspace_and_ctrl_u(): + result, _ = _compose_in_pty( + ("flex-auth",), + "caf\u00e9x\b!\x15final\r".encode(), + ) + assert result == ("flex-auth", "final") + + +def test_composer_ctrl_c_cancels_and_restores_terminal(): + result, output = _compose_in_pty(("flex-auth",), b"draft\x03") + assert result is None + assert b"^C\r\n" in output + + +def test_composer_requires_a_terminal(): + read_fd, write_fd = os.pipe() + try: + with pytest.raises(ComposerError, match="requires a terminal"): + compose_message(("flex-auth",), input_fd=read_fd, output_fd=write_fd) + finally: + os.close(read_fd) + os.close(write_fd) + + +def test_composer_requires_a_recipient(): + with pytest.raises(ComposerError, match="no recipient"): + compose_message(()) diff --git a/tests/test_manual_messaging.py b/tests/test_manual_messaging.py index 2dbbd94..4f69ff5 100644 --- a/tests/test_manual_messaging.py +++ b/tests/test_manual_messaging.py @@ -55,6 +55,66 @@ def test_bare_reply_targets_latest_inbound_sender(tmp_path, monkeypatch, capsys) store.close() +def test_interactive_reply_shows_latest_and_cycles_session_recipients(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("TAMQ_REPO", "audit-core") + monkeypatch.setenv( + "TAMQ_RECIPIENTS", '["audit-core","railiance-platform","flex-auth"]' + ) + monkeypatch.setattr("tamq.cli.ping", service_is_down) + monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) + store = Store(tmp_path / "state" / "tamq.sqlite3") + store.add("flex-auth", "audit-core", "latest inbound") + store.close() + seen = [] + + def compose(recipients): + seen.append(recipients) + return "railiance-platform", "What's up?" + + monkeypatch.setattr("tamq.cli.compose_message", compose) + assert main(["reply"]) == 0 + message_id = capsys.readouterr().out.strip() + + assert seen == [("flex-auth", "railiance-platform")] + store = Store(tmp_path / "state" / "tamq.sqlite3") + row = next(row for row in store.list() if row["message_id"] == message_id) + assert (row["sender_repo"], row["target_repo"], row["body"]) == ( + "audit-core", "railiance-platform", "What's up?" + ) + store.close() + + +def test_interactive_reply_uses_first_peer_without_history(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("TAMQ_REPO", "audit-core") + monkeypatch.setenv("TAMQ_RECIPIENTS", '["audit-core","flex-auth"]') + monkeypatch.setattr("tamq.cli.ping", service_is_down) + monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) + monkeypatch.setattr( + "tamq.cli.compose_message", lambda recipients: (recipients[0], "hello") + ) + + assert main(["reply"]) == 0 + message_id = capsys.readouterr().out.strip() + store = Store(tmp_path / "state" / "tamq.sqlite3") + row = next(row for row in store.list() if row["message_id"] == message_id) + assert row["target_repo"] == "flex-auth" + store.close() + + +def test_interactive_reply_cancellation_queues_nothing(tmp_path, monkeypatch): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("TAMQ_REPO", "audit-core") + monkeypatch.setenv("TAMQ_RECIPIENTS", '["audit-core","flex-auth"]') + monkeypatch.setattr("tamq.cli.compose_message", lambda recipients: None) + + assert main(["reply"]) == 130 + store = Store(tmp_path / "state" / "tamq.sqlite3") + assert store.list() == [] + store.close() + + def test_bare_reply_requires_managed_identity_and_prior_sender(tmp_path, monkeypatch, capsys): monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) monkeypatch.setattr("tamq.cli.ping", service_is_down) diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py index 620eca6..3f0f093 100644 --- a/tests/test_tmux_manager.py +++ b/tests/test_tmux_manager.py @@ -181,6 +181,9 @@ def test_address_commands_preserve_message_arguments(tmp_path): check=True, ) assert result.stdout == "reply -- Some reply! $value --literal\n" + generic = (command_dir / "@").read_text(encoding="utf-8") + assert 'TAMQ_RECIPIENTS=' in generic + assert '["audit-core"]' in generic def test_address_commands_reject_unsafe_repository_names(tmp_path): diff --git a/workplans/TAMQ-WP-0009-interactive-recipient-composer.md b/workplans/TAMQ-WP-0009-interactive-recipient-composer.md index 5837f83..c38f4f5 100644 --- a/workplans/TAMQ-WP-0009-interactive-recipient-composer.md +++ b/workplans/TAMQ-WP-0009-interactive-recipient-composer.md @@ -4,7 +4,7 @@ type: workplan title: "Interactive recipient-aware message composer" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social planning_priority: P0 @@ -36,7 +36,7 @@ retains normal shell quoting rules. ```task id: TAMQ-WP-0009-T01 -status: progress +status: done priority: high state_hub_task_id: "320e4c7c-9c44-5023-8c60-7b270a0276f5" ``` @@ -49,7 +49,7 @@ Exclude the current repository from the interactive recipient cycle. ```task id: TAMQ-WP-0009-T02 -status: todo +status: done priority: high state_hub_task_id: "6973e046-f8b3-50d1-958b-4216af656624" ``` @@ -63,7 +63,7 @@ exit path. Ctrl-C cancels without queuing a message. ```task id: TAMQ-WP-0009-T03 -status: todo +status: done priority: high state_hub_task_id: "85b8bb9f-cf0d-5d95-9598-1cec8449ac17" ``` @@ -72,3 +72,17 @@ Cover recipient ordering, shim metadata, non-TTY rejection, punctuation, editing and cancellation with pseudo-terminal tests. Run the full checks, install the build, exercise the composer in the live two-window session, and update README and SCOPE. + +## Completion evidence + +- `make check`: 109 tests passed, including pseudo-terminal coverage for + literal apostrophes, recipient cycling with draft preservation, UTF-8-aware + backspace, Ctrl-U, Ctrl-C, and terminal restoration. +- Coverage: 76% overall and 76% for the new composer module. +- `make install`: installed `tmux-amq==0.1.0` and refreshed the live session's + tamq-owned command shims. +- Live `flex-auth` composer: delivered `What's up? WP0009-live-smoke` without + shell quoting, then retained `cycle-draft` while Tab changed the visible + recipient from `@audit-core: ` to `@railiance-platform: `. Ctrl-C cancelled + that draft. The controlled smoke message was acknowledged; existing user + messages were not.