feat: add hash routing for pushy agents

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

tests/test_broker.py

@@ -9,3 +9,13 @@ def test_broker_preserves_identity(tmp_path):
     row = store.list()[0]
     assert row["sender_repo"] == "net-kingdom"
     assert row["endpoint_id"] == "tmux-amq-42"
+
+
+def test_broker_accepts_hash_address_alias(tmp_path):
+    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"

tests/test_cli.py

@@ -209,10 +209,11 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys):
     class Manager:
         def preflight(self, repos, command):
             assert repos == ["a", "b"]
+            assert command == "codex"
             return "plan"
 
         def ensure_plan(self, plan, *, tap=True):
-            assert tap is False
+            assert tap is True
             return endpoint
 
         def rollback(self, value):
@@ -231,10 +232,13 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys):
     )
     monkeypatch.setattr("tamq.cli.request", register)
 
-    assert main(["start", "--detach", "--mode", "pushy", "a", "b"]) == 0
+    assert main([
+        "start", "--detach", "--mode", "pushy", "--command", "codex", "a", "b"
+    ]) == 0
     summary = json.loads(capsys.readouterr().out)
     assert summary["mode"] == "pushy"
     assert summary["delivery_mode"] == "pushy"
+    assert summary["input_observation_requested"] is True
     assert requests[0]["delivery_mode"] == "pushy"
 
 

tests/test_ptytap.py

@@ -13,7 +13,7 @@ class RecordingBroker:
 
     def inspect_line(self, line):
         self.lines.append(line)
-        return line.startswith("@")
+        return line.startswith(("@", "#"))
 
 
 def test_input_observer_accepts_raw_terminal_carriage_returns():
@@ -22,11 +22,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"#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 == ["#activity-core: hello", "plain line"]
+    assert observed == ["#activity-core: hello"]
 
 
 def test_copy_winsize_preserves_rows_columns_and_pixels():

tests/test_routing.py

@@ -1,11 +1,17 @@
+import pytest
+
 from tamq.routing import parse_address_line
 
 
-def test_direct_address():
-    assert parse_address_line("@railiance-platform: do something!").body == "do something!"
-    assert parse_address_line("@railiance-platform: do something!").target_repo == "railiance-platform"
+@pytest.mark.parametrize("prefix", ["@", "#"])
+def test_direct_address(prefix):
+    routed = parse_address_line(f"{prefix}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

tests/test_service_delivery.py

@@ -74,7 +74,7 @@ def test_service_pushy_mode_submits_once_and_marks_injected(tmp_path, monkeypatc
     assert control.submitted == [
         (
             "tamq:repo-b",
-            f"#repo-a: first\\x0asecond [{message_id}]",
+            f"# from repo-a: first\\x0asecond [{message_id}]",
         )
     ]
     assert store.list()[0]["state"] == "injected"

tests/test_terminal_output.py

@@ -21,7 +21,7 @@ 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]"
+        "# from repo-a: first\\x0asecond\\x1b[31m [m-1]"
     )
 
 

tests/test_tmux_integration.py

@@ -242,10 +242,10 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch
         service = Service(store=store)
         service._deliver_once()
 
-        expected = f"#flex-auth: What's next?\\x0asecond [{message_id}]"
+        expected = f"# from flex-auth: What's next?\\x0asecond [{message_id}]"
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
-            capture = manager._run("capture-pane", "-p", "-t", target)
+            capture = manager._run("capture-pane", "-p", "-J", "-t", target)
             if expected in capture:
                 break
             time.sleep(0.05)
@@ -254,8 +254,128 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch
         assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0
 
         service._deliver_once()
-        repeated_capture = manager._run("capture-pane", "-p", "-t", target)
+        repeated_capture = manager._run("capture-pane", "-p", "-J", "-t", target)
         assert repeated_capture.count(expected) == 1
     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_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch):
