feat: add readable duplex messaging

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

tests/test_broker.py

@@ -2,53 +2,69 @@ from tamq.broker import BrokerIdentity, InputBroker
 from tamq.store import Store
 
 
-def test_broker_preserves_identity(tmp_path):
+def test_broker_preserves_operator_identity_and_provenance(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
     store = Store(tmp_path / "queue.sqlite3")
     broker = InputBroker(store, BrokerIdentity("tmux-amq-42", "net-kingdom"))
-    assert broker.inspect_line("@railiance-platform: hello") is not None
+    assert broker.inspect_operator_line("To:railiance-platform: hello") is not None
     row = store.list()[0]
     assert row["sender_repo"] == "net-kingdom"
     assert row["endpoint_id"] == "tmux-amq-42"
+    assert row["provenance"] == "operator_input"
 
 
-def test_broker_accepts_hash_address_alias(tmp_path):
+def test_broker_accepts_worker_output_with_distinct_provenance(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
     store = Store(tmp_path / "queue.sqlite3")
-    broker = InputBroker(store, BrokerIdentity("tmux-amq-42-boot", "net-kingdom"))
-    assert broker.inspect_line("#railiance-platform: hello") is not None
-    row = store.list()[0]
-    assert row["sender_repo"] == "net-kingdom"
-    assert row["target_repo"] == "railiance-platform"
-    assert row["endpoint_id"] == "tmux-amq-42-boot"
+    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
+    assert broker.inspect_worker_line("To:railiance-platform: hello") is not None
+    assert store.list()[0]["provenance"] == "worker_output"
 
 
-def test_broker_rejects_legacy_injected_envelope_feedback(tmp_path):
+def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path):
     store = Store(tmp_path / "queue.sqlite3")
-    message_id = store.add(
-        "flex-auth",
-        "audit-core",
-        "Hello worker agent!",
-        endpoint="tmux-amq-42-boot",
-    )
+    store.register_endpoint("ep", 9, "tamq", ["repo"], "output")
+    notices = []
     broker = InputBroker(
-        store, BrokerIdentity("tmux-amq-42-boot", "audit-core")
+        store, BrokerIdentity("ep", "repo"), notify=notices.append
     )
-
-    assert (
-        broker.inspect_line(
-            f"#flex-auth: Hello worker agent! [{message_id}]"
-        )
-        is None
-    )
-    assert len(store.list()) == 1
+    broker.inspect_worker_line("Cmd: mode=trigger")
+    assert store.endpoint("ep")["delivery_mode"] == "output"
+    broker.inspect_operator_line("Cmd: mode=trigger")
+    assert store.endpoint("ep")["delivery_mode"] == "trigger"
+    assert notices[-1] == "From:tamq: Mode set to trigger."
 
 
-def test_broker_does_not_block_unknown_receipt_like_user_text(tmp_path):
+def test_limits_block_at_equality_and_reset_current_window(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
     store = Store(tmp_path / "queue.sqlite3")
+    notices = []
     broker = InputBroker(
-        store, BrokerIdentity("tmux-amq-42-boot", "audit-core")
-    )
-
-    assert broker.inspect_line(
-        "#flex-auth: Please inspect [m-00000000-0000-0000-0000-000000000000]"
+        store,
+        BrokerIdentity("ep", "repo"),
+        limits=(1, 10, 10),
+        notify=notices.append,
     )
+    assert broker.inspect_operator_line("To:target: first") is not None
+    assert broker.inspect_operator_line("To:target: blocked") is None
+    assert "Running linecount 1 limited by 1 lines of messages" in notices[-1]
     assert len(store.list()) == 1
+    broker.inspect_operator_line("Cmd: reset-limits")
+    assert broker.inspect_operator_line("To:target: after reset") is not None
+
+
+def test_input_limit_includes_the_current_operator_line(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    store = Store(tmp_path / "queue.sqlite3")
+    notices = []
+    broker = InputBroker(
+        store,
+        BrokerIdentity("ep", "repo"),
+        limits=(8, 1, 32768),
+        notify=notices.append,
+    )
+    assert broker.inspect_operator_line("To:target: blocked") is None
+    assert notices[-1] == (
+        "From:tamq: Messaging blocked! Running linecount 1 limited by 1 lines "
+        "of input in this terminal. Use 'Cmd: reset-limits' to unblock."
+    )

tests/test_broker_delivery.py

@@ -16,6 +16,6 @@ def test_pending_delivery_marks_injected(tmp_path):
     control = FakeControl()
     count = InputBroker(store, BrokerIdentity("tmux-amq-42", "net-kingdom")).deliver_pending(control, window_for_repo=lambda repo: f"tamq:{repo}")
     assert count == 1
-    assert control.calls == [("tamq:railiance-platform", "#net-kingdom: hello")]
+    assert control.calls == [("tamq:railiance-platform", "From:net-kingdom: hello")]
     assert store.list()[0]["message_id"] == message_id
     assert store.list()[0]["state"] == "injected"

tests/test_broker_registry.py

@@ -6,5 +6,5 @@ from tamq.store import Store
 def test_broker_rejects_unregistered_target(tmp_path, monkeypatch):
     monkeypatch.setattr(broker, "validate_targets", lambda repos: (_ for _ in ()).throw(broker.RegistryError("unknown")))
     store = Store(tmp_path / "queue.sqlite3")
-    assert InputBroker(store, BrokerIdentity("ep", "source")).inspect_line("@missing: hello") is None
+    assert InputBroker(store, BrokerIdentity("ep", "source")).inspect_line("To:missing: hello") is None
     assert store.list() == []

tests/test_cleanup.py

@@ -42,10 +42,12 @@ def test_cleanup_is_dry_run_then_removes_only_owned_runtime(
     state, runtime = configure_paths(tmp_path, monkeypatch)
     command_dir = state / "commands" / "tamq"
     command_dir.mkdir(parents=True)
-    generated = command_dir / "@audit-core"
+    generated = command_dir / "To:audit-core:"
     generated.write_text(f"#!/bin/sh\n{SHIM_MARKER}\n", encoding="utf-8")
     unrelated = command_dir / "keep-me"
     unrelated.write_text("operator file\n", encoding="utf-8")
+    binary = command_dir / "binary"
+    binary.write_bytes(b"\xff\x00\x80")
     for name in ("tamq.sock", "tamq.pid", "tamq.lock"):
         (runtime / name).write_text("stale\n", encoding="utf-8")
     tmux_socket_dir = tmp_path / "tmux" / f"tmux-{os.getuid()}"
@@ -96,6 +98,7 @@ def test_cleanup_is_dry_run_then_removes_only_owned_runtime(
     assert manager.identity is None
     assert not generated.exists()
     assert unrelated.exists()
+    assert binary.exists()
     assert not any((runtime / name).exists() for name in ("tamq.sock", "tamq.pid", "tamq.lock"))
     assert not stale_tmux_socket.exists()
     assert live_tmux_socket.exists()

tests/test_cli.py

@@ -18,9 +18,11 @@ 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] [--mode MODE] REPO" in output
+    assert "tamq [--detach] [--command COMMAND] [--mode MODE]" in output
     assert "With no --command" in output
-    assert "@TARGET: MESSAGE" in output
+    assert "To:TARGET: MESSAGE" in output
+    assert "Cmd: mode=trigger" in output
+    assert "[--maxmsg N] [--maxin N] [--maxout N]" in output
 
 
 def test_attach_delegates_to_tmux(monkeypatch):
@@ -46,12 +48,15 @@ def test_start_parser_supports_command_and_cmd_alias():
     neutral = parser.parse_args(["start", "a", "b"])
     inbox_only = parser.parse_args(["start", "--no-display", "a"])
     pushy = parser.parse_args(["start", "--mode", "pushy", "a"])
+    trigger = parser.parse_args(["start", "--mode", "trigger", "--maxmsg", "3", "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"
+    assert trigger.mode == "trigger"
+    assert trigger.maxmsg == 3
 
 
 def test_repository_first_and_command_first_start_shorthand():
@@ -82,6 +87,9 @@ def test_repository_first_and_command_first_start_shorthand():
         "--mode=pushy",
         "flex-auth",
     ]
+    assert normalize_argv(["--maxmsg=3", "flex-auth"]) == [
+        "start", "--maxmsg=3", "flex-auth"
+    ]
     assert normalize_argv(["status"]) == ["status"]
     assert normalize_argv(["--verbose", "flex-auth", "audit-core"]) == [
         "--verbose",
@@ -173,8 +181,9 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys):
         def preflight(self, repos, command):
             return "plan"
 
-        def ensure_plan(self, plan, *, tap=True):
-            assert tap is False
+        def ensure_plan(self, plan, *, tap=True, tap_limits=None):
+            assert tap is True
+            assert tap_limits == (8, 1024, 32768)
             return endpoint
 
         def rollback(self, value):
@@ -212,8 +221,9 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys):
             assert command == "codex"
             return "plan"
 
-        def ensure_plan(self, plan, *, tap=True):
+        def ensure_plan(self, plan, *, tap=True, tap_limits=None):
             assert tap is True
+            assert tap_limits == (8, 1024, 32768)
             return endpoint
 
         def rollback(self, value):
@@ -240,6 +250,7 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys):
     assert summary["delivery_mode"] == "pushy"
     assert summary["input_observation_requested"] is True
     assert requests[0]["delivery_mode"] == "pushy"
+    assert requests[0]["maxmsg"] == 8
 
 
 def test_tap_requires_explicit_command_and_service(capsys):
@@ -255,3 +266,8 @@ def test_tap_requires_explicit_command_and_service(capsys):
     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
+
+
+def test_start_rejects_nonpositive_line_limits(capsys):
+    assert main(["start", "--maxmsg", "0", "a"]) == 2
+    assert "maxmsg must be a positive integer" in capsys.readouterr().err

tests/test_cli_helpers.py

@@ -4,7 +4,11 @@ from tamq.cli import (
     ensure_pushy_service,
     parse_size,
 )
-from tamq.service import PUSHY_FRAMING_CAPABILITY
+from tamq.service import (
+    LINE_LIMITS_CAPABILITY,
+    PUSHY_FRAMING_CAPABILITY,
+    TRIGGER_CAPABILITY,
+)
 
 
 def test_parse_size():
@@ -13,7 +17,7 @@ def test_parse_size():
 
 
 def test_manual_service_restarts_legacy_broker(monkeypatch):
-    capabilities = iter([[], ["manual_delivery"]])
+    capabilities = iter([[], ["manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY]])
     starts = []
     stops = []
 
@@ -33,7 +37,7 @@ def test_manual_service_restarts_legacy_broker(monkeypatch):
 def test_output_service_requires_terminal_output_capability(monkeypatch):
     capabilities = iter([
         ["manual_delivery"],
-        ["manual_delivery", "terminal_output"],
+        ["manual_delivery", "terminal_output", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY],
     ])
     starts = []
     stops = []
@@ -58,6 +62,8 @@ def test_pushy_service_requires_pushy_input_capability(monkeypatch):
             "terminal_output",
             "pushy_input",
             PUSHY_FRAMING_CAPABILITY,
+            TRIGGER_CAPABILITY,
+            LINE_LIMITS_CAPABILITY,
         ],
     ])
     starts = []

tests/test_composer.py

@@ -1,101 +0,0 @@
-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_config_command.py

@@ -6,6 +6,9 @@ from tamq.cli import main
 def test_config_command(monkeypatch, capsys, tmp_path):
     monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
     assert main(["config"]) == 0
-    output = json.loads(capsys.readouterr().out)
-    assert output["database"].endswith("tamq.sqlite3")
-    assert output["policy_profile"] == "default"
+    result = json.loads(capsys.readouterr().out)
+    assert result["maxmsg"] == 8
+    assert result["maxin"] == 1024
+    assert result["maxout"] == 32768
+    assert result["database"].endswith("tamq.sqlite3")
+    assert result["policy_profile"] == "default"

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] [--mode MODE] REPO" in root_help
+        assert "tamq [--detach] [--command COMMAND] [--mode MODE]" in root_help
         started = json.loads(
             run(
                 str(tamq),
@@ -178,7 +178,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path):
             "send-keys",
             "-t",
             "tamq:railiance-platform",
-            "@activity-core: installed-message",
+            "To:activity-core: installed-message",
             "C-m",
         )
         deadline = time.monotonic() + 5
@@ -194,7 +194,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path):
             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:
+            if inbox and inbox[-1]["displayed_at"] is not None and "From:railiance-platform/o: installed-message" in target_after:
                 break
             time.sleep(0.05)
         assert inbox, run(
@@ -206,10 +206,10 @@ def test_isolated_installed_tool_session_smoke(tmp_path):
         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"
+            "From:railiance-platform/o: installed-message\n"
         )
         assert target_after != target_before
-        assert f"#railiance-platform: installed-message [{message_id}]" in target_after
+        assert "From:railiance-platform/o: installed-message" in target_after
 
         assert run(str(tamq), "ack", message_id).stdout.strip() == message_id
         assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == ""
@@ -226,7 +226,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path):
             "send-keys",
             "-t",
             "tamq:railiance-platform",
