feat: add reliable trigger and worker message blocks

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

tests/test_broker.py

@@ -17,7 +17,8 @@ def test_broker_accepts_worker_output_with_distinct_provenance(tmp_path, monkeyp
     monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
     store = Store(tmp_path / "queue.sqlite3")
     broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
-    assert broker.inspect_worker_line("To:railiance-platform: hello") is not None
+    assert broker.inspect_worker_line("To:railiance-platform: hello") is None
+    assert broker.inspect_worker_line("") is not None
     assert store.list()[0]["provenance"] == "worker_output"
 
 
@@ -25,12 +26,42 @@ def test_broker_accepts_case_insensitive_framed_worker_output(tmp_path, monkeypa
     monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
     store = Store(tmp_path / "queue.sqlite3")
     broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
-    assert broker.inspect_worker_line("• tO:railiance-platform: hello") is not None
+    assert broker.inspect_worker_line("• tO:railiance-platform: hello") is None
+    assert broker.inspect_worker_line("") is not None
     row = store.list()[0]
     assert row["body"] == "hello"
     assert row["provenance"] == "worker_output"
 
 
+def test_broker_collects_worker_followup_lines_until_empty(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    store = Store(tmp_path / "queue.sqlite3")
+    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
+
+    assert broker.inspect_worker_line("• To:railiance-platform: Review this") is None
+    assert broker.inspect_worker_line("  Context: AUTH-WP-4.") is None
+    assert broker.inspect_worker_line("  Reply with accepted or blocked.") is None
+    assert broker.inspect_worker_line("") is not None
+
+    row = store.list()[0]
+    assert row["body"] == (
+        "Review this\nContext: AUTH-WP-4.\nReply with accepted or blocked."
+    )
+
+
+def test_new_worker_address_flushes_previous_block(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    store = Store(tmp_path / "queue.sqlite3")
+    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
+    broker.inspect_worker_line("To:first: one")
+    assert broker.inspect_worker_line("To:second: two") is not None
+    assert broker.flush_worker_message() is not None
+    assert [(row["target_repo"], row["body"]) for row in store.list()] == [
+        ("first", "one"),
+        ("second", "two"),
+    ]
+
+
 def test_case_insensitive_delivery_envelopes_are_not_rescanned(tmp_path):
     store = Store(tmp_path / "queue.sqlite3")
     broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))

tests/test_control.py

@@ -23,7 +23,9 @@ def test_pane_tty_is_resolved_through_tmux(monkeypatch):
         return type("Result", (), {"returncode": 0, "stdout": "/dev/pts/7|12|8|24|0\n", "stderr": ""})()
 
     monkeypatch.setattr(control.subprocess, "run", run)
-    client = ControlModeClient("tamq", tmux_command=("tmux", "-L", "test"))
+    client = ControlModeClient(
+        "tamq", tmux_command=("tmux", "-L", "test"), submit_delay=0.15
+    )
     assert client.pane_tty("tamq:audit-core") == "/dev/pts/7"
     assert calls == [[
         "tmux", "-L", "test", "display-message", "-p", "-t",
@@ -49,6 +51,8 @@ def test_submit_sends_literal_input_then_one_enter(monkeypatch):
             "send-keys", "-t", "tamq:audit-core", "-l", "--",
             "#flex-auth: What's up? [m-1]",
             ";",
+            "run-shell", "sleep 0.150",
+            ";",
             "send-keys", "-t", "tamq:audit-core", "Enter",
         ]
     ]

tests/test_line_limits.py

@@ -25,6 +25,7 @@ def test_window_counters_survive_reopen_and_isolate_repositories(tmp_path, monke
         notify=blocked.append,
     )
     assert left.inspect_worker_line("To:target: blocked") is None
+    assert left.inspect_worker_line("") is None
     assert "lines of messages" in blocked[-1]
 
 
@@ -43,7 +44,8 @@ def test_atomic_message_admission_does_not_cross_limit(tmp_path, monkeypatch):
                 BrokerIdentity("session-1", "source"),
                 limits=(1, 100, 100),
             )
-            return broker.inspect_worker_line(f"To:target: {body}") is not None
+            broker.inspect_worker_line(f"To:target: {body}")
+            return broker.inspect_worker_line("") is not None
         finally:
             store.close()
 

tests/test_manual_messaging.py

@@ -15,7 +15,7 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk
     monkeypatch.setattr("tamq.cli.ping", service_is_down)
     monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None)
 
-    assert main(["send", "To:audit-core:", "review", "this"]) == 0
+    assert main(["send", "tO:audit-core:", "review", "this"]) == 0
     message_id = capsys.readouterr().out.strip()
 
     assert main(["inbox", "--repo", "audit-core", "--json"]) == 0
