feat: display messages as terminal output

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

src/tamq/cli.py

@@ -27,12 +27,13 @@ from .control import ControlModeClient
 from .diagnostics import configure
 from .policy import load_profile
 from .registry import RegistryError, validate_targets
+from .terminal import format_comment
 
 
 SUBCOMMANDS = frozenset(
     "start attach serve stop status ping inbox history inspect ack send export replay purge completion db-version config tap".split()
 )
-START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service"})
+START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service", "--no-display"})
 GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"})
 
 
@@ -99,15 +100,15 @@ def stop_service_process() -> int | None:
     return pid
 
 
-def ensure_manual_service() -> bool:
-    """Ensure the broker understands neutral endpoints before registration."""
+def ensure_service_capabilities(required: set[str]) -> bool:
+    """Restart an older broker once before registering a newer endpoint mode."""
     if not ensure_service():
         return False
     try:
         capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", [])
     except (OSError, json.JSONDecodeError):
         return False
-    if "manual_delivery" in capabilities:
+    if required.issubset(capabilities):
         return True
     stop_service_process()
     if not ensure_service():
@@ -116,7 +117,15 @@ def ensure_manual_service() -> bool:
         capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", [])
     except (OSError, json.JSONDecodeError):
         return False
-    return "manual_delivery" in capabilities
+    return required.issubset(capabilities)
+
+
+def ensure_manual_service() -> bool:
+    return ensure_service_capabilities({"manual_delivery"})
+
+
+def ensure_output_service() -> bool:
+    return ensure_service_capabilities({"manual_delivery", "terminal_output"})
 
 
 def preflight_runtime_paths() -> None:
@@ -133,24 +142,9 @@ def attach_session(session: str) -> int:
     return subprocess.run(command, check=False).returncode
 
 
-def _terminal_safe(text: str) -> str:
-    return "".join(
-        character
-        if character.isprintable()
-        else f"\\x{ord(character):02x}"
-        for character in text
-    )
-
-
 def format_comment_message(row: sqlite3.Row) -> str:
     """Render a durable message so every displayed line remains a shell comment."""