+    repo_a = tmp_path / "railiance-platform"
+    repo_b = tmp_path / "activity-core"
+    bin_dir = tmp_path / "bin"
+    for path in (repo_a, repo_b, bin_dir):
+        path.mkdir()
+    gita = bin_dir / "gita"
+    gita.write_text(
+        "#!/bin/sh\n"
+        "if [ \"$1\" = ls ]; then printf 'railiance-platform activity-core\\n'; exit 0; fi\n"
+        "exit 2\n",
+        encoding="utf-8",
+    )
+    gita.chmod(0o755)
+    fixture = tmp_path / "agent_fixture.py"
+    fixture.write_text(
+        "import sys\n"
+        "print('AGENT-READY', flush=True)\n"
+        "for line in sys.stdin:\n"
+        "    print('AGENT:' + line.rstrip('\\r\\n'), flush=True)\n",
+        encoding="utf-8",
+    )
+    project_src = str(Path(__file__).resolve().parents[1] / "src")
+    existing_pythonpath = os.environ.get("PYTHONPATH")
+    monkeypatch.setenv(
+        "PYTHONPATH",
+        project_src if not existing_pythonpath else f"{project_src}:{existing_pythonpath}",
+    )
+    monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
+    monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))
+    socket_name = f"tamq-hash-{os.getpid()}-{uuid4().hex[:8]}"
+    session = f"tamq-hash-{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",
+    )
+    plan = LaunchPlan(
+        ("railiance-platform", "activity-core"),
+        {
+            "railiance-platform": str(repo_a),
+            "activity-core": str(repo_b),
+        },
+        (sys.executable, str(fixture)),
+    )
+    store = None
+
+    try:
+        endpoint = manager.ensure_plan(plan, tap=True)
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            captures = {
+                repo: manager._run("capture-pane", "-p", "-t", f"{session}:{repo}")
+                for repo in plan.repos
+            }
+            if all("AGENT-READY" in output for output in captures.values()):
+                break
+            time.sleep(0.05)
+        assert all("AGENT-READY" in output for output in captures.values())
+
+        store = Store(tmp_path / "state" / "tamq.sqlite3")
+        store.register_endpoint(
+            endpoint.instance_key,
+            endpoint.pid,
+            endpoint.session,
+            endpoint.repos,
+            "pushy",
+        )
+        manager._run(
+            "send-keys",
+            "-t",
+            f"{session}:railiance-platform",
+            "-l",
+            "--",
+            "#activity-core: Hello!",
+        )
+        manager._run("send-keys", "-t", f"{session}:railiance-platform", "Enter")
+
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            rows = store.list()
+            if rows:
+                break
+            time.sleep(0.05)
+        assert len(rows) == 1
+        row = rows[0]
+        assert row["sender_repo"] == "railiance-platform"
+        assert row["target_repo"] == "activity-core"
+        assert row["body"] == "Hello!"
+        assert row["endpoint_id"] == endpoint.instance_key
+
+        service = Service(store=store)
+        service._deliver_once()
+        expected = f"# from railiance-platform: Hello! [{row['message_id']}]"
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            target_capture = manager._run(
+                "capture-pane", "-p", "-J", "-t", f"{session}:activity-core"
+            )
+            if f"AGENT:{expected}" in target_capture:
+                break
+            time.sleep(0.05)
+        assert f"AGENT:{expected}" in target_capture
+        assert store.list()[0]["state"] == "injected"
+
+        service._deliver_once()
+        time.sleep(0.1)
+        assert len(store.list()) == 1
+        assert manager._run(
+            "capture-pane", "-p", "-J", "-t", f"{session}:activity-core"
+        ).count(f"AGENT:{expected}") == 1
+    finally:
+        if store is not None:
+            store.close()
+        manager._run("kill-server", check=False)

tests/test_tmux_manager.py

@@ -37,6 +37,7 @@ def test_tmux_manager_builds_tap_windows(tmp_path, monkeypatch):
     tap_calls = [call for call in calls if call and call[0] == "send-keys"]
     assert len(tap_calls) == 2
     assert all("tamq tap" in " ".join(call) for call in tap_calls)
+    assert all(endpoint.instance_id in " ".join(call) for call in tap_calls)
     assert all("codex --quiet" in " ".join(call) for call in tap_calls)
     assert ("select-window", "-t", "tamq-test:a") in calls
     assert any(call and call[0] == "set-option" for call in calls)