From 4d915a26179945ce43fdae4c8c94c6022990c09a Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 24 Aug 2026 01:18:40 +0200 Subject: [PATCH] Bootstrap tamq local tmux agent message queue Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02f34-ead4-7a60-85cd-d7f08e22fa0e --- .forgejo/workflows/ci-smoke.yaml | 18 ++ .gitignore | 7 + Makefile | 9 + README.md | 71 +++++- pyproject.toml | 22 ++ src/tamq/__init__.py | 3 + src/tamq/broker.py | 55 +++++ src/tamq/cli.py | 313 +++++++++++++++++++++++++++ src/tamq/config.py | 46 ++++ src/tamq/control.py | 45 ++++ src/tamq/diagnostics.py | 27 +++ src/tamq/policy.py | 43 ++++ src/tamq/ptytap.py | 55 +++++ src/tamq/registry.py | 39 ++++ src/tamq/routing.py | 20 ++ src/tamq/service.py | 199 +++++++++++++++++ src/tamq/store.py | 171 +++++++++++++++ src/tamq/tmux.py | 77 +++++++ tests/test_ack.py | 9 + tests/test_broker.py | 11 + tests/test_broker_delivery.py | 21 ++ tests/test_broker_registry.py | 10 + tests/test_cli.py | 18 ++ tests/test_cli_helpers.py | 6 + tests/test_completion.py | 9 + tests/test_completion_dynamic.py | 7 + tests/test_config_command.py | 11 + tests/test_config_defaults.py | 20 ++ tests/test_config_paths.py | 6 + tests/test_config_toml.py | 11 + tests/test_control.py | 12 + tests/test_diagnostics.py | 9 + tests/test_disconnect.py | 17 ++ tests/test_endpoint.py | 7 + tests/test_endpoint_authorization.py | 25 +++ tests/test_endpoints.py | 15 ++ tests/test_leases.py | 12 + tests/test_message_limits.py | 9 + tests/test_peer_auth.py | 5 + tests/test_policy.py | 14 ++ tests/test_policy_ack.py | 7 + tests/test_protocol.py | 21 ++ tests/test_purge.py | 29 +++ tests/test_registry.py | 16 ++ tests/test_registry_paths.py | 10 + tests/test_replay.py | 31 +++ tests/test_request.py | 19 ++ tests/test_routing.py | 11 + tests/test_send_endpoint.py | 8 + tests/test_service.py | 24 ++ tests/test_service_delivery.py | 31 +++ tests/test_service_registry.py | 23 ++ tests/test_service_send_registry.py | 22 ++ tests/test_shutdown.py | 25 +++ tests/test_stop.py | 11 + tests/test_store.py | 15 ++ tests/test_tmux_manager.py | 20 ++ uv.lock | 76 +++++++ 58 files changed, 1882 insertions(+), 1 deletion(-) create mode 100644 .forgejo/workflows/ci-smoke.yaml create mode 100644 .gitignore create mode 100644 Makefile create mode 100644 pyproject.toml create mode 100644 src/tamq/__init__.py create mode 100644 src/tamq/broker.py create mode 100644 src/tamq/cli.py create mode 100644 src/tamq/config.py create mode 100644 src/tamq/control.py create mode 100644 src/tamq/diagnostics.py create mode 100644 src/tamq/policy.py create mode 100644 src/tamq/ptytap.py create mode 100644 src/tamq/registry.py create mode 100644 src/tamq/routing.py create mode 100644 src/tamq/service.py create mode 100644 src/tamq/store.py create mode 100644 src/tamq/tmux.py create mode 100644 tests/test_ack.py create mode 100644 tests/test_broker.py create mode 100644 tests/test_broker_delivery.py create mode 100644 tests/test_broker_registry.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_cli_helpers.py create mode 100644 tests/test_completion.py create mode 100644 tests/test_completion_dynamic.py create mode 100644 tests/test_config_command.py create mode 100644 tests/test_config_defaults.py create mode 100644 tests/test_config_paths.py create mode 100644 tests/test_config_toml.py create mode 100644 tests/test_control.py create mode 100644 tests/test_diagnostics.py create mode 100644 tests/test_disconnect.py create mode 100644 tests/test_endpoint.py create mode 100644 tests/test_endpoint_authorization.py create mode 100644 tests/test_endpoints.py create mode 100644 tests/test_leases.py create mode 100644 tests/test_message_limits.py create mode 100644 tests/test_peer_auth.py create mode 100644 tests/test_policy.py create mode 100644 tests/test_policy_ack.py create mode 100644 tests/test_protocol.py create mode 100644 tests/test_purge.py create mode 100644 tests/test_registry.py create mode 100644 tests/test_registry_paths.py create mode 100644 tests/test_replay.py create mode 100644 tests/test_request.py create mode 100644 tests/test_routing.py create mode 100644 tests/test_send_endpoint.py create mode 100644 tests/test_service.py create mode 100644 tests/test_service_delivery.py create mode 100644 tests/test_service_registry.py create mode 100644 tests/test_service_send_registry.py create mode 100644 tests/test_shutdown.py create mode 100644 tests/test_stop.py create mode 100644 tests/test_store.py create mode 100644 tests/test_tmux_manager.py create mode 100644 uv.lock diff --git a/.forgejo/workflows/ci-smoke.yaml b/.forgejo/workflows/ci-smoke.yaml new file mode 100644 index 0000000..e918d72 --- /dev/null +++ b/.forgejo/workflows/ci-smoke.yaml @@ -0,0 +1,18 @@ +name: tamq-ci + +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - run: python -m pip install -e '.[dev]' + - run: pytest -q + - run: python -m tamq.cli --help + - run: python -m tamq.cli --version diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c31e32e --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +.pytest_cache/ +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..997767f --- /dev/null +++ b/Makefile @@ -0,0 +1,9 @@ +.PHONY: test check + +test: + uv run pytest + +check: + uv run pytest + git diff --check + python3 -m compileall -q src diff --git a/README.md b/README.md index 4cd559c..185466a 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,72 @@ # tmux-amq -A local tmux based agentic message queue for repository worker agent coordination. \ No newline at end of file +Tmux Agentic Message Queueing (`tamq`) is a local, durable message queue for +agent workers running in gita-registered repositories. + +The initial implementation is being bootstrapped from the standalone queue and +CLI contract. It will provide tmux endpoint lifecycle, local SQLite history and +leases, direct `@repo:` routing, JSONL export/replay, and a Unix-socket attach +protocol for a later coordination-engine adapter. + +## Development + +```bash +uv run pytest +uv run tamq --help +uv run tamq --version +uv run tamq start --cmd codex net-kingdom railiance-platform +uv run tamq attach +``` + +The bootstrap currently includes the local SQLite queue and a tmux endpoint +manager, strict gita target validation, and a control-mode client for tmux +operations/injection, and the PTY tap for full-duplex input observation. The +control-mode client alone does not expose arbitrary pane input; `tamq tap` is +the supported input-broker hook. `tamq send` remains available as an explicit +fallback. coordination-engine integration remains a later phase. + +## Terminal architecture + +`tamq` separates terminal coordination into three layers: + +1. tmux control mode is the topology and output/control stream; +2. the local broker assigns endpoint/source identity and durably queues intent; +3. `tamq tap` is a full-duplex PTY proxy around the agent process. It forwards + bytes unchanged while observing complete input lines for `@repo:` routing. + +This keeps tmux-specific topology concerns separate from reusable terminal I/O +observation and message identity. + +Pending messages are delivered through the control-mode client as +`#sender-repo: body`, then marked `injected` in SQLite. Delivery is +endpoint-scoped and uses the same message ID for retry/deduplication. +The visible endpoint label is `tmux-amq-`; each boot also receives an +instance nonce so PID reuse cannot collide with prior leases or receipts. + +## Configuration + +Configuration is read from `${XDG_CONFIG_HOME:-~/.config}/tamq/config.toml` (or +`$TAMQ_CONFIG`). Environment variables such as `TAMQ_SOCKET` and +`TAMQ_STATE_DIR` take precedence. The purge and startup advisory defaults can +be tuned without changing command lines: + +```toml +[tamq] +state_dir = "/run/user/1000/tamq" +purge_before = "365d" +purge_max_size = "100MB" +history_max_size = "100MB" +delivery_poll_interval = "0.5" + +[policy.profiles.diagnostics] +safety_gated_max_attempts = 2 +delivery_ack_mode = "acknowledged" +``` + +Retries remain safety-gated and capped at nine attempts. Use `--policy-profile` +to select a profile; `--orwell` enables explicitly unsafe local diagnostics. + +The Unix socket service now supports structured `ping`, `register`, `send`, and +`history` operations. Endpoint registrations and messages are persisted in the +same local SQLite database; delivery remains delegated to the control-mode +adapter. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e526dd5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "tmux-amq" +version = "0.1.0" +description = "Tmux Agentic Message Queueing for local repository workers" +requires-python = ">=3.11" +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest>=8"] + +[project.scripts] +tamq = "tamq.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +pythonpath = ["src"] diff --git a/src/tamq/__init__.py b/src/tamq/__init__.py new file mode 100644 index 0000000..c0260aa --- /dev/null +++ b/src/tamq/__init__.py @@ -0,0 +1,3 @@ +"""Tmux Agentic Message Queueing.""" + +__version__ = "0.1.0" diff --git a/src/tamq/broker.py b/src/tamq/broker.py new file mode 100644 index 0000000..c7ddc5a --- /dev/null +++ b/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 diff --git a/src/tamq/cli.py b/src/tamq/cli.py new file mode 100644 index 0000000..9197c3d --- /dev/null +++ b/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()) diff --git a/src/tamq/config.py b/src/tamq/config.py new file mode 100644 index 0000000..0dab7ff --- /dev/null +++ b/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")) diff --git a/src/tamq/control.py b/src/tamq/control.py new file mode 100644 index 0000000..3b3f798 --- /dev/null +++ b/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("'", "'\\''") + "'" diff --git a/src/tamq/diagnostics.py b/src/tamq/diagnostics.py new file mode 100644 index 0000000..3c092d2 --- /dev/null +++ b/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 diff --git a/src/tamq/policy.py b/src/tamq/policy.py new file mode 100644 index 0000000..cd0e79b --- /dev/null +++ b/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) diff --git a/src/tamq/ptytap.py b/src/tamq/ptytap.py new file mode 100644 index 0000000..50a9208 --- /dev/null +++ b/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 diff --git a/src/tamq/registry.py b/src/tamq/registry.py new file mode 100644 index 0000000..604420b --- /dev/null +++ b/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)}") diff --git a/src/tamq/routing.py b/src/tamq/routing.py new file mode 100644 index 0000000..8d15fab --- /dev/null +++ b/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)) diff --git a/src/tamq/service.py b/src/tamq/service.py new file mode 100644 index 0000000..c81a009 --- /dev/null +++ b/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 diff --git a/src/tamq/store.py b/src/tamq/store.py new file mode 100644 index 0000000..2422fc4 --- /dev/null +++ b/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)) diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py new file mode 100644 index 0000000..9ce202b --- /dev/null +++ b/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) diff --git a/tests/test_ack.py b/tests/test_ack.py new file mode 100644 index 0000000..d1c6973 --- /dev/null +++ b/tests/test_ack.py @@ -0,0 +1,9 @@ +from tamq.store import Store + + +def test_acknowledge_message(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add("a", "b", "body") + assert store.acknowledge(message_id) + assert store.list()[0]["state"] == "acknowledged" + assert not store.acknowledge("missing") diff --git a/tests/test_broker.py b/tests/test_broker.py new file mode 100644 index 0000000..327d478 --- /dev/null +++ b/tests/test_broker.py @@ -0,0 +1,11 @@ +from tamq.broker import BrokerIdentity, InputBroker +from tamq.store import Store + + +def test_broker_preserves_identity(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + broker = InputBroker(store, BrokerIdentity("tmux-amq-42", "net-kingdom")) + assert broker.inspect_line("@railiance-platform: hello") is not None + row = store.list()[0] + assert row["sender_repo"] == "net-kingdom" + assert row["endpoint_id"] == "tmux-amq-42" diff --git a/tests/test_broker_delivery.py b/tests/test_broker_delivery.py new file mode 100644 index 0000000..f8e9d66 --- /dev/null +++ b/tests/test_broker_delivery.py @@ -0,0 +1,21 @@ +from tamq.broker import BrokerIdentity, InputBroker +from tamq.store import Store + + +class FakeControl: + def __init__(self): + self.calls = [] + + def inject(self, window, text): + self.calls.append((window, text)) + + +def test_pending_delivery_marks_injected(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add("net-kingdom", "railiance-platform", "hello", endpoint="tmux-amq-42") + control = FakeControl() + count = InputBroker(store, BrokerIdentity("tmux-amq-42", "net-kingdom")).deliver_pending(control, window_for_repo=lambda repo: f"tamq:{repo}") + assert count == 1 + assert control.calls == [("tamq:railiance-platform", "#net-kingdom: hello")] + assert store.list()[0]["message_id"] == message_id + assert store.list()[0]["state"] == "injected" diff --git a/tests/test_broker_registry.py b/tests/test_broker_registry.py new file mode 100644 index 0000000..7bccbff --- /dev/null +++ b/tests/test_broker_registry.py @@ -0,0 +1,10 @@ +from tamq import broker +from tamq.broker import BrokerIdentity, InputBroker +from tamq.store import Store + + +def test_broker_rejects_unregistered_target(tmp_path, monkeypatch): + monkeypatch.setattr(broker, "validate_targets", lambda repos: (_ for _ in ()).throw(broker.RegistryError("unknown"))) + store = Store(tmp_path / "queue.sqlite3") + assert InputBroker(store, BrokerIdentity("ep", "source")).inspect_line("@missing: hello") is None + assert store.list() == [] diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..0173619 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,18 @@ +from tamq.cli import main +import pytest + + +def test_help_and_version(capsys): + with pytest.raises(SystemExit) as exc: + main(["--version"]) + assert exc.value.code == 0 + assert "0.1.0" in capsys.readouterr().out + assert main([]) == 0 + assert "usage:" in capsys.readouterr().out + + +def test_attach_delegates_to_tmux(monkeypatch): + calls = [] + monkeypatch.setattr("tamq.cli.subprocess.run", lambda args, check=False: calls.append((args, check)) or type("R", (), {"returncode": 0})()) + assert main(["attach", "--session", "demo"]) == 0 + assert calls == [(["tmux", "attach-session", "-t", "demo"], False)] diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py new file mode 100644 index 0000000..8c4f54c --- /dev/null +++ b/tests/test_cli_helpers.py @@ -0,0 +1,6 @@ +from tamq.cli import parse_size + + +def test_parse_size(): + assert parse_size("100MB") == 100_000_000 + assert parse_size("2KB") == 2_000 diff --git a/tests/test_completion.py b/tests/test_completion.py new file mode 100644 index 0000000..3c560cf --- /dev/null +++ b/tests/test_completion.py @@ -0,0 +1,9 @@ +from tamq.cli import completion_script + + +def test_completion_scripts_include_commands(): + for shell in ("bash", "zsh", "fish"): + script = completion_script(shell) + assert "start" in script + assert "attach" in script + assert "replay" in script diff --git a/tests/test_completion_dynamic.py b/tests/test_completion_dynamic.py new file mode 100644 index 0000000..c4944dc --- /dev/null +++ b/tests/test_completion_dynamic.py @@ -0,0 +1,7 @@ +from tamq.cli import completion_script + + +def test_completion_mentions_gita(): + assert "gita ls" in completion_script("bash") + assert "gita ls" in completion_script("zsh") + assert "gita ls" in completion_script("fish") diff --git a/tests/test_config_command.py b/tests/test_config_command.py new file mode 100644 index 0000000..43fecb9 --- /dev/null +++ b/tests/test_config_command.py @@ -0,0 +1,11 @@ +import json + +from tamq.cli import main + + +def test_config_command(monkeypatch, capsys, tmp_path): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + assert main(["config"]) == 0 + output = json.loads(capsys.readouterr().out) + assert output["database"].endswith("tamq.sqlite3") + assert output["policy_profile"] == "default" diff --git a/tests/test_config_defaults.py b/tests/test_config_defaults.py new file mode 100644 index 0000000..b6e3264 --- /dev/null +++ b/tests/test_config_defaults.py @@ -0,0 +1,20 @@ +from tamq.config import setting + + +def test_toml_defaults(monkeypatch, tmp_path): + config = tmp_path / "config.toml" + config.write_text("[tamq]\npurge_before='30d'\nhistory_max_size='12MB'\n") + monkeypatch.setenv("TAMQ_CONFIG", str(config)) + assert setting("purge_before", "365d") == "30d" + assert setting("history_max_size", "100MB") == "12MB" + assert setting("missing", "fallback") == "fallback" + + +def test_delivery_interval_is_configurable(monkeypatch, tmp_path): + config = tmp_path / "config.toml" + config.write_text("[tamq]\ndelivery_poll_interval='2.5'\n") + monkeypatch.setenv("TAMQ_CONFIG", str(config)) + from tamq.service import Service + from tamq.store import Store + service = Service(store=Store(tmp_path / "state.sqlite3")) + assert service.poll_interval == 2.5 diff --git a/tests/test_config_paths.py b/tests/test_config_paths.py new file mode 100644 index 0000000..ea48e58 --- /dev/null +++ b/tests/test_config_paths.py @@ -0,0 +1,6 @@ +from tamq.config import lock_path + + +def test_lock_path_override(monkeypatch, tmp_path): + monkeypatch.setenv("TAMQ_LOCKFILE", str(tmp_path / "tamq.lock")) + assert lock_path() == tmp_path / "tamq.lock" diff --git a/tests/test_config_toml.py b/tests/test_config_toml.py new file mode 100644 index 0000000..8bba668 --- /dev/null +++ b/tests/test_config_toml.py @@ -0,0 +1,11 @@ +from tamq.config import socket_path, state_dir + + +def test_toml_paths(monkeypatch, tmp_path): + config = tmp_path / "config.toml" + config.write_text(f"[tamq]\nsocket='{tmp_path / 'socket'}'\nstate_dir='{tmp_path / 'state'}'\n") + monkeypatch.setenv("TAMQ_CONFIG", str(config)) + monkeypatch.delenv("TAMQ_SOCKET", raising=False) + monkeypatch.delenv("TAMQ_STATE_DIR", raising=False) + assert socket_path() == tmp_path / "socket" + assert state_dir() == tmp_path / "state" diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..e0b75ac --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,12 @@ +from tamq.control import shell_quote +from tamq.tmux import shell_join + + +def test_shell_quote(): + quoted = shell_quote("hello 'world'") + assert quoted.startswith("'") and quoted.endswith("'") + assert "\\'" in quoted + + +def test_tmux_shell_join_quotes_arguments(): + assert shell_join(["tamq", "tap", "--repo", "net-kingdom", "--", "codex"]) == "tamq tap --repo net-kingdom -- codex" diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py new file mode 100644 index 0000000..fb1ee52 --- /dev/null +++ b/tests/test_diagnostics.py @@ -0,0 +1,9 @@ +from tamq.diagnostics import configure + + +def test_orwell_log_is_private(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path)) + configure(orwell=True) + warning = capsys.readouterr().err + assert "--orwell" in warning + assert (tmp_path / "tamq" / "tamq-orwell.log").stat().st_mode & 0o077 == 0 diff --git a/tests/test_disconnect.py b/tests/test_disconnect.py new file mode 100644 index 0000000..c985401 --- /dev/null +++ b/tests/test_disconnect.py @@ -0,0 +1,17 @@ +from tamq.store import Store + + +def test_disconnect_endpoint(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep", 1, "tamq", ["a"]) + store.disconnect_endpoint("ep") + assert store.endpoint("ep") is None + + +def test_disconnect_all_marks_active_endpoints(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep-1", 1, "tamq", ["a"]) + store.register_endpoint("ep-2", 2, "tamq", ["b"]) + store.disconnect_all() + assert store.endpoint("ep-1") is None + assert store.endpoint("ep-2") is None diff --git a/tests/test_endpoint.py b/tests/test_endpoint.py new file mode 100644 index 0000000..9963903 --- /dev/null +++ b/tests/test_endpoint.py @@ -0,0 +1,7 @@ +from tamq.tmux import Endpoint + + +def test_endpoint_instance_key(): + endpoint = Endpoint("tamq", 42, ["net-kingdom"], "tmux-amq-42-boot") + assert endpoint.endpoint_id == "tmux-amq-42" + assert endpoint.instance_key == "tmux-amq-42-boot" diff --git a/tests/test_endpoint_authorization.py b/tests/test_endpoint_authorization.py new file mode 100644 index 0000000..e4c7cba --- /dev/null +++ b/tests/test_endpoint_authorization.py @@ -0,0 +1,25 @@ +import asyncio +import json + +from tamq.service import Service +from tamq.store import Store +from tamq import service + + +def test_send_rejects_endpoint_target_not_attached(tmp_path, monkeypatch): + monkeypatch.setattr(service, "validate_targets", lambda repos: None) + async def run(): + path = tmp_path / "tamq.sock" + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep", 1, "tamq", ["source"]) + instance = Service(path, store) + server = await asyncio.start_unix_server(instance.handle, path=str(path)) + try: + reader, writer = await asyncio.open_unix_connection(str(path)) + payload = {"op":"send","endpoint_id":"ep","sender_repo":"source","target_repo":"other","body":"hello"} + writer.write((json.dumps(payload) + "\n").encode()); await writer.drain() + assert json.loads(await reader.readline())["ok"] is False + writer.close(); await writer.wait_closed() + finally: + server.close(); await server.wait_closed(); instance.close() + asyncio.run(run()) diff --git a/tests/test_endpoints.py b/tests/test_endpoints.py new file mode 100644 index 0000000..a479ee9 --- /dev/null +++ b/tests/test_endpoints.py @@ -0,0 +1,15 @@ +from tamq.store import Store + + +def test_endpoint_registration(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-1-boot", 1, "tamq", ["a"]) + row = store.endpoints()[0] + assert row["endpoint_id"] == "tmux-amq-1-boot" + assert row["pid"] == 1 + + +def test_endpoint_resolves_visible_pid_id(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-1-boot", 1, "tamq", ["a"]) + assert store.endpoint("tmux-amq-1")["endpoint_id"] == "tmux-amq-1-boot" diff --git a/tests/test_leases.py b/tests/test_leases.py new file mode 100644 index 0000000..aece348 --- /dev/null +++ b/tests/test_leases.py @@ -0,0 +1,12 @@ +from tamq.store import Store + + +def test_message_lease_claim_and_release(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add("a", "b", "body") + lease = store.claim(message_id, "tmux-amq-1") + assert lease is not None + assert store.claim(message_id, "tmux-amq-2") is None + assert store.renew(message_id, lease) + assert store.release(message_id, lease) + assert store.list()[0]["state"] == "injected" diff --git a/tests/test_message_limits.py b/tests/test_message_limits.py new file mode 100644 index 0000000..bdec730 --- /dev/null +++ b/tests/test_message_limits.py @@ -0,0 +1,9 @@ +import pytest + +from tamq.store import Store + + +def test_message_body_limit(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + with pytest.raises(ValueError): + store.add("a", "b", "x" * 8193) diff --git a/tests/test_peer_auth.py b/tests/test_peer_auth.py new file mode 100644 index 0000000..222a73d --- /dev/null +++ b/tests/test_peer_auth.py @@ -0,0 +1,5 @@ +from tamq.service import Service + + +def test_peer_auth_method_exists(): + assert hasattr(Service, "_peer_allowed") diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..f68e466 --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,14 @@ +from tamq.policy import load_profile + + +def test_default_policy(tmp_path): + profile = load_profile(tmp_path / "missing.toml") + assert profile.name == "default" + assert "secrets" in profile.require_human + + +def test_configured_policy(tmp_path): + path = tmp_path / "config.toml" + path.write_text("[policy.profiles.dev]\nallow=['repo_inspect']\nrequire_human=['destructive']\nsafety_gated_max_attempts=2\n") + profile = load_profile(path, "dev") + assert profile.safety_gated_max_attempts == 2 diff --git a/tests/test_policy_ack.py b/tests/test_policy_ack.py new file mode 100644 index 0000000..36a3b9b --- /dev/null +++ b/tests/test_policy_ack.py @@ -0,0 +1,7 @@ +from tamq.policy import load_profile + + +def test_ack_mode_profile(tmp_path): + path = tmp_path / "config.toml" + path.write_text("[policy.profiles.safe]\ndelivery_ack_mode='acknowledged'\n") + assert load_profile(path, "safe").delivery_ack_mode == "acknowledged" diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..ec89606 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,21 @@ +import asyncio +import json + +from tamq.service import Service +from tamq.store import Store + + +def test_incompatible_protocol_rejected(tmp_path): + async def run(): + path = tmp_path / "tamq.sock" + service = Service(path, Store(tmp_path / "queue.sqlite3")) + server = await asyncio.start_unix_server(service.handle, path=str(path)) + try: + reader, writer = await asyncio.open_unix_connection(str(path)) + writer.write(b'{"op":"ping","protocol":"2.0"}\n'); await writer.drain() + response = json.loads(await reader.readline()) + assert response["ok"] is False + writer.close(); await writer.wait_closed() + finally: + server.close(); await server.wait_closed(); service.close() + asyncio.run(run()) diff --git a/tests/test_purge.py b/tests/test_purge.py new file mode 100644 index 0000000..31f316c --- /dev/null +++ b/tests/test_purge.py @@ -0,0 +1,29 @@ +import json + +from tamq.cli import main +from tamq.store import Store + + +def test_purge_dry_run_does_not_delete(tmp_path, monkeypatch, capsys): + state = tmp_path / "state" + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + store = Store(state / "tamq.sqlite3") + store.add("a", "b", "body") + store.close() + assert main(["purge"]) == 0 + assert "Dry run" in capsys.readouterr().out + store = Store(state / "tamq.sqlite3") + assert len(store.list()) == 1 + + +def test_purge_confirmed_deletes(tmp_path, monkeypatch, capsys): + state = tmp_path / "state" + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + store = Store(state / "tamq.sqlite3") + store.add("a", "b", "body") + store.db.execute("UPDATE messages SET created_at=created_at-400*86400") + store.db.commit() + store.close() + assert main(["purge", "--yes"]) == 0 + assert "Purged" in capsys.readouterr().out + assert len(Store(state / "tamq.sqlite3").list()) == 0 diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..65545a8 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,16 @@ +from tamq import registry + + +def test_validate_targets(monkeypatch): + monkeypatch.setattr(registry, "registered_repositories", lambda: {"net-kingdom"}) + registry.validate_targets(["net-kingdom"]) + + +def test_reject_unknown_targets(monkeypatch): + monkeypatch.setattr(registry, "registered_repositories", lambda: {"net-kingdom"}) + try: + registry.validate_targets(["missing"]) + except registry.RegistryError as exc: + assert "missing" in str(exc) + else: + raise AssertionError("unknown target was accepted") diff --git a/tests/test_registry_paths.py b/tests/test_registry_paths.py new file mode 100644 index 0000000..2614a9f --- /dev/null +++ b/tests/test_registry_paths.py @@ -0,0 +1,10 @@ +from tamq import registry + + +def test_repository_paths(monkeypatch): + class Result: + returncode = 0 + stdout = "remote,net-kingdom,/home/worsch/net-kingdom\n" + stderr = "" + monkeypatch.setattr(registry.subprocess, "run", lambda *args, **kwargs: Result()) + assert registry.repository_paths()["net-kingdom"] == "/home/worsch/net-kingdom" diff --git a/tests/test_replay.py b/tests/test_replay.py new file mode 100644 index 0000000..8afbf95 --- /dev/null +++ b/tests/test_replay.py @@ -0,0 +1,31 @@ +import json + +from tamq.cli import main +from tamq.store import Store + + +def test_replay_reports_batch(tmp_path, monkeypatch, capsys): + source = tmp_path / "messages.jsonl" + source.write_text(json.dumps({"message_id": "old-1", "sender_repo": "a", "target_repo": "b", "body": "hello"}) + "\n") + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + assert main(["replay", str(source)]) == 0 + output = json.loads(capsys.readouterr().out) + assert output["batch_id"].startswith("replay-") + assert output["count"] == 1 + + +def test_replay_preserves_endpoint(tmp_path, monkeypatch, capsys): + source = tmp_path / "messages.jsonl" + source.write_text(json.dumps({"sender_repo": "a", "target_repo": "b", "body": "hello", "endpoint_id": "tmux-amq-42"}) + "\n") + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + assert main(["replay", str(source)]) == 0 + row = Store(tmp_path / "state" / "tamq.sqlite3").list()[0] + assert row["endpoint_id"] == "tmux-amq-42" + + +def test_replay_rejects_bad_json(tmp_path, monkeypatch, capsys): + source = tmp_path / "bad.jsonl" + source.write_text("not-json\n") + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + assert main(["replay", str(source)]) == 2 + assert "cannot replay" in capsys.readouterr().err diff --git a/tests/test_request.py b/tests/test_request.py new file mode 100644 index 0000000..af7c76f --- /dev/null +++ b/tests/test_request.py @@ -0,0 +1,19 @@ +import asyncio +import json + +from tamq.service import request + + +def test_request_round_trip(tmp_path): + async def run(): + path = tmp_path / "tamq.sock" + async def handle(reader, writer): + payload = json.loads(await reader.readline()) + writer.write((json.dumps({"ok": payload["op"] == "ping"}) + "\n").encode()) + await writer.drain(); writer.close(); await writer.wait_closed() + server = await asyncio.start_unix_server(handle, path=str(path)) + try: + assert await request({"op": "ping"}, path) == {"ok": True} + finally: + server.close(); await server.wait_closed() + asyncio.run(run()) diff --git a/tests/test_routing.py b/tests/test_routing.py new file mode 100644 index 0000000..8df82f7 --- /dev/null +++ b/tests/test_routing.py @@ -0,0 +1,11 @@ +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" + + +def test_ordinary_input_is_unchanged(): + assert parse_address_line("hello @repo: not at start") is None + assert parse_address_line("@repo:") is None diff --git a/tests/test_send_endpoint.py b/tests/test_send_endpoint.py new file mode 100644 index 0000000..84a39f2 --- /dev/null +++ b/tests/test_send_endpoint.py @@ -0,0 +1,8 @@ +import json + +from tamq.cli import build_parser + + +def test_send_accepts_endpoint_id(): + args = build_parser().parse_args(["send", "--endpoint-id", "ep", "@repo:", "hello"]) + assert args.endpoint_id == "ep" diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..a93d9dd --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,24 @@ +import asyncio +import json + +from tamq.service import Service +from tamq.store import Store + + +def test_register_and_send(tmp_path): + async def run(): + path = tmp_path / "tamq.sock" + service = Service(path, Store(tmp_path / "queue.sqlite3")) + server = await asyncio.start_unix_server(service.handle, path=str(path)) + try: + reader, writer = await asyncio.open_unix_connection(str(path)) + writer.write((json.dumps({"op": "register", "endpoint_id": "tmux-amq-42", "pid": 42, "session": "tamq", "repos": ["net-kingdom"]}) + "\n").encode()); await writer.drain() + assert json.loads(await reader.readline())["ok"] is True + writer.close(); await writer.wait_closed() + reader, writer = await asyncio.open_unix_connection(str(path)) + writer.write(b'{"op":"send","sender_repo":"net-kingdom","target_repo":"railiance-platform","body":"hello"}\n'); await writer.drain() + assert json.loads(await reader.readline())["state"] == "pending" + writer.close(); await writer.wait_closed() + finally: + server.close(); await server.wait_closed(); service.close() + asyncio.run(run()) diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py new file mode 100644 index 0000000..5ec4ea7 --- /dev/null +++ b/tests/test_service_delivery.py @@ -0,0 +1,31 @@ +from tamq.service import Service +from tamq.store import Store + + +class FakeControl: + instances = [] + + def __init__(self, session): + self.session = session + self.injected = [] + self.__class__.instances.append(self) + + def start(self): + return None + + def inject(self, window, text): + self.injected.append((window, text)) + + def close(self): + return None + + +def test_service_delivery_injects_pending_messages(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"]) + message_id = store.add("repo-a", "repo-b", "continue") + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + Service(store=store)._deliver_once() + assert FakeControl.instances[-1].injected == [("tamq:repo-b", "#repo-a: continue")] + assert store.list()[0]["message_id"] == message_id + assert store.list()[0]["state"] == "injected" diff --git a/tests/test_service_registry.py b/tests/test_service_registry.py new file mode 100644 index 0000000..b7955f7 --- /dev/null +++ b/tests/test_service_registry.py @@ -0,0 +1,23 @@ +import asyncio +import json + +from tamq.service import Service +from tamq.store import Store +from tamq import service + + +def test_register_rejects_unknown_repo(tmp_path, monkeypatch): + monkeypatch.setattr(service, "validate_targets", lambda repos: (_ for _ in ()).throw(service.RegistryError("unknown"))) + async def run(): + socket = tmp_path / "tamq.sock" + instance = Service(socket, Store(tmp_path / "queue.sqlite3")) + server = await asyncio.start_unix_server(instance.handle, path=str(socket)) + try: + reader, writer = await asyncio.open_unix_connection(str(socket)) + writer.write((json.dumps({"op": "register", "endpoint_id": "tmux-amq-1", "pid": 1, "session": "tamq", "repos": ["missing"]}) + "\n").encode()); await writer.drain() + response = json.loads(await reader.readline()) + assert response["ok"] is False + writer.close(); await writer.wait_closed() + finally: + server.close(); await server.wait_closed(); instance.close() + asyncio.run(run()) diff --git a/tests/test_service_send_registry.py b/tests/test_service_send_registry.py new file mode 100644 index 0000000..52e9e7b --- /dev/null +++ b/tests/test_service_send_registry.py @@ -0,0 +1,22 @@ +import asyncio +import json + +from tamq import service +from tamq.service import Service +from tamq.store import Store + + +def test_send_rejects_unknown_target(tmp_path, monkeypatch): + monkeypatch.setattr(service, "validate_targets", lambda repos: (_ for _ in ()).throw(service.RegistryError("unknown"))) + async def run(): + socket = tmp_path / "tamq.sock" + instance = Service(socket, Store(tmp_path / "queue.sqlite3")) + server = await asyncio.start_unix_server(instance.handle, path=str(socket)) + try: + reader, writer = await asyncio.open_unix_connection(str(socket)) + writer.write(b'{"op":"send","sender_repo":"a","target_repo":"missing","body":"hello"}\n'); await writer.drain() + assert json.loads(await reader.readline())["ok"] is False + writer.close(); await writer.wait_closed() + finally: + server.close(); await server.wait_closed(); instance.close() + asyncio.run(run()) diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py new file mode 100644 index 0000000..9546a0b --- /dev/null +++ b/tests/test_shutdown.py @@ -0,0 +1,25 @@ +import asyncio + +from tamq.service import Service +from tamq.store import Store + + +def test_service_shutdown_removes_socket(tmp_path, monkeypatch): + monkeypatch.setenv("TAMQ_PIDFILE", str(tmp_path / "tamq.pid")) + async def run(): + socket = tmp_path / "tamq.sock" + service = Service(socket, Store(tmp_path / "queue.sqlite3")) + task = asyncio.create_task(service.run()) + for _ in range(20): + if socket.exists(): + break + await asyncio.sleep(0.01) + assert socket.exists() + service.server.close() + await service.server.wait_closed() + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + asyncio.run(run()) diff --git a/tests/test_stop.py b/tests/test_stop.py new file mode 100644 index 0000000..22dac43 --- /dev/null +++ b/tests/test_stop.py @@ -0,0 +1,11 @@ +from tamq.cli import main + + +def test_stop_cleans_stale_pidfile(tmp_path, monkeypatch, capsys): + pidfile = tmp_path / "tamq.pid" + pidfile.write_text("999999") + monkeypatch.setattr("tamq.cli.pid_path", lambda: pidfile) + monkeypatch.setattr("tamq.cli.os.kill", lambda pid, sig: (_ for _ in ()).throw(ProcessLookupError)) + assert main(["stop"]) == 1 + assert not pidfile.exists() + assert "not running" in capsys.readouterr().err diff --git a/tests/test_store.py b/tests/test_store.py new file mode 100644 index 0000000..6f6f966 --- /dev/null +++ b/tests/test_store.py @@ -0,0 +1,15 @@ +import json + +from tamq.store import Store + + +def test_message_history_and_jsonl(tmp_path): + store = Store(tmp_path / "tamq.sqlite3") + message_id = store.add("net-kingdom", "railiance-platform", "hello") + rows = store.list(target="railiance-platform") + assert rows[0]["message_id"] == message_id + output = tmp_path / "messages.jsonl" + store.export(rows, output) + assert json.loads(output.read_text())["body"] == "hello" + assert store.db.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + store.close() diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py new file mode 100644 index 0000000..cccfb55 --- /dev/null +++ b/tests/test_tmux_manager.py @@ -0,0 +1,20 @@ +from tamq import tmux + + +def test_tmux_manager_builds_tap_windows(monkeypatch): + calls = [] + manager = tmux.TmuxManager("tamq-test") + monkeypatch.setattr(tmux, "validate_targets", lambda repos: None) + monkeypatch.setattr(tmux, "repository_paths", lambda: {"a": "/tmp/a", "b": "/tmp/b"}) + def run(*args, check=True): + calls.append(args) + if args[:2] == ("list-windows", "-t"): + return "__tamq_boot" + if args[:2] == ("display-message", "-p"): + return "42" + return "" + monkeypatch.setattr(manager, "_run", run) + monkeypatch.setattr(tmux.subprocess, "run", lambda *args, **kwargs: type("R", (), {"returncode": 1})()) + endpoint = manager.ensure(["a", "b"], "codex") + assert endpoint.endpoint_id == "tmux-amq-42" + assert any("tamq tap" in " ".join(call) for call in calls if call and call[0] == "send-keys") diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..1ec9d65 --- /dev/null +++ b/uv.lock @@ -0,0 +1,76 @@ +version = 1 +requires-python = ">=3.11" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335 }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484 }, +] + +[[package]] +name = "packaging" +version = "26.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956 }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538 }, +] + +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147 }, +] + +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536 }, +] + +[[package]] +name = "tmux-amq" +version = "0.1.0" +source = { editable = "." } + +[package.optional-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [{ name = "pytest", marker = "extra == 'dev'", specifier = ">=8" }]