@@ -51,7 +51,7 @@ def test_human_inbox_uses_readable_origin_framing(tmp_path, monkeypatch, capsys)
 
     assert main(["inbox", "--repo", "audit-core"]) == 0
     assert capsys.readouterr().out == (
-        "From:flex-auth/o: first\\x0asecond\\x1b[31m\n"
+        "From:flex-auth/o: first\nFrom:flex-auth/o: second\\x1b[31m\n"
     )
 
 

tests/test_ptytap.py

@@ -104,6 +104,24 @@ def test_output_observer_preserves_full_screen_program_gutters():
     assert lines == ["› To:target: operator echo", "• tO:target: worker reply"]
 
 
+def test_output_observer_emits_one_empty_line_for_crlf_terminator():
+    lines = []
+    observer = TerminalOutputObserver(lines.append)
+    observer.feed(b"To:target: first\r\nfollowup\r\n\r\nordinary\r\n")
+    assert lines == ["To:target: first", "followup", "", "ordinary"]
+
+
+def test_output_observer_infers_empty_full_screen_row_from_cursor_gap():
+    lines = []
+    observer = TerminalOutputObserver(lines.append)
+    observer.feed(
+        b"\x1b[4;1H\xe2\x80\xa2 To:target: first"
+        b"\x1b[5;1H  followup"
+        b"\x1b[7;1Hprompt"
+    )
+    assert lines == ["• To:target: first", "  followup", ""]
+
+
 def test_copy_winsize_preserves_rows_columns_and_pixels():
     source_master, source_slave = pty.openpty()
     target_master, target_slave = pty.openpty()

tests/test_ptytap_integration.py

@@ -113,7 +113,9 @@ def test_real_pty_attributes_worker_output_and_suppresses_operator_echo(
         "print('READY', flush=True)\n"
         "for line in sys.stdin:\n"
         "    print(line.rstrip('\\r\\n'), flush=True)\n"
-        "    print('\\x1b[5;1H• tO:target: worker reply\\x1b[6;1H', flush=True)\n",
+        "    print('\\x1b[5;1H• tO:target: worker reply'\n"
+        "          '\\x1b[6;1H  Context: integration fixture.'\n"
+        "          '\\x1b[8;1H', flush=True)\n",
         encoding="utf-8",
     )
     bin_dir = tmp_path / "bin"
@@ -171,7 +173,7 @@ def test_real_pty_attributes_worker_output_and_suppresses_operator_echo(
             time.sleep(0.05)
         assert [(row["body"], row["provenance"]) for row in rows] == [
             ("operator request", "operator_input"),
-            ("worker reply", "worker_output"),
+            ("worker reply\nContext: integration fixture.", "worker_output"),
         ]
         assert all(row["endpoint_id"] == endpoint.instance_id for row in rows)
     finally:

tests/test_service_delivery.py

@@ -69,7 +69,7 @@ def test_service_pushy_mode_places_once_without_enter(tmp_path, monkeypatch):
     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 = Service(store=store, input_grace=0)
 
     service._deliver_once()
     control = FakeControl.instances[-1]
@@ -78,7 +78,7 @@ def test_service_pushy_mode_places_once_without_enter(tmp_path, monkeypatch):
     assert control.placed == [
         (
             "tamq:repo-b",
-            "From:repo-a: first\\x0asecond",
+            "From:repo-a: first\nFrom:repo-a: second",
         )
     ]
     assert control.submitted == []
@@ -90,7 +90,7 @@ def test_service_trigger_mode_submits_exactly_once(tmp_path, monkeypatch):
     store.register_endpoint("ep", 9, "tamq", ["repo-b"], "trigger")
     store.add("repo-a", "repo-b", "go", provenance="operator_input")
     monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl)
-    service = Service(store=store)
+    service = Service(store=store, input_grace=0)
     service._deliver_once()
     control = FakeControl.instances[-1]
     service._deliver_once()
@@ -108,11 +108,24 @@ def test_failed_pushy_submission_remains_pending(tmp_path, monkeypatch):
             raise RuntimeError("tmux rejected input")
 
     monkeypatch.setattr("tamq.service.ControlModeClient", FailingControl)
-    Service(store=store)._deliver_once()
+    Service(store=store, input_grace=0)._deliver_once()
 
     assert store.list()[0]["state"] == "pending"
 
 
+def test_input_delivery_waits_for_new_endpoint_grace(tmp_path, monkeypatch):
+    store = Store(tmp_path / "queue.sqlite3")
+    store.register_endpoint("ep", 9, "tamq", ["repo-b"], "trigger")
+    store.add("repo-a", "repo-b", "first")
+    monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl)
+    service = Service(store=store, input_grace=10)
+
+    service._deliver_once()
+
+    assert FakeControl.instances[-1].submitted == []
+    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