-            "@activity-core filtered-message",
+            "To:activity-core: filtered-message",
             "C-m",
         )
         deadline = time.monotonic() + 5
@@ -242,7 +242,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path):
             str(tamq), "inbox", "--repo", "activity-core", "--filter",
             f"cat >> {message_log}",
         )
-        assert "#railiance-platform: filtered-message [m-" in message_log.read_text(
+        assert "From:railiance-platform/o: filtered-message" in message_log.read_text(
             encoding="utf-8"
         )
         assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == ""

tests/test_line_limits.py

@@ -0,0 +1,65 @@
+from concurrent.futures import ThreadPoolExecutor
+
+from tamq.broker import BrokerIdentity, InputBroker
+from tamq.store import Store
+
+
+def test_window_counters_survive_reopen_and_isolate_repositories(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    path = tmp_path / "queue.sqlite3"
+    store = Store(path)
+    left = InputBroker(store, BrokerIdentity("session-1", "left"), limits=(1, 10, 10))
+    right = InputBroker(store, BrokerIdentity("session-1", "right"), limits=(1, 10, 10))
+    assert left.inspect_operator_line("To:target: left") is not None
+    assert right.inspect_operator_line("To:target: right") is not None
+    store.close()
+
+    reopened = Store(path)
+    assert reopened.line_state("session-1", "left")["messages"] == 1
+    assert reopened.line_state("session-1", "right")["messages"] == 1
+    blocked = []
+    left = InputBroker(
+        reopened,
+        BrokerIdentity("session-1", "left"),
+        limits=(1, 10, 10),
+        notify=blocked.append,
+    )
+    assert left.inspect_worker_line("To:target: blocked") is None
+    assert "lines of messages" in blocked[-1]
+
+
+def test_atomic_message_admission_does_not_cross_limit(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    path = tmp_path / "queue.sqlite3"
+    seed = Store(path)
+    seed.configure_window("session-1", "source", (1, 100, 100))
+    seed.close()
+
+    def send(body):
+        store = Store(path)
+        try:
+            broker = InputBroker(
+                store,
+                BrokerIdentity("session-1", "source"),
+                limits=(1, 100, 100),
+            )
+            return broker.inspect_worker_line(f"To:target: {body}") is not None
+        finally:
+            store.close()
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        accepted = list(executor.map(send, ("one", "two")))
+    assert sorted(accepted) == [False, True]
+    final = Store(path)
+    assert len(final.list()) == 1
+    assert final.line_state("session-1", "source")["messages"] == 1
+
+
+def test_new_session_generation_starts_a_fresh_ledger(tmp_path, monkeypatch):
+    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
+    store = Store(tmp_path / "queue.sqlite3")
+    first = InputBroker(store, BrokerIdentity("generation-1", "source"), limits=(1, 10, 10))
+    second = InputBroker(store, BrokerIdentity("generation-2", "source"), limits=(1, 10, 10))
+    assert first.inspect_operator_line("To:target: first") is not None
+    assert second.inspect_operator_line("To:target: second") is not None
+    assert len(store.list()) == 2

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", "@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
@@ -24,7 +24,7 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk
     assert row["sender_repo"] == "flex-auth"
     assert row["target_repo"] == "audit-core"
     assert row["body"] == "review this"
-    assert row["state"] == "pending"
+    assert row["provenance"] == "operator_input"
 
     assert main(["ack", message_id]) == 0
     capsys.readouterr()
@@ -32,103 +32,6 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk
     assert capsys.readouterr().out == ""
 
 
-def test_bare_reply_targets_latest_inbound_sender(tmp_path, monkeypatch, capsys):
-    monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
-    monkeypatch.setenv("TAMQ_REPO", "audit-core")
-    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", "older")
-    latest = store.add("railiance-platform", "audit-core", "latest")
-    store.acknowledge(latest)
-    store.add("audit-core", "audit-core", "self note")
-    store.close()
-
-    assert main(["reply", "--", "--looks-like-an-option", "thanks"]) == 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["sender_repo"] == "audit-core"
-    assert row["target_repo"] == "railiance-platform"
-    assert row["body"] == "--looks-like-an-option thanks"
-    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)
-    monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None)
-    monkeypatch.delenv("TAMQ_REPO", raising=False)
-
-    assert main(["reply", "hello"]) == 2
-    assert "managed window" in capsys.readouterr().err
-
-    monkeypatch.setenv("TAMQ_REPO", "audit-core")
-    assert main(["reply", "hello"]) == 2
-    assert "no counterparty" in capsys.readouterr().err
-
-
 def test_inbox_requires_repository_outside_managed_window(tmp_path, monkeypatch, capsys):
     monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
     monkeypatch.delenv("TAMQ_REPO", raising=False)
@@ -136,16 +39,19 @@ def test_inbox_requires_repository_outside_managed_window(tmp_path, monkeypatch,
     assert "requires --repo" in capsys.readouterr().err
 
 
-def test_human_inbox_is_comment_safe_on_every_line(tmp_path, monkeypatch, capsys):
+def test_human_inbox_uses_readable_origin_framing(tmp_path, monkeypatch, capsys):
     monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
     monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None)
     store = Store(tmp_path / "state" / "tamq.sqlite3")
-    message_id = store.add("flex-auth", "audit-core", "first\nsecond\x1b[31m")
+    store.add(
+        "flex-auth", "audit-core", "first\nsecond\x1b[31m",
+        provenance="operator_input",
+    )
     store.close()
 
     assert main(["inbox", "--repo", "audit-core"]) == 0
     assert capsys.readouterr().out == (
-        f"#flex-auth: first\n# second\\x1b[31m [{message_id}]\n"
+        "From:flex-auth/o: first\\x0asecond\\x1b[31m\n"
     )
 
 
@@ -153,8 +59,8 @@ def test_inbox_filter_logs_and_acknowledges_successful_messages(tmp_path, monkey
     monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
     monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None)
     store = Store(tmp_path / "state" / "tamq.sqlite3")
-    first_id = store.add("flex-auth", "audit-core", "review 'quoted' $value")
-    second_id = store.add("railiance-platform", "audit-core", "second")
+    store.add("flex-auth", "audit-core", "review 'quoted' $value", provenance="operator_input")
+    store.add("railiance-platform", "audit-core", "second", provenance="worker_output")
     store.close()
     log = tmp_path / "messages.log"
 
@@ -163,8 +69,8 @@ def test_inbox_filter_logs_and_acknowledges_successful_messages(tmp_path, monkey
         f"cat >> {shlex.quote(str(log))}",
     ]) == 0
     assert log.read_text(encoding="utf-8") == (
-        f"#flex-auth: review 'quoted' $value [{first_id}]\n"
-        f"#railiance-platform: second [{second_id}]\n"
+        "From:flex-auth/o: review 'quoted' $value\n"
+        "From:railiance-platform: second\n"
     )
     assert main(["inbox", "--repo", "audit-core"]) == 0
     assert capsys.readouterr().out == ""
