fix: prevent mixed-version pushy feedback

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

src/tamq/broker.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 
 from .routing import RoutedMessage, parse_address_line
@@ -7,6 +8,12 @@ from .store import Store
 from .registry import RegistryError, validate_targets
 
 
+LEGACY_DELIVERY_SUFFIX = re.compile(
+    r"\s+\[(m-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\]$",
+    re.IGNORECASE,
+)
+
+
 @dataclass(frozen=True)
 class BrokerIdentity:
     endpoint_id: str
@@ -24,6 +31,15 @@ class InputBroker:
         routed = parse_address_line(line)
         if routed is None:
             return None
+        receipt = LEGACY_DELIVERY_SUFFIX.search(routed.body)
+        if line.startswith("#") and receipt:
+            prior = self.store.message(receipt.group(1))
+            if (
+                prior is not None
+                and prior["sender_repo"] == routed.target_repo
+                and prior["target_repo"] == self.identity.source_repo
+            ):
+                return None
         try:
             validate_targets([routed.target_repo])
         except RegistryError:

src/tamq/cli.py

@@ -18,7 +18,7 @@ from pathlib import Path
 
 from . import __version__
 from .config import db_path, pid_path, lock_path, config_path, socket_path, state_dir, setting
-from .service import Service, ping, request
+from .service import PUSHY_FRAMING_CAPABILITY, Service, ping, request
 from .store import Store
 from .tmux import TmuxError, TmuxManager
 from .broker import BrokerIdentity, InputBroker
@@ -130,7 +130,9 @@ def ensure_output_service() -> bool:
 
 
 def ensure_pushy_service() -> bool:
-    return ensure_service_capabilities({"manual_delivery", "pushy_input"})
+    return ensure_service_capabilities(
+        {"manual_delivery", "pushy_input", PUSHY_FRAMING_CAPABILITY}
+    )
 
 
 def preflight_runtime_paths() -> None:

src/tamq/service.py

@@ -16,6 +16,16 @@ from .control import ControlModeClient
 from .terminal import format_pushy_input, terminal_frame, write_terminal_output
 
 PROTOCOL_VERSION = "0.1"
+PUSHY_FRAMING_CAPABILITY = "pushy_input_non_routable_v1"
+SERVICE_CAPABILITIES = [
+    "register",
+    "send",
+    "history",
+    "manual_delivery",
+    "terminal_output",
+    "pushy_input",
+    PUSHY_FRAMING_CAPABILITY,
+]
 
 
 class Service:
@@ -144,7 +154,7 @@ class Service:
             if protocol.split(".")[0] != PROTOCOL_VERSION.split(".")[0]:
                 response = {"ok": False, "error": f"incompatible protocol: {protocol}", "protocol": PROTOCOL_VERSION}
             elif op == "ping":
-                response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery", "terminal_output", "pushy_input"]}
+                response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": SERVICE_CAPABILITIES}
             elif op == "register":
                 required = ("endpoint_id", "pid", "session", "repos")
                 if any(key not in request for key in required):

src/tamq/store.py

@@ -153,6 +153,11 @@ class Store:
         where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
         return list(self.db.execute(f"SELECT * FROM messages{where} ORDER BY created_at", values))
 
+    def message(self, message_id: str) -> sqlite3.Row | None:
+        return self.db.execute(
+            "SELECT * FROM messages WHERE message_id=?", (message_id,)
+        ).fetchone()
+
     def latest_counterparty(self, target: str) -> str | None:
         row = self.db.execute(
             "SELECT sender_repo FROM messages "