@@ -14,7 +14,7 @@ from tamq.terminal import (
 
 def test_delivery_format_escapes_controls_and_marks_operator_origin():
     assert format_delivery("repo-a", "first\nsecond\x1b[31m", "operator_input") == (
-        "From:repo-a/o: first\\x0asecond\\x1b[31m"
+        "From:repo-a/o: first\nFrom:repo-a/o: second\\x1b[31m"
     )
     assert format_delivery("repo-a", "worker", "worker_output") == (
         "From:repo-a: worker"

tests/test_tmux_integration.py

@@ -239,28 +239,93 @@ def test_real_tmux_pushy_mode_places_one_readable_input_without_enter(tmp_path,
             "pushy",
         )
         store.add("flex-auth", "audit-core", "What's next?\nsecond", provenance="operator_input")
-        service = Service(store=store)
+        service = Service(store=store, input_grace=0)
         service._deliver_once()
 
-        expected = "From:flex-auth/o: What's next?\\x0asecond"
+        expected_lines = ["From:flex-auth/o: What's next?", "From:flex-auth/o: second"]
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
             capture = manager._run("capture-pane", "-p", "-J", "-t", target)
-            if expected in capture:
+            if all(expected in capture for expected in expected_lines):
                 break
             time.sleep(0.05)
-        assert expected in capture
+        assert all(expected in capture for expected in expected_lines)
         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", "-J", "-t", target)
-        assert repeated_capture.count(expected) == 1
+        assert all(repeated_capture.count(expected) == 1 for expected in expected_lines)
     finally:
         store.close()
         manager._run("kill-server", check=False)
 
 
+@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
+def test_real_tmux_trigger_clears_terminal_paste_window_before_enter(tmp_path):
+    repo = tmp_path / "audit-core"
+    repo.mkdir()
+    fixture = tmp_path / "paste_sensitive_fixture.py"
+    fixture.write_text(
+        "import os, time, tty\n"
+        "tty.setraw(0)\n"
+        "os.write(1, b'READY\\r\\n')\n"
+        "body = bytearray()\n"
+        "last_text = 0.0\n"
+        "while True:\n"
+        "    data = os.read(0, 4096)\n"
+        "    now = time.monotonic()\n"
+        "    for byte in data:\n"
+        "        if byte == 13:\n"
+        "            label = b'SUBMITTED:' if now - last_text >= 0.1 else b'NEWLINE:'\n"
+        "            os.write(1, label + bytes(body) + b'\\r\\n')\n"
+        "            raise SystemExit(0)\n"
+        "        body.append(byte)\n"
+        "        last_text = now\n",
+        encoding="utf-8",
+    )
+    socket_name = f"tamq-trigger-paste-{uuid4().hex[:8]}"
+    session = f"tamq-trigger-paste-{uuid4().hex[:8]}"
+    tmux_command = ("tmux", "-L", socket_name)
+    manager = TmuxManager(
+        session,
+        tmux_command=tmux_command,
+        tamq_command=(sys.executable, "-m", "tamq.cli"),
+        command_dir=tmp_path / "commands",
+    )
+    try:
+        manager.ensure_plan(
+            LaunchPlan(
+                ("audit-core",),
+                {"audit-core": str(repo)},
+                (sys.executable, str(fixture)),
+            )
+        )
+        target = f"{session}:audit-core"
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            capture = manager._run("capture-pane", "-p", "-t", target)
+            if "READY" in capture:
+                break
+            time.sleep(0.05)
+        assert "READY" in capture
+
+        ControlModeClient(
+            session, tmux_command=tmux_command, submit_delay=0.15
+        ).submit(target, "From:flex-auth: first delivery")
+
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            capture = manager._run("capture-pane", "-p", "-t", target)
+            if "SUBMITTED:" in capture:
+                break
+            time.sleep(0.05)
+        assert "SUBMITTED:From:flex-auth: first delivery" in capture
+        assert "NEWLINE:" not in capture
+    finally:
+        manager._run("kill-server", check=False)
+
+
 @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
 def test_real_tmux_to_route_triggers_once_without_feedback(tmp_path, monkeypatch):
     repo_a = tmp_path / "railiance-platform"
@@ -355,7 +420,7 @@ def test_real_tmux_to_route_triggers_once_without_feedback(tmp_path, monkeypatch
         assert row["body"] == "Hello!"
         assert row["endpoint_id"] == endpoint.instance_key
 
-        service = Service(store=store)
+        service = Service(store=store, input_grace=0)
         service._deliver_once()
         expected = "From:railiance-platform/o: Hello!"
         deadline = time.monotonic() + 5

tests/test_tmux_manager.py

@@ -103,7 +103,7 @@ def test_tap_refuses_managed_session_without_duplex_protocol(tmp_path, monkeypat
     plan = tmux.LaunchPlan(("a",), {"a": str(tmp_path)}, ("sh",))
     monkeypatch.setattr(manager, "_existing_session_identity", lambda: (42, "tmux-amq-42-old"))
     monkeypatch.setattr(manager, "_run", lambda *args, **kwargs: "")
-    with pytest.raises(tmux.TmuxError, match="predates readable duplex"):
+    with pytest.raises(tmux.TmuxError, match="predates worker message blocks"):
         manager.ensure_plan(plan, tap=True)