@@ -183,7 +89,6 @@ def test_failed_inbox_filter_leaves_current_and_later_messages_pending(tmp_path,
     assert main(["inbox", "--repo", "audit-core", "--json"]) == 0
     rows = [json.loads(line) for line in capsys.readouterr().out.splitlines()]
     assert [row["message_id"] for row in rows] == [first_id, second_id]
-    assert all(row["state"] == "pending" for row in rows)
 
 
 def test_inbox_filter_rejects_non_pending_and_json_modes(tmp_path, monkeypatch, capsys):

tests/test_ptytap.py

@@ -4,7 +4,13 @@ import pty
 import struct
 import termios
 
-from tamq.ptytap import PtyTap, copy_winsize, observed_line, write_all
+from tamq.ptytap import (
+    PtyTap,
+    TerminalOutputObserver,
+    copy_winsize,
+    observed_line,
+    write_all,
+)
 
 
 class RecordingBroker:
@@ -13,7 +19,7 @@ class RecordingBroker:
 
     def inspect_line(self, line):
         self.lines.append(line)
-        return line.startswith(("@", "#"))
+        return line.startswith("To:")
 
 
 def test_input_observer_accepts_raw_terminal_carriage_returns():
@@ -22,11 +28,11 @@ def test_input_observer_accepts_raw_terminal_carriage_returns():
     tap = PtyTap(["true"], broker, on_line=observed.append)
     buffer = bytearray()
 
-    tap._observe_input(buffer, b"#activity-core: hel")
+    tap._observe_input(buffer, b"To:activity-core: hel")
     tap._observe_input(buffer, b"lo\rplain line\n")
 
-    assert broker.lines == ["#activity-core: hello", "plain line"]
-    assert observed == ["#activity-core: hello"]
+    assert broker.lines == ["To:activity-core: hello", "plain line"]
+    assert observed == ["To:activity-core: hello"]
 
 
 def test_input_observer_accepts_enhanced_terminal_enter_sequences():
@@ -35,17 +41,17 @@ def test_input_observer_accepts_enhanced_terminal_enter_sequences():
     tap = PtyTap(["true"], broker, on_line=observed.append)
     buffer = bytearray()
 
-    tap._observe_input(buffer, b"#activity-core: CSI-u\x1b[13;1")
+    tap._observe_input(buffer, b"To:activity-core: CSI-u\x1b[13;1")
     assert broker.lines == []
-    tap._observe_input(buffer, b":1u#other: modify\x1b[27;1;13~")
+    tap._observe_input(buffer, b":1uTo:other: modify\x1b[27;1;13~")
     tap._observe_input(buffer, b"plain\x1bOM")
 
     assert broker.lines == [
-        "#activity-core: CSI-u",
-        "#other: modify",
+        "To:activity-core: CSI-u",
+        "To:other: modify",
         "plain",
     ]
-    assert observed == ["#activity-core: CSI-u", "#other: modify"]
+    assert observed == ["To:activity-core: CSI-u", "To:other: modify"]
 
 
 def test_input_observer_ignores_terminal_replies_before_typed_line():
@@ -55,20 +61,36 @@ def test_input_observer_ignores_terminal_replies_before_typed_line():
     buffer = bytearray()
 
     tap._observe_input(buffer, b"\x1b[9;1R\x1b[?1;2;4c")
-    tap._observe_input(buffer, b"#activity-core: Helol\x7f\x7flo\r")
+    tap._observe_input(buffer, b"To:activity-core: Helol\x7f\x7flo\r")
 
-    assert broker.lines == ["#activity-core: Hello"]
-    assert observed == ["#activity-core: Hello"]
+    assert broker.lines == ["To:activity-core: Hello"]
+    assert observed == ["To:activity-core: Hello"]
 
 
 def test_observed_line_removes_common_terminal_protocol_wrappers():
     raw = (
-        b"\x1b[200~#target: pasted\x1b[201~"
+        b"\x1b[200~To:target: pasted\x1b[201~"
         b"\x1b]10;rgb:ffff/ffff/ffff\x07"
         b"\x1bP1$r0m\x1b\\"
     )
 
-    assert observed_line(raw) == "#target: pasted"
+    assert observed_line(raw) == "To:target: pasted"
+
+
+def test_output_observer_normalizes_ansi_and_deduplicates_redraws():
+    lines = []
+    observer = TerminalOutputObserver(lines.append)
+    observer.feed(b"plain\r\n\x1b[31mTo:target: worker\x1b[0m\r")
+    observer.feed(b"To:target: worker\r")
+    assert lines == ["plain", "To:target: worker"]
+
+
+def test_output_observer_fails_closed_on_recent_operator_echo():
+    lines = []
+    observer = TerminalOutputObserver(lines.append)
+    observer.note_operator_line("To:target: operator")
+    observer.feed(b"To:target: operator\r\nTo:target: worker\r\n")
+    assert lines == ["To:target: worker"]
 
 
 def test_copy_winsize_preserves_rows_columns_and_pixels():

tests/test_ptytap_integration.py

@@ -8,6 +8,7 @@ from uuid import uuid4
 import pytest
 
 from tamq.tmux import LaunchPlan, TmuxManager
+from tamq.store import Store
 
 
 @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
@@ -98,3 +99,82 @@ def test_tap_propagates_terminal_size_resize_and_raw_mouse_input(tmp_path, monke
         assert f"RESIZE {resized}" in output
     finally:
         manager._run("kill-server", check=False)
+
+
+@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
+def test_real_pty_attributes_worker_output_and_suppresses_operator_echo(
+    tmp_path, monkeypatch
+):
+    source = tmp_path / "source"
+    source.mkdir()
+    fixture = tmp_path / "duplex_fixture.py"
+    fixture.write_text(
+        "import sys\n"
+        "print('READY', flush=True)\n"
+        "for line in sys.stdin:\n"
+        "    print(line.rstrip('\\r\\n'), flush=True)\n"
+        "    print('To:target: worker reply', flush=True)\n",
+        encoding="utf-8",
+    )
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    gita = bin_dir / "gita"
+    gita.write_text(
+        "#!/bin/sh\n"
+        "if [ \"$1\" = ls ]; then printf 'source target\\n'; exit 0; fi\n"
+        "exit 2\n",
+        encoding="utf-8",
+    )
+    gita.chmod(0o755)
+    project_src = str(Path(__file__).resolve().parents[1] / "src")
+    monkeypatch.setenv("PYTHONPATH", project_src)
+    monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
+    state = tmp_path / "state"
+    monkeypatch.setenv("TAMQ_STATE_DIR", str(state))
+    socket_name = f"tamq-duplex-{uuid4().hex[:8]}"
+    session = f"tamq-duplex-{uuid4().hex[:8]}"
+    manager = TmuxManager(
+        session,
+        tmux_command=("tmux", "-L", socket_name),
+        tamq_command=(sys.executable, "-m", "tamq.cli"),
+        command_dir=tmp_path / "commands",
+    )
+    plan = LaunchPlan(
+        ("source",), {"source": str(source)}, (sys.executable, str(fixture))
+    )
+    store = None
+    try:
+        endpoint = manager.ensure_plan(plan, tap=True, tap_limits=(8, 1024, 32768))
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            if "READY" in manager._run(
+                "capture-pane", "-p", "-t", f"{session}:source"
+            ):
+                break
+            time.sleep(0.05)
+        manager._run(
+            "send-keys",
+            "-t",
+            f"{session}:source",
+            "-l",
+            "--",
+            "To:target: operator request",
+        )
+        manager._run("send-keys", "-t", f"{session}:source", "Enter")
+        store = Store(state / "tamq.sqlite3")
+        deadline = time.monotonic() + 5
+        rows = []
+        while time.monotonic() < deadline:
+            rows = store.list()
+            if len(rows) >= 2:
+                break
+            time.sleep(0.05)
+        assert [(row["body"], row["provenance"]) for row in rows] == [
+            ("operator request", "operator_input"),
+            ("worker reply", "worker_output"),
+        ]
+        assert all(row["endpoint_id"] == endpoint.instance_id for row in rows)
+    finally:
+        if store is not None:
+            store.close()
+        manager._run("kill-server", check=False)

tests/test_routing.py

@@ -1,17 +1,30 @@
 import pytest
 
-from tamq.routing import parse_address_line
+from tamq.routing import parse_address_line, parse_command_line
 
 
-@pytest.mark.parametrize("prefix", ["@", "#"])
-def test_direct_address(prefix):
-    routed = parse_address_line(f"{prefix}railiance-platform: do something!")
+def test_direct_address_uses_readable_case_sensitive_grammar():
+    routed = parse_address_line("To:railiance-platform: do something!")
     assert routed.body == "do something!"
     assert routed.target_repo == "railiance-platform"
 
 
-def test_ordinary_input_is_unchanged():
-    assert parse_address_line("hello @repo: not at start") is None
-    assert parse_address_line("@repo:") is None
-    assert parse_address_line("# from repo: inbound envelope") is None
-    assert parse_address_line("##repo: not an address") is None
+@pytest.mark.parametrize(
+    "line",
+    [
+        "@repo: removed",
+        "#repo: removed",
+        "to:repo: wrong case",
+        "hello To:repo: not at start",
+        "To:repo:",
+        "From:repo: inbound envelope",
+    ],
+)
+def test_non_protocol_and_legacy_input_is_not_routed(line):
+    assert parse_address_line(line) is None
+
+
+def test_operator_command_grammar():
+    assert parse_command_line("Cmd: mode=trigger") == "mode=trigger"
+    assert parse_command_line("Cmd: reset-limits") == "reset-limits"
+    assert parse_command_line("cmd: mode=trigger") is None

tests/test_send_endpoint.py

@@ -4,5 +4,5 @@ from tamq.cli import build_parser
 
 
 def test_send_accepts_endpoint_id():
-    args = build_parser().parse_args(["send", "--endpoint-id", "ep", "@repo:", "hello"])
+    args = build_parser().parse_args(["send", "--endpoint-id", "ep", "To:repo:", "hello"])
     assert args.endpoint_id == "ep"

tests/test_service_delivery.py

@@ -9,6 +9,7 @@ class FakeControl:
     def __init__(self, session):
         self.session = session
         self.injected = []
+        self.placed = []
         self.submitted = []
         self.__class__.instances.append(self)
 
@@ -24,6 +25,9 @@ class FakeControl:
     def submit(self, window, text):
         self.submitted.append((window, text))
 
+    def place(self, window, text):
+        self.placed.append((window, text))
+
     def pane_display(self, window):
         return PaneDisplay(
             tty_path=f"/dev/pts/{window.rsplit(':', 1)[-1]}",
@@ -43,7 +47,7 @@ def test_service_delivery_injects_pending_messages(tmp_path, monkeypatch):
     message_id = store.add("repo-a", "repo-b", "continue")
     monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl)
     Service(store=store)._deliver_once()
-    assert FakeControl.instances[-1].injected == [("tamq:repo-b", "#repo-a: continue")]
+    assert FakeControl.instances[-1].injected == [("tamq:repo-b", "From:repo-a: continue")]
     assert store.list()[0]["message_id"] == message_id
     assert store.list()[0]["state"] == "injected"
 
@@ -60,7 +64,7 @@ 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):
+def test_service_pushy_mode_places_once_without_enter(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")
@@ -71,22 +75,36 @@ def test_service_pushy_mode_submits_once_and_marks_injected(tmp_path, monkeypatc
     control = FakeControl.instances[-1]
     service._deliver_once()
 
-    assert control.submitted == [
+    assert control.placed == [
         (
             "tamq:repo-b",
-            f"# from repo-a: first\\x0asecond [{message_id}]",
+            "From:repo-a: first\\x0asecond",
         )
     ]
+    assert control.submitted == []
     assert store.list()[0]["state"] == "injected"
 
 
+def test_service_trigger_mode_submits_exactly_once(tmp_path, monkeypatch):
+    store = Store(tmp_path / "queue.sqlite3")
+    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._deliver_once()
+    control = FakeControl.instances[-1]
+    service._deliver_once()
+    assert control.submitted == [("tamq:repo-b", "From:repo-a/o: go")]
+    assert control.placed == []
+
+
 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):
+        def place(self, window, text):
             raise RuntimeError("tmux rejected input")
 
     monkeypatch.setattr("tamq.service.ControlModeClient", FailingControl)
@@ -113,7 +131,7 @@ def test_service_writes_output_once_and_retains_pending_ack(tmp_path, monkeypatc
     assert writes == [
         (
             "/dev/pts/repo-b",
-            f"\r\n#repo-a: continue [{message_id}]\r\n",
+            "\r\nFrom:repo-a: continue\r\n",
         )
     ]
     row = store.list()[0]

tests/test_store.py

@@ -41,7 +41,7 @@ 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] == "3"
+    ).fetchone()[0] == "4"
     assert "displayed_at" in {
         row["name"] for row in store.db.execute("PRAGMA table_info(messages)")
     }
@@ -59,15 +59,3 @@ def test_mark_displayed_releases_lease_without_acknowledging(tmp_path):
     assert row["displayed_at"] is not None
     assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0
     store.close()
-
-
-def test_latest_counterparty_uses_latest_inbound_message_regardless_of_state(tmp_path):
-    store = Store(tmp_path / "queue.sqlite3")
-    store.add("audit-core", "audit-core", "self note")
-    older = store.add("flex-auth", "audit-core", "first")
-    store.acknowledge(older)
-    store.add("railiance-platform", "audit-core", "latest")
-
-    assert store.latest_counterparty("audit-core") == "railiance-platform"
-    assert store.latest_counterparty("unknown") is None
-    store.close()

tests/test_terminal_output.py

@@ -6,55 +6,48 @@ import pytest
 
 from tamq.terminal import (
     TerminalOutputError,
-    format_comment,
-    format_pushy_input,
+    format_delivery,
     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]"
+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"
     )
-
-
-def test_pushy_input_is_one_sanitized_shell_comment():
-    assert format_pushy_input("repo-a", "first\nsecond\x1b[31m", "m-1") == (
-        "# from repo-a: first\\x0asecond\\x1b[31m [m-1]"
+    assert format_delivery("repo-a", "worker", "worker_output") == (
+        "From:repo-a: worker"
     )
 
 
 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
+        "repo-a", "hello", "operator_input", cursor_y=8, pane_height=24
     ) == (
         "\x1b7\x1b[1;8r\x1b[8;1H"
-        "\n\r#repo-a: first\n\r# second [m-1]"
+        "\n\rFrom:repo-a/o: hello"
         "\x1b[r\x1b8"
     )
 
 
 def test_terminal_frame_falls_back_when_stable_region_is_not_safe():
-    expected = "\r\n#repo-a: hello [m-1]\r\n"
-    assert terminal_frame("repo-a", "hello", "m-1", cursor_y=0, pane_height=24) == expected
+    expected = "\r\nFrom:repo-a: hello\r\n"
+    assert terminal_frame("repo-a", "hello", "worker_output", cursor_y=0, pane_height=24) == expected
     assert terminal_frame(
-        "repo-a", "hello", "m-1", cursor_y=8, pane_height=24, alternate_on=True
+        "repo-a", "hello", "worker_output", cursor_y=8, pane_height=24, alternate_on=True
     ) == expected
-    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"))
+        write_terminal_output(tty_path, terminal_frame("repo-a", "hello", "worker_output"))
 
         readable, _, _ = select.select([master], [], [], 1)
         assert readable == [master]
-        assert b"#repo-a: hello [m-1]" in os.read(master, 4096)
+        assert b"From:repo-a: hello" in os.read(master, 4096)
         readable_input, _, _ = select.select([slave], [], [], 0)
         assert readable_input == []
     finally:

tests/test_tmux_integration.py

@@ -105,10 +105,10 @@ def test_real_tmux_starts_two_repo_windows_and_reuses_them(tmp_path, monkeypatch
                 capture = manager._run(
                     "capture-pane", "-p", "-t", f"{session}:activity-core"
                 )
-                if "#local: integration-message" in capture:
+                if "From:local: integration-message" in capture:
                     break
                 time.sleep(0.05)
-            assert "#local: integration-message" in capture
+            assert "From:local: integration-message" in capture
         finally:
             control.close()
             store.close()
@@ -173,7 +173,7 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path):
             terminal_frame(
                 "flex-auth",
                 "stable-message",
-                "m-stable",
+                "worker_output",
                 cursor_y=before.cursor_y,
                 pane_height=before.pane_height,
                 alternate_on=before.alternate_on,
@@ -183,11 +183,11 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path):
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
             after_capture = manager._run("capture-pane", "-p", "-t", target)
-            if "#flex-auth: stable-message [m-stable]" in after_capture:
+            if "From:flex-auth: stable-message" in after_capture:
                 break
             time.sleep(0.05)
         after = control.pane_display(target)
-        assert "#flex-auth: stable-message [m-stable]" in after_capture
+        assert "From:flex-auth: stable-message" in after_capture
         before_input = next(line for line in before_capture.splitlines() if "PARTIAL-INPUT" in line)
         after_lines = after_capture.splitlines()
         assert any("PARTIAL-INPUT" in line for line in after_lines), repr(after_capture)
@@ -195,14 +195,14 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path):
             index for index, line in enumerate(after_lines) if "PARTIAL-INPUT" in line
         )
         assert after_lines[after_input_index] == before_input
