From 74b2f27997f499c844b62ea0f2757dc782fbd05e Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 24 Aug 2026 20:10:08 +0200 Subject: [PATCH] feat: make sessions terminal neutral Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- .claude/rules/stack-and-commands.md | 7 +- AGENTS.md | 9 +- INTENT.md | 8 +- README.md | 91 +++++--- SCOPE.md | 48 +++-- WORK-RECORDS.md | 10 +- src/tamq/cli.py | 195 +++++++++++++----- src/tamq/service.py | 15 +- src/tamq/store.py | 33 ++- src/tamq/tmux.py | 51 +++-- tests/test_cli.py | 42 +++- tests/test_cli_helpers.py | 20 +- tests/test_install_target.py | 103 +++++++-- tests/test_manual_messaging.py | 37 ++++ tests/test_ptytap_integration.py | 2 +- tests/test_service_delivery.py | 14 +- tests/test_store.py | 30 +++ tests/test_tmux_integration.py | 4 +- tests/test_tmux_manager.py | 33 ++- ...-0005-terminal-neutral-manual-messaging.md | 38 +++- 20 files changed, 608 insertions(+), 182 deletions(-) create mode 100644 tests/test_manual_messaging.py diff --git a/.claude/rules/stack-and-commands.md b/.claude/rules/stack-and-commands.md index 1a20fba..bb3a36c 100644 --- a/.claude/rules/stack-and-commands.md +++ b/.claude/rules/stack-and-commands.md @@ -14,8 +14,11 @@ uv sync --extra dev # Install/upgrade the user command make install -# Start the operator alpha session -tamq start --command codex railiance-platform activity-core +# Start neutral repository shells +tamq flex-auth audit-core + +# Run an explicit initial command only when requested +tamq --command codex flex-auth audit-core # Run tests uv run pytest diff --git a/AGENTS.md b/AGENTS.md index 6b34e06..dad00fe 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,7 +2,7 @@ ## Repo Identity -**Purpose:** Local durable message queue and tmux endpoint runtime for coordinating agent workers across gita-registered repositories. +**Purpose:** Repository-aware tmux sessions and durable local messaging without assuming what runs in a pane. **Domain:** communication **Repo slug:** tmux-amq @@ -267,9 +267,10 @@ To create a new workplan: - Install development dependencies with `uv sync --extra dev`. - Install or upgrade the user command with `make install`. -- The operator alpha acceptance command is - `tamq start --command codex railiance-platform activity-core`; it must open - both exact gita paths and launch Codex once in each new window. +- The terminal-neutral acceptance command is `tamq flex-auth audit-core`; it + must open untouched shells in both exact gita paths and inject no pane input. +- Explicit `--command` runs exactly the requested initial command. Manual + `tamq send`/`tamq inbox` exchange must remain durable and non-invasive. - Run the test suite with `uv run pytest` or `make test`. - Before handoff, run `make check`; it combines tests, whitespace validation, and Python bytecode compilation. diff --git a/INTENT.md b/INTENT.md index c09ed78..d2c1c30 100644 --- a/INTENT.md +++ b/INTENT.md @@ -7,7 +7,10 @@ updated: "2026-08-24" ## Why it exists -Provide a local, durable, inspectable transport for direct messages between agent workers running in tmux, with strict repository targeting, explicit endpoint lifecycle, bounded retry behavior, and future interoperability with coordination-engine. +Provide repository-aware tmux sessions and a local, durable, inspectable +transport for direct messages between their users or processes, with strict +targeting, explicit endpoint lifecycle, safe delivery boundaries, and future +interoperability with coordination-engine. ## Governing principle @@ -17,7 +20,8 @@ workplan before implementation. ## What it enables -- Agent workers can address one another by registered repository slug. +- Terminal users and processes can address one another by registered repository + slug without tamq assuming what runs in a pane. - Messages, leases, delivery state, and acknowledgements survive process restarts. - Operators can inspect, export, replay, and purge local queue history. - A later coordination-engine adapter can use a stable local socket boundary diff --git a/README.md b/README.md index 870e486..c553e09 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # tmux-amq -Tmux Agentic Message Queueing (`tamq`) is a local, durable message queue for -agent workers running in gita-registered repositories. +Tmux Agentic Message Queueing (`tamq`) provides repository-aware tmux sessions +and a local, durable message queue. It does not assume who or what uses a pane. The local alpha provides tmux endpoint lifecycle, local SQLite history and leases, direct `@repo:` routing, JSONL export/replay, and a Unix-socket protocol @@ -10,8 +10,7 @@ for a later coordination-engine adapter. ## Install and start a local session Prerequisites must be available on `PATH`: Python 3.11+, [uv](https://docs.astral.sh/uv/), -tmux, gita, and the agent command you want to run (`codex` by default). Each -repository must already have an exact gita slug and path. +tmux, and gita. Each repository must already have an exact gita slug and path. Install or upgrade `tamq` as an isolated user tool: @@ -21,47 +20,74 @@ command -v tamq tamq --version ``` -Start an attached two-repository Codex session: +Start an attached two-repository shell session: ```bash -tamq start --command codex railiance-platform activity-core +tamq flex-auth audit-core ``` This creates or reuses the managed `tamq` session, opens same-named windows in -the exact gita paths for `railiance-platform` and `activity-core`, launches Codex -behind `tamq tap` in each new window, and attaches to the first window. Codex is -the default, so this is equivalent: +the exact gita paths, leaves tmux's ordinary interactive shells untouched, and +attaches to the first window. The explicit form is equivalent: ```bash -tamq start railiance-platform activity-core +tamq start flex-auth audit-core ``` Set up the session without attaching, then attach later: ```bash -tamq start --detach --command codex railiance-platform activity-core +tamq start --detach flex-auth audit-core tamq status tamq attach ``` -Repeated starts reuse existing windows and do not launch a second agent in -them. `--cmd` remains an alias for `--command`. A quoted command may include -arguments, for example `--command 'codex --profile local'`. +No initial command runs unless requested. To run one explicitly in each newly +created window, use `--command`; tamq runs exactly that command and makes no +assumption about its purpose: -An already-running pane keeps the `tamq tap` code that launched it. After -upgrading from an earlier alpha, recreate the managed session once so terminal -mode and resize fixes take effect: +```bash +tamq --command codex flex-auth audit-core +tamq start --command 'htop --tree' flex-auth audit-core +``` + +`--cmd` remains an alias for `--command`. Repeated starts reuse existing +windows and never run another initial command in them. + +## Exchange messages manually + +Each managed shell exports its own repository slug as `TAMQ_REPO`. From the +`flex-auth` window, queue a message: + +```bash +tamq send '@audit-core: please review the auth boundary' +``` + +In the `audit-core` window, inspect and acknowledge it: + +```bash +tamq inbox +tamq ack +``` + +Outside a managed window, use `tamq inbox --repo audit-core` and optionally +`--json`. Manual messages remain durable and pending until acknowledged. They +are never injected as terminal keystrokes, so they cannot corrupt a command +being typed in the target pane. + +After upgrading from the earlier Codex-default alpha, recreate the managed +session once so existing panes are replaced by neutral shells and the broker is +re-registered in manual delivery mode: ```bash tamq stop tmux kill-session -t tamq -tamq start --command codex railiance-platform activity-core +tamq flex-auth audit-core ``` -For degraded tmux-only use, `tamq start --no-service ...` launches the selected -agent directly, without `tamq tap`, socket endpoint registration, or message -delivery. Stop the broker and remove the managed tmux session explicitly when -finished: +For degraded tmux-only use, `tamq start --no-service ...` skips socket endpoint +registration and durable messaging. Stop the broker and remove the managed tmux +session explicitly when finished: ```bash tamq stop @@ -77,16 +103,14 @@ anywhere with `uv tool uninstall tmux-amq`. uv run pytest uv run tamq --help uv run tamq --version -uv run tamq start --cmd codex net-kingdom railiance-platform +uv run tamq start net-kingdom railiance-platform uv run tamq attach ``` -The alpha 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. +The alpha includes the local SQLite queue, neutral tmux endpoint manager, strict +gita target validation, manual inbox, control-mode client, and an explicit PTY +tap for integration experiments. Coordination-engine integration remains a +later phase. ## Terminal architecture @@ -94,16 +118,17 @@ fallback. coordination-engine integration remains a later phase. 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 +3. Explicit `tamq tap` mode is a full-duplex PTY proxy around a command. It forwards bytes unchanged in raw terminal mode, propagates terminal resize and lifecycle signals, and observes 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. +Neutral endpoints use manual delivery: messages stay in the durable inbox and +never become pane input. The legacy pane-delivery path is available only with +the explicit `tamq start --tap --command ...` opt-in; it remains subject to the +retry and acknowledgement limitations tracked by `TAMQ-WP-0003`. The visible endpoint label is `tmux-amq-`; each boot also receives an instance nonce so PID reuse cannot collide with prior leases or receipts. diff --git a/SCOPE.md b/SCOPE.md index 09aec29..2cc0ada 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -2,20 +2,23 @@ ## One-liner -Local durable message queue and tmux endpoint runtime for coordinating agent workers across gita-registered repositories. +Repository-aware tmux sessions and durable local messaging for users or +processes working across gita-registered repositories. ## Core Idea -Keep tmux topology, terminal input observation, and durable local message state -behind a small CLI and Unix-socket protocol that agents can use without owning -a distributed coordination service. +Keep neutral tmux topology and durable local message state behind a small CLI +and Unix-socket protocol. Pane occupants may be humans, shells, tools, or agents; +tamq does not choose or infer them. ## In Scope - Local SQLite message history, leases, endpoint registrations, delivery state, acknowledgements, replay, export, and bounded purging. -- Managed tmux endpoint lifecycle, control-mode output/injection, and the - full-duplex `tamq tap` PTY broker. +- Managed neutral-shell tmux lifecycle and explicit initial commands. +- Durable manual send/inbox/acknowledgement with per-window repository identity. +- Explicit opt-in control-mode pane delivery and the full-duplex `tamq tap` PTY + broker for integration experiments. - Exact `gita` repository validation and direct `@repo: message` routing. - Unix-socket operations for local clients and a future coordination-engine adapter. @@ -27,6 +30,8 @@ a distributed coordination service. - Owning goal planning, workflow scheduling, or cross-host coordination; those belong to `coordination-engine` and its consumers. - Acting as a network-accessible or multi-host message broker. +- Selecting a coding agent, starting one implicitly, or assuming a pane accepts + machine-generated terminal input. - Bypassing `gita` registration or injecting arbitrary pane input outside the supported tap/control-mode boundaries. - Owning tmux, Codex, State Hub, or adjacent repositories' lifecycle. @@ -40,41 +45,42 @@ transport. | Intent capability | State | Evidence and remaining gap | | --- | --- | --- | | Direct repository addressing | Implemented | Exact `gita` validation and `@repo:` parsing are covered by tests. | -| Durable, inspectable local queue | Implemented | SQLite history, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, and purge are present. | +| Durable, inspectable local queue | Implemented | SQLite history, manual inbox, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, and purge are present. | | Local socket service | Implemented | Peer-credential checks and structured ping/register/send/history/ack/endpoints/disconnect operations are tested. | -| Tmux endpoint lifecycle and delivery | Implemented for local alpha | Real isolated-tmux and installed-package tests prove exact repository windows, stable reuse, control-mode injection, durable history transition, service stop, and bounded cleanup. | -| Full-duplex input observation | Implemented for local alpha | `tamq tap` uses raw outer-terminal mode, copies and propagates window dimensions, forwards lifecycle signals, preserves mouse/control bytes, and observes CR/LF address lines. A real tmux fixture covers geometry, resize, and raw mouse input. | +| Neutral tmux session lifecycle | Implemented for local alpha | Repository-first startup opens untouched shells at exact gita paths, exports per-window identity, and runs no initial command unless `--command` is explicit. Stable reuse, service restart, and cleanup are covered by the installed-package test. | +| Safe manual messaging | Implemented for local alpha | Manual endpoints never inject pending messages into panes. Installed testing proves send/inbox/ack exchange while the target pane remains unchanged; legacy rows migrate to manual mode. | +| Full-duplex input observation | Explicit opt-in | `--tap --command ...` enables the PTY integration path. It is absent from neutral startup and remains covered for geometry, resize, raw mouse input, and lifecycle behavior. | | Bounded retry behavior | Not enforced | Failed injection remains pending and becomes claimable after lease expiry, but no attempt counter or terminal failure state applies the configured cap. | | Acknowledgement policy | Partially implemented | Explicit acknowledgement and the configuration field exist; delivery always marks a successful tmux injection as `injected`, irrespective of `delivery_ack_mode`. | | Coordination-engine interoperability | Not implemented | The adapter contract and implementation remain in `TAMQ-WP-0002`. | ## Practical Usability -The installable alpha path was exercised successfully on 2026-08-24 with both -isolated fixtures and the operator's registered repositories. `make install` -produced a user-level `tamq` command; `tamq start --detach --command codex -railiance-platform activity-core` created two same-named live panes rooted at -their exact gita paths. Repeating the command retained the same session instance -and pane PIDs. The automated installed-package path also proved message -injection, history, status, service stop, and cleanup. +The terminal-neutral alpha path was exercised successfully on 2026-08-24 with +an isolated installed tool. Repository-first startup created two ordinary +shells at exact gita paths without sending initial keystrokes. The test proved +per-window repository identity, stable reuse, manual send/inbox/ack exchange, +zero target-pane mutation, service restart, endpoint disappearance, explicit +initial-command startup, and cleanup. Suitable today: - Local queue, history, export/replay, and diagnostic use. -- Interactive local Codex sessions over one or more gita-registered repositories. -- Controlled message-routing experiments between managed tmux endpoints. +- Interactive local shell or explicitly commanded sessions over one or more + gita-registered repositories. +- Durable manual message exchange between managed repository windows. - Developing and testing the future coordination-engine adapter against the local socket boundary. Not yet suitable: -- Unattended or high-confidence delivery where bounded retries and positive - recipient acknowledgement are required. +- Unattended pane injection where bounded retries and positive recipient + acknowledgement are required. - Production-style operation without longer-running crash/terminal soak tests and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 69 passing tests and 72% statement coverage. Coverage +The suite currently has 77 passing tests and 74% statement coverage. Coverage is strongest in durable storage and registry handling, and weakest in the PTY tap and CLI orchestration; PTY statement coverage increased from 23% to 33%, while subprocess behavior is primarily proven by the real-tmux test. The diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 3fe18d8..cb1664e 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -13,7 +13,7 @@ | workplan | TAMQ-WP-0002 | active | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | | workplan | TAMQ-WP-0003 | active | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | workplan | TAMQ-WP-0004 | finished | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | -| workplan | TAMQ-WP-0005 | active | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | +| workplan | TAMQ-WP-0005 | finished | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | | task | TAMQ-WP-ADHOC-2026-08-24-T01 | done | — | workplans/ADHOC-2026-08-24.md | | task | TAMQ-WP-0001-T01 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | | task | TAMQ-WP-0001-T02 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | @@ -30,7 +30,7 @@ | task | TAMQ-WP-0004-T03 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | task | TAMQ-WP-0004-T04 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | task | TAMQ-WP-0004-T05 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | -| task | TAMQ-WP-0005-T01 | progress | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | -| task | TAMQ-WP-0005-T02 | todo | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | -| task | TAMQ-WP-0005-T03 | todo | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | -| task | TAMQ-WP-0005-T04 | wait | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | +| task | TAMQ-WP-0005-T01 | done | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | +| task | TAMQ-WP-0005-T02 | done | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | +| task | TAMQ-WP-0005-T03 | done | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | +| task | TAMQ-WP-0005-T04 | done | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | diff --git a/src/tamq/cli.py b/src/tamq/cli.py index 8476937..701488e 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -25,6 +25,14 @@ from .ptytap import PtyTap from .control import ControlModeClient from .diagnostics import configure from .policy import load_profile +from .registry import RegistryError, validate_targets + + +SUBCOMMANDS = frozenset( + "start attach serve stop status ping inbox history inspect ack send export replay purge completion db-version config tap".split() +) +START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service"}) +GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"}) def parse_size(value: str) -> int: @@ -70,6 +78,46 @@ def ensure_service() -> bool: return False +def stop_service_process() -> int | None: + try: + pid = int(pid_path().read_text(encoding="utf-8")) + except (FileNotFoundError, ValueError): + return None + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + pid_path().unlink(missing_ok=True) + return None + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + try: + os.kill(pid, 0) + except ProcessLookupError: + break + time.sleep(0.05) + return pid + + +def ensure_manual_service() -> bool: + """Ensure the broker understands neutral endpoints before registration.""" + if not ensure_service(): + return False + try: + capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", []) + except (OSError, json.JSONDecodeError): + return False + if "manual_delivery" in capabilities: + return True + stop_service_process() + if not ensure_service(): + return False + try: + capabilities = asyncio.run(request({"op": "ping"})).get("capabilities", []) + except (OSError, json.JSONDecodeError): + return False + return "manual_delivery" in capabilities + + def preflight_runtime_paths() -> None: """Ensure local state/runtime parents are available before tmux mutation.""" parents = {db_path().parent, socket_path().parent, pid_path().parent, lock_path().parent} @@ -87,7 +135,7 @@ def attach_session(session: str) -> int: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="tamq", - description="Tmux Agentic Message Queueing for local repository workers.", + description="Repository-aware tmux sessions with durable local messaging.", ) parser.add_argument("--version", "-V", action="version", version=__version__) parser.add_argument("--orwell", action="store_true", help="enable unsafe local diagnostics") @@ -97,7 +145,8 @@ def build_parser() -> argparse.ArgumentParser: 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("--command", "--cmd", dest="agent_command", default="codex", help="agent command for new windows (default: codex)") + start.add_argument("--command", "--cmd", dest="initial_command", default=None, help="explicit initial command for newly created windows (default: ordinary shell)") + start.add_argument("--tap", action="store_true", help="explicitly opt in to PTY input observation and pane message injection") start.add_argument("--detach", action="store_true", help="detach after startup") start.add_argument("--no-service", action="store_true", help="open tmux windows without service registration or messaging") attach = subparsers.add_parser("attach", help="attach to the managed tmux session") @@ -110,14 +159,19 @@ def build_parser() -> argparse.ArgumentParser: history = subparsers.add_parser("history", help="inspect local message history") history.add_argument("--repo", dest="target_repo") history.add_argument("--state") + inbox = subparsers.add_parser("inbox", help="show durable messages for a repository without injecting terminal input") + inbox.add_argument("--repo", dest="target_repo", help="target repository (default: TAMQ_REPO in a managed window)") + inbox.add_argument("--all", action="store_true", help="include non-pending messages") + inbox.add_argument("--json", action="store_true", help="emit one JSON object per message") 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 = subparsers.add_parser("ack", help="acknowledge one durable 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") + send.add_argument("--from", dest="sender_repo", help="sender repository (default: TAMQ_REPO or local)") export = subparsers.add_parser("export", help="export history as JSONL") export.add_argument("--output", required=True) export.add_argument("--repo", dest="target_repo") @@ -134,16 +188,41 @@ def build_parser() -> argparse.ArgumentParser: 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 = subparsers.add_parser("tap", help="explicitly run a command 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 --") + tap.add_argument("wrapped_command", nargs=argparse.REMAINDER, help="command after --") return parser +def normalize_argv(argv: list[str]) -> list[str]: + """Allow repository-first startup while retaining explicit subcommands.""" + if not argv: + return argv + index = 0 + while index < len(argv): + token = argv[index] + if token in GLOBAL_FLAGS or token.startswith("--policy-profile="): + index += 1 + continue + if token == "--policy-profile" and index + 1 < len(argv): + index += 2 + continue + break + if index == len(argv): + return argv + candidate = argv[index] + if candidate in START_OPTIONS or ( + not candidate.startswith("-") and candidate not in SUBCOMMANDS + ): + return [*argv[:index], "start", *argv[index:]] + return argv + + def main(argv: list[str] | None = None) -> int: parser = build_parser() - args = parser.parse_args(argv) + raw_argv = list(sys.argv[1:] if argv is None else argv) + args = parser.parse_args(normalize_argv(raw_argv)) configure(args.orwell, args.verbose) try: profile = load_profile(selected=args.policy_profile or os.environ.get("TAMQ_POLICY_PROFILE", "default")) @@ -158,25 +237,12 @@ def main(argv: list[str] | None = None) -> int: 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) + pid = stop_service_process() + if pid is not None: 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 + print("tamq service is not running", file=sys.stderr) + return 1 if args.command == "serve": try: asyncio.run(Service().run()) except KeyboardInterrupt: return 0 @@ -190,11 +256,11 @@ def main(argv: list[str] | None = None) -> int: 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) + command = list(args.wrapped_command) if command and command[0] == "--": command = command[1:] if not command: - print("tamq tap requires an agent command after --", file=sys.stderr) + print("tamq tap requires a command after --", file=sys.stderr) return 2 store = Store(db_path()) try: @@ -202,19 +268,25 @@ def main(argv: list[str] | None = None) -> int: finally: store.close() if args.command == "start": + if args.tap and args.initial_command is None: + print("tamq: --tap requires an explicit --command", file=sys.stderr) + return 2 + if args.tap and args.no_service: + print("tamq: --tap cannot be combined with --no-service", file=sys.stderr) + return 2 manager = TmuxManager() try: if not args.no_service: preflight_runtime_paths() - launch_plan = manager.preflight(args.repos, args.agent_command) + launch_plan = manager.preflight(args.repos, args.initial_command) except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) return 2 - 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) + if not args.no_service and not ensure_manual_service(): + print("tamq service failed to start with manual-delivery safety; use --no-service to open repos without messaging", file=sys.stderr) return 1 try: - endpoint = manager.ensure_plan(launch_plan, tap=not args.no_service) + endpoint = manager.ensure_plan(launch_plan, tap=args.tap) except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) return 2 @@ -222,20 +294,22 @@ def main(argv: list[str] | None = None) -> int: registered = False if not args.no_service: try: - registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos})) + delivery_mode = "pane" if args.tap else "manual" + registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos, "delivery_mode": delivery_mode})) if not registration.get("ok"): raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}") registered = True - 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() + if args.tap: + 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() except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as exc: manager.rollback(endpoint) print(f"tamq: {exc}", file=sys.stderr) @@ -248,6 +322,7 @@ def main(argv: list[str] | None = None) -> int: "delivered": delivered, "service": not args.no_service, "messaging": registered, + "delivery_mode": "none" if args.no_service else ("pane" if args.tap else "manual"), } print(json.dumps(summary), flush=True) if args.detach: @@ -256,7 +331,7 @@ def main(argv: list[str] | None = None) -> int: store = Store(db_path()) try: advisory = history_advisory(store) - if advisory and args.command in {"start", "serve", "status", "history", "send"}: + if advisory and args.command in {"start", "serve", "status", "history", "inbox", "send"}: print(advisory, file=sys.stderr) if args.command == "send": text = " ".join([args.address, *args.body]).strip() @@ -264,18 +339,44 @@ def main(argv: list[str] | None = None) -> int: 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})) + sender = args.sender_repo or os.environ.get("TAMQ_REPO") or "local" + try: + validate_targets([target]) + if sender != "local": + validate_targets([sender]) + except RegistryError as exc: + print(f"tamq: {exc}", file=sys.stderr); return 2 + if asyncio.run(ping()): + payload = {"op": "send", "sender_repo": sender, "target_repo": target, "body": body} + if args.endpoint_id: + payload["endpoint_id"] = args.endpoint_id + response = asyncio.run(request(payload)) if not response.get("ok"): print(f"tamq: {response.get('error', 'send failed')}", file=sys.stderr); return 1 print(response["message_id"]); return 0 + if args.endpoint_id: + print("tamq: service is not running", file=sys.stderr); return 1 try: - print(store.add("local", target, body)) + print(store.add(sender, target, body)) except ValueError as exc: print(f"tamq: {exc}", file=sys.stderr); return 2 return 0 + if args.command == "inbox": + target = args.target_repo or os.environ.get("TAMQ_REPO") + if not target: + print("tamq inbox requires --repo outside a managed tamq window", file=sys.stderr) + return 2 + try: + validate_targets([target]) + except RegistryError as exc: + print(f"tamq: {exc}", file=sys.stderr); return 2 + rows = store.list(target, None if args.all else "pending") + for row in rows: + if args.json: + print(json.dumps(dict(row), sort_keys=True)) + else: + print(f"{row['message_id']} {row['sender_repo']} -> {row['target_repo']}: {row['body']}") + 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 @@ -332,7 +433,7 @@ def main(argv: list[str] | None = None) -> int: 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" + commands = " ".join(sorted(SUBCOMMANDS)) if shell == "bash": return f"""_tamq_complete() {{ local commands=\"{commands}\" diff --git a/src/tamq/service.py b/src/tamq/service.py index bb3e339..28afcf3 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -78,6 +78,8 @@ class Service: if not control.session_exists(expected_pid=int(endpoint["pid"])): self.store.disconnect_endpoint(endpoint["endpoint_id"]) continue + if endpoint["delivery_mode"] != "pane": + continue pending = [row for row in self.store.list(state="pending") if row["endpoint_id"] in (None, endpoint["endpoint_id"]) and row["target_repo"] in repos] if not pending: continue @@ -107,7 +109,7 @@ class Service: if protocol.split(".")[0] != PROTOCOL_VERSION.split(".")[0]: response = {"ok": False, "error": f"incompatible protocol: {protocol}", "protocol": PROTOCOL_VERSION} elif op == "ping": - response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history"]} + response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery"]} elif op == "register": required = ("endpoint_id", "pid", "session", "repos") if any(key not in request for key in required): @@ -116,7 +118,14 @@ class Service: 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"])) + delivery_mode = request.get("delivery_mode", "manual") + self.store.register_endpoint( + endpoint_id, + int(request["pid"]), + request["session"], + list(request["repos"]), + delivery_mode, + ) 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)} @@ -127,6 +136,8 @@ class Service: else: try: validate_targets([request["target_repo"]]) + if request["sender_repo"] != "local": + validate_targets([request["sender_repo"]]) endpoint_id = request.get("endpoint_id") if endpoint_id: endpoint = self.store.endpoint(endpoint_id) diff --git a/src/tamq/store.py b/src/tamq/store.py index 4899743..f3ac634 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -7,7 +7,7 @@ from pathlib import Path from typing import Iterable from uuid import uuid4 -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 class Store: @@ -38,6 +38,7 @@ class Store: pid INTEGER NOT NULL, session TEXT NOT NULL, repos TEXT NOT NULL, + delivery_mode TEXT NOT NULL DEFAULT 'manual', connected_at REAL NOT NULL, disconnected_at REAL ); @@ -49,23 +50,43 @@ class Store: expires_at REAL NOT NULL ); """) - self.db.execute("INSERT OR IGNORE INTO metadata(key,value) VALUES('schema_version',?)", (str(SCHEMA_VERSION),)) + endpoint_columns = { + row["name"] for row in self.db.execute("PRAGMA table_info(endpoints)") + } + if "delivery_mode" not in endpoint_columns: + self.db.execute( + "ALTER TABLE endpoints ADD COLUMN delivery_mode TEXT NOT NULL DEFAULT 'manual'" + ) + self.db.execute( + "INSERT INTO metadata(key,value) VALUES('schema_version',?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (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: + def register_endpoint( + self, + endpoint_id: str, + pid: int, + session: str, + repos: list[str], + delivery_mode: str = "manual", + ) -> None: import json + if delivery_mode not in {"manual", "pane"}: + raise ValueError(f"invalid delivery mode: {delivery_mode}") self.db.execute( "UPDATE endpoints SET disconnected_at=strftime('%s','now') " "WHERE pid=? AND session=? AND endpoint_id<>? AND disconnected_at IS NULL", (pid, session, endpoint_id), ) 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)), + "INSERT INTO endpoints(endpoint_id,pid,session,repos,delivery_mode,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, delivery_mode=excluded.delivery_mode, connected_at=excluded.connected_at, disconnected_at=NULL", + (endpoint_id, pid, session, json.dumps(repos), delivery_mode), ) self.db.commit() diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py index 6987ada..344fa03 100644 --- a/src/tamq/tmux.py +++ b/src/tamq/tmux.py @@ -84,21 +84,23 @@ class TmuxManager: ) return pid, instance_id - def preflight(self, repos: list[str], command: str = "codex") -> LaunchPlan: + def preflight(self, repos: list[str], command: str | None = None) -> LaunchPlan: if not repos: raise TmuxError("at least one repository is required") if not self.tmux_command or shutil.which(self.tmux_command[0]) is None: raise TmuxError("tmux is required; install it and ensure 'tmux' is on PATH") if shutil.which("gita") is None: raise TmuxError("gita is required; install it and ensure 'gita' is on PATH") - try: - command_parts = tuple(shlex.split(command)) - except ValueError as exc: - raise TmuxError(f"invalid agent command: {exc}") from exc - if not command_parts: - raise TmuxError("agent command must not be empty") - if shutil.which(command_parts[0]) is None: - raise TmuxError(f"agent command not found on PATH: {command_parts[0]}") + command_parts: tuple[str, ...] = () + if command is not None: + try: + command_parts = tuple(shlex.split(command)) + except ValueError as exc: + raise TmuxError(f"invalid initial command: {exc}") from exc + if not command_parts: + raise TmuxError("initial command must not be empty") + if shutil.which(command_parts[0]) is None: + raise TmuxError(f"initial command not found on PATH: {command_parts[0]}") unique_repos = tuple(dict.fromkeys(repos)) try: validate_targets(list(unique_repos)) @@ -117,10 +119,12 @@ class TmuxManager: self._existing_session_identity() return LaunchPlan(unique_repos, selected_paths, command_parts) - def ensure(self, repos: list[str], command: str = "codex") -> Endpoint: + def ensure(self, repos: list[str], command: str | None = None) -> Endpoint: return self.ensure_plan(self.preflight(repos, command)) - def ensure_plan(self, plan: LaunchPlan, *, tap: bool = True) -> Endpoint: + def ensure_plan(self, plan: LaunchPlan, *, tap: bool = False) -> Endpoint: + if tap and not plan.command: + raise TmuxError("PTY tap requires an explicit initial command") created_session = False created_windows: list[str] = [] try: @@ -129,6 +133,7 @@ class TmuxManager: first_repo = plan.repos[0] self._run( "new-session", "-d", "-s", self.session, "-n", "__tamq_boot", + "-e", f"TAMQ_REPO={first_repo}", "-c", plan.paths[first_repo], ) created_session = True @@ -155,18 +160,22 @@ class TmuxManager: self._run("rename-window", "-t", f"{self.session}:__tamq_boot", repo) windows[windows.index("__tamq_boot")] = repo else: - self._run("new-window", "-d", "-t", self.session, "-n", repo, "-c", plan.paths[repo]) + self._run( + "new-window", "-d", "-t", self.session, "-n", repo, + "-e", f"TAMQ_REPO={repo}", "-c", plan.paths[repo], + ) windows.append(repo) created_windows.append(repo) - command = ( - [ - *self.tamq_command, - "tap", "--repo", repo, "--endpoint", endpoint_id, "--", *plan.command, - ] - if tap - else list(plan.command) - ) - self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(command), "C-m") + if plan.command: + command = ( + [ + *self.tamq_command, + "tap", "--repo", repo, "--endpoint", endpoint_id, "--", *plan.command, + ] + if tap + else list(plan.command) + ) + self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(command), "C-m") self._run("select-window", "-t", f"{self.session}:{plan.repos[0]}") return Endpoint( self.session, diff --git a/tests/test_cli.py b/tests/test_cli.py index 9b9ff4c..911b7ae 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,6 +1,6 @@ import json -from tamq.cli import build_parser, main +from tamq.cli import build_parser, main, normalize_argv import pytest @@ -33,9 +33,32 @@ def test_start_parser_supports_command_and_cmd_alias(): parser = build_parser() canonical = parser.parse_args(["start", "--command", "codex --quiet", "a", "b"]) compatibility = parser.parse_args(["start", "--cmd", "claude", "a"]) - assert canonical.agent_command == "codex --quiet" + neutral = parser.parse_args(["start", "a", "b"]) + assert canonical.initial_command == "codex --quiet" assert canonical.repos == ["a", "b"] - assert compatibility.agent_command == "claude" + assert compatibility.initial_command == "claude" + assert neutral.initial_command is None + + +def test_repository_first_and_command_first_start_shorthand(): + assert normalize_argv(["flex-auth", "audit-core"]) == [ + "start", + "flex-auth", + "audit-core", + ] + assert normalize_argv(["--command", "codex", "flex-auth"]) == [ + "start", + "--command", + "codex", + "flex-auth", + ] + assert normalize_argv(["status"]) == ["status"] + assert normalize_argv(["--verbose", "flex-auth", "audit-core"]) == [ + "--verbose", + "start", + "flex-auth", + "audit-core", + ] def test_detached_no_service_start_skips_socket_registration(monkeypatch, capsys): @@ -54,7 +77,7 @@ def test_detached_no_service_start_skips_socket_registration(monkeypatch, capsys class Manager: def preflight(self, repos, command): assert repos == ["a", "b"] - assert command == "codex" + assert command is None return "plan" def ensure_plan(self, plan, *, tap=True): @@ -121,7 +144,7 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys): return "plan" def ensure_plan(self, plan, *, tap=True): - assert tap is True + assert tap is False return endpoint def rollback(self, value): @@ -132,8 +155,15 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys): monkeypatch.setattr("tamq.cli.TmuxManager", Manager) monkeypatch.setattr("tamq.cli.preflight_runtime_paths", lambda: None) - monkeypatch.setattr("tamq.cli.ensure_service", lambda: True) + monkeypatch.setattr("tamq.cli.ensure_manual_service", lambda: True) monkeypatch.setattr("tamq.cli.request", failed_registration) assert main(["start", "--detach", "a"]) == 1 assert rolled_back == [endpoint] assert "endpoint registration failed: denied" in capsys.readouterr().err + + +def test_tap_requires_explicit_command_and_service(capsys): + assert main(["start", "--tap", "a"]) == 2 + assert "--tap requires an explicit --command" in capsys.readouterr().err + assert main(["start", "--tap", "--no-service", "--command", "sh", "a"]) == 2 + assert "--tap cannot be combined with --no-service" in capsys.readouterr().err diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index 8c4f54c..46f0337 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -1,6 +1,24 @@ -from tamq.cli import parse_size +from tamq.cli import ensure_manual_service, parse_size def test_parse_size(): assert parse_size("100MB") == 100_000_000 assert parse_size("2KB") == 2_000 + + +def test_manual_service_restarts_legacy_broker(monkeypatch): + capabilities = iter([[], ["manual_delivery"]]) + starts = [] + stops = [] + + async def request(payload): + assert payload == {"op": "ping"} + return {"ok": True, "capabilities": next(capabilities)} + + monkeypatch.setattr("tamq.cli.ensure_service", lambda: starts.append(True) or True) + monkeypatch.setattr("tamq.cli.stop_service_process", lambda: stops.append(True) or 42) + monkeypatch.setattr("tamq.cli.request", request) + + assert ensure_manual_service() is True + assert len(starts) == 2 + assert stops == [True] diff --git a/tests/test_install_target.py b/tests/test_install_target.py index 801f89f..3b4994c 100644 --- a/tests/test_install_target.py +++ b/tests/test_install_target.py @@ -94,34 +94,59 @@ def test_isolated_installed_tool_session_smoke(tmp_path): started = json.loads( run( str(tamq), - "start", - "--detach", - "--command", - "alpha-agent", "railiance-platform", "activity-core", + "--detach", ).stdout ) assert started["repos"] == ["railiance-platform", "activity-core"] + assert started["delivery_mode"] == "manual" rows = run( *tmux, "list-windows", "-t", "tamq", "-F", - "#{window_name}|#{pane_current_path}|#{pane_pid}", + "#{window_name}|#{pane_current_path}|#{pane_pid}|#{pane_current_command}", ).stdout.splitlines() - parsed = {name: (path, pid) for name, path, pid in (row.split("|", 2) for row in rows)} + parsed = { + name: (path, pid, command) + for name, path, pid, command in (row.split("|", 3) for row in rows) + } assert parsed["railiance-platform"][0] == str(repo_a) assert parsed["activity-core"][0] == str(repo_b) + assert all(command != "alpha-agent" for _, _, command in parsed.values()) + assert all( + "installed-alpha-ready" + not in run(*tmux, "capture-pane", "-p", "-t", f"tamq:{repo}").stdout + for repo in ("railiance-platform", "activity-core") + ) + + for repo in ("railiance-platform", "activity-core"): + run( + *tmux, + "send-keys", + "-t", + f"tamq:{repo}", + "printf 'TAMQ_REPO=%s\\n' \"$TAMQ_REPO\"", + "C-m", + ) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + identity_captures = { + repo: run(*tmux, "capture-pane", "-p", "-t", f"tamq:{repo}").stdout + for repo in ("railiance-platform", "activity-core") + } + if all(f"TAMQ_REPO={repo}" in output for repo, output in identity_captures.items()): + break + time.sleep(0.05) + assert all(f"TAMQ_REPO={repo}" in output for repo, output in identity_captures.items()) repeated = json.loads( run( str(tamq), "start", "--detach", - "--command", - "alpha-agent", "railiance-platform", "activity-core", ).stdout @@ -133,7 +158,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): "-t", "tamq", "-F", - "#{window_name}|#{pane_current_path}|#{pane_pid}", + "#{window_name}|#{pane_current_path}|#{pane_pid}|#{pane_current_command}", ).stdout.splitlines() assert repeated_rows == rows pre_delivery_status = json.loads(run(str(tamq), "status").stdout) @@ -141,22 +166,49 @@ def test_isolated_installed_tool_session_smoke(tmp_path): assert [item["endpoint_id"] for item in pre_delivery_status["endpoints"]] == [ started["instance_id"] ] + assert pre_delivery_status["endpoints"][0]["delivery_mode"] == "manual" - message_id = run( - str(tamq), "send", "@activity-core:", "installed-message" - ).stdout.strip() + target_before = run( + *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" + ).stdout + run( + *tmux, + "send-keys", + "-t", + "tamq:railiance-platform", + f"{tamq} send '@activity-core: installed-message'", + "C-m", + ) deadline = time.monotonic() + 5 - history = [] + inbox = [] while time.monotonic() < deadline: - history = [ + inbox = [ json.loads(line) - for line in run(str(tamq), "history", "--repo", "activity-core").stdout.splitlines() + for line in run( + str(tamq), "inbox", "--repo", "activity-core", "--json" + ).stdout.splitlines() ] - if history and history[-1]["state"] == "injected": + if inbox: break time.sleep(0.05) - assert history[-1]["message_id"] == message_id - assert history[-1]["state"] == "injected" + assert inbox[-1]["sender_repo"] == "railiance-platform" + assert inbox[-1]["body"] == "installed-message" + assert inbox[-1]["state"] == "pending" + message_id = inbox[-1]["message_id"] + target_after = run( + *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" + ).stdout + assert target_after == target_before + + assert run(str(tamq), "ack", message_id).stdout.strip() == message_id + assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == "" + all_messages = [ + json.loads(line) + for line in run( + str(tamq), "inbox", "--repo", "activity-core", "--all", "--json" + ).stdout.splitlines() + ] + assert all_messages[-1]["state"] == "acknowledged" status = json.loads(run(str(tamq), "status").stdout) assert status["service"] is True assert [item["endpoint_id"] for item in status["endpoints"]] == [started["instance_id"]] @@ -168,8 +220,6 @@ def test_isolated_installed_tool_session_smoke(tmp_path): str(tamq), "start", "--detach", - "--command", - "alpha-agent", "railiance-platform", "activity-core", ).stdout @@ -181,7 +231,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): "-t", "tamq", "-F", - "#{window_name}|#{pane_current_path}|#{pane_pid}", + "#{window_name}|#{pane_current_path}|#{pane_pid}|#{pane_current_command}", ).stdout.splitlines() == rows run(*tmux, "kill-session", "-t", "tamq") @@ -205,6 +255,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ).stdout ) assert recovered["instance_id"] != started["instance_id"] + assert recovered["delivery_mode"] == "manual" recovered_rows = run( *tmux, "list-windows", @@ -217,6 +268,16 @@ def test_isolated_installed_tool_session_smoke(tmp_path): f"railiance-platform|{repo_a}", f"activity-core|{repo_b}", ] + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + captures = [ + run(*tmux, "capture-pane", "-p", "-t", f"tamq:{repo}").stdout + for repo in ("railiance-platform", "activity-core") + ] + if all("installed-alpha-ready" in output for output in captures): + break + time.sleep(0.05) + assert all("installed-alpha-ready" in output for output in captures) assert run(str(tamq), "stop").returncode == 0 finally: subprocess.run( diff --git a/tests/test_manual_messaging.py b/tests/test_manual_messaging.py new file mode 100644 index 0000000..dc58480 --- /dev/null +++ b/tests/test_manual_messaging.py @@ -0,0 +1,37 @@ +import json + +from tamq.cli import main + + +async def service_is_down(): + return False + + +def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("TAMQ_REPO", "flex-auth") + monkeypatch.setattr("tamq.cli.ping", service_is_down) + monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) + + assert main(["send", "@audit-core:", "review", "this"]) == 0 + message_id = capsys.readouterr().out.strip() + + assert main(["inbox", "--repo", "audit-core", "--json"]) == 0 + row = json.loads(capsys.readouterr().out) + assert row["message_id"] == message_id + assert row["sender_repo"] == "flex-auth" + assert row["target_repo"] == "audit-core" + assert row["body"] == "review this" + assert row["state"] == "pending" + + assert main(["ack", message_id]) == 0 + capsys.readouterr() + assert main(["inbox", "--repo", "audit-core"]) == 0 + assert capsys.readouterr().out == "" + + +def test_inbox_requires_repository_outside_managed_window(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.delenv("TAMQ_REPO", raising=False) + assert main(["inbox"]) == 2 + assert "requires --repo" in capsys.readouterr().err diff --git a/tests/test_ptytap_integration.py b/tests/test_ptytap_integration.py index 7a374ea..2dcf23b 100644 --- a/tests/test_ptytap_integration.py +++ b/tests/test_ptytap_integration.py @@ -67,7 +67,7 @@ def test_tap_propagates_terminal_size_resize_and_raw_mouse_input(tmp_path, monke pytest.fail(f"timed out waiting for {marker!r}; pane contained:\n{output}") try: - manager.ensure_plan(plan) + manager.ensure_plan(plan, tap=True) output = wait_for("READY") dimensions = manager._run( "display-message", diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py index c90baa0..a5bfc19 100644 --- a/tests/test_service_delivery.py +++ b/tests/test_service_delivery.py @@ -25,7 +25,7 @@ class FakeControl: 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"]) + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "pane") message_id = store.add("repo-a", "repo-b", "continue") monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) Service(store=store)._deliver_once() @@ -34,6 +34,18 @@ def test_service_delivery_injects_pending_messages(tmp_path, monkeypatch): assert store.list()[0]["state"] == "injected" +def test_service_never_injects_manual_endpoint_messages(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "manual") + store.add("repo-a", "repo-b", "continue") + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + + Service(store=store)._deliver_once() + + assert FakeControl.instances[-1].injected == [] + assert store.list()[0]["state"] == "pending" + + def test_service_disconnects_disappeared_tmux_endpoint(tmp_path, monkeypatch): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"]) diff --git a/tests/test_store.py b/tests/test_store.py index 6f6f966..de43c98 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -1,4 +1,5 @@ import json +import sqlite3 from tamq.store import Store @@ -13,3 +14,32 @@ def test_message_history_and_jsonl(tmp_path): assert json.loads(output.read_text())["body"] == "hello" assert store.db.execute("PRAGMA journal_mode").fetchone()[0] == "wal" store.close() + + +def test_store_migrates_legacy_endpoint_rows_to_manual_delivery(tmp_path): + path = tmp_path / "legacy.sqlite3" + db = sqlite3.connect(path) + db.executescript( + """ + CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); + INSERT INTO metadata VALUES('schema_version', '1'); + CREATE TABLE 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 + ); + INSERT INTO endpoints VALUES('legacy', 1, 'tamq', '[\"repo\"]', 1, NULL); + """ + ) + db.commit() + db.close() + + store = Store(path) + assert store.endpoints()[0]["delivery_mode"] == "manual" + assert store.db.execute( + "SELECT value FROM metadata WHERE key='schema_version'" + ).fetchone()[0] == "2" + store.close() diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index b0f0166..1fa9eca 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -38,7 +38,7 @@ def test_real_tmux_starts_two_repo_windows_and_reuses_them(tmp_path, monkeypatch ) try: - endpoint = manager.ensure_plan(plan) + endpoint = manager.ensure_plan(plan, tap=True) assert endpoint.created_session is True assert endpoint.created_windows == ("railiance-platform", "activity-core") @@ -65,7 +65,7 @@ def test_real_tmux_starts_two_repo_windows_and_reuses_them(tmp_path, monkeypatch time.sleep(0.05) assert all("alpha-ready" in capture for capture in captures) - reused = manager.ensure_plan(plan) + reused = manager.ensure_plan(plan, tap=True) assert reused.created_session is False assert reused.created_windows == () reused_rows = manager._run( diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py index e56782c..1a192e7 100644 --- a/tests/test_tmux_manager.py +++ b/tests/test_tmux_manager.py @@ -21,7 +21,8 @@ def test_tmux_manager_builds_tap_windows(tmp_path, monkeypatch): return "" monkeypatch.setattr(manager, "_run", run) monkeypatch.setattr(tmux.subprocess, "run", lambda *args, **kwargs: type("R", (), {"returncode": 1})()) - endpoint = manager.ensure(["a", "b"], "codex --quiet") + plan = manager.preflight(["a", "b"], "codex --quiet") + endpoint = manager.ensure_plan(plan, tap=True) assert endpoint.endpoint_id == "tmux-amq-42" assert endpoint.instance_id.startswith("tmux-amq-42-") assert endpoint.repos == ["a", "b"] @@ -118,6 +119,34 @@ def test_tmux_only_plan_starts_agent_without_tap(tmp_path, monkeypatch): assert "tamq tap" not in send +def test_neutral_plan_starts_shell_without_sending_keystrokes(tmp_path, monkeypatch): + manager = tmux.TmuxManager("tamq-test") + calls = [] + plan = tmux.LaunchPlan(("a", "b"), {"a": str(tmp_path), "b": str(tmp_path)}, ()) + + def run(*args, check=True): + calls.append(args) + if args[:2] == ("display-message", "-p"): + return "42" + if args[:2] == ("list-windows", "-t"): + return "__tamq_boot" + return "" + + monkeypatch.setattr(manager, "_run", run) + monkeypatch.setattr( + tmux.subprocess, + "run", + lambda *args, **kwargs: type("R", (), {"returncode": 1})(), + ) + manager.ensure_plan(plan) + + assert not any(call and call[0] == "send-keys" for call in calls) + new_session = next(call for call in calls if call and call[0] == "new-session") + new_window = next(call for call in calls if call and call[0] == "new-window") + assert "TAMQ_REPO=a" in new_session + assert "TAMQ_REPO=b" in new_window + + def test_preflight_rejects_missing_agent(tmp_path, monkeypatch): repo = tmp_path / "a" repo.mkdir() @@ -125,7 +154,7 @@ def test_preflight_rejects_missing_agent(tmp_path, monkeypatch): monkeypatch.setattr(tmux, "validate_targets", lambda repos: None) monkeypatch.setattr(tmux, "repository_paths", lambda: {"a": str(repo)}) monkeypatch.setattr(tmux.shutil, "which", lambda command: None if command == "missing-agent" else f"/bin/{command}") - with pytest.raises(tmux.TmuxError, match="agent command not found"): + with pytest.raises(tmux.TmuxError, match="initial command not found"): manager.preflight(["a"], "missing-agent") diff --git a/workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md b/workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md index f48de2c..c3cf6d3 100644 --- a/workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md +++ b/workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md @@ -4,7 +4,7 @@ type: workplan title: "Terminal-neutral sessions and safe manual messaging" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social planning_priority: P0 @@ -26,6 +26,12 @@ interactive program—owns a pane. tamq start flex-auth audit-core ``` +The repository-first shorthand is equivalent: + +```bash +tamq flex-auth audit-core +``` + This opens or reuses ordinary interactive shells in the exact registered repository paths. It runs no initial command, installs no PTY observer, and never injects a queued message into an active terminal input line. @@ -52,7 +58,7 @@ The sender defaults to the repository identity of the current managed window. ```task id: TAMQ-WP-0005-T01 -status: progress +status: done priority: high state_hub_task_id: "84d3b151-3a0a-56ca-8c00-9000ecbb4751" ``` @@ -63,11 +69,15 @@ command. Set a trustworthy per-window `TAMQ_REPO` identity without changing the user's shell or terminal behavior. Retain explicit compatibility aliases where they do not weaken the neutral default. +Completed with repository-first and explicit `start` forms, no default command, +untouched tmux shells, per-window exported `TAMQ_REPO`, and exact opt-in +`--command`/`--cmd` execution only for newly created windows. + ## Separate durable messaging from pane input ```task id: TAMQ-WP-0005-T02 -status: todo +status: done priority: high state_hub_task_id: "6c0f04a6-2af1-53b7-9976-325a9483a838" ``` @@ -77,11 +87,18 @@ from turning their queued messages into `send-keys` input. Make any legacy PTY observation/injection path explicit opt-in, visible in status, and absent from the default startup path. +Completed with schema-versioned endpoint `delivery_mode`. New and migrated +endpoints default to `manual`; the broker never injects their pending messages. +Pane observation/injection requires both explicit `--tap` and `--command`, is +reported as `pane` mode, and cannot be combined with `--no-service`. Startup +also restarts a legacy broker that lacks the manual-delivery capability before +registration. + ## Add a practical manual inbox ```task id: TAMQ-WP-0005-T03 -status: todo +status: done priority: high state_hub_task_id: "80bb61bb-8182-58d3-9067-3530026a59a4" ``` @@ -91,11 +108,15 @@ automation, use the current window's `TAMQ_REPO` as sender identity, validate targets, and preserve explicit acknowledgement. Manual exchange must work without corrupting either terminal's current input. +Completed with `tamq inbox`, repository-context sender identity, local target +validation, human-readable and JSON output, pending-by-default filtering, and +the existing explicit acknowledgement command. + ## Prove and document the neutral workflow ```task id: TAMQ-WP-0005-T04 -status: wait +status: done priority: high state_hub_task_id: "08ca9746-97d1-56d8-badd-5a5ac2c8e147" ``` @@ -105,3 +126,10 @@ per-window repository identity, durable send/inbox/ack exchange, and proof that pending manual messages never appear as pane input. Update the install/start quickstart and practical-usability assessment, then record the local acceptance evidence. This task follows T01-T03. + +Completed with the isolated installed-package acceptance. It proves untouched +shells at exact paths, stable reuse, per-window identities, manual +send/inbox/ack exchange, byte-for-byte unchanged target-pane output, service +restart and endpoint recovery, and exact explicit-command startup. The full +suite passes 77 tests with 74% statement coverage; README, INTENT, SCOPE, agent +workflow, and command rules now state the terminal-neutral boundary.