feat: add shell-native message routing

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

src/tamq/cli.py

@@ -7,6 +7,7 @@ import asyncio
 import json
 import os
 import signal
+import sqlite3
 import fcntl
 from uuid import uuid4
 import subprocess
@@ -132,10 +133,42 @@ 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)
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="tamq",
         description="Repository-aware tmux sessions with durable local messaging.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""session shorthand:
+  tamq [--detach] [--command COMMAND] 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.
+
+manual messaging from a managed shell:
+  @TARGET: MESSAGE...
+  tamq inbox [--filter COMMAND]
+""",
     )
     parser.add_argument("--version", "-V", action="version", version=__version__)
     parser.add_argument("--orwell", action="store_true", help="enable unsafe local diagnostics")
@@ -163,6 +196,11 @@ def build_parser() -> argparse.ArgumentParser:
     inbox.add_argument("--repo", dest="target_repo", help="target repository (default: TAMQ_REPO in a managed window)")
     inbox.add_argument("--all", action="store_true", help="include non-pending messages")
     inbox.add_argument("--json", action="store_true", help="emit one JSON object per message")
+    inbox.add_argument(
+        "--filter",
+        dest="filter_command",
+        help="consume each pending comment through COMMAND and acknowledge it after a zero exit",
+    )
     inspect = subparsers.add_parser("inspect", help="inspect one message")
     inspect.add_argument("message_id")
     ack = subparsers.add_parser("ack", help="acknowledge one durable message")
@@ -370,12 +408,42 @@ def main(argv: list[str] | None = None) -> int:
                 validate_targets([target])
             except RegistryError as exc:
                 print(f"tamq: {exc}", file=sys.stderr); return 2
+            if args.filter_command is not None and not args.filter_command.strip():
+                print("tamq: --filter command must not be empty", file=sys.stderr)
+                return 2
+            if args.filter_command and (args.all or args.json):
+                print("tamq: --filter cannot be combined with --all or --json", file=sys.stderr)
+                return 2
             rows = store.list(target, None if args.all else "pending")
             for row in rows:
                 if args.json:
                     print(json.dumps(dict(row), sort_keys=True))
+                elif args.filter_command:
+                    environment = os.environ.copy()
+                    environment.update(
+                        {
+                            "TAMQ_MESSAGE_ID": row["message_id"],
+                            "TAMQ_SENDER_REPO": row["sender_repo"],
+                            "TAMQ_TARGET_REPO": row["target_repo"],
+                        }
+                    )
+                    result = subprocess.run(
+                        args.filter_command,
+                        shell=True,
+                        input=format_comment_message(row) + "\n",
+                        text=True,
+                        env=environment,
+                        check=False,
+                    )
+                    if result.returncode:
+                        print(
+                            f"tamq: filter failed for {row['message_id']} with exit status {result.returncode}; message remains pending",
+                            file=sys.stderr,
+                        )
+                        return 1
+                    store.acknowledge(row["message_id"])
                 else:
-                    print(f"{row['message_id']}  {row['sender_repo']} -> {row['target_repo']}: {row['body']}")
+                    print(format_comment_message(row))
             return 0
         if args.command == "history":
             for row in store.list(args.target_repo, args.state): print(json.dumps(dict(row), sort_keys=True))