-        assert after_lines[after_input_index - 1] == "#flex-auth: stable-message [m-stable]"
+        assert after_lines[after_input_index - 1] == "From:flex-auth: stable-message"
         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):
+def test_real_tmux_pushy_mode_places_one_readable_input_without_enter(tmp_path, monkeypatch):
     repo = tmp_path / "audit-core"
     repo.mkdir()
     socket_name = f"tamq-pushy-{os.getpid()}-{uuid4().hex[:8]}"
@@ -238,11 +238,11 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch
             endpoint.repos,
             "pushy",
         )
-        message_id = store.add("flex-auth", "audit-core", "What's next?\nsecond")
+        store.add("flex-auth", "audit-core", "What's next?\nsecond", provenance="operator_input")
         service = Service(store=store)
         service._deliver_once()
 
-        expected = f"# from flex-auth: What's next?\\x0asecond [{message_id}]"
+        expected = "From:flex-auth/o: What's next?\\x0asecond"
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
             capture = manager._run("capture-pane", "-p", "-J", "-t", target)
@@ -262,7 +262,7 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch
 
 
 @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
-def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch):
+def test_real_tmux_to_route_triggers_once_without_feedback(tmp_path, monkeypatch):
     repo_a = tmp_path / "railiance-platform"
     repo_b = tmp_path / "activity-core"
     bin_dir = tmp_path / "bin"
