Bootstrap tamq local tmux agent message queue

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02f34-ead4-7a60-85cd-d7f08e22fa0e

src/tamq/__init__.py

@@ -0,0 +1,3 @@
+"""Tmux Agentic Message Queueing."""
+
+__version__ = "0.1.0"

src/tamq/broker.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .routing import RoutedMessage, parse_address_line
+from .store import Store
+from .registry import RegistryError, validate_targets
+
+
+@dataclass(frozen=True)
+class BrokerIdentity:
+    endpoint_id: str
+    source_repo: str
+
+
+class InputBroker:
+    """Convert tap input into durable, identity-bearing outbound messages."""
+
+    def __init__(self, store: Store, identity: BrokerIdentity):
+        self.store = store
+        self.identity = identity
+
+    def inspect_line(self, line: str) -> RoutedMessage | None:
+        routed = parse_address_line(line)
+        if routed is None:
+            return None
+        try:
+            validate_targets([routed.target_repo])
+        except RegistryError:
+            return None
+        self.store.add(
+            self.identity.source_repo,
+            routed.target_repo,
+            routed.body,
+            endpoint=self.identity.endpoint_id,
+        )
+        return routed
+
+    def deliver_pending(self, control, *, window_for_repo):
+        """Inject pending messages through the authoritative control client."""
+        delivered = 0
+        for row in self.store.list(state="pending"):
+            if row["endpoint_id"] not in (None, self.identity.endpoint_id):
+                continue
+            target = window_for_repo(row["target_repo"])
+            lease_id = self.store.claim(row["message_id"], self.identity.endpoint_id)
+            if lease_id is None:
+                continue
+            try:
+                control.inject(target, f"#{row['sender_repo']}: {row['body']}")
+            except Exception:
+                continue
+            self.store.release(row["message_id"], lease_id, "injected")
+            delivered += 1
+        return delivered

src/tamq/cli.py

