feat: add interactive recipient composer

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc

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(())

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)

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):