@@ -330,7 +330,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch
             endpoint.pid,
             endpoint.session,
             endpoint.repos,
-            "pushy",
+            "trigger",
         )
         manager._run(
             "send-keys",
@@ -338,7 +338,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch
             f"{session}:railiance-platform",
             "-l",
             "--",
-            "#activity-core: Hello!",
+            "To:activity-core: Hello!",
         )
         manager._run("send-keys", "-t", f"{session}:railiance-platform", "Enter")
 
@@ -357,7 +357,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch
 
         service = Service(store=store)
         service._deliver_once()
-        expected = f"# from railiance-platform: Hello! [{row['message_id']}]"
+        expected = "From:railiance-platform/o: Hello!"
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
             target_capture = manager._run(
@@ -369,9 +369,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch
         assert f"AGENT:{expected}" in target_capture
         assert store.list()[0]["state"] == "injected"
 
-        legacy_envelope = (
-            f"#railiance-platform: Hello! [{row['message_id']}]"
-        )
+        legacy_envelope = "#railiance-platform: Hello!"
         manager._run(
             "send-keys",
             "-t",

tests/test_tmux_manager.py

@@ -98,6 +98,15 @@ def test_preflight_rejects_foreign_session(tmp_path, monkeypatch):
         manager.preflight(["a"])
 
 
+def test_tap_refuses_managed_session_without_duplex_protocol(tmp_path, monkeypatch):
+    manager = tmux.TmuxManager("tamq-test", command_dir=tmp_path / "commands")
+    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"):
+        manager.ensure_plan(plan, tap=True)
+
+
 def test_tmux_only_plan_starts_agent_without_tap(tmp_path, monkeypatch):
     manager = tmux.TmuxManager("tamq-test", command_dir=tmp_path / "commands")
     calls = []
@@ -152,39 +161,27 @@ def test_neutral_plan_starts_shell_without_sending_keystrokes(tmp_path, monkeypa
     assert "TAMQ_REPO=b" in new_window
     assert f"PATH={command_dir}{os.pathsep}" in " ".join(new_session)
     assert f"PATH={command_dir}{os.pathsep}" in " ".join(new_window)
-    assert (command_dir / "@a").is_file()
-    assert (command_dir / "@a:").is_file()
-    assert (command_dir / "@b").is_file()
-    assert (command_dir / "@b:").is_file()
-    assert (command_dir / "@").is_file()
+    assert (command_dir / "To:a:").is_file()
+    assert (command_dir / "To:b:").is_file()
+    assert (command_dir / "Cmd:").is_file()
+    assert not any(path.name.startswith("@") for path in command_dir.iterdir())
 
 
-def test_address_commands_preserve_message_arguments(tmp_path):
+def test_protocol_commands_absorb_forwarded_shell_lines(tmp_path):
     command_dir = tmp_path / "commands"
     manager = tmux.TmuxManager(
         "tamq-test", tamq_command=("echo",), command_dir=command_dir,
     )
     manager._install_address_commands(["audit-core"])
 
-    for name in ("@audit-core", "@audit-core:"):
+    for name in ("To:audit-core:", "Cmd:"):
         result = subprocess.run(
-            [str(command_dir / name), "Some message!", "$value", "; literal", "--from"],
+            [str(command_dir / name), "Some message!", "$value"],
             text=True,
             capture_output=True,
             check=True,
         )
-        assert result.stdout == "send -- @audit-core: Some message! $value ; literal --from\n"
-
-    result = subprocess.run(
-        [str(command_dir / "@"), "Some reply!", "$value", "--literal"],
-        text=True,
-        capture_output=True,
-        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
+        assert result.stdout == ""
 
 
 def test_address_commands_reject_unsafe_repository_names(tmp_path):
@@ -196,7 +193,7 @@ def test_address_commands_reject_unsafe_repository_names(tmp_path):
 def test_address_commands_do_not_replace_unowned_commands(tmp_path):
     command_dir = tmp_path / "commands"
     command_dir.mkdir()
-    existing = command_dir / "@audit-core"
+    existing = command_dir / "To:audit-core:"
     existing.write_text("#!/bin/sh\necho mine\n", encoding="utf-8")
     manager = tmux.TmuxManager("tamq-test", command_dir=command_dir)
 
@@ -205,10 +202,10 @@ def test_address_commands_do_not_replace_unowned_commands(tmp_path):
     assert existing.read_text(encoding="utf-8") == "#!/bin/sh\necho mine\n"
 
 
-def test_address_commands_do_not_replace_unowned_bare_reply(tmp_path):
+def test_address_commands_do_not_replace_unowned_command_lane(tmp_path):
     command_dir = tmp_path / "commands"
     command_dir.mkdir()
-    existing = command_dir / "@"
+    existing = command_dir / "Cmd:"
     existing.write_text("#!/bin/sh\necho mine\n", encoding="utf-8")
     manager = tmux.TmuxManager("tamq-test", command_dir=command_dir)