@@ -0,0 +1,313 @@
+"""Initial tamq command-line surface."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import signal
+import fcntl
+from uuid import uuid4
+import subprocess
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+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 .store import Store
+from .tmux import TmuxError, TmuxManager
+from .broker import BrokerIdentity, InputBroker
+from .ptytap import PtyTap
+from .control import ControlModeClient
+from .diagnostics import configure
+from .policy import load_profile
+
+
+def parse_size(value: str) -> int:
+    units = {"B": 1, "KB": 1000, "MB": 1000**2, "GB": 1000**3}
+    text = value.upper()
+    for unit, multiplier in sorted(units.items(), key=lambda item: -len(item[0])):
+        if text.endswith(unit):
+            return int(float(text[:-len(unit)]) * multiplier)
+    return int(text)
+
+
+def history_advisory(store: Store) -> str | None:
+    size, oldest = store.history_stats()
+    limit_text = setting("history_max_size", "100MB")
+    limit = parse_size(limit_text)
+    if size <= limit:
+        return None
+    since = time.strftime("%y%m%d", time.localtime(oldest)) if oldest else "unknown"
+    return f"Size of history is {size / 1000**2:.0f} MB since {since}, exceeding {limit_text}. Consider using 'tamq purge'."
+
+
+def ensure_service() -> bool:
+    if asyncio.run(ping()):
+        return True
+    path = lock_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a+") as lock:
+        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
+        if asyncio.run(ping()):
+            return True
+        subprocess.Popen(
+            [sys.executable, "-m", "tamq.cli", "serve"],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+            env=os.environ.copy(),
+        )
+        for _ in range(20):
+            if asyncio.run(ping()):
+                return True
+            time.sleep(0.05)
+        return False
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="tamq",
+        description="Tmux Agentic Message Queueing for local repository workers.",
+    )
+    parser.add_argument("--version", "-V", action="version", version=__version__)
+    parser.add_argument("--orwell", action="store_true", help="enable unsafe local diagnostics")
+    parser.add_argument("--verbose", action="store_true", help="enable debug diagnostics")
+    parser.add_argument("--policy-profile", default=None, help="select configured policy profile")
+    subparsers = parser.add_subparsers(dest="command")
+
+    start = subparsers.add_parser("start", help="start or reuse a tamq endpoint")
+    start.add_argument("repos", nargs="*", help="gita-registered repository slugs")
+    start.add_argument("--cmd", default="codex", help="agent command for new windows")
+    start.add_argument("--detach", action="store_true", help="detach after startup")
+    start.add_argument("--no-service", action="store_true", help="skip service startup")
+    attach = subparsers.add_parser("attach", help="attach to the managed tmux session")
+    attach.add_argument("--session", default="tamq", help="tmux session name")
+
+    subparsers.add_parser("serve", help="run the local service in the foreground")
+    subparsers.add_parser("stop", help="stop the local service")
+    subparsers.add_parser("status", help="show endpoint health")
+    subparsers.add_parser("ping", help="check local service liveness")
+    history = subparsers.add_parser("history", help="inspect local message history")
+    history.add_argument("--repo", dest="target_repo")
+    history.add_argument("--state")
+    inspect = subparsers.add_parser("inspect", help="inspect one message")
+    inspect.add_argument("message_id")
+    ack = subparsers.add_parser("ack", help="acknowledge one injected message")
+    ack.add_argument("message_id")
+    send = subparsers.add_parser("send", help="queue a direct message")
+    send.add_argument("address", help="@repo: message")
+    send.add_argument("body", nargs="*", help="message body when address is a repo slug")
+    send.add_argument("--endpoint-id")
+    export = subparsers.add_parser("export", help="export history as JSONL")
+    export.add_argument("--output", required=True)
+    export.add_argument("--repo", dest="target_repo")
+    export.add_argument("--state")
+    replay = subparsers.add_parser("replay", help="replay a JSONL message file")
+    replay.add_argument("file")
+    replay.add_argument("--endpoint-id", help="target a registered tmux-amq endpoint")
+    purge = subparsers.add_parser("purge", help="dry-run or remove history")
+    purge.add_argument("--before", default=None)
+    purge.add_argument("--max-size", default=None)
+    purge.add_argument("--yes", action="store_true")
+    purge.add_argument("--orwell", action="store_true")
+    completion = subparsers.add_parser("completion", help="emit shell completion script")
+    completion.add_argument("shell", choices=("bash", "zsh", "fish"))
+    subparsers.add_parser("db-version", help="show SQLite schema version")
+    subparsers.add_parser("config", help="show effective configuration paths and policy")
+    tap = subparsers.add_parser("tap", help="run an agent behind the full-duplex PTY tap")
+    tap.add_argument("--repo", required=True, help="source gita repository slug")
+    tap.add_argument("--endpoint", required=True, help="tmux-amq endpoint identity")
+    tap.add_argument("agent_command", nargs=argparse.REMAINDER, help="agent command after --")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    configure(args.orwell, args.verbose)
+    try:
+        profile = load_profile(selected=args.policy_profile or os.environ.get("TAMQ_POLICY_PROFILE", "default"))
+    except ValueError as exc:
+        print(f"tamq: {exc}", file=sys.stderr)
+        return 2
+    if args.command is None:
+        parser.print_help()
+        return 0
+    if args.command == "db-version":
+        store = Store(db_path()); print(store.db.execute("SELECT value FROM metadata WHERE key='schema_version'").fetchone()[0]); store.close(); return 0
+    if args.command == "ping":
+        return 0 if asyncio.run(ping()) else 1
+    if args.command == "stop":
+        try:
+            pid = int(pid_path().read_text(encoding="utf-8"))
+            os.kill(pid, signal.SIGTERM)
+            deadline = time.monotonic() + 2.0
+            while time.monotonic() < deadline:
+                try:
+                    os.kill(pid, 0)
+                except ProcessLookupError:
+                    break
+                time.sleep(0.05)
+            print(f"stopped tamq service {pid}")
+            return 0
+        except ProcessLookupError:
+            pid_path().unlink(missing_ok=True)
+            print("tamq service is not running", file=sys.stderr)
+            return 1
+        except (FileNotFoundError, ValueError):
+            print("tamq service is not running", file=sys.stderr)
+            return 1
+    if args.command == "serve":
+        try: asyncio.run(Service().run())
+        except KeyboardInterrupt: return 0
+        return 0
+    if args.command == "attach":
+        result = subprocess.run(["tmux", "attach-session", "-t", args.session], check=False)
+        return result.returncode
+    if args.command == "completion":
+        print(completion_script(args.shell), end="")
+        return 0
+    if args.command == "config":
+        print(json.dumps({"config": str(config_path()), "state_dir": str(state_dir()), "database": str(db_path()), "socket": str(socket_path()), "pidfile": str(pid_path()), "lockfile": str(lock_path()), "policy_profile": profile.name, "delivery_ack_mode": profile.delivery_ack_mode}, sort_keys=True))
+        return 0
+    if args.command == "tap":
+        command = list(args.agent_command)
+        if command and command[0] == "--":
+            command = command[1:]
+        if not command:
+            print("tamq tap requires an agent command after --", file=sys.stderr)
+            return 2
+        store = Store(db_path())
+        try:
+            return PtyTap(command, InputBroker(store, BrokerIdentity(args.endpoint, args.repo))).run()
+        finally:
+            store.close()
+    if args.command == "start":
+        if not args.no_service and not ensure_service():
+            print("tamq service failed to start; use --no-service to open repos without messaging", file=sys.stderr)
+            return 1
+        try:
+            endpoint = TmuxManager().ensure(args.repos, args.cmd)
+        except (TmuxError, OSError) as exc:
+            print(f"tamq: {exc}", file=sys.stderr)
+            return 2
+        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}))
+        if not registration.get("ok"):
+            print(f"tamq: endpoint registration failed: {registration.get('error', 'unknown error')}", file=sys.stderr)
+            return 1
+        control = ControlModeClient(endpoint.session)
+        queue_store = Store(db_path())
+        try:
+            control.start()
+            delivered = InputBroker(queue_store, BrokerIdentity(endpoint.instance_key, endpoint.repos[0])).deliver_pending(
+                control, window_for_repo=lambda repo: f"{endpoint.session}:{repo}"
+            )
+        finally:
+            control.close()
+            queue_store.close()
+        print(json.dumps({"endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "session": endpoint.session, "repos": endpoint.repos, "delivered": delivered}))
+        return 0
+    store = Store(db_path())
+    try:
+        advisory = history_advisory(store)
+        if advisory and args.command in {"start", "serve", "status", "history", "send"}:
+            print(advisory, file=sys.stderr)
+        if args.command == "send":
+            text = " ".join([args.address, *args.body]).strip()
+            if not text.startswith("@") or ":" not in text:
+                print("tamq send expects @repo: message", file=sys.stderr); return 2
+            target, body = text[1:].split(":", 1); body = body.strip()
+            if not target or not body: print("tamq send requires a target and body", file=sys.stderr); return 2
+            if args.endpoint_id:
+                if not asyncio.run(ping()):
+                    print("tamq: service is not running", file=sys.stderr); return 1
+                response = asyncio.run(request({"op": "send", "endpoint_id": args.endpoint_id, "sender_repo": "local", "target_repo": target, "body": body}))
+                if not response.get("ok"):
+                    print(f"tamq: {response.get('error', 'send failed')}", file=sys.stderr); return 1
+                print(response["message_id"]); return 0
+            try:
+                print(store.add("local", target, body))
+            except ValueError as exc:
+                print(f"tamq: {exc}", file=sys.stderr); return 2
+            return 0
+        if args.command == "history":
+            for row in store.list(args.target_repo, args.state): print(json.dumps(dict(row), sort_keys=True))
+            return 0
+        if args.command == "inspect":
+            rows = [row for row in store.list() if row["message_id"] == args.message_id]
+            if not rows:
+                print("tamq: message not found", file=sys.stderr)
+                return 1
+            print(json.dumps(dict(rows[0]), sort_keys=True)); return 0
+        if args.command == "ack":
+            if not store.acknowledge(args.message_id):
+                print("tamq: message not found", file=sys.stderr); return 1
+            print(args.message_id); return 0
+        if args.command == "export":
+            store.export(store.list(args.target_repo, args.state), Path(args.output)); return 0
+        if args.command == "replay":
+            batch_id = f"replay-{uuid4()}"
+            count = 0
+            if args.endpoint_id and store.endpoint(args.endpoint_id) is None:
+                print(f"tamq: endpoint is not registered: {args.endpoint_id}", file=sys.stderr)
+                return 1
+            try:
+                lines = Path(args.file).read_text(encoding="utf-8").splitlines()
+                for line in lines:
+                    if not line.strip():
+                        continue
+                    item = json.loads(line)
+                    provenance = json.dumps({"batch_id": batch_id, "original_message_id": item.get("message_id")}, sort_keys=True)
+                    endpoint = args.endpoint_id or item.get("endpoint_id")
+                    store.add(item.get("sender_repo", "replay"), item["target_repo"], item["body"], endpoint=endpoint, provenance=provenance)
+                    count += 1
+            except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                print(f"tamq: cannot replay {args.file}: {exc}", file=sys.stderr)
+                return 2
+            print(json.dumps({"batch_id": batch_id, "count": count})); return 0
+        if args.command == "purge":
+            before_text = args.before or setting("purge_before", "365d")
+            max_size_text = args.max_size or setting("purge_max_size", "100MB")
+            before = time.time() - int(before_text[:-1]) * 86400 if before_text.endswith("d") else None
+            if not args.yes:
+                matching = [row for row in store.list() if before is None or row["created_at"] < before]
+                print(f"Dry run: {len(matching)} message(s) match; re-run with --yes to purge.")
+                return 0
+            count = store.purge(before=before, max_bytes=parse_size(max_size_text))
+            print(f"Purged {count} message(s)."); return 0
+        if args.command == "status":
+            live = asyncio.run(ping())
+            endpoints = asyncio.run(request({"op": "endpoints"})) if live else {"endpoints": []}
+            print(json.dumps({"version": __version__, "db": str(db_path()), "messages": len(store.list()), "pending": len(store.list(state="pending")), "leases": store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0], "service": live, "policy_profile": profile.name, "delivery_ack_mode": profile.delivery_ack_mode, "endpoints": endpoints.get("endpoints", [])}, sort_keys=True)); return 0
+        print(f"tamq {__version__}: command '{args.command}' is not implemented yet", file=sys.stderr)
+        return 2
+    finally:
+        store.close()
+
+
+def completion_script(shell: str) -> str:
+    commands = "start attach serve stop status ping history inspect ack send export replay purge completion db-version config tap"
+    if shell == "bash":
+        return f"""_tamq_complete() {{
+  local commands=\"{commands}\"
+  if [[ ${{COMP_WORDS[1]}} == start || ${{COMP_WORDS[1]}} == send ]]; then
+    COMPREPLY=( $(compgen -W \"$(gita ls 2>/dev/null)\" -- \"${{COMP_WORDS[COMP_CWORD]}}\") )
+  else
+    COMPREPLY=( $(compgen -W \"$commands\" -- \"${{COMP_WORDS[COMP_CWORD]}}\") )
+  fi
+}}
+complete -F _tamq_complete tamq
+"""
+    if shell == "zsh":
+        return f"#compdef tamq\n_arguments '1:command:(({commands}))' '*:repo:->repos'\nif [[ $state == repos ]]; then _values 'gita repo' $(gita ls 2>/dev/null); fi\n"
+    return f"complete -c tamq -f -n '__fish_seen_subcommand_from start send' -a '(gita ls 2>/dev/null)'\ncomplete -c tamq -f -a '{commands}'\n"
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/tamq/config.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import os
+import tomllib
+from pathlib import Path
+
+
+def config_path() -> Path:
+    return Path(os.environ.get("TAMQ_CONFIG", Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "tamq" / "config.toml"))
+
+
+def _setting(name: str) -> str | None:
+    path = config_path()
+    if not path.exists():
+        return None
+    try:
+        with path.open("rb") as stream:
+            value = tomllib.load(stream).get("tamq", {}).get(name)
+            return value if isinstance(value, str) else None
+    except (OSError, tomllib.TOMLDecodeError):
+        return None
+
+
+def setting(name: str, default: str) -> str:
+    """Read a scalar [tamq] setting, falling back to a stable default."""
+    return _setting(name) or default
+
+
+def state_dir() -> Path:
+    return Path(os.environ.get("TAMQ_STATE_DIR", _setting("state_dir") or Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")) / "tamq"))
+
+
+def db_path() -> Path:
+    return state_dir() / "tamq.sqlite3"
+
+
+def socket_path() -> Path:
+    return Path(os.environ.get("TAMQ_SOCKET", _setting("socket") or Path(os.environ.get("XDG_RUNTIME_DIR", "/tmp")) / "tamq.sock"))
+
+
+def pid_path() -> Path:
+    return Path(os.environ.get("TAMQ_PIDFILE", _setting("pidfile") or Path(os.environ.get("XDG_RUNTIME_DIR", "/tmp")) / "tamq.pid"))
+
+
+def lock_path() -> Path:
+    return Path(os.environ.get("TAMQ_LOCKFILE", _setting("lockfile") or Path(os.environ.get("XDG_RUNTIME_DIR", "/tmp")) / "tamq.lock"))

src/tamq/control.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+
+
+class ControlModeError(RuntimeError):
+    pass
+
+
+@dataclass
+class ControlModeClient:
+    session: str
+    process: subprocess.Popen[str] | None = None
+
+    def start(self) -> None:
+        if self.process is not None:
+            return
+        self.process = subprocess.Popen(
+            ["tmux", "-C", "attach-session", "-t", self.session],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+        )
+
+    def command(self, value: str) -> None:
+        if self.process is None or self.process.stdin is None:
+            raise ControlModeError("control-mode client is not started")
+        self.process.stdin.write(value.rstrip("\n") + "\n")
+        self.process.stdin.flush()
+
+    def inject(self, window: str, text: str) -> None:
+        # send-keys -l preserves punctuation and whitespace in the message.
+        self.command(f"send-keys -t {window} -l -- {shell_quote(text)}")
+
+    def close(self) -> None:
+        if self.process is not None:
+            self.process.terminate()
+            self.process = None
+
+
+def shell_quote(value: str) -> str:
+    return "'" + value.replace("'", "'\\''") + "'"

src/tamq/diagnostics.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import logging
+import os
+import sys
+from pathlib import Path
+
+
+def orwell_path() -> Path:
+    state = Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")) / "tamq"
+    state.mkdir(parents=True, exist_ok=True)
+    return state / "tamq-orwell.log"
+
+
+def configure(orwell: bool = False, verbose: bool = False) -> logging.Logger:
+    logger = logging.getLogger("tamq")
+    logger.handlers.clear()
+    logger.setLevel(logging.DEBUG if verbose or orwell else logging.INFO)
+    if orwell:
+        path = orwell_path()
+        handler = logging.FileHandler(path, encoding="utf-8")
+        os.chmod(path, 0o600)
+        logger.addHandler(handler)
+        print("WARNING: --orwell enables sensitive local diagnostics; purge tamq-orwell.log after use.", file=sys.stderr)
+    else:
+        logger.addHandler(logging.StreamHandler(sys.stderr))
+    return logger

src/tamq/policy.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+
+from .config import config_path
+
+
+@dataclass(frozen=True)
+class PolicyProfile:
+    name: str
+    allow: frozenset[str]
+    require_human: frozenset[str]
+    safety_gated_max_attempts: int = 0
+    delivery_ack_mode: str = "injected"
+
+
+DEFAULT = PolicyProfile(
+    "default",
+    frozenset({"statehub_read", "repo_inspect", "repo_edit", "local_checks", "state_updates", "coordination_messages", "tmux_wake", "transient_retry"}),
+    frozenset({"secrets", "destructive", "live_external", "scope_change", "ambiguous_ownership", "external_publish", "financial", "legal"}),
+)
+
+
+def load_profile(path: Path | None = None, selected: str = "default") -> PolicyProfile:
+    path = path or config_path()
+    if not path.exists():
+        return DEFAULT
+    with path.open("rb") as stream:
+        data = tomllib.load(stream)
+    profile = data.get("policy", {}).get("profiles", {}).get(selected)
+    if not profile:
+        if selected == "default":
+            return DEFAULT
+        raise ValueError(f"unknown policy profile: {selected}")
+    attempts = int(profile.get("safety_gated_max_attempts", 0))
+    if not 0 <= attempts <= 9:
+        raise ValueError("safety_gated_max_attempts must be between 0 and 9")
+    ack_mode = profile.get("delivery_ack_mode", "injected")
+    if ack_mode not in {"injected", "acknowledged"}:
+        raise ValueError("delivery_ack_mode must be injected or acknowledged")
+    return PolicyProfile(selected, frozenset(profile.get("allow", [])), frozenset(profile.get("require_human", [])), attempts, ack_mode)

src/tamq/ptytap.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import os
+import pty
+import select
+import signal
+import sys
+from collections.abc import Callable, Sequence
+
+from .broker import InputBroker
+
+
+class PtyTap:
+    """Full-duplex PTY proxy. Input is observed, never rewritten."""
+
+    def __init__(self, command: Sequence[str], broker: InputBroker, *, on_line: Callable[[str], None] | None = None):
+        self.command = list(command)
+        self.broker = broker
+        self.on_line = on_line
+
+    def run(self) -> int:
+        pid, master = pty.fork()
+        if pid == 0:
+            os.execvp(self.command[0], self.command)
+        input_buffer = bytearray()
+        try:
+            while True:
+                readable, _, _ = select.select([master, sys.stdin], [], [])
+                if master in readable:
+                    try:
+                        data = os.read(master, 65536)
+                    except OSError:
+                        break
+                    if not data:
+                        break
+                    os.write(sys.stdout.fileno(), data)
+                if sys.stdin in readable:
+                    data = os.read(sys.stdin.fileno(), 65536)
+                    if not data:
+                        break
+                    os.write(master, data)
+                    input_buffer.extend(data)
+                    while b"\n" in input_buffer:
+                        raw, _, rest = input_buffer.partition(b"\n")
+                        input_buffer = bytearray(rest)
+                        line = raw.decode("utf-8", errors="replace")
+                        routed = self.broker.inspect_line(line)
+                        if routed and self.on_line:
+                            self.on_line(line)
+        finally:
+            try:
+                _, status = os.waitpid(pid, 0)
+                return os.waitstatus_to_exitcode(status)
+            except ChildProcessError:
+                return 0

src/tamq/registry.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import subprocess
+
+
+class RegistryError(RuntimeError):
+    pass
+
+
+def registered_repositories() -> set[str]:
+    """Return exact gita slugs from the local registry."""
+    result = subprocess.run(["gita", "ls"], text=True, capture_output=True, check=False)
+    if result.returncode:
+        raise RegistryError(result.stderr.strip() or "gita registry unavailable")
+    repos: set[str] = set()
+    # gita may render one slug per line or a whitespace-separated compact list.
+    for slug in result.stdout.split():
+        if slug and not slug.startswith("["):
+            repos.add(slug)
+    return repos
+
+
+def repository_paths() -> dict[str, str]:
+    result = subprocess.run(["gita", "freeze"], text=True, capture_output=True, check=False)
+    if result.returncode:
+        raise RegistryError(result.stderr.strip() or "gita registry unavailable")
+    paths: dict[str, str] = {}
+    for line in result.stdout.splitlines():
+        fields = line.split(",")
+        if len(fields) >= 3:
+            paths[fields[1]] = fields[2]
+    return paths
+
+
+def validate_targets(targets: list[str]) -> None:
+    registered = registered_repositories()
+    unknown = sorted(set(targets) - registered)
+    if unknown:
+        raise RegistryError(f"unregistered gita repository target(s): {', '.join(unknown)}")

src/tamq/routing.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+ADDRESS = re.compile(r"^@([a-z0-9][a-z0-9._-]*):(?:[ \t]*)(.+)$", re.IGNORECASE)
+
+
+@dataclass(frozen=True)
+class RoutedMessage:
+    target_repo: str
+    body: str
+
+
+def parse_address_line(line: str) -> RoutedMessage | None:
+    """Parse one complete direct-address line; ordinary input returns None."""
+    match = ADDRESS.match(line.rstrip("\r\n"))
+    if not match:
+        return None
+    return RoutedMessage(target_repo=match.group(1), body=match.group(2))

src/tamq/service.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import signal
+import socket
+import struct
+from pathlib import Path
+
+from .config import socket_path, pid_path, setting
+from .config import db_path
+from .store import Store
+from .registry import RegistryError, validate_targets
+from .control import ControlModeClient
+
+PROTOCOL_VERSION = "0.1"
+
+
+class Service:
+    def __init__(self, path: Path | None = None, store: Store | None = None, poll_interval: float | None = None):
+        self.path = path or socket_path()
+        self.store = store or Store(db_path())
+        self.pidfile = pid_path()
+        self.server: asyncio.AbstractServer | None = None
+        configured_interval = setting("delivery_poll_interval", "0.5")
+        try:
+            self.poll_interval = poll_interval if poll_interval is not None else max(0.05, float(configured_interval))
+        except ValueError:
+            self.poll_interval = poll_interval if poll_interval is not None else 0.5
+
+    async def run(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.pidfile.write_text(str(os.getpid()), encoding="utf-8")
+        os.chmod(self.pidfile, 0o600)
+        if self.path.exists():
+            self.path.unlink()
+        self.server = await asyncio.start_unix_server(self.handle, path=str(self.path))
+        os.chmod(self.path, 0o600)
+        stopped = asyncio.Event()
+        loop = asyncio.get_running_loop()
+        for sig in (signal.SIGTERM, signal.SIGINT):
+            try:
+                loop.add_signal_handler(sig, stopped.set)
+            except (NotImplementedError, RuntimeError):
+                pass
+        async with self.server:
+            delivery = asyncio.create_task(self._delivery_loop(stopped))
+            try:
+                await stopped.wait()
+            finally:
+                delivery.cancel()
+                await asyncio.gather(delivery, return_exceptions=True)
+                self.server.close()
+                await self.server.wait_closed()
+                self.store.disconnect_all()
+                self.path.unlink(missing_ok=True)
+                self.pidfile.unlink(missing_ok=True)
+                self.store.close()
+
+    async def _delivery_loop(self, stopped: asyncio.Event) -> None:
+        """Continuously inject pending messages into registered live endpoints."""
+        while not stopped.is_set():
+            try:
+                await asyncio.to_thread(self._deliver_once)
+            except Exception:
+                # A disappearing tmux session must not take down the queue.
+                pass
+            try:
+                await asyncio.wait_for(stopped.wait(), timeout=self.poll_interval)
+            except asyncio.TimeoutError:
+                continue
+
+    def _deliver_once(self) -> None:
+        for endpoint in self.store.endpoints():
+            repos = json.loads(endpoint["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]
+            if not pending:
+                continue
+            control = ControlModeClient(endpoint["session"])
+            try:
+                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"]}')
+                    except Exception:
+                        continue
+                    self.store.release(row["message_id"], lease, "injected")
+            finally:
+                control.close()
+
+    async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        if not self._peer_allowed(writer):
+            writer.write(b'{"ok":false,"error":"unauthorized peer"}\n')
+            await writer.drain(); writer.close(); await writer.wait_closed(); return
+        line = await reader.readline()
+        try:
+            request = json.loads(line or b"{}")
+            op = request.get("op")
+            protocol = request.get("protocol", PROTOCOL_VERSION)
+            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"]}
+            elif op == "register":
+                required = ("endpoint_id", "pid", "session", "repos")
+                if any(key not in request for key in required):
+                    response = {"ok": False, "error": "register requires endpoint_id, pid, session, repos"}
+                else:
+                    try:
+                        validate_targets(list(request["repos"]))
+                        endpoint_id = request.get("instance_id", request["endpoint_id"])
+                        self.store.register_endpoint(endpoint_id, int(request["pid"]), request["session"], list(request["repos"]))
+                        response = {"ok": True, "endpoint_id": request["endpoint_id"], "instance_id": endpoint_id, "protocol": PROTOCOL_VERSION}
+                    except (RegistryError, ValueError) as exc:
+                        response = {"ok": False, "error": str(exc)}
+            elif op == "send":
+                required = ("sender_repo", "target_repo", "body")
+                if any(key not in request for key in required):
+                    response = {"ok": False, "error": "send requires sender_repo, target_repo, body"}
+                else:
+                    try:
+                        validate_targets([request["target_repo"]])
+                        endpoint_id = request.get("endpoint_id")
+                        if endpoint_id:
+                            endpoint = self.store.endpoint(endpoint_id)
+                            if endpoint is None:
+                                raise RegistryError("endpoint is not registered")
+                            import json as _json
+                            if request["target_repo"] not in _json.loads(endpoint["repos"]):
+                                raise RegistryError("target is not attached to endpoint")
+                            endpoint_id = endpoint["endpoint_id"]
+                        message_id = self.store.add(request["sender_repo"], request["target_repo"], request["body"], endpoint=endpoint_id)
+                        response = {"ok": True, "message_id": message_id, "state": "pending"}
+                    except (RegistryError, ValueError) as exc:
+                        response = {"ok": False, "error": str(exc)}
+            elif op == "history":
+                response = {"ok": True, "messages": [dict(row) for row in self.store.list(request.get("target_repo"), request.get("state"))]}
+            elif op == "ack":
+                if not request.get("message_id"):
+                    response = {"ok": False, "error": "ack requires message_id"}
+                else:
+                    ok = self.store.acknowledge(request["message_id"])
+                    response = {"ok": ok, "message_id": request["message_id"], "state": "acknowledged" if ok else "missing"}
+            elif op == "endpoints":
+                response = {"ok": True, "endpoints": [dict(row) for row in self.store.endpoints()]}
+            elif op == "disconnect":
+                endpoint_id = request.get("endpoint_id")
+                if not endpoint_id:
+                    response = {"ok": False, "error": "disconnect requires endpoint_id"}
+                else:
+                    self.store.disconnect_endpoint(endpoint_id)
+                    response = {"ok": True, "endpoint_id": endpoint_id, "state": "disconnected"}
+            else:
+                response = {"ok": False, "error": "unsupported operation"}
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            response = {"ok": False, "error": "invalid JSON"}
+        writer.write((json.dumps(response) + "\n").encode())
+        await writer.drain()
+        writer.close()
+        await writer.wait_closed()
+
+    @staticmethod
+    def _peer_allowed(writer: asyncio.StreamWriter) -> bool:
+        sock = writer.get_extra_info("socket")
+        if sock is None or not hasattr(socket, "SO_PEERCRED"):
+            return True
+        try:
+            _, uid, _ = struct.unpack("3i", sock.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i")))
+            return uid == os.getuid()
+        except OSError:
+            return False
+
+    def close(self) -> None:
+        self.store.close()
+
+
+async def ping(path: Path | None = None) -> bool:
+    try:
+        reader, writer = await asyncio.open_unix_connection(str(path or socket_path()))
+        writer.write(b'{"op":"ping"}\n'); await writer.drain()
+        result = json.loads(await reader.readline())
+        writer.close(); await writer.wait_closed()
+        return bool(result.get("ok"))
+    except (OSError, json.JSONDecodeError):
+        return False
+
+
+async def request(payload: dict, path: Path | None = None) -> dict:
+    reader, writer = await asyncio.open_unix_connection(str(path or socket_path()))
+    writer.write((json.dumps(payload) + "\n").encode())
+    await writer.drain()
+    response = json.loads(await reader.readline())
+    writer.close()
+    await writer.wait_closed()
+    return response

src/tamq/store.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+from pathlib import Path
+from typing import Iterable
+from uuid import uuid4
+
+SCHEMA_VERSION = 1
+
+
+class Store:
+    def __init__(self, path: Path):
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self.db = sqlite3.connect(path, timeout=5)
+        self.db.row_factory = sqlite3.Row
+        self.db.execute("PRAGMA journal_mode=WAL")
+        self.db.execute("PRAGMA foreign_keys=ON")
+        self.db.execute("PRAGMA busy_timeout=5000")
+        self.db.executescript("""
+        CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+        CREATE TABLE IF NOT EXISTS messages (
+          message_id TEXT PRIMARY KEY,
+          sender_repo TEXT NOT NULL,
+          target_repo TEXT NOT NULL,
+          body TEXT NOT NULL,
+          created_at REAL NOT NULL,
+          state TEXT NOT NULL,
+          endpoint_id TEXT,
+          provenance TEXT,
+          injected_at REAL,
+          acknowledged_at REAL
+        );
+        CREATE INDEX IF NOT EXISTS messages_target_state ON messages(target_repo, state);
+        CREATE TABLE IF NOT EXISTS endpoints (
+          endpoint_id TEXT PRIMARY KEY,
+          pid INTEGER NOT NULL,
+          session TEXT NOT NULL,
+          repos TEXT NOT NULL,
+          connected_at REAL NOT NULL,
+          disconnected_at REAL
+        );
+        CREATE TABLE IF NOT EXISTS leases (
+          message_id TEXT PRIMARY KEY REFERENCES messages(message_id) ON DELETE CASCADE,
+          lease_id TEXT NOT NULL,
+          endpoint_id TEXT NOT NULL,
+          acquired_at REAL NOT NULL,
+          expires_at REAL NOT NULL
+        );
+        """)
+        self.db.execute("INSERT OR IGNORE INTO metadata(key,value) VALUES('schema_version',?)", (str(SCHEMA_VERSION),))
+        self.db.commit()
+
+    def close(self) -> None:
+        self.db.close()
+
+    def register_endpoint(self, endpoint_id: str, pid: int, session: str, repos: list[str]) -> None:
+        import json
+        self.db.execute(
+            "INSERT INTO endpoints(endpoint_id,pid,session,repos,connected_at,disconnected_at) VALUES(?,?,?,?,strftime('%s','now'),NULL) "
+            "ON CONFLICT(endpoint_id) DO UPDATE SET pid=excluded.pid, session=excluded.session, repos=excluded.repos, connected_at=excluded.connected_at, disconnected_at=NULL",
+            (endpoint_id, pid, session, json.dumps(repos)),
+        )
+        self.db.commit()
+
+    def disconnect_endpoint(self, endpoint_id: str) -> None:
+        self.db.execute("UPDATE endpoints SET disconnected_at=strftime('%s','now') WHERE endpoint_id=?", (endpoint_id,))
+        self.db.commit()
+
+    def disconnect_all(self) -> None:
+        self.db.execute("UPDATE endpoints SET disconnected_at=strftime('%s','now') WHERE disconnected_at IS NULL")
+        self.db.commit()
+
+    def endpoints(self) -> list[sqlite3.Row]:
+        return list(self.db.execute("SELECT * FROM endpoints ORDER BY endpoint_id"))
+
+    def endpoint(self, endpoint_id: str) -> sqlite3.Row | None:
+        row = self.db.execute("SELECT * FROM endpoints WHERE endpoint_id=? AND disconnected_at IS NULL", (endpoint_id,)).fetchone()
+        if row is not None:
+            return row
+        if endpoint_id.startswith("tmux-amq-"):
+            try:
+                pid = int(endpoint_id.removeprefix("tmux-amq-").split("-", 1)[0])
+            except ValueError:
+                return None
+            return self.db.execute("SELECT * FROM endpoints WHERE pid=? AND disconnected_at IS NULL ORDER BY connected_at DESC LIMIT 1", (pid,)).fetchone()
+        return None
+
+    def history_stats(self) -> tuple[int, float | None]:
+        size = 0
+        try:
+            size = self.db.execute("SELECT page_count * page_size FROM pragma_page_count(), pragma_page_size()").fetchone()[0]
+        except sqlite3.DatabaseError:
+            pass
+        oldest = self.db.execute("SELECT MIN(created_at) FROM messages").fetchone()[0]
+        return size, oldest
+
+    def add(self, sender: str, target: str, body: str, *, endpoint: str | None = None, provenance: str | None = None) -> str:
+        if len(body.encode("utf-8")) > 8192:
+            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),
+        )
+        self.db.commit()
+        return message_id
+
+    def list(self, target: str | None = None, state: str | None = None) -> list[sqlite3.Row]:
+        clauses, values = [], []
+        if target:
+            clauses.append("target_repo=?"); values.append(target)
+        if state:
+            clauses.append("state=?"); values.append(state)
+        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 set_state(self, message_id: str, state: str) -> None:
+        column = {"injected": "injected_at", "acknowledged": "acknowledged_at"}.get(state)
+        if column:
+            self.db.execute(f"UPDATE messages SET state=?, {column}=? WHERE message_id=?", (state, time.time(), message_id))
+        else:
+            self.db.execute("UPDATE messages SET state=? WHERE message_id=?", (state, message_id))
+        self.db.commit()
+
+    def acknowledge(self, message_id: str) -> bool:
+        row = self.db.execute("SELECT 1 FROM messages WHERE message_id=?", (message_id,)).fetchone()
+        if row is None:
+            return False
+        self.set_state(message_id, "acknowledged")
+        return True
+
+    def claim(self, message_id: str, endpoint_id: str, ttl: float = 30.0) -> str | None:
+        now = time.time()
+        lease_id = f"lease-{uuid4()}"
+        with self.db:
+            self.db.execute("DELETE FROM leases WHERE expires_at < ?", (now,))
+            try:
+                self.db.execute("INSERT INTO leases VALUES(?,?,?,?,?)", (message_id, lease_id, endpoint_id, now, now + ttl))
+            except sqlite3.IntegrityError:
+                return None
+        return lease_id
+
+    def renew(self, message_id: str, lease_id: str, ttl: float = 30.0) -> bool:
+        with self.db:
+            result = self.db.execute("UPDATE leases SET expires_at=? WHERE message_id=? AND lease_id=?", (time.time() + ttl, message_id, lease_id))
+        return result.rowcount == 1
+
+    def release(self, message_id: str, lease_id: str, state: str = "injected") -> 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.set_state(message_id, state)
+        return result.rowcount == 1
+
+    def export(self, rows: Iterable[sqlite3.Row], path: Path) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("w", encoding="utf-8") as stream:
+            for row in rows:
+                stream.write(json.dumps(dict(row), sort_keys=True) + "\n")
+
+    def purge(self, before: float | None = None, max_bytes: int | None = None) -> int:
+        rows = self.list()
+        ids = [r["message_id"] for r in rows if before is not None and r["created_at"] < before]
+        if max_bytes is not None and self.db.execute("SELECT page_count * page_size FROM pragma_page_count(), pragma_page_size()").fetchone()[0] > max_bytes:
+            ids += [r["message_id"] for r in rows if r["message_id"] not in ids][:max(0, len(rows) // 2)]
+        if ids:
+            self.db.executemany("DELETE FROM messages WHERE message_id=?", ((i,) for i in set(ids)))
+            self.db.commit()
+        return len(set(ids))

src/tamq/tmux.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import os
+import shlex
+from uuid import uuid4
+import subprocess
+from dataclasses import dataclass
+
+from .registry import repository_paths, validate_targets
+
+
+class TmuxError(RuntimeError):
+    pass
+
+
+@dataclass
+class Endpoint:
+    session: str
+    pid: int
+    repos: list[str]
+    instance_id: str = ""
+
+    @property
+    def endpoint_id(self) -> str:
+        return f"tmux-amq-{self.pid}"
+
+    @property
+    def instance_key(self) -> str:
+        return self.instance_id or self.endpoint_id
+
+
+class TmuxManager:
+    def __init__(self, session: str = "tamq"):
+        self.session = session
+
+    def _run(self, *args: str, check: bool = True) -> str:
+        result = subprocess.run(["tmux", *args], text=True, capture_output=True, check=False)
+        if check and result.returncode:
+            raise TmuxError(result.stderr.strip() or "tmux command failed")
+        return result.stdout.strip()
+
+    def ensure(self, repos: list[str], command: str = "codex") -> Endpoint:
+        if not repos:
+            raise TmuxError("at least one repository is required")
+        try:
+            validate_targets(repos)
+        except Exception as exc:
+            raise TmuxError(str(exc)) from exc
+        result = subprocess.run(["tmux", "has-session", "-t", self.session], capture_output=True)
+        if result.returncode != 0:
+            self._run("new-session", "-d", "-s", self.session, "-n", "__tamq_boot", "-c", os.getcwd(), "sleep", "infinity")
+        pid_text = self._run("display-message", "-p", "-t", self.session, "#{pid}")
+        try:
+            pid = int(pid_text)
+        except ValueError as exc:
+            raise TmuxError(f"unable to determine tmux PID: {pid_text!r}") from exc
+        endpoint = Endpoint(self.session, pid, list(dict.fromkeys(repos)), f"tmux-amq-{pid}-{uuid4().hex[:12]}")
+        paths = repository_paths()
+        windows = self._run("list-windows", "-t", self.session, "-F", "#{window_name}").splitlines()
+        for index, repo in enumerate(endpoint.repos):
+            if repo not in windows:
+                if index == 0 and "__tamq_boot" in windows:
+                    self._run("rename-window", "-t", f"{self.session}:__tamq_boot", repo)
+                    windows[windows.index("__tamq_boot")] = repo
+                else:
+                    try:
+                        repo_path = paths[repo]
+                    except KeyError as exc:
+                        raise TmuxError(f"gita path missing for registered repository: {repo}") from exc
+                    self._run("new-window", "-d", "-t", self.session, "-n", repo, "-c", repo_path, "sleep", "infinity")
+                tap = ["tamq", "tap", "--repo", repo, "--endpoint", endpoint.endpoint_id, "--", command]
+                self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(tap), "C-m")
+        return endpoint
+
+
+def shell_join(parts: list[str]) -> str:
+    return " ".join(shlex.quote(part) for part in parts)