feat: add experimental pushy delivery mode

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

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] [--no-display] REPO" in output
+    assert "tamq [--detach] [--command COMMAND] [--mode MODE] REPO" in output
     assert "With no --command" in output
     assert "@TARGET: MESSAGE" in output
 
@@ -45,11 +45,13 @@ def test_start_parser_supports_command_and_cmd_alias():
     compatibility = parser.parse_args(["start", "--cmd", "claude", "a"])
     neutral = parser.parse_args(["start", "a", "b"])
     inbox_only = parser.parse_args(["start", "--no-display", "a"])
+    pushy = parser.parse_args(["start", "--mode", "pushy", "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
+    assert pushy.mode == "pushy"
 
 
 def test_repository_first_and_command_first_start_shorthand():
@@ -69,6 +71,17 @@ def test_repository_first_and_command_first_start_shorthand():
         "--no-display",
         "flex-auth",
     ]
+    assert normalize_argv(["--mode", "pushy", "flex-auth"]) == [
+        "start",
+        "--mode",
+        "pushy",
+        "flex-auth",
+    ]
+    assert normalize_argv(["--mode=pushy", "flex-auth"]) == [
+        "start",
+        "--mode=pushy",
+        "flex-auth",
+    ]
     assert normalize_argv(["status"]) == ["status"]
     assert normalize_argv(["--verbose", "flex-auth", "audit-core"]) == [
         "--verbose",
@@ -179,6 +192,52 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys):
     assert "endpoint registration failed: denied" in capsys.readouterr().err
 
 
+def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys):
+    endpoint = type(
+        "Endpoint",
+        (),
+        {
+            "endpoint_id": "tmux-amq-42",
+            "instance_key": "tmux-amq-42-boot",
+            "pid": 42,
+            "session": "tamq",
+            "repos": ["a", "b"],
+        },
+    )()
+    requests = []
+
+    class Manager:
+        def preflight(self, repos, command):
+            assert repos == ["a", "b"]
+            return "plan"
+
+        def ensure_plan(self, plan, *, tap=True):
+            assert tap is False
+            return endpoint
+
+        def rollback(self, value):
+            pytest.fail("successful registration must not roll back")
+
+    async def register(payload):
+        requests.append(payload)
+        return {"ok": True}
+
+    monkeypatch.setattr("tamq.cli.TmuxManager", Manager)
+    monkeypatch.setattr("tamq.cli.preflight_runtime_paths", lambda: None)
+    monkeypatch.setattr("tamq.cli.ensure_pushy_service", lambda: True)
+    monkeypatch.setattr(
+        "tamq.cli.ensure_output_service",
+        lambda: pytest.fail("output capability must not select pushy mode"),
+    )
+    monkeypatch.setattr("tamq.cli.request", register)
+
+    assert main(["start", "--detach", "--mode", "pushy", "a", "b"]) == 0
+    summary = json.loads(capsys.readouterr().out)
+    assert summary["mode"] == "pushy"
+    assert summary["delivery_mode"] == "pushy"
+    assert requests[0]["delivery_mode"] == "pushy"
+
+
 def test_tap_requires_explicit_command_and_service(capsys):
     assert main(["start", "--tap", "a"]) == 2
     assert "--tap requires an explicit --command" in capsys.readouterr().err
@@ -186,3 +245,9 @@ def test_tap_requires_explicit_command_and_service(capsys):
     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
+    assert main(["start", "--no-display", "--mode", "pushy", "a"]) == 2
+    assert "conflicts with the selected --mode" in capsys.readouterr().err
+    assert main(["start", "--tap", "--mode", "pushy", "--command", "sh", "a"]) == 2
+    assert "--tap cannot be combined with --mode" in capsys.readouterr().err
+    assert main(["start", "--no-service", "--mode", "pushy", "a"]) == 2
+    assert "--mode cannot be combined with --no-service" in capsys.readouterr().err

tests/test_cli_helpers.py