-    sender = _terminal_safe(str(row["sender_repo"]))
-    body = str(row["body"])
-    lines = body.splitlines() or [""]
-    rendered = [f"#{sender}: {_terminal_safe(lines[0])}"]
-    rendered.extend(f"# {_terminal_safe(line)}" for line in lines[1:])
-    rendered[-1] += f" [{_terminal_safe(str(row['message_id']))}]"
-    return "\n".join(rendered)
+    return format_comment(row["sender_repo"], row["body"], row["message_id"])
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -159,11 +153,12 @@ def build_parser() -> argparse.ArgumentParser:
         description="Repository-aware tmux sessions with durable local messaging.",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog="""session shorthand:
-  tamq [--detach] [--command COMMAND] REPO [REPO ...]
+  tamq [--detach] [--command COMMAND] [--no-display] REPO [REPO ...]
 
   With no --command, tamq opens ordinary repository shells. --command is an
-  explicit initial command for newly created windows; it does not opt in to
-  terminal observation or message injection.
+  explicit initial command for newly created windows. Messages appear as
+  sanitized terminal output by default; --no-display keeps them inbox-only.
+  Neither mode injects message bytes into pane input.
 
 manual messaging from a managed shell:
   @TARGET: MESSAGE...
@@ -180,6 +175,7 @@ manual messaging from a managed shell:
     start.add_argument("repos", nargs="*", help="gita-registered repository slugs")
     start.add_argument("--command", "--cmd", dest="initial_command", default=None, help="explicit initial command for newly created windows (default: ordinary shell)")
     start.add_argument("--tap", action="store_true", help="explicitly opt in to PTY input observation and pane message injection")
+    start.add_argument("--no-display", action="store_true", help="keep messages in the durable inbox without writing target terminal output")
     start.add_argument("--detach", action="store_true", help="detach after startup")
     start.add_argument("--no-service", action="store_true", help="open tmux windows without service registration or messaging")
     attach = subparsers.add_parser("attach", help="attach to the managed tmux session")
@@ -312,6 +308,9 @@ def main(argv: list[str] | None = None) -> int:
         if args.tap and args.no_service:
             print("tamq: --tap cannot be combined with --no-service", file=sys.stderr)
             return 2
+        if args.tap and args.no_display:
+            print("tamq: --tap cannot be combined with --no-display", file=sys.stderr)
+            return 2
         manager = TmuxManager()
         try:
             if not args.no_service:
@@ -320,8 +319,11 @@ def main(argv: list[str] | None = None) -> int:
         except (TmuxError, OSError) as exc:
             print(f"tamq: {exc}", file=sys.stderr)
             return 2
-        if not args.no_service and not ensure_manual_service():
-            print("tamq service failed to start with manual-delivery safety; use --no-service to open repos without messaging", file=sys.stderr)
+        service_ready = (
+            ensure_manual_service() if args.no_display else ensure_output_service()
+        ) if not args.no_service else True
+        if not service_ready:
+            print("tamq service failed to start with the required delivery capability; use --no-service to open repos without messaging", file=sys.stderr)
             return 1
         try:
             endpoint = manager.ensure_plan(launch_plan, tap=args.tap)
@@ -332,7 +334,7 @@ def main(argv: list[str] | None = None) -> int:
         registered = False
         if not args.no_service:
             try:
-                delivery_mode = "pane" if args.tap else "manual"
+                delivery_mode = "pane" if args.tap else ("manual" if args.no_display else "output")
                 registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos, "delivery_mode": delivery_mode}))
                 if not registration.get("ok"):
                     raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}")
@@ -360,7 +362,7 @@ def main(argv: list[str] | None = None) -> int:
             "delivered": delivered,
             "service": not args.no_service,
             "messaging": registered,
-            "delivery_mode": "none" if args.no_service else ("pane" if args.tap else "manual"),
+            "delivery_mode": "none" if args.no_service else ("pane" if args.tap else ("manual" if args.no_display else "output")),
         }
         print(json.dumps(summary), flush=True)
         if args.detach:

src/tamq/control.py

@@ -34,6 +34,18 @@ class ControlModeClient:
             return False
         return expected_pid is None or result.stdout.strip() == str(expected_pid)
 
+    def pane_tty(self, window: str) -> str:
+        result = subprocess.run(
+            [*self._command(), "display-message", "-p", "-t", window, "#{pane_tty}"],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        tty_path = result.stdout.strip()
+        if result.returncode != 0 or not tty_path:
+            raise ControlModeError(result.stderr.strip() or f"cannot resolve pane tty: {window}")
+        return tty_path
+
     def start(self) -> None:
         if self.process is not None:
             return

src/tamq/service.py

@@ -13,6 +13,7 @@ from .config import db_path
 from .store import Store
 from .registry import RegistryError, validate_targets
 from .control import ControlModeClient
+from .terminal import terminal_frame, write_terminal_output
 
 PROTOCOL_VERSION = "0.1"
 
@@ -59,7 +60,7 @@ class Service:
                 self.store.close()
 
     async def _delivery_loop(self, stopped: asyncio.Event) -> None:
-        """Continuously inject pending messages into registered live endpoints."""
+        """Continuously deliver pending messages to registered live endpoints."""
         while not stopped.is_set():
             try:
                 self._deliver_once()
@@ -78,22 +79,42 @@ class Service:
             if not control.session_exists(expected_pid=int(endpoint["pid"])):
                 self.store.disconnect_endpoint(endpoint["endpoint_id"])
                 continue
-            if endpoint["delivery_mode"] != "pane":
+            delivery_mode = endpoint["delivery_mode"]
+            if delivery_mode == "manual":
                 continue
-            pending = [row for row in self.store.list(state="pending") if row["endpoint_id"] in (None, endpoint["endpoint_id"]) and row["target_repo"] in repos]
+            pending = [
+                row for row in self.store.list(state="pending")
+                if row["endpoint_id"] in (None, endpoint["endpoint_id"])
+                and row["target_repo"] in repos
+                and (delivery_mode == "pane" or row["displayed_at"] is None)
+            ]
             if not pending:
                 continue
             try:
-                control.start()
+                if delivery_mode == "pane":
+                    control.start()
                 for row in pending:
                     lease = self.store.claim(row["message_id"], endpoint["endpoint_id"])
                     if lease is None:
                         continue
                     try:
-                        control.inject(f'{endpoint["session"]}:{row["target_repo"]}', f'#{row["sender_repo"]}: {row["body"]}')
+                        target = f'{endpoint["session"]}:{row["target_repo"]}'
+                        if delivery_mode == "output":
+                            tty_path = control.pane_tty(target)
+                            write_terminal_output(
+                                tty_path,
+                                terminal_frame(
+                                    row["sender_repo"], row["body"], row["message_id"]
+                                ),
+                            )
+                        else:
+                            control.inject(target, f'#{row["sender_repo"]}: {row["body"]}')
                     except Exception:
                         continue
-                    self.store.release(row["message_id"], lease, "injected")
+                    if delivery_mode == "output":
+                        self.store.mark_displayed(row["message_id"], lease)
+                    else:
+                        self.store.release(row["message_id"], lease, "injected")
             finally:
                 control.close()
 
@@ -109,7 +130,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"]}
+                response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery", "terminal_output"]}
             elif op == "register":
                 required = ("endpoint_id", "pid", "session", "repos")
                 if any(key not in request for key in required):

src/tamq/store.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from typing import Iterable
 from uuid import uuid4
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 
 
 class Store:
@@ -30,7 +30,8 @@ class Store:
           endpoint_id TEXT,
           provenance TEXT,
           injected_at REAL,
-          acknowledged_at REAL
+          acknowledged_at REAL,
+          displayed_at REAL
         );
         CREATE INDEX IF NOT EXISTS messages_target_state ON messages(target_repo, state);
         CREATE TABLE IF NOT EXISTS endpoints (
@@ -57,6 +58,11 @@ class Store:
             self.db.execute(
                 "ALTER TABLE endpoints ADD COLUMN delivery_mode TEXT NOT NULL DEFAULT 'manual'"
             )
+        message_columns = {
+            row["name"] for row in self.db.execute("PRAGMA table_info(messages)")
+        }
+        if "displayed_at" not in message_columns:
+            self.db.execute("ALTER TABLE messages ADD COLUMN displayed_at REAL")
         self.db.execute(
             "INSERT INTO metadata(key,value) VALUES('schema_version',?) "
             "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
@@ -76,7 +82,7 @@ class Store:
         delivery_mode: str = "manual",
     ) -> None:
         import json
-        if delivery_mode not in {"manual", "pane"}:
+        if delivery_mode not in {"manual", "output", "pane"}:
             raise ValueError(f"invalid delivery mode: {delivery_mode}")
         self.db.execute(
             "UPDATE endpoints SET disconnected_at=strftime('%s','now') "
@@ -131,8 +137,9 @@ class Store:
             raise ValueError("message body exceeds 8 KiB limit")
         message_id = f"m-{uuid4()}"
         self.db.execute(
-            "INSERT INTO messages VALUES(?,?,?,?,?,?,?,?,?,?)",
-            (message_id, sender, target, body, time.time(), "pending", endpoint, provenance, None, None),
+            "INSERT INTO messages(message_id,sender_repo,target_repo,body,created_at,state,endpoint_id,provenance,injected_at,acknowledged_at,displayed_at) "
+            "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
+            (message_id, sender, target, body, time.time(), "pending", endpoint, provenance, None, None, None),
         )
         self.db.commit()
         return message_id
@@ -161,6 +168,19 @@ class Store:
         self.set_state(message_id, "acknowledged")
         return True
 
+    def mark_displayed(self, message_id: str, lease_id: str) -> bool:
+        with self.db:
+            result = self.db.execute(
+                "DELETE FROM leases WHERE message_id=? AND lease_id=?",
+                (message_id, lease_id),
+            )
+            if result.rowcount:
+                self.db.execute(
+                    "UPDATE messages SET displayed_at=? WHERE message_id=?",
+                    (time.time(), message_id),
+                )
+        return result.rowcount == 1
+
     def claim(self, message_id: str, endpoint_id: str, ttl: float = 30.0) -> str | None:
         now = time.time()
         lease_id = f"lease-{uuid4()}"

src/tamq/terminal.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import os
+import re
+import stat
+from pathlib import Path
+
+
+class TerminalOutputError(RuntimeError):
+    pass
+
+
+def terminal_safe(text: str) -> str:
+    """Escape terminal controls while retaining printable Unicode verbatim."""
+    return "".join(
+        character
+        if character.isprintable()
+        else f"\\x{ord(character):02x}"
+        for character in text
+    )
+
+
+def format_comment(sender: str, body: str, message_id: str) -> str:
+    """Render a message so every line is visibly comment-prefixed."""
+    lines = body.splitlines() or [""]
+    rendered = [f"#{terminal_safe(sender)}: {terminal_safe(lines[0])}"]
+    rendered.extend(f"# {terminal_safe(line)}" for line in lines[1:])
+    rendered[-1] += f" [{terminal_safe(message_id)}]"
+    return "\n".join(rendered)
+
+
+def terminal_frame(sender: str, body: str, message_id: str) -> str:
+    """Frame asynchronous output away from the current visual input line."""
+    content = format_comment(sender, body, message_id).replace("\n", "\r\n")
+    return f"\r\n{content}\r\n"
+
+
+def write_terminal_output(tty_path: str | Path, text: str) -> None:
+    """Write output to a tmux pane PTY without placing bytes on its stdin."""
+    try:
+        path = Path(tty_path).resolve(strict=True)
+    except OSError as exc:
+        raise TerminalOutputError(f"terminal device is unavailable: {tty_path}") from exc
+    if re.fullmatch(r"/dev/pts/[0-9]+", str(path)) is None:
+        raise TerminalOutputError(f"refusing non-PTY terminal path: {path}")
+    flags = os.O_WRONLY | os.O_NOCTTY | os.O_NONBLOCK | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as exc:
+        raise TerminalOutputError(f"cannot open terminal output: {path}") from exc
+    try:
+        details = os.fstat(descriptor)
+        if not stat.S_ISCHR(details.st_mode) or details.st_uid != os.getuid():
+            raise TerminalOutputError(f"terminal device is not owned by this user: {path}")
+        payload = text.encode("utf-8")
+        offset = 0
+        while offset < len(payload):
+            written = os.write(descriptor, payload[offset:])
+            if written <= 0:
+                raise TerminalOutputError(f"terminal output made no progress: {path}")
+            offset += written
+    except OSError as exc:
+        raise TerminalOutputError(f"cannot write terminal output: {path}") from exc
+    finally:
+        os.close(descriptor)