@@ -1,4 +1,9 @@
-from tamq.cli import ensure_manual_service, ensure_output_service, parse_size
+from tamq.cli import (
+    ensure_manual_service,
+    ensure_output_service,
+    ensure_pushy_service,
+    parse_size,
+)
 
 
 def test_parse_size():
@@ -42,3 +47,23 @@ def test_output_service_requires_terminal_output_capability(monkeypatch):
     assert ensure_output_service() is True
     assert len(starts) == 2
     assert stops == [True]
+
+
+def test_pushy_service_requires_pushy_input_capability(monkeypatch):
+    capabilities = iter([
+        ["manual_delivery", "terminal_output"],
+        ["manual_delivery", "terminal_output", "pushy_input"],
+    ])
+    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_pushy_service() is True
+    assert len(starts) == 2
+    assert stops == [True]

tests/test_control.py

@@ -1,5 +1,7 @@
 from tamq import control
-from tamq.control import ControlModeClient, shell_quote
+import pytest
+
+from tamq.control import ControlModeClient, ControlModeError, shell_quote
 from tamq.tmux import shell_join
 
 
@@ -27,3 +29,39 @@ def test_pane_tty_is_resolved_through_tmux(monkeypatch):
         "tmux", "-L", "test", "display-message", "-p", "-t",
         "tamq:audit-core", "#{pane_tty}|#{cursor_x}|#{cursor_y}|#{pane_height}|#{alternate_on}",
     ]]
+
+
+def test_submit_sends_literal_input_then_one_enter(monkeypatch):
+    calls = []
+
+    def run(command, **kwargs):
+        calls.append(command)
+        return type("Result", (), {"returncode": 0, "stderr": ""})()
+
+    monkeypatch.setattr(control.subprocess, "run", run)
+    client = ControlModeClient("tamq", tmux_command=("tmux", "-L", "test"))
+
+    client.submit("tamq:audit-core", "#flex-auth: What's up? [m-1]")
+
+    assert calls == [
+        [
+            "tmux", "-L", "test",
+            "send-keys", "-t", "tamq:audit-core", "-l", "--",
+            "#flex-auth: What's up? [m-1]",
+            ";",
+            "send-keys", "-t", "tamq:audit-core", "Enter",
+        ]
+    ]
+
+
+def test_submit_surfaces_tmux_rejection(monkeypatch):
+    monkeypatch.setattr(
+        control.subprocess,
+        "run",
+        lambda *args, **kwargs: type(
+            "Result", (), {"returncode": 1, "stderr": "missing pane"}
+        )(),
+    )
+    client = ControlModeClient("tamq")
+    with pytest.raises(ControlModeError, match="missing pane"):
+        client.submit("tamq:missing", "message")

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] [--no-display] REPO" in root_help
+        assert "tamq [--detach] [--command COMMAND] [--mode MODE] REPO" in root_help
         started = json.loads(
             run(
                 str(tamq),

tests/test_service_delivery.py

@@ -9,6 +9,7 @@ class FakeControl:
     def __init__(self, session):
         self.session = session
         self.injected = []
+        self.submitted = []
         self.__class__.instances.append(self)
 
     def start(self):
@@ -20,6 +21,9 @@ class FakeControl:
     def inject(self, window, text):
         self.injected.append((window, text))
 
+    def submit(self, window, text):
+        self.submitted.append((window, text))
+
     def pane_display(self, window):
         return PaneDisplay(
             tty_path=f"/dev/pts/{window.rsplit(':', 1)[-1]}",
@@ -56,6 +60,41 @@ def test_service_never_injects_manual_endpoint_messages(tmp_path, monkeypatch):
     assert store.list()[0]["state"] == "pending"
 
 
+def test_service_pushy_mode_submits_once_and_marks_injected(tmp_path, monkeypatch):
+    store = Store(tmp_path / "queue.sqlite3")
+    store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "pushy")
+    message_id = store.add("repo-a", "repo-b", "first\nsecond")
+    monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl)
+    service = Service(store=store)
+
+    service._deliver_once()
+    control = FakeControl.instances[-1]
+    service._deliver_once()
+
+    assert control.submitted == [
+        (
+            "tamq:repo-b",
+            f"#repo-a: first\\x0asecond [{message_id}]",
+        )
+    ]
+    assert store.list()[0]["state"] == "injected"
+
+
+def test_failed_pushy_submission_remains_pending(tmp_path, monkeypatch):
+    store = Store(tmp_path / "queue.sqlite3")
+    store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "pushy")
+    store.add("repo-a", "repo-b", "continue")
+
+    class FailingControl(FakeControl):
+        def submit(self, window, text):
+            raise RuntimeError("tmux rejected input")
+
+    monkeypatch.setattr("tamq.service.ControlModeClient", FailingControl)
+    Service(store=store)._deliver_once()
+
+    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")

tests/test_terminal_output.py

@@ -7,6 +7,7 @@ import pytest
 from tamq.terminal import (
     TerminalOutputError,
     format_comment,
+    format_pushy_input,
     terminal_frame,
     write_terminal_output,
 )
@@ -18,6 +19,12 @@ def test_comment_format_escapes_controls_and_prefixes_every_line():
     )
 
 
+def test_pushy_input_is_one_sanitized_shell_comment():
+    assert format_pushy_input("repo-a", "first\nsecond\x1b[31m", "m-1") == (
+        "#repo-a: first\\x0asecond\\x1b[31m [m-1]"
+    )
+
+
 def test_terminal_frame_scrolls_only_rows_above_the_cursor():
     assert terminal_frame(
         "repo-a", "first\nsecond", "m-1", cursor_y=8, pane_height=24

tests/test_tmux_integration.py

@@ -9,6 +9,7 @@ import pytest
 
 from tamq.broker import BrokerIdentity, InputBroker
 from tamq.control import ControlModeClient
+from tamq.service import Service
 from tamq.store import Store
 from tamq.terminal import terminal_frame, write_terminal_output
 from tamq.tmux import LaunchPlan, TmuxManager
@@ -198,3 +199,63 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path):
         assert (after.cursor_x, after.cursor_y) == (before.cursor_x, before.cursor_y)
     finally:
         manager._run("kill-server", check=False)
+
+
+@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
+def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch):
+    repo = tmp_path / "audit-core"
+    repo.mkdir()
+    socket_name = f"tamq-pushy-{os.getpid()}-{uuid4().hex[:8]}"
+    session = f"tamq-pushy-{uuid4().hex[:8]}"
+    monkeypatch.setenv("TAMQ_TMUX_SOCKET", socket_name)
+    manager = TmuxManager(
+        session,
+        tmux_command=("tmux", "-L", socket_name),
+        tamq_command=(sys.executable, "-m", "tamq.cli"),
+        command_dir=tmp_path / "commands",
+    )
+    target = f"{session}:audit-core"
+    store = Store(tmp_path / "queue.sqlite3")
+
+    try:
+        endpoint = manager.ensure_plan(
+            LaunchPlan(("audit-core",), {"audit-core": str(repo)}, ())
+        )
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            current_command = manager._run(
+                "display-message", "-p", "-t", target, "#{pane_current_command}"
+            )
+            if current_command in {"bash", "sh", "zsh", "fish"}:
+                break
+            time.sleep(0.05)
+        assert current_command in {"bash", "sh", "zsh", "fish"}
+
+        store.register_endpoint(
+            endpoint.instance_key,
+            endpoint.pid,
+            endpoint.session,
+            endpoint.repos,
+            "pushy",
+        )
+        message_id = store.add("flex-auth", "audit-core", "What's next?\nsecond")
+        service = Service(store=store)
+        service._deliver_once()
+
+        expected = f"#flex-auth: What's next?\\x0asecond [{message_id}]"
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            capture = manager._run("capture-pane", "-p", "-t", target)
+            if expected in capture:
+                break
+            time.sleep(0.05)
+        assert expected in capture
+        assert store.list()[0]["state"] == "injected"
+        assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0
+
+        service._deliver_once()
+        repeated_capture = manager._run("capture-pane", "-p", "-t", target)
+        assert repeated_capture.count(expected) == 1
+    finally:
+        store.close()
+        manager._run("kill-server", check=False)