diff --git a/README.md b/README.md index 32d0d04..61190f9 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ 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 +leases, readable `To:` routing, JSONL export/replay, and a Unix-socket protocol for a later coordination-engine adapter. ## Install and start a local session @@ -27,8 +27,8 @@ tamq flex-auth audit-core ``` This creates or reuses the managed `tamq` session, opens same-named windows in -the exact gita paths, leaves tmux's ordinary interactive shells untouched, and -attaches to the first window. The explicit form is equivalent: +the exact gita paths, starts an ordinary shell behind tamq's transparent PTY +observer, and attaches to the first window. The explicit form is equivalent: ```bash tamq start flex-auth audit-core @@ -64,111 +64,80 @@ tamq --mode inbox flex-auth audit-core `--no-display` remains a compatibility alias for `--mode inbox`. -Experimental `pushy` mode submits each routed message to the target pane as -input. It is intended for coding-agent interfaces that queue user prompts: +Experimental `pushy` mode places a routed message in the target input buffer +without Enter. `trigger` places the same input and submits it exactly once: ```bash tamq --mode pushy --command codex flex-auth audit-core +tamq --mode trigger --command codex flex-auth audit-core ``` -For newly created windows with an explicit command, pushy mode automatically -places that command behind tamq's transparent PTY tap. Inside the coding-agent -input, use `#repo:` to avoid interfaces that reserve `@` for file mentions or -templates: +All messaging-enabled windows use the same terminal-neutral PTY observer, +whether they contain a shell, a tool, or an explicitly selected coding agent. +An operator or a worker can emit the same readable line: ```text -#audit-core: Please review the authentication change. +To:audit-core: Please review the authentication change. ``` -The tap observes this complete line and queues the same durable route as -`@audit-core: ...`; it forwards the source input unchanged to its occupant. -`#repo:` is therefore an agent/tap convention, not a shell command—at an -ordinary shell it remains a comment. Existing windows are never respawned, so -recreate a session that was originally started without pushy observation. -Terminal protocol replies and local backspace editing are removed only from -tamq's observation copy, so strict routing remains reliable in TUIs such as -Codex while the wrapped program still receives the original byte stream. - -The submitted target line is sanitized, sender-labelled, deliberately -non-routable, and then followed by exactly one Enter key: +Operator input is forwarded unchanged and is delivered with `/o`; a line +originating in worker output has no suffix: ```text -# from flex-auth: please review the auth boundary [m-...] +From:flex-auth/o: Please review the authentication change. +From:flex-auth: Worker-generated message. ``` -The `# from` shape cannot match the outbound `#repo:` grammar, preventing a -tapped recipient from bouncing the message back. Its leading `#` also makes an -empty ordinary shell prompt treat the line as a comment. Pushy mode cannot -determine whether a pane is an agent, a shell, or whether someone is already -editing input: it can append to that input and submit the combined line. Use it -only for panes whose occupant is known to accept or queue asynchronous prompts. -Switching an existing endpoint to pushy may also submit pending messages that -have never been displayed. - -Pushy startup requires a capability for this non-routable framing and restarts -an older broker that only advertised generic pushy input. As a second circuit -breaker, the tap refuses a legacy `#sender: body [message-id]` line when that -identifier belongs to the corresponding durable inbound delivery. This keeps -mixed-version local processes from reflecting a message between panes. +`From:` is never routable. The observer suppresses recent exact operator lines +when a TUI echoes or redraws them, so the echo cannot become a second worker +message. Ambiguous exact echoes fail closed. Pushy and trigger still cannot +infer the foreground program or protect input already being edited, so both +remain explicit experiments. Startup requires the readable-duplex, trigger, +and line-limit capabilities and replaces an incompatible old broker. ## Exchange messages manually -Each managed shell exports its own repository slug as `TAMQ_REPO`. From the -`flex-auth` window, queue a message using the repository command installed for -the session: +Each managed window has its own repository identity. From operator input in the +`flex-auth` window, queue a message with: ```bash -@audit-core: please review the auth boundary +To:audit-core: please review the auth boundary ``` -The spelling without the trailing colon is equivalent: +The same line at the start of worker output queues a worker-originated message. +The `@`, `#`, reply shorthand, and interactive recipient composer from earlier +alphas have been removed. -```bash -@audit-core please review the auth boundary -``` - -Run bare `@` to open the inline composer. It shows the latest sender as the -recipient, and ordinary prose is read directly rather than parsed by the shell: +Operator input can change allowlisted runtime state: ```text -$ @ -@audit-core: What's up? +Cmd: mode=trigger +Cmd: maxmsg=16 +Cmd: maxin=2048 +Cmd: maxout=65536 +Cmd: reset-limits ``` -Press Tab to cycle through the other repository windows while preserving the -draft, Enter to send, or Ctrl-C to cancel without queuing. The initial recipient -is the sender of the latest durable inbound message, including an already -acknowledged message; without history, it is the first peer window. -Self-addressed messages are ignored. - -The one-line fast path remains available, but its text is parsed by the shell -and therefore follows ordinary shell quoting rules: +`Cmd:` is recognized only on the operator-input path; identical worker output +is inert. The transparent input contract means the line also remains visible +to the foreground program. Small tamq-owned shell absorbers prevent ordinary +shells from reporting `To:repo:` and `Cmd:` as missing commands; tamq refuses +to overwrite unrelated files. The explicit CLI send form is: ```bash -@ "What's up?" -tamq reply "What's up?" -``` - -These are tamq-owned executable commands beside the installed `tamq` command, -not shell-specific aliases. Set `TAMQ_COMMAND_DIR` before startup to select a -different writable command directory already present on your shell's `PATH`. -Tamq refuses to overwrite unrelated commands. The shims only use tamq's durable -send operation. The long form remains available: - -```bash -tamq send '@audit-core: please review the auth boundary' +tamq send 'To:audit-core: please review the auth boundary' ``` With normal `output` delivery, the target pane visibly receives: ```text -#flex-auth: please review the auth boundary [m-...] +From:flex-auth/o: please review the auth boundary ``` This uses the pane's tmux-reported `/dev/pts/` device—the same Unix terminal-output mechanism underlying tools such as `write(1)`. For an ordinary shell with screen rows above its cursor, tamq confines scrolling to those rows, -writes the comment immediately above the input row, and restores the cursor. +writes the delivery immediately above the input row, and restores the cursor. Thus a partially typed command remains in place. If no safe row exists, tamq uses ordinary line output; alternate-screen programs receive the conservative fallback and may redraw over it. Neither path uses `send-keys`, sends Enter, or @@ -182,11 +151,11 @@ tamq inbox tamq ack ``` -Human inbox output is safe to paste into an ordinary shell because every line -is a comment. It includes the durable id needed by `ack`: +Human inbox output uses the same framing. Use `--json` or `tamq history` to get +the durable id needed by `ack`: ```text -#flex-auth: please review the auth boundary [m-...] +From:flex-auth/o: please review the auth boundary ``` To explicitly consume pending messages through a command, use an inbox filter: @@ -195,7 +164,7 @@ To explicitly consume pending messages through a command, use an inbox filter: tamq inbox --filter 'cat >> msg.log' ``` -The command runs once per pending message with the comment form on standard +The command runs once per pending message with the readable `From:` form on standard input. `TAMQ_MESSAGE_ID`, `TAMQ_SENDER_REPO`, and `TAMQ_TARGET_REPO` are set in its environment. A zero exit acknowledges that message; a non-zero exit leaves it and all later messages pending. Filters never run in the background and @@ -203,10 +172,23 @@ cannot be combined with `--all` or `--json`. Outside a managed window, use `tamq inbox --repo audit-core` and optionally `--json`. Output-displayed messages remain durable and pending until -acknowledged. Neither `output` nor `inbox` mode injects terminal keystrokes, so -they cannot execute or alter a command being typed in the target pane. Pushy -mode intentionally crosses that boundary and records accepted submissions as -`injected`. +acknowledged. Neither `output` nor `inbox` mode injects terminal keystrokes. +Pushy places input without Enter; trigger adds exactly one Enter. Both input +modes record accepted delivery as `injected`. + +Every session window has independent, session-lifetime running counters for +accepted outbound messages, operator input lines, and normalized worker output +lines. Defaults are 8, 1024, and 32768 respectively: + +```bash +tamq --maxmsg 16 --maxin 2048 --maxout 65536 flex-auth audit-core +``` + +The same defaults can be set as `maxmsg`, `maxin`, and `maxout` under `[tamq]` +in the configuration file. A message is rejected once any counter is equal to +its limit. Tamq displays the applicable count and `Cmd: reset-limits` resets all +three counters for the current window. Counters survive tap and broker restarts +within the same managed-session generation and are visible in `tamq status`. After upgrading from an earlier alpha, recreate the managed session once so existing panes inherit the neutral shell contract, repository command `PATH`, @@ -237,8 +219,9 @@ tamq cleanup --yes Confirmed cleanup stops only the verified tamq broker, closes only a tmux session carrying tamq's management marker, disconnects transient endpoints, -clears leases, removes the configured socket/PID/lock files, deletes only -tamq-generated `@repo` shims, and removes owned stale `tamq-*` tmux sockets. +clears leases and line counters, removes the configured socket/PID/lock files, +deletes only tamq-generated protocol absorbers (including legacy `@repo` +shims), and removes owned stale `tamq-*` tmux sockets. Durable message history and unrelated tmux sessions or files are preserved. The command is idempotent; an ownership mismatch is reported instead of being removed. @@ -276,9 +259,10 @@ 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. 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:`/`#repo:` routing. +3. `tamq tap` is a full-duplex PTY proxy around every messaging-enabled shell + or explicit command. It forwards bytes unchanged in raw terminal mode, + propagates resize and lifecycle signals, observes operator input, and + normalizes worker output for start-of-line `To:` routing. This keeps tmux-specific topology concerns separate from reusable terminal I/O observation and message identity. @@ -286,10 +270,10 @@ observation and message identity. Normal endpoints use terminal-output delivery: each message is written once to the target pane's PTY output and stays pending in the durable inbox until acknowledged. `--mode inbox` selects inbox-only manual mode. Neither becomes -pane input. Experimental `--mode pushy` sends one checked tmux command list -containing sanitized literal input followed by Enter, then records the message -as `injected`. When pushy mode creates explicitly commanded windows, the same -PTY tap observes outbound `@repo:` and `#repo:` lines without rewriting them. +pane input. Experimental `--mode pushy` places one sanitized `From:` line +without Enter; `--mode trigger` performs the same placement and submits once. +The PTY observer records explicit operator/worker provenance and suppresses +echoed operator lines before they can route again. The older pane-delivery experiment remains available only with the explicit `tamq start --tap --command ...` opt-in. Both input paths remain subject to the retry and acknowledgement limitations tracked by `TAMQ-WP-0003`. @@ -310,6 +294,9 @@ purge_before = "365d" purge_max_size = "100MB" history_max_size = "100MB" delivery_poll_interval = "0.5" +maxmsg = 8 +maxin = 1024 +maxout = 32768 [policy.profiles.diagnostics] safety_gated_max_attempts = 2 diff --git a/SCOPE.md b/SCOPE.md index 0e371d3..2b300eb 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -16,17 +16,19 @@ tamq does not choose or infer them. - Local SQLite message history, leases, endpoint registrations, delivery state, acknowledgements, replay, export, and bounded purging. - Managed neutral-shell tmux lifecycle and explicit initial commands. -- Durable manual send/inbox/acknowledgement with per-window repository identity, - shell-native direct and latest-counterparty reply commands, comment-safe - display, a recipient-aware inline composer, and explicit pull-time filters. +- Durable send/inbox/acknowledgement with per-window repository identity, + readable `To:`/`From:` framing, explicit operator/worker provenance, and + pull-time filters. - Sanitized one-time output notifications through target tmux pane PTYs, with inbox-only delivery as an explicit option and no foreground-process input. -- Explicit experimental pushy delivery that submits a sanitized comment and - Enter to target pane input for known queue-capable interactive programs, - with `#repo:` routing observed through the PTY tap. +- Explicit experimental pushy input placement and trigger submission for known + queue-capable interactive programs. - 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:`/`#repo:` routing. +- Exact `gita` validation and direct start-of-line `To:repo:` routing from + operator input or normalized worker output. +- Operator-only allowlisted `Cmd:` runtime changes and atomic, session-lifetime + per-window message/input/output line budgets. - Unix-socket operations for local clients and a future coordination-engine adapter. - Policy profiles, safety-gated retries, local diagnostics, tests, packaging, @@ -55,40 +57,39 @@ transport. | Intent capability | State | Evidence and remaining gap | | --- | --- | --- | -| Direct repository addressing | Implemented for local alpha | Exact `gita` validation, per-session `@repo`/`@repo:` shell commands, tapped `#repo:` agent-input routing, a bare `@` composer with latest-counterparty default and Tab recipient cycling, and long-form parsing are covered. | +| Direct repository addressing | Implemented for local alpha | Exact `gita` validation and case-sensitive `To:repo: body` parsing are shared across operator input and worker output. Legacy `@/#` addressing and the composer are removed. | | Durable, inspectable local queue | Implemented | SQLite history, manual inbox, leases, endpoint records, inspect/history, JSONL export/replay, acknowledgement, age/size purge, and exact reflected-chain purge are present. | | Local socket service | Implemented | Peer-credential checks and structured ping/register/send/history/ack/endpoints/disconnect operations are tested. | | 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. | | Emergency local cleanup | Implemented for local alpha | `tamq cleanup` dry-runs by default; confirmed cleanup verifies ownership before stopping the broker or marked session, clears only transient DB state, removes configured runtime files/generated shims/owned stale tmux sockets, and preserves history. | -| Safe manual messaging | Implemented for local alpha | Normal endpoints write one sanitized comment above a stable shell input row without injecting stdin; conservative fallback handles the first row and alternate screens. Messages remain pending until acknowledgement. Inbox-only manual mode is explicit with `--no-display`. | -| Experimental pushy delivery | Explicit opt-in | `--mode pushy --command ...` observes outbound `#repo:`/`@repo:` lines and submits a non-routable `# from sender:` envelope plus Enter through a checked tmux command. Startup rejects stale framing capabilities, and the tap recognizes durable legacy delivery receipts as a feedback circuit breaker. It marks successful delivery `injected`, but cannot identify pane occupants or protect input already being edited. | -| 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. | +| Safe output messaging | Implemented for local alpha | Normal endpoints write one sanitized `From:` line above a stable shell input row without injecting stdin; `/o` marks operator origin. Messages remain pending until acknowledgement. Inbox-only mode is explicit. | +| Experimental pushy and trigger delivery | Explicit opt-in | Pushy places a non-routable `From:` line without Enter; trigger adds exactly one Enter. Both are capability-gated and mark accepted delivery `injected`, but cannot identify pane occupants or protect input already being edited. | +| Full-duplex observation | Implemented for managed messaging | Every messaging-enabled new window runs its neutral shell or explicit command behind the PTY observer. It preserves geometry, resize, mouse input, and raw forwarding; normalizes worker output, deduplicates redraws, and fails closed on exact recent operator echoes. | +| Runtime commands and generation budgets | Implemented for local alpha | Operator-only `Cmd:` changes mode or per-window limits and resets the current ledger. Defaults are 8 message, 1024 input, and 32768 output lines. Admission and counter increments are atomic and survive service/tap restarts in one session generation. | | Bounded retry behavior | Not enforced | Failed output or 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 | Terminal output remains pending until explicit acknowledgement, while legacy pane injection becomes `injected`; the configured `delivery_ack_mode` does not yet govern both paths. | | Coordination-engine interoperability | Not implemented | The adapter contract and implementation remain in `TAMQ-WP-0002`. | ## Practical Usability -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, shell-native addressing, -latest-counterparty replies, comment-safe target output above an unchanged -partial input line and cursor, inbox/filter/ack exchange, zero target-input -mutation in safe modes, service restart, endpoint disappearance, explicit -initial-command startup, and cleanup. Isolated pushy-mode coverage additionally -proves one `#repo:` route, one non-routable target submission, and no feedback -message. +The terminal-neutral alpha path is exercised with an isolated installed tool +and real tmux/PTYs. Repository-first startup creates two observed ordinary +shells at exact gita paths without selecting an agent. Tests prove per-window +identity, stable reuse, operator and worker `To:` routing, exact `/o` +attribution, echoed-input suppression, stable target output above unchanged +partial input, inbox/filter/ack exchange, pushy placement, trigger submission, +line-limit/reset behavior, service restart, explicit-command startup, and +cleanup. Suitable today: - Local queue, history, export/replay, and diagnostic use. - Interactive local shell or explicitly commanded sessions over one or more gita-registered repositories. -- Durable manual message exchange between managed repository windows, including +- Durable message exchange between managed repository windows, including explicit pull-time loggers and filters. -- Controlled experiments with pushy delivery to coding-agent interfaces known - to queue asynchronous user prompts. +- Controlled experiments with pushy placement or trigger submission to + interfaces known to queue asynchronous prompts. - Developing and testing the future coordination-engine adapter against the local socket boundary. @@ -102,13 +103,11 @@ Not yet suitable: and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 131 passing tests and 78% 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 -Forgejo CI job installs tmux and uv, runs the real-tmux and isolated -installed-package session tests with a deterministic gita fixture, and retains -CLI help/version smoke checks on Python 3.11. +The suite currently has 133 passing tests. It includes atomic counter races, +pseudo-terminal normalization, real tmux pushy/trigger behavior, operator-echo +suppression, and an isolated installed-package workflow with deterministic gita +fixtures. Forgejo CI installs tmux and uv and retains CLI help/version smoke +checks on Python 3.11. ## Next Usability Gates diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index bff26be..06b2952 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -23,7 +23,7 @@ | workplan | TAMQ-WP-0011 | finished | — | workplans/TAMQ-WP-0011-hash-routing-for-pushy-agents.md | | workplan | TAMQ-WP-0012 | finished | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | | workplan | TAMQ-WP-0013 | finished | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | -| workplan | TAMQ-WP-0014 | ready | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| workplan | TAMQ-WP-0014 | finished | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | | task | TAMQ-WP-ADHOC-2026-08-24-T01 | done | — | workplans/ADHOC-2026-08-24.md | | task | TAMQ-WP-ADHOC-2026-08-25-T01 | done | — | workplans/ADHOC-2026-08-25.md | | task | TAMQ-WP-0001-T01 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | @@ -74,10 +74,10 @@ | task | TAMQ-WP-0013-T01 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | | task | TAMQ-WP-0013-T02 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | | task | TAMQ-WP-0013-T03 | done | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | -| task | TAMQ-WP-0014-T01 | todo | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T02 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T03 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T04 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T05 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T06 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | -| task | TAMQ-WP-0014-T07 | wait | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T01 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T02 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T03 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T04 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T05 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T06 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | +| task | TAMQ-WP-0014-T07 | done | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | diff --git a/src/tamq/broker.py b/src/tamq/broker.py index 11778bf..5979c6e 100644 --- a/src/tamq/broker.py +++ b/src/tamq/broker.py @@ -1,17 +1,12 @@ from __future__ import annotations -import re +from collections.abc import Callable from dataclasses import dataclass -from .routing import RoutedMessage, parse_address_line -from .store import Store from .registry import RegistryError, validate_targets - - -LEGACY_DELIVERY_SUFFIX = re.compile( - r"\s+\[(m-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\]$", - re.IGNORECASE, -) +from .routing import RoutedMessage, parse_address_line, parse_command_line +from .store import LimitBlock, Store +from .terminal import format_delivery @dataclass(frozen=True) @@ -21,39 +16,117 @@ class BrokerIdentity: class InputBroker: - """Convert tap input into durable, identity-bearing outbound messages.""" + """Convert attributed PTY lines into durable outbound messages.""" - def __init__(self, store: Store, identity: BrokerIdentity): + def __init__( + self, + store: Store, + identity: BrokerIdentity, + *, + limits: tuple[int, int, int] = (8, 1024, 32768), + notify: Callable[[str], None] | None = None, + ): self.store = store self.identity = identity + self.notify = notify + self.store.configure_window(identity.endpoint_id, identity.source_repo, limits) - def inspect_line(self, line: str) -> RoutedMessage | None: + def _notice(self, text: str) -> None: + if self.notify is not None: + self.notify(f"From:tamq: {text}") + + def _blocked(self, blocked: LimitBlock) -> None: + self._notice( + "Messaging blocked! Running linecount " + f"{blocked.count} limited by {blocked.limit} lines of {blocked.dimension} " + "in this terminal. Use 'Cmd: reset-limits' to unblock." + ) + + def _route(self, line: str, provenance: str) -> RoutedMessage | None: routed = parse_address_line(line) if routed is None: return None - receipt = LEGACY_DELIVERY_SUFFIX.search(routed.body) - if line.startswith("#") and receipt: - prior = self.store.message(receipt.group(1)) - if ( - prior is not None - and prior["sender_repo"] == routed.target_repo - and prior["target_repo"] == self.identity.source_repo - ): - return None try: validate_targets([routed.target_repo]) except RegistryError: return None - self.store.add( + result = self.store.admit_message( self.identity.source_repo, routed.target_repo, routed.body, endpoint=self.identity.endpoint_id, + source_repo=self.identity.source_repo, + provenance=provenance, ) + if isinstance(result, LimitBlock): + self._blocked(result) + return None return routed + def inspect_operator_line(self, line: str) -> RoutedMessage | None: + """Observe one submitted operator line; it is still forwarded unchanged.""" + if line.startswith("From:"): + return None + self.store.count_line( + self.identity.endpoint_id, self.identity.source_repo, "input" + ) + command = parse_command_line(line) + if command is not None: + self._command(command) + return None + return self._route(line, "operator_input") + + def inspect_worker_line(self, line: str) -> RoutedMessage | None: + """Observe worker output; Cmd is intentionally inert on this path.""" + if line.startswith("From:"): + return None + self.store.count_line( + self.identity.endpoint_id, self.identity.source_repo, "output" + ) + return self._route(line, "worker_output") + + def inspect_line(self, line: str) -> RoutedMessage | None: + """Compatibility entry point for operator-input observers.""" + return self.inspect_operator_line(line) + + def _command(self, command: str) -> None: + if command == "reset-limits": + self.store.reset_limits( + self.identity.endpoint_id, self.identity.source_repo + ) + self._notice("Limits reset for this terminal.") + return + if "=" not in command: + self._notice(f"Unknown command: {command}") + return + name, value = (part.strip() for part in command.split("=", 1)) + if name == "mode": + if value not in {"inbox", "output", "pushy", "trigger"}: + self._notice(f"Invalid mode: {value}") + return + if not self.store.set_endpoint_mode(self.identity.endpoint_id, value): + self._notice("Cannot change mode before endpoint registration.") + return + self._notice(f"Mode set to {value}.") + return + if name in {"maxmsg", "maxin", "maxout"}: + try: + parsed = int(value) + self.store.set_window_limit( + self.identity.endpoint_id, + self.identity.source_repo, + name, + parsed, + ) + except ValueError as exc: + self._notice(str(exc)) + return + self._notice(f"{name} set to {parsed} for this terminal.") + return + self._notice(f"Unknown command: {name}") + def deliver_pending(self, control, *, window_for_repo): - """Inject pending messages through the authoritative control client.""" + """Place 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): @@ -63,7 +136,12 @@ class InputBroker: if lease_id is None: continue try: - control.inject(target, f"#{row['sender_repo']}: {row['body']}") + control.inject( + target, + format_delivery( + row["sender_repo"], row["body"], row["provenance"] + ), + ) except Exception: continue self.store.release(row["message_id"], lease_id, "injected") diff --git a/src/tamq/cleanup.py b/src/tamq/cleanup.py index fb96dc0..c73620f 100644 --- a/src/tamq/cleanup.py +++ b/src/tamq/cleanup.py @@ -13,7 +13,8 @@ from .store import Store from .tmux import TmuxError, TmuxManager -SHIM_MARKER = "# tamq-address-command v1" +SHIM_MARKER = "# tamq-protocol-command v2" +LEGACY_SHIM_MARKER = "# tamq-address-command v1" def _read_pid() -> int | None: @@ -45,14 +46,15 @@ def _generated_shims(directory: Path) -> list[Path]: return [] shims: list[Path] = [] for candidate in candidates: - if not candidate.name.startswith("@") or candidate.is_symlink(): + if candidate.is_symlink(): continue try: - if candidate.is_file() and SHIM_MARKER in candidate.read_text( - encoding="utf-8" + content = candidate.read_text(encoding="utf-8") + if candidate.is_file() and ( + SHIM_MARKER in content or LEGACY_SHIM_MARKER in content ): shims.append(candidate) - except OSError: + except (OSError, UnicodeError): continue return sorted(shims) diff --git a/src/tamq/cli.py b/src/tamq/cli.py index dfe9473..31a8227 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -18,24 +18,30 @@ 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 PUSHY_FRAMING_CAPABILITY, Service, ping, request +from .service import ( + LINE_LIMITS_CAPABILITY, + PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, + 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 .composer import ComposerError, compose_message, recipient_order from .cleanup import cleanup_runtime from .diagnostics import configure from .policy import load_profile from .registry import RegistryError, validate_targets -from .terminal import format_comment +from .terminal import format_delivery SUBCOMMANDS = frozenset( - "start attach serve stop cleanup status ping inbox history inspect ack send reply export replay purge completion db-version config tap".split() + "start attach serve stop cleanup 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", "--no-display", "--mode"}) +START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service", "--no-display", "--mode", "--maxmsg", "--maxin", "--maxout"}) GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"}) @@ -123,16 +129,26 @@ def ensure_service_capabilities(required: set[str]) -> bool: def ensure_manual_service() -> bool: - return ensure_service_capabilities({"manual_delivery"}) + return ensure_service_capabilities( + {"manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY} + ) def ensure_output_service() -> bool: - return ensure_service_capabilities({"manual_delivery", "terminal_output"}) + return ensure_service_capabilities( + {"manual_delivery", "terminal_output", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY} + ) def ensure_pushy_service() -> bool: return ensure_service_capabilities( - {"manual_delivery", "pushy_input", PUSHY_FRAMING_CAPABILITY} + { + "manual_delivery", + "pushy_input", + PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, + LINE_LIMITS_CAPABILITY, + } ) @@ -150,9 +166,18 @@ def attach_session(session: str) -> int: return subprocess.run(command, check=False).returncode -def format_comment_message(row: sqlite3.Row) -> str: - """Render a durable message so every displayed line remains a shell comment.""" - return format_comment(row["sender_repo"], row["body"], row["message_id"]) +def format_inbox_message(row: sqlite3.Row) -> str: + return format_delivery(row["sender_repo"], row["body"], row["provenance"]) + + +def positive_limit(value: str | int, name: str) -> int: + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{name} must be a positive integer") from exc + if parsed <= 0: + raise ValueError(f"{name} must be a positive integer") + return parsed def build_parser() -> argparse.ArgumentParser: @@ -161,18 +186,19 @@ def build_parser() -> argparse.ArgumentParser: description="Repository-aware tmux sessions with durable local messaging.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""session shorthand: - tamq [--detach] [--command COMMAND] [--mode MODE] REPO [REPO ...] + tamq [--detach] [--command COMMAND] [--mode MODE] + [--maxmsg N] [--maxin N] [--maxout N] REPO [REPO ...] With no --command, tamq opens ordinary repository shells. --command is an explicit initial command for newly created windows. Messages appear as sanitized terminal output by default. --mode inbox keeps messages durable - without display; experimental --mode pushy submits them to target input. + without display. --mode pushy places target input; --mode trigger submits it. + Line-limit defaults are 8 messages, 1024 input, and 32768 output lines. -manual messaging from a managed shell: - @TARGET: MESSAGE... - @ compose with a visible, Tab-selectable recipient - @ MESSAGE... fast reply (ordinary shell quoting applies) - #TARGET: MESSAGE... agent-input alias in a tapped pushy session +duplex messaging from a managed terminal: + To:TARGET: MESSAGE... operator input or worker output + Cmd: mode=trigger operator-only runtime command + Cmd: reset-limits reset this window's running counters tamq inbox [--filter COMMAND] """, ) @@ -186,7 +212,10 @@ manual messaging from a managed shell: start.add_argument("repos", nargs="*", help="gita-registered repository slugs") start.add_argument("--command", "--cmd", dest="initial_command", default=None, help="explicit initial command for newly created windows (default: ordinary shell)") start.add_argument("--tap", action="store_true", help="explicitly opt in to PTY input observation and pane message injection") - start.add_argument("--mode", choices=("output", "inbox", "pushy"), help="message delivery mode (default: output; pushy is experimental)") + start.add_argument("--mode", choices=("output", "inbox", "pushy", "trigger"), help="message delivery mode (default: output; pushy/trigger are experimental)") + start.add_argument("--maxmsg", type=int, help="maximum accepted outbound messages per window (default: config or 8)") + start.add_argument("--maxin", type=int, help="maximum operator input lines per window (default: config or 1024)") + start.add_argument("--maxout", type=int, help="maximum worker output lines per window (default: config or 32768)") start.add_argument("--no-display", action="store_true", help="compatibility alias for --mode inbox") 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") @@ -224,12 +253,10 @@ manual messaging from a managed shell: 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("address", help="To: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)") - reply = subparsers.add_parser("reply", help="reply to the latest sender for the current repository") - reply.add_argument("body", nargs="*", help="message body; omit to open the interactive composer") export = subparsers.add_parser("export", help="export history as JSONL") export.add_argument("--output", required=True) export.add_argument("--repo", dest="target_repo") @@ -254,6 +281,10 @@ manual messaging from a managed shell: 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("--maxmsg", type=int, default=None) + tap.add_argument("--maxin", type=int, default=None) + tap.add_argument("--maxout", type=int, default=None) + tap.add_argument("--ready-file", type=Path, default=None, help=argparse.SUPPRESS) tap.add_argument("wrapped_command", nargs=argparse.REMAINDER, help="command after --") return parser @@ -275,7 +306,7 @@ def normalize_argv(argv: list[str]) -> list[str]: if index == len(argv): return argv candidate = argv[index] - if candidate in START_OPTIONS or candidate.startswith("--mode=") or ( + if candidate in START_OPTIONS or any(candidate.startswith(f"{name}=") for name in ("--mode", "--maxmsg", "--maxin", "--maxout")) or ( not candidate.startswith("-") and candidate not in SUBCOMMANDS ): return [*argv[:index], "start", *argv[index:]] @@ -324,7 +355,16 @@ def main(argv: list[str] | None = None) -> int: 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)) + try: + configured_limits = { + "maxmsg": positive_limit(setting("maxmsg", "8"), "maxmsg"), + "maxin": positive_limit(setting("maxin", "1024"), "maxin"), + "maxout": positive_limit(setting("maxout", "32768"), "maxout"), + } + except ValueError as exc: + print(f"tamq: {exc}", file=sys.stderr) + return 2 + 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, **configured_limits}, sort_keys=True)) return 0 if args.command == "tap": command = list(args.wrapped_command) @@ -335,10 +375,31 @@ def main(argv: list[str] | None = None) -> int: return 2 store = Store(db_path()) try: - return PtyTap(command, InputBroker(store, BrokerIdentity(args.endpoint, args.repo))).run() + limits = ( + positive_limit(args.maxmsg if args.maxmsg is not None else setting("maxmsg", "8"), "maxmsg"), + positive_limit(args.maxin if args.maxin is not None else setting("maxin", "1024"), "maxin"), + positive_limit(args.maxout if args.maxout is not None else setting("maxout", "32768"), "maxout"), + ) + return PtyTap( + command, + InputBroker(store, BrokerIdentity(args.endpoint, args.repo), limits=limits), + ready_file=args.ready_file, + ).run() + except ValueError as exc: + print(f"tamq: {exc}", file=sys.stderr) + return 2 finally: store.close() if args.command == "start": + try: + limits = ( + positive_limit(args.maxmsg if args.maxmsg is not None else setting("maxmsg", "8"), "maxmsg"), + positive_limit(args.maxin if args.maxin is not None else setting("maxin", "1024"), "maxin"), + positive_limit(args.maxout if args.maxout is not None else setting("maxout", "32768"), "maxout"), + ) + except ValueError as exc: + print(f"tamq: {exc}", file=sys.stderr) + return 2 if args.no_display and args.mode not in (None, "inbox"): print("tamq: --no-display conflicts with the selected --mode", file=sys.stderr) return 2 @@ -347,6 +408,7 @@ def main(argv: list[str] | None = None) -> int: "inbox": "manual", "output": "output", "pushy": "pushy", + "trigger": "trigger", }[selected_mode] if args.tap and args.initial_command is None: print("tamq: --tap requires an explicit --command", file=sys.stderr) @@ -363,9 +425,7 @@ def main(argv: list[str] | None = None) -> int: if args.no_service and args.mode is not None: print("tamq: --mode cannot be combined with --no-service", file=sys.stderr) return 2 - tap_enabled = args.tap or ( - selected_mode == "pushy" and args.initial_command is not None - ) + tap_enabled = not args.no_service manager = TmuxManager() try: if not args.no_service: @@ -378,7 +438,7 @@ def main(argv: list[str] | None = None) -> int: service_ready = True elif selected_mode == "inbox": service_ready = ensure_manual_service() - elif selected_mode == "pushy": + elif selected_mode in {"pushy", "trigger"}: service_ready = ensure_pushy_service() else: service_ready = ensure_output_service() @@ -386,7 +446,15 @@ def main(argv: list[str] | None = None) -> int: print("tamq service failed to start with the required delivery capability; use --no-service to open repos without messaging", file=sys.stderr) return 1 try: - endpoint = manager.ensure_plan(launch_plan, tap=tap_enabled) + if tap_enabled: + endpoint = manager.ensure_plan( + launch_plan, tap=True, tap_limits=limits + ) + wait_taps = getattr(manager, "wait_taps", None) + if wait_taps is not None: + wait_taps(endpoint) + else: + endpoint = manager.ensure_plan(launch_plan, tap=False) except (TmuxError, OSError) as exc: print(f"tamq: {exc}", file=sys.stderr) return 2 @@ -395,7 +463,7 @@ def main(argv: list[str] | None = None) -> int: if not args.no_service: try: delivery_mode = "pane" if args.tap else endpoint_delivery_mode - 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})) + 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, "maxmsg": limits[0], "maxin": limits[1], "maxout": limits[2]})) if not registration.get("ok"): raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}") registered = True @@ -427,6 +495,7 @@ def main(argv: list[str] | None = None) -> int: "delivery_mode": "none" if args.no_service else ( "pane" if args.tap else endpoint_delivery_mode ), + "limits": {"messages": limits[0], "input": limits[1], "output": limits[2]}, } print(json.dumps(summary), flush=True) if args.detach: @@ -435,53 +504,17 @@ 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", "inbox", "send", "reply"}: + if advisory and args.command in {"start", "serve", "status", "history", "inbox", "send"}: print(advisory, file=sys.stderr) - if args.command in {"send", "reply"}: + if args.command == "send": endpoint_id = None - if args.command == "reply": - sender = os.environ.get("TAMQ_REPO") - if not sender: - print("tamq reply requires a managed window with TAMQ_REPO", file=sys.stderr) - return 2 - latest = store.latest_counterparty(sender) - if args.body: - body = " ".join(args.body).strip() - target = latest - if target is None: - print(f"tamq: no counterparty has sent a message to {sender}", file=sys.stderr) - return 2 - else: - try: - encoded_recipients = json.loads(os.environ.get("TAMQ_RECIPIENTS", "[]")) - except json.JSONDecodeError: - encoded_recipients = [] - session_repos = ( - encoded_recipients - if isinstance(encoded_recipients, list) - and all(isinstance(repo, str) for repo in encoded_recipients) - else [] - ) - recipients = recipient_order(sender, latest, session_repos) - try: - composed = compose_message(recipients) - except ComposerError as exc: - print(f"tamq: {exc}", file=sys.stderr) - return 2 - if composed is None: - return 130 - target, body = composed - else: - 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 - sender = args.sender_repo or os.environ.get("TAMQ_REPO") or "local" - endpoint_id = args.endpoint_id - if not body: - print("tamq reply requires a message body", file=sys.stderr) - return 2 + text = " ".join([args.address, *args.body]).strip() + if not text.startswith("To:") or text.count(":") < 2: + print("tamq send expects To:repo: message", file=sys.stderr); return 2 + target, body = text[3:].split(":", 1); body = body.strip() + if not target or not body: print("tamq send requires a target and body", file=sys.stderr); return 2 + sender = args.sender_repo or os.environ.get("TAMQ_REPO") or "local" + endpoint_id = args.endpoint_id try: validate_targets([target]) if sender != "local": @@ -489,7 +522,7 @@ def main(argv: list[str] | None = None) -> int: 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} + payload = {"op": "send", "sender_repo": sender, "target_repo": target, "body": body, "provenance": "operator_input"} if endpoint_id: payload["endpoint_id"] = endpoint_id response = asyncio.run(request(payload)) @@ -499,7 +532,7 @@ def main(argv: list[str] | None = None) -> int: if endpoint_id: print("tamq: service is not running", file=sys.stderr); return 1 try: - print(store.add(sender, target, body)) + print(store.add(sender, target, body, provenance="operator_input")) except ValueError as exc: print(f"tamq: {exc}", file=sys.stderr); return 2 return 0 @@ -534,7 +567,7 @@ def main(argv: list[str] | None = None) -> int: result = subprocess.run( args.filter_command, shell=True, - input=format_comment_message(row) + "\n", + input=format_inbox_message(row) + "\n", text=True, env=environment, check=False, @@ -547,7 +580,7 @@ def main(argv: list[str] | None = None) -> int: return 1 store.acknowledge(row["message_id"]) else: - print(format_comment_message(row)) + print(format_inbox_message(row)) return 0 if args.command == "history": for row in store.list(args.target_repo, args.state): print(json.dumps(dict(row), sort_keys=True)) @@ -618,7 +651,7 @@ def main(argv: list[str] | None = None) -> int: 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(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", []), "window_counters": [dict(row) for row in store.counters()]}, sort_keys=True)); return 0 print(f"tamq {__version__}: command '{args.command}' is not implemented yet", file=sys.stderr) return 2 finally: diff --git a/src/tamq/composer.py b/src/tamq/composer.py deleted file mode 100644 index 0c0c07f..0000000 --- a/src/tamq/composer.py +++ /dev/null @@ -1,131 +0,0 @@ -from __future__ import annotations - -import os -import re -import sys -import termios -import tty -from collections.abc import Sequence - - -class ComposerError(RuntimeError): - pass - - -def recipient_order( - current_repo: str, - latest_counterparty: str | None, - session_repos: Sequence[str], -) -> tuple[str, ...]: - """Put the latest sender first, then the remaining peer repositories.""" - ordered: list[str] = [] - for repo in ([latest_counterparty] if latest_counterparty else []): - if ( - repo != current_repo - and repo not in ordered - and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is not None - ): - ordered.append(repo) - for repo in session_repos: - if ( - repo != current_repo - and repo not in ordered - and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is not None - ): - ordered.append(repo) - return tuple(ordered) - - -def _remove_last_character(value: bytearray) -> None: - if not value: - return - removed = value.pop() - if removed & 0xC0 == 0x80: - while value and value[-1] & 0xC0 == 0x80: - value.pop() - if value and value[-1] & 0xC0 == 0xC0: - value.pop() - - -def compose_message( - recipients: Sequence[str], - *, - input_fd: int | None = None, - output_fd: int | None = None, -) -> tuple[str, str] | None: - """Read one literal terminal line while Tab cycles the visible recipient.""" - if not recipients: - raise ComposerError("no recipient is available") - try: - source = sys.stdin.fileno() if input_fd is None else input_fd - destination = sys.stderr.fileno() if output_fd is None else output_fd - except (AttributeError, OSError) as exc: - raise ComposerError("interactive composition requires a terminal") from exc - if not os.isatty(source) or not os.isatty(destination): - raise ComposerError("interactive composition requires a terminal") - - saved = termios.tcgetattr(source) - draft = bytearray() - recipient_index = 0 - escape_state = 0 - - def write(value: bytes) -> None: - offset = 0 - while offset < len(value): - offset += os.write(destination, value[offset:]) - - def render() -> None: - prompt = f"@{recipients[recipient_index]}: ".encode("utf-8") - write(b"\r\x1b[2K" + prompt + draft) - - try: - tty.setraw(source) - render() - while True: - value = os.read(source, 1) - if not value: - write(b"\r\n") - return None - byte = value[0] - if escape_state == 1: - escape_state = 2 if byte in (ord("["), ord("O")) else 0 - continue - if escape_state == 2: - if 0x40 <= byte <= 0x7E: - escape_state = 0 - continue - if byte == 0x1B: - escape_state = 1 - elif byte == 0x09: - recipient_index = (recipient_index + 1) % len(recipients) - render() - elif byte in (0x0A, 0x0D): - try: - body = draft.decode("utf-8") - except UnicodeDecodeError: - write(b"\a") - continue - if not body.strip(): - write(b"\a") - continue - write(b"\r\n") - return recipients[recipient_index], body - elif byte == 0x03: - write(b"^C\r\n") - return None - elif byte == 0x04 and not draft: - write(b"\r\n") - return None - elif byte in (0x08, 0x7F): - _remove_last_character(draft) - render() - elif byte == 0x15: - draft.clear() - render() - elif byte >= 0x20: - draft.extend(value) - write(value) - else: - write(b"\a") - finally: - termios.tcsetattr(source, termios.TCSADRAIN, saved) diff --git a/src/tamq/control.py b/src/tamq/control.py index c06503a..b8287d0 100644 --- a/src/tamq/control.py +++ b/src/tamq/control.py @@ -111,6 +111,19 @@ class ControlModeClient: result.stderr.strip() or f"cannot submit pane input: {window}" ) + def place(self, window: str, text: str) -> None: + """Place literal text in pane input without submitting it.""" + result = subprocess.run( + [*self._command(), "send-keys", "-t", window, "-l", "--", text], + text=True, + capture_output=True, + check=False, + ) + if result.returncode: + raise ControlModeError( + result.stderr.strip() or f"cannot place pane input: {window}" + ) + def close(self) -> None: if self.process is not None: process = self.process diff --git a/src/tamq/ptytap.py b/src/tamq/ptytap.py index d89c299..c378d9a 100644 --- a/src/tamq/ptytap.py +++ b/src/tamq/ptytap.py @@ -9,9 +9,14 @@ import signal import sys import termios import tty +import codecs +import time +from collections import deque from collections.abc import Callable, Sequence +from pathlib import Path from .broker import InputBroker +from .terminal import terminal_safe ENHANCED_ENTER = re.compile( @@ -43,13 +48,99 @@ def observed_line(raw: bytes) -> str: return "".join(text) +class TerminalOutputObserver: + """Normalize PTY output into logical lines without interpreting its content.""" + + _CURSOR_BOUNDARIES = frozenset("ABCDEFGHfJKd") + + def __init__(self, on_line: Callable[[str], None], *, echo_seconds: float = 5.0): + self.on_line = on_line + self.echo_seconds = echo_seconds + self.decoder = codecs.getincrementaldecoder("utf-8")("replace") + self.text: list[str] = [] + self.state = "normal" + self.string_kind = "" + self.operator_lines: deque[tuple[str, float]] = deque(maxlen=32) + self.last_redraw: str | None = None + + def note_operator_line(self, line: str) -> None: + self.operator_lines.append((line, time.monotonic() + self.echo_seconds)) + + def _is_recent_operator_echo(self, line: str) -> bool: + now = time.monotonic() + while self.operator_lines and self.operator_lines[0][1] < now: + self.operator_lines.popleft() + return any(candidate == line for candidate, _ in self.operator_lines) + + def _emit(self, *, redraw: bool = False) -> None: + line = "".join(self.text) + self.text.clear() + if not line: + return + if self._is_recent_operator_echo(line): + return + if redraw and line == self.last_redraw: + return + self.last_redraw = line if redraw else None + self.on_line(line) + + def feed(self, data: bytes) -> None: + for character in self.decoder.decode(data): + if self.state == "normal": + if character == "\x1b": + self.state = "esc" + elif character in "\r\n": + self._emit(redraw=character == "\r") + elif character in ("\b", "\x7f"): + if self.text: + self.text.pop() + elif character == "\t" or character >= " ": + self.text.append(character) + elif self.state == "esc": + if character == "[": + self.state = "csi" + elif character in "]P_^": + self.state = "string" + self.string_kind = character + else: + self.state = "normal" + elif self.state == "csi": + if "@" <= character <= "~": + if character in self._CURSOR_BOUNDARIES: + self._emit(redraw=True) + self.state = "normal" + elif self.state == "string": + if character == "\x07" and self.string_kind == "]": + self.state = "normal" + elif character == "\x1b": + self.state = "string_esc" + elif self.state == "string_esc": + self.state = "normal" if character == "\\" else "string" + + def flush(self) -> None: + self.decoder.decode(b"", final=True) + self._emit() + + 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): + def __init__(self, command: Sequence[str], broker: InputBroker, *, on_line: Callable[[str], None] | None = None, ready_file: Path | None = None): self.command = list(command) self.broker = broker self.on_line = on_line + self.ready_file = ready_file + self.output_observer = TerminalOutputObserver(self._observe_worker_line) + + def _observe_worker_line(self, line: str) -> None: + inspector = getattr(self.broker, "inspect_worker_line", None) + routed = inspector(line) if inspector is not None else None + if routed and self.on_line: + self.on_line(line) + + def _local_notice(self, line: str) -> None: + payload = f"\x1b7\r\n{terminal_safe(line)}\r\n\x1b8".encode("utf-8") + write_all(sys.stdout.fileno(), payload) def _observe_input(self, buffer: bytearray, data: bytes) -> None: buffer.extend(data) @@ -67,7 +158,9 @@ class PtyTap: raw = bytes(buffer[:start]) del buffer[:end] line = observed_line(raw) - routed = self.broker.inspect_line(line) + self.output_observer.note_operator_line(line) + inspector = getattr(self.broker, "inspect_operator_line", self.broker.inspect_line) + routed = inspector(line) if routed and self.on_line: self.on_line(line) @@ -107,6 +200,10 @@ class PtyTap: status = 0 try: + self.broker.notify = self._local_notice + if self.ready_file is not None: + self.ready_file.parent.mkdir(parents=True, exist_ok=True) + self.ready_file.write_text(str(os.getpid()), encoding="utf-8") if saved_terminal is not None: tty.setraw(stdin_fd) saved_handlers[signal.SIGWINCH] = signal.signal(signal.SIGWINCH, resize) @@ -129,6 +226,7 @@ class PtyTap: if not data: break write_all(stdout_fd, data) + self.output_observer.feed(data) if input_open and stdin_fd in readable: data = os.read(stdin_fd, 65536) if not data: @@ -147,6 +245,9 @@ class PtyTap: pass raise finally: + if self.ready_file is not None: + self.ready_file.unlink(missing_ok=True) + self.output_observer.flush() for signum, handler in saved_handlers.items(): signal.signal(signum, handler) if saved_terminal is not None: diff --git a/src/tamq/routing.py b/src/tamq/routing.py index d8e25cb..aed9867 100644 --- a/src/tamq/routing.py +++ b/src/tamq/routing.py @@ -3,7 +3,8 @@ from __future__ import annotations import re from dataclasses import dataclass -ADDRESS = re.compile(r"^[@#]([a-z0-9][a-z0-9._-]*):(?:[ \t]*)(.+)$", re.IGNORECASE) +ADDRESS = re.compile(r"^To:([a-z0-9][a-z0-9._-]*):[ \t]*(.+)$") +COMMAND = re.compile(r"^Cmd:[ \t]*(\S(?:.*\S)?)?[ \t]*$") @dataclass(frozen=True) @@ -18,3 +19,11 @@ def parse_address_line(line: str) -> RoutedMessage | None: if not match: return None return RoutedMessage(target_repo=match.group(1), body=match.group(2)) + + +def parse_command_line(line: str) -> str | None: + """Return an operator command body; worker output must never call this.""" + match = COMMAND.match(line.rstrip("\r\n")) + if not match or not match.group(1): + return None + return match.group(1) diff --git a/src/tamq/service.py b/src/tamq/service.py index f2dca71..658de09 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -10,13 +10,15 @@ from pathlib import Path from .config import socket_path, pid_path, setting from .config import db_path -from .store import Store +from .store import LimitBlock, Store from .registry import RegistryError, validate_targets from .control import ControlModeClient -from .terminal import format_pushy_input, terminal_frame, write_terminal_output +from .terminal import format_delivery, terminal_frame, write_terminal_output PROTOCOL_VERSION = "0.1" -PUSHY_FRAMING_CAPABILITY = "pushy_input_non_routable_v1" +PUSHY_FRAMING_CAPABILITY = "readable_duplex_framing_v1" +TRIGGER_CAPABILITY = "trigger_input_v1" +LINE_LIMITS_CAPABILITY = "session_window_line_limits_v1" SERVICE_CAPABILITIES = [ "register", "send", @@ -25,6 +27,8 @@ SERVICE_CAPABILITIES = [ "terminal_output", "pushy_input", PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, + LINE_LIMITS_CAPABILITY, ] @@ -116,23 +120,37 @@ class Service: terminal_frame( row["sender_repo"], row["body"], - row["message_id"], + row["provenance"], cursor_y=display.cursor_y, pane_height=display.pane_height, alternate_on=display.alternate_on, ), ) elif delivery_mode == "pushy": - control.submit( + control.place( target, - format_pushy_input( + format_delivery( row["sender_repo"], row["body"], - row["message_id"], + row["provenance"], + ), + ) + elif delivery_mode == "trigger": + control.submit( + target, + format_delivery( + row["sender_repo"], + row["body"], + row["provenance"], ), ) else: - control.inject(target, f'#{row["sender_repo"]}: {row["body"]}') + control.inject( + target, + format_delivery( + row["sender_repo"], row["body"], row["provenance"] + ), + ) except Exception: continue if delivery_mode == "output": @@ -175,6 +193,11 @@ class Service: request["session"], list(request["repos"]), delivery_mode, + ( + int(request.get("maxmsg", 8)), + int(request.get("maxin", 1024)), + int(request.get("maxout", 32768)), + ), ) response = {"ok": True, "endpoint_id": request["endpoint_id"], "instance_id": endpoint_id, "protocol": PROTOCOL_VERSION} except (RegistryError, ValueError) as exc: @@ -197,8 +220,35 @@ class Service: 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"} + provenance = request.get("provenance", "operator_input") + if endpoint_id and request["sender_repo"] != "local": + result = self.store.admit_message( + request["sender_repo"], + request["target_repo"], + request["body"], + endpoint=endpoint_id, + source_repo=request["sender_repo"], + provenance=provenance, + ) + if isinstance(result, LimitBlock): + response = { + "ok": False, + "error": "Messaging blocked! Running linecount " + f"{result.count} limited by {result.limit} lines of {result.dimension} " + "in this terminal. Use 'Cmd: reset-limits' to unblock.", + "blocked": result.dimension, + } + else: + response = {"ok": True, "message_id": result, "state": "pending"} + else: + message_id = self.store.add( + request["sender_repo"], + request["target_repo"], + request["body"], + endpoint=endpoint_id, + provenance=provenance, + ) + response = {"ok": True, "message_id": message_id, "state": "pending"} except (RegistryError, ValueError) as exc: response = {"ok": False, "error": str(exc)} elif op == "history": diff --git a/src/tamq/store.py b/src/tamq/store.py index 86ab7f9..1e93d03 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -4,17 +4,25 @@ import json import re import sqlite3 import time +from dataclasses import dataclass from pathlib import Path from typing import Iterable from uuid import uuid4 -SCHEMA_VERSION = 3 +SCHEMA_VERSION = 4 RECEIPT_SUFFIX = re.compile( r"\s+\[(m-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\]$", re.IGNORECASE, ) +@dataclass(frozen=True) +class LimitBlock: + dimension: str + count: int + limit: int + + class Store: def __init__(self, path: Path): path.parent.mkdir(parents=True, exist_ok=True) @@ -48,6 +56,18 @@ class Store: connected_at REAL NOT NULL, disconnected_at REAL ); + CREATE TABLE IF NOT EXISTS window_counters ( + endpoint_id TEXT NOT NULL, + repo TEXT NOT NULL, + messages INTEGER NOT NULL DEFAULT 0, + input INTEGER NOT NULL DEFAULT 0, + output INTEGER NOT NULL DEFAULT 0, + maxmsg INTEGER NOT NULL DEFAULT 8, + maxin INTEGER NOT NULL DEFAULT 1024, + maxout INTEGER NOT NULL DEFAULT 32768, + updated_at REAL NOT NULL, + PRIMARY KEY(endpoint_id, repo) + ); CREATE TABLE IF NOT EXISTS leases ( message_id TEXT PRIMARY KEY REFERENCES messages(message_id) ON DELETE CASCADE, lease_id TEXT NOT NULL, @@ -85,10 +105,12 @@ class Store: session: str, repos: list[str], delivery_mode: str = "manual", + limits: tuple[int, int, int] = (8, 1024, 32768), ) -> None: import json - if delivery_mode not in {"manual", "output", "pane", "pushy"}: + if delivery_mode not in {"manual", "output", "pane", "pushy", "trigger"}: raise ValueError(f"invalid delivery mode: {delivery_mode}") + self._validate_limits(limits) self.db.execute( "UPDATE endpoints SET disconnected_at=strftime('%s','now') " "WHERE pid=? AND session=? AND endpoint_id<>? AND disconnected_at IS NULL", @@ -99,8 +121,113 @@ class Store: "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), ) + for repo in repos: + self.configure_window(endpoint_id, repo, limits, commit=False) self.db.commit() + @staticmethod + def _validate_limits(limits: tuple[int, int, int]) -> None: + if any(isinstance(value, bool) or value <= 0 for value in limits): + raise ValueError("message, input, and output limits must be positive integers") + + def configure_window( + self, + endpoint_id: str, + repo: str, + limits: tuple[int, int, int] = (8, 1024, 32768), + *, + commit: bool = True, + ) -> None: + """Create a session-window ledger or update its limits without resetting it.""" + self._validate_limits(limits) + self.db.execute( + "INSERT INTO window_counters(endpoint_id,repo,maxmsg,maxin,maxout,updated_at) " + "VALUES(?,?,?,?,?,?) ON CONFLICT(endpoint_id,repo) DO UPDATE SET " + "maxmsg=excluded.maxmsg,maxin=excluded.maxin,maxout=excluded.maxout,updated_at=excluded.updated_at", + (endpoint_id, repo, *limits, time.time()), + ) + if commit: + self.db.commit() + + def line_state(self, endpoint_id: str, repo: str) -> sqlite3.Row: + row = self.db.execute( + "SELECT * FROM window_counters WHERE endpoint_id=? AND repo=?", + (endpoint_id, repo), + ).fetchone() + if row is None: + self.configure_window(endpoint_id, repo) + row = self.db.execute( + "SELECT * FROM window_counters WHERE endpoint_id=? AND repo=?", + (endpoint_id, repo), + ).fetchone() + assert row is not None + return row + + def counters(self, endpoint_id: str | None = None) -> list[sqlite3.Row]: + if endpoint_id is None: + return list( + self.db.execute( + "SELECT * FROM window_counters ORDER BY endpoint_id,repo" + ) + ) + return list( + self.db.execute( + "SELECT * FROM window_counters WHERE endpoint_id=? ORDER BY repo", + (endpoint_id,), + ) + ) + + def count_line(self, endpoint_id: str, repo: str, direction: str) -> int: + if direction not in {"input", "output"}: + raise ValueError(f"invalid line direction: {direction}") + with self.db: + self.db.execute( + "INSERT INTO window_counters(endpoint_id,repo,updated_at) VALUES(?,?,?) " + "ON CONFLICT(endpoint_id,repo) DO NOTHING", + (endpoint_id, repo, time.time()), + ) + self.db.execute( + f"UPDATE window_counters SET {direction}={direction}+1,updated_at=? " + "WHERE endpoint_id=? AND repo=?", + (time.time(), endpoint_id, repo), + ) + return int(self.line_state(endpoint_id, repo)[direction]) + + def reset_limits(self, endpoint_id: str, repo: str) -> None: + with self.db: + self.db.execute( + "INSERT INTO window_counters(endpoint_id,repo,updated_at) VALUES(?,?,?) " + "ON CONFLICT(endpoint_id,repo) DO UPDATE SET messages=0,input=0,output=0,updated_at=excluded.updated_at", + (endpoint_id, repo, time.time()), + ) + + def set_window_limit(self, endpoint_id: str, repo: str, name: str, value: int) -> None: + column = {"maxmsg": "maxmsg", "maxin": "maxin", "maxout": "maxout"}.get(name) + if column is None: + raise ValueError(f"unknown limit: {name}") + self._validate_limits((value, value, value)) + with self.db: + self.db.execute( + "INSERT INTO window_counters(endpoint_id,repo,updated_at) VALUES(?,?,?) " + "ON CONFLICT(endpoint_id,repo) DO NOTHING", + (endpoint_id, repo, time.time()), + ) + self.db.execute( + f"UPDATE window_counters SET {column}=?,updated_at=? WHERE endpoint_id=? AND repo=?", + (value, time.time(), endpoint_id, repo), + ) + + def set_endpoint_mode(self, endpoint_id: str, mode: str) -> bool: + delivery_mode = "manual" if mode == "inbox" else mode + if delivery_mode not in {"manual", "output", "pushy", "trigger"}: + raise ValueError(f"invalid delivery mode: {mode}") + with self.db: + result = self.db.execute( + "UPDATE endpoints SET delivery_mode=? WHERE endpoint_id=? AND disconnected_at IS NULL", + (delivery_mode, endpoint_id), + ) + return result.rowcount == 1 + 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() @@ -119,6 +246,7 @@ class Store: ) with self.db: self.db.execute("DELETE FROM leases") + self.db.execute("DELETE FROM window_counters") self.db.execute( "UPDATE endpoints SET disconnected_at=strftime('%s','now') " "WHERE disconnected_at IS NULL" @@ -165,6 +293,55 @@ class Store: self.db.commit() return message_id + def admit_message( + self, + sender: str, + target: str, + body: str, + *, + endpoint: str, + source_repo: str, + provenance: str, + ) -> str | LimitBlock: + """Atomically check source-window budgets and create one message.""" + if len(body.encode("utf-8")) > 8192: + raise ValueError("message body exceeds 8 KiB limit") + self.db.execute("BEGIN IMMEDIATE") + try: + self.db.execute( + "INSERT INTO window_counters(endpoint_id,repo,updated_at) VALUES(?,?,?) " + "ON CONFLICT(endpoint_id,repo) DO NOTHING", + (endpoint, source_repo, time.time()), + ) + state = self.db.execute( + "SELECT * FROM window_counters WHERE endpoint_id=? AND repo=?", + (endpoint, source_repo), + ).fetchone() + assert state is not None + for count_name, limit_name, label in ( + ("messages", "maxmsg", "messages"), + ("input", "maxin", "input"), + ("output", "maxout", "output"), + ): + if int(state[count_name]) >= int(state[limit_name]): + self.db.rollback() + return LimitBlock(label, int(state[count_name]), int(state[limit_name])) + message_id = f"m-{uuid4()}" + self.db.execute( + "INSERT INTO messages(message_id,sender_repo,target_repo,body,created_at,state,endpoint_id,provenance,injected_at,acknowledged_at,displayed_at) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?)", + (message_id, sender, target, body, time.time(), "pending", endpoint, provenance, None, None, None), + ) + self.db.execute( + "UPDATE window_counters SET messages=messages+1,updated_at=? WHERE endpoint_id=? AND repo=?", + (time.time(), endpoint, source_repo), + ) + self.db.commit() + return message_id + except BaseException: + self.db.rollback() + raise + def list(self, target: str | None = None, state: str | None = None) -> list[sqlite3.Row]: clauses, values = [], [] if target: @@ -219,15 +396,6 @@ class Store: ) return int(count) - def latest_counterparty(self, target: str) -> str | None: - row = self.db.execute( - "SELECT sender_repo FROM messages " - "WHERE target_repo=? AND sender_repo<>? " - "ORDER BY created_at DESC, rowid DESC LIMIT 1", - (target, target), - ).fetchone() - return None if row is None else str(row["sender_repo"]) - def set_state(self, message_id: str, state: str) -> None: column = {"injected": "injected_at", "acknowledged": "acknowledged_at"}.get(state) if column: diff --git a/src/tamq/terminal.py b/src/tamq/terminal.py index 95a0026..0746a5f 100644 --- a/src/tamq/terminal.py +++ b/src/tamq/terminal.py @@ -20,34 +20,23 @@ def terminal_safe(text: str) -> str: ) -def format_comment(sender: str, body: str, message_id: str) -> str: - """Render a message so every line is visibly comment-prefixed.""" - lines = body.splitlines() or [""] - rendered = [f"#{terminal_safe(sender)}: {terminal_safe(lines[0])}"] - rendered.extend(f"# {terminal_safe(line)}" for line in lines[1:]) - rendered[-1] += f" [{terminal_safe(message_id)}]" - return "\n".join(rendered) - - -def format_pushy_input(sender: str, body: str, message_id: str) -> str: - """Render one shell-safe, non-routable line for pushy delivery.""" - return ( - f"# from {terminal_safe(sender)}: {terminal_safe(body)} " - f"[{terminal_safe(message_id)}]" - ) +def format_delivery(sender: str, body: str, provenance: str | None = None) -> str: + """Render one readable, non-routable delivery line.""" + suffix = "/o" if provenance == "operator_input" else "" + return f"From:{terminal_safe(sender)}{suffix}: {terminal_safe(body)}" def terminal_frame( sender: str, body: str, - message_id: str, + provenance: str | None = None, *, cursor_y: int | None = None, pane_height: int | None = None, alternate_on: bool = False, ) -> str: """Frame output above the cursor when a safe scroll region is available.""" - content = format_comment(sender, body, message_id).replace("\n", "\r\n") + content = format_delivery(sender, body, provenance).replace("\n", "\r\n") line_count = content.count("\r\n") + 1 if ( cursor_y is not None diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py index f8424fe..37c4967 100644 --- a/src/tamq/tmux.py +++ b/src/tamq/tmux.py @@ -1,12 +1,12 @@ from __future__ import annotations -import json import os import re import shlex import shutil from uuid import uuid4 import subprocess +import time from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path @@ -75,32 +75,30 @@ class TmuxManager: self.command_dir = state_dir() / "commands" / session def _install_address_commands(self, repos: Sequence[str]) -> None: - """Install tamq-owned commands without modifying a user's shell files.""" + """Install shell absorbers while the PTY tap performs protocol routing.""" self.command_dir.mkdir(parents=True, exist_ok=True, mode=0o700) if not os.access(self.command_dir, os.W_OK | os.X_OK): raise TmuxError(f"address command directory is not writable: {self.command_dir}") - scripts: dict[str, str] = {} + for existing in self.command_dir.iterdir(): + if not existing.is_file() or existing.is_symlink(): + continue + try: + content = existing.read_text(encoding="utf-8") + except OSError: + continue + if "# tamq-address-command v1" in content: + existing.unlink() + scripts: dict[str, str] = { + "Cmd:": "#!/bin/sh\n# tamq-protocol-command v2\nexit 0\n", + } for repo in repos: if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", repo) is None: raise TmuxError( f"repository name cannot be exposed as a shell command: {repo!r}" ) - target = f"@{repo}:" - script = ( - "#!/bin/sh\n" - "# tamq-address-command v1\n" - f"exec {shell_join(list(self.tamq_command))} send -- {shlex.quote(target)} \"$@\"\n" + scripts[f"To:{repo}:"] = ( + "#!/bin/sh\n# tamq-protocol-command v2\nexit 0\n" ) - for name in (f"@{repo}", target): - scripts[name] = script - recipient_data = shlex.quote(json.dumps(list(repos), separators=(",", ":"))) - scripts["@"] = ( - "#!/bin/sh\n" - "# tamq-address-command v1\n" - f"TAMQ_RECIPIENTS={recipient_data}\n" - "export TAMQ_RECIPIENTS\n" - f"exec {shell_join(list(self.tamq_command))} reply -- \"$@\"\n" - ) for name in scripts: destination = self.command_dir / name if destination.exists() or destination.is_symlink(): @@ -110,7 +108,7 @@ class TmuxManager: existing = destination.read_text(encoding="utf-8") except OSError as exc: raise TmuxError(f"cannot inspect existing command: {destination}") from exc - if "# tamq-address-command v1" not in existing: + if "# tamq-protocol-command v2" not in existing: raise TmuxError(f"refusing to replace existing command: {destination}") for name, script in scripts.items(): destination = self.command_dir / name @@ -124,6 +122,25 @@ class TmuxManager: command_path = f"{self.command_dir}{os.pathsep}{path}" if path else str(self.command_dir) return "-e", f"TAMQ_REPO={repo}", "-e", f"PATH={command_path}" + def _tap_ready_file(self, instance_id: str, repo: str) -> Path: + return state_dir() / "tap-ready" / f"{instance_id}--{repo}" + + def wait_taps(self, endpoint: Endpoint, timeout: float = 3.0) -> None: + """Wait until every newly launched tap owns the pane input path.""" + pending = { + repo: self._tap_ready_file(endpoint.instance_key, repo) + for repo in endpoint.created_windows + } + deadline = time.monotonic() + timeout + while pending and time.monotonic() < deadline: + pending = {repo: path for repo, path in pending.items() if not path.exists()} + if pending: + time.sleep(0.01) + if pending: + raise TmuxError( + "PTY tap did not become ready for: " + ", ".join(sorted(pending)) + ) + def _run(self, *args: str, check: bool = True) -> str: result = subprocess.run([*self.tmux_command, *args], text=True, capture_output=True, check=False) if check and result.returncode: @@ -192,13 +209,27 @@ class TmuxManager: 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 = False) -> Endpoint: - if tap and not plan.command: - raise TmuxError("PTY tap requires an explicit initial command") + def ensure_plan( + self, + plan: LaunchPlan, + *, + tap: bool = False, + tap_limits: tuple[int, int, int] = (8, 1024, 32768), + ) -> Endpoint: created_session = False created_windows: list[str] = [] try: identity = self._existing_session_identity() + if identity is not None and tap: + duplex_protocol = self._run( + "show-options", "-v", "-t", self.session, + "@tamq_duplex_protocol", check=False, + ) + if duplex_protocol != "1": + raise TmuxError( + "existing managed session predates readable duplex messaging; " + "run 'tamq cleanup --yes' before recreating it" + ) if identity is None: first_repo = plan.repos[0] self._run( @@ -220,6 +251,8 @@ class TmuxManager: instance_id = f"{endpoint_id}-{uuid4().hex[:12]}" self._run("set-option", "-t", self.session, "@tamq_instance_id", instance_id) self._run("set-option", "-t", self.session, "@tamq_managed", "1") + if tap: + self._run("set-option", "-t", self.session, "@tamq_duplex_protocol", "1") windows = self._run("list-windows", "-t", self.session, "-F", "#{window_name}").splitlines() addressable_repos = [repo for repo in windows if repo != "__tamq_boot"] addressable_repos.extend(repo for repo in plan.repos if repo not in addressable_repos) @@ -242,14 +275,28 @@ class TmuxManager: ) windows.append(repo) created_windows.append(repo) - if plan.command: + if plan.command or tap: + wrapped_command = list(plan.command) + if not wrapped_command: + wrapped_command = [ + os.environ.get("SHELL") + or shutil.which("bash") + or "/bin/sh" + ] + ready_file = self._tap_ready_file(instance_id, repo) + ready_file.unlink(missing_ok=True) command = ( [ - *self.tamq_command, - "tap", "--repo", repo, "--endpoint", instance_id, "--", *plan.command, + *self.tamq_command, "tap", "--repo", repo, + "--endpoint", instance_id, + "--maxmsg", str(tap_limits[0]), + "--maxin", str(tap_limits[1]), + "--maxout", str(tap_limits[2]), + "--ready-file", str(ready_file), + "--", *wrapped_command, ] if tap - else list(plan.command) + else wrapped_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]}") diff --git a/tests/test_broker.py b/tests/test_broker.py index ae955b3..5f2f08e 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -2,53 +2,69 @@ from tamq.broker import BrokerIdentity, InputBroker from tamq.store import Store -def test_broker_preserves_identity(tmp_path): +def test_broker_preserves_operator_identity_and_provenance(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) 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 + assert broker.inspect_operator_line("To:railiance-platform: hello") is not None row = store.list()[0] assert row["sender_repo"] == "net-kingdom" assert row["endpoint_id"] == "tmux-amq-42" + assert row["provenance"] == "operator_input" -def test_broker_accepts_hash_address_alias(tmp_path): +def test_broker_accepts_worker_output_with_distinct_provenance(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) store = Store(tmp_path / "queue.sqlite3") - broker = InputBroker(store, BrokerIdentity("tmux-amq-42-boot", "net-kingdom")) - assert broker.inspect_line("#railiance-platform: hello") is not None - row = store.list()[0] - assert row["sender_repo"] == "net-kingdom" - assert row["target_repo"] == "railiance-platform" - assert row["endpoint_id"] == "tmux-amq-42-boot" + broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom")) + assert broker.inspect_worker_line("To:railiance-platform: hello") is not None + assert store.list()[0]["provenance"] == "worker_output" -def test_broker_rejects_legacy_injected_envelope_feedback(tmp_path): +def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path): store = Store(tmp_path / "queue.sqlite3") - message_id = store.add( - "flex-auth", - "audit-core", - "Hello worker agent!", - endpoint="tmux-amq-42-boot", - ) + store.register_endpoint("ep", 9, "tamq", ["repo"], "output") + notices = [] broker = InputBroker( - store, BrokerIdentity("tmux-amq-42-boot", "audit-core") + store, BrokerIdentity("ep", "repo"), notify=notices.append ) - - assert ( - broker.inspect_line( - f"#flex-auth: Hello worker agent! [{message_id}]" - ) - is None - ) - assert len(store.list()) == 1 + broker.inspect_worker_line("Cmd: mode=trigger") + assert store.endpoint("ep")["delivery_mode"] == "output" + broker.inspect_operator_line("Cmd: mode=trigger") + assert store.endpoint("ep")["delivery_mode"] == "trigger" + assert notices[-1] == "From:tamq: Mode set to trigger." -def test_broker_does_not_block_unknown_receipt_like_user_text(tmp_path): +def test_limits_block_at_equality_and_reset_current_window(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) store = Store(tmp_path / "queue.sqlite3") + notices = [] broker = InputBroker( - store, BrokerIdentity("tmux-amq-42-boot", "audit-core") - ) - - assert broker.inspect_line( - "#flex-auth: Please inspect [m-00000000-0000-0000-0000-000000000000]" + store, + BrokerIdentity("ep", "repo"), + limits=(1, 10, 10), + notify=notices.append, ) + assert broker.inspect_operator_line("To:target: first") is not None + assert broker.inspect_operator_line("To:target: blocked") is None + assert "Running linecount 1 limited by 1 lines of messages" in notices[-1] assert len(store.list()) == 1 + broker.inspect_operator_line("Cmd: reset-limits") + assert broker.inspect_operator_line("To:target: after reset") is not None + + +def test_input_limit_includes_the_current_operator_line(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) + store = Store(tmp_path / "queue.sqlite3") + notices = [] + broker = InputBroker( + store, + BrokerIdentity("ep", "repo"), + limits=(8, 1, 32768), + notify=notices.append, + ) + assert broker.inspect_operator_line("To:target: blocked") is None + assert notices[-1] == ( + "From:tamq: Messaging blocked! Running linecount 1 limited by 1 lines " + "of input in this terminal. Use 'Cmd: reset-limits' to unblock." + ) diff --git a/tests/test_broker_delivery.py b/tests/test_broker_delivery.py index f8e9d66..a9aaadf 100644 --- a/tests/test_broker_delivery.py +++ b/tests/test_broker_delivery.py @@ -16,6 +16,6 @@ def test_pending_delivery_marks_injected(tmp_path): 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 control.calls == [("tamq:railiance-platform", "From: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 index 7bccbff..596d635 100644 --- a/tests/test_broker_registry.py +++ b/tests/test_broker_registry.py @@ -6,5 +6,5 @@ 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 InputBroker(store, BrokerIdentity("ep", "source")).inspect_line("To:missing: hello") is None assert store.list() == [] diff --git a/tests/test_cleanup.py b/tests/test_cleanup.py index a2e1608..d955b17 100644 --- a/tests/test_cleanup.py +++ b/tests/test_cleanup.py @@ -42,10 +42,12 @@ def test_cleanup_is_dry_run_then_removes_only_owned_runtime( state, runtime = configure_paths(tmp_path, monkeypatch) command_dir = state / "commands" / "tamq" command_dir.mkdir(parents=True) - generated = command_dir / "@audit-core" + generated = command_dir / "To:audit-core:" generated.write_text(f"#!/bin/sh\n{SHIM_MARKER}\n", encoding="utf-8") unrelated = command_dir / "keep-me" unrelated.write_text("operator file\n", encoding="utf-8") + binary = command_dir / "binary" + binary.write_bytes(b"\xff\x00\x80") for name in ("tamq.sock", "tamq.pid", "tamq.lock"): (runtime / name).write_text("stale\n", encoding="utf-8") tmux_socket_dir = tmp_path / "tmux" / f"tmux-{os.getuid()}" @@ -96,6 +98,7 @@ def test_cleanup_is_dry_run_then_removes_only_owned_runtime( assert manager.identity is None assert not generated.exists() assert unrelated.exists() + assert binary.exists() assert not any((runtime / name).exists() for name in ("tamq.sock", "tamq.pid", "tamq.lock")) assert not stale_tmux_socket.exists() assert live_tmux_socket.exists() diff --git a/tests/test_cli.py b/tests/test_cli.py index e43b7d5..aeadc40 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -18,9 +18,11 @@ def test_root_help_exposes_repository_shorthand_and_command(capsys): main(["--help"]) assert exc.value.code == 0 output = capsys.readouterr().out - assert "tamq [--detach] [--command COMMAND] [--mode MODE] REPO" in output + assert "tamq [--detach] [--command COMMAND] [--mode MODE]" in output assert "With no --command" in output - assert "@TARGET: MESSAGE" in output + assert "To:TARGET: MESSAGE" in output + assert "Cmd: mode=trigger" in output + assert "[--maxmsg N] [--maxin N] [--maxout N]" in output def test_attach_delegates_to_tmux(monkeypatch): @@ -46,12 +48,15 @@ def test_start_parser_supports_command_and_cmd_alias(): neutral = parser.parse_args(["start", "a", "b"]) inbox_only = parser.parse_args(["start", "--no-display", "a"]) pushy = parser.parse_args(["start", "--mode", "pushy", "a"]) + trigger = parser.parse_args(["start", "--mode", "trigger", "--maxmsg", "3", "a"]) assert canonical.initial_command == "codex --quiet" assert canonical.repos == ["a", "b"] assert compatibility.initial_command == "claude" assert neutral.initial_command is None assert inbox_only.no_display is True assert pushy.mode == "pushy" + assert trigger.mode == "trigger" + assert trigger.maxmsg == 3 def test_repository_first_and_command_first_start_shorthand(): @@ -82,6 +87,9 @@ def test_repository_first_and_command_first_start_shorthand(): "--mode=pushy", "flex-auth", ] + assert normalize_argv(["--maxmsg=3", "flex-auth"]) == [ + "start", "--maxmsg=3", "flex-auth" + ] assert normalize_argv(["status"]) == ["status"] assert normalize_argv(["--verbose", "flex-auth", "audit-core"]) == [ "--verbose", @@ -173,8 +181,9 @@ def test_registration_failure_rolls_back_created_windows(monkeypatch, capsys): def preflight(self, repos, command): return "plan" - def ensure_plan(self, plan, *, tap=True): - assert tap is False + def ensure_plan(self, plan, *, tap=True, tap_limits=None): + assert tap is True + assert tap_limits == (8, 1024, 32768) return endpoint def rollback(self, value): @@ -212,8 +221,9 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys): assert command == "codex" return "plan" - def ensure_plan(self, plan, *, tap=True): + def ensure_plan(self, plan, *, tap=True, tap_limits=None): assert tap is True + assert tap_limits == (8, 1024, 32768) return endpoint def rollback(self, value): @@ -240,6 +250,7 @@ def test_pushy_mode_registers_explicit_delivery_mode(monkeypatch, capsys): assert summary["delivery_mode"] == "pushy" assert summary["input_observation_requested"] is True assert requests[0]["delivery_mode"] == "pushy" + assert requests[0]["maxmsg"] == 8 def test_tap_requires_explicit_command_and_service(capsys): @@ -255,3 +266,8 @@ def test_tap_requires_explicit_command_and_service(capsys): assert "--tap cannot be combined with --mode" in capsys.readouterr().err assert main(["start", "--no-service", "--mode", "pushy", "a"]) == 2 assert "--mode cannot be combined with --no-service" in capsys.readouterr().err + + +def test_start_rejects_nonpositive_line_limits(capsys): + assert main(["start", "--maxmsg", "0", "a"]) == 2 + assert "maxmsg must be a positive integer" in capsys.readouterr().err diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index bed0b09..6860182 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -4,7 +4,11 @@ from tamq.cli import ( ensure_pushy_service, parse_size, ) -from tamq.service import PUSHY_FRAMING_CAPABILITY +from tamq.service import ( + LINE_LIMITS_CAPABILITY, + PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, +) def test_parse_size(): @@ -13,7 +17,7 @@ def test_parse_size(): def test_manual_service_restarts_legacy_broker(monkeypatch): - capabilities = iter([[], ["manual_delivery"]]) + capabilities = iter([[], ["manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY]]) starts = [] stops = [] @@ -33,7 +37,7 @@ def test_manual_service_restarts_legacy_broker(monkeypatch): def test_output_service_requires_terminal_output_capability(monkeypatch): capabilities = iter([ ["manual_delivery"], - ["manual_delivery", "terminal_output"], + ["manual_delivery", "terminal_output", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY], ]) starts = [] stops = [] @@ -58,6 +62,8 @@ def test_pushy_service_requires_pushy_input_capability(monkeypatch): "terminal_output", "pushy_input", PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, + LINE_LIMITS_CAPABILITY, ], ]) starts = [] diff --git a/tests/test_composer.py b/tests/test_composer.py deleted file mode 100644 index 80a8d62..0000000 --- a/tests/test_composer.py +++ /dev/null @@ -1,101 +0,0 @@ -import os -import pty -import select -import termios -import threading -import time - -import pytest - -from tamq.composer import ComposerError, compose_message, recipient_order - - -def _read_until(master: int, marker: bytes, timeout: float = 2) -> bytes: - output = bytearray() - deadline = time.monotonic() + timeout - while marker not in output and time.monotonic() < deadline: - readable, _, _ = select.select([master], [], [], 0.05) - if readable: - output.extend(os.read(master, 4096)) - return bytes(output) - - -def _compose_in_pty(recipients, typed): - master, slave = pty.openpty() - original = termios.tcgetattr(slave) - result = {} - - def run(): - try: - result["value"] = compose_message( - recipients, input_fd=slave, output_fd=slave - ) - except Exception as exc: # surfaced in the test thread - result["error"] = exc - - thread = threading.Thread(target=run) - try: - thread.start() - output = bytearray(_read_until(master, b": ")) - os.write(master, typed) - thread.join(timeout=2) - assert not thread.is_alive() - output.extend(_read_until(master, b"\r\n", timeout=0.2)) - assert "error" not in result - assert termios.tcgetattr(slave) == original - return result["value"], bytes(output) - finally: - os.close(master) - os.close(slave) - - -def test_recipient_order_prefers_latest_then_session_peers(): - assert recipient_order( - "audit-core", - "flex-auth", - ["audit-core", "railiance-platform", "flex-auth", "railiance-platform"], - ) == ("flex-auth", "railiance-platform") - assert recipient_order( - "audit-core", None, ["audit-core", "railiance-platform"] - ) == ("railiance-platform",) - assert recipient_order("audit-core", "bad\x1btarget", ["../bad"]) == () - - -def test_composer_accepts_literal_prose_and_cycles_recipient_with_tab(): - result, output = _compose_in_pty( - ("flex-auth", "railiance-platform"), - b"What's up?\t Really?\r", - ) - - assert result == ("railiance-platform", "What's up? Really?") - assert b"@flex-auth: " in output - assert b"@railiance-platform: What's up?" in output - - -def test_composer_supports_unicode_backspace_and_ctrl_u(): - result, _ = _compose_in_pty( - ("flex-auth",), - "caf\u00e9x\b!\x15final\r".encode(), - ) - assert result == ("flex-auth", "final") - - -def test_composer_ctrl_c_cancels_and_restores_terminal(): - result, output = _compose_in_pty(("flex-auth",), b"draft\x03") - assert result is None - assert b"^C\r\n" in output - - -def test_composer_requires_a_terminal(): - read_fd, write_fd = os.pipe() - try: - with pytest.raises(ComposerError, match="requires a terminal"): - compose_message(("flex-auth",), input_fd=read_fd, output_fd=write_fd) - finally: - os.close(read_fd) - os.close(write_fd) - - -def test_composer_requires_a_recipient(): - with pytest.raises(ComposerError, match="no recipient"): - compose_message(()) diff --git a/tests/test_config_command.py b/tests/test_config_command.py index 43fecb9..5a3265c 100644 --- a/tests/test_config_command.py +++ b/tests/test_config_command.py @@ -6,6 +6,9 @@ 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" + result = json.loads(capsys.readouterr().out) + assert result["maxmsg"] == 8 + assert result["maxin"] == 1024 + assert result["maxout"] == 32768 + assert result["database"].endswith("tamq.sqlite3") + assert result["policy_profile"] == "default" diff --git a/tests/test_install_target.py b/tests/test_install_target.py index 43b5a32..daec056 100644 --- a/tests/test_install_target.py +++ b/tests/test_install_target.py @@ -92,7 +92,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ) assert run(str(tamq), "--version").stdout.strip() == "0.1.0" root_help = run(str(tamq), "--help").stdout - assert "tamq [--detach] [--command COMMAND] [--mode MODE] REPO" in root_help + assert "tamq [--detach] [--command COMMAND] [--mode MODE]" in root_help started = json.loads( run( str(tamq), @@ -178,7 +178,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): "send-keys", "-t", "tamq:railiance-platform", - "@activity-core: installed-message", + "To:activity-core: installed-message", "C-m", ) deadline = time.monotonic() + 5 @@ -194,7 +194,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): target_after = run( *tmux, "capture-pane", "-p", "-t", "tamq:activity-core" ).stdout - if inbox and inbox[-1]["displayed_at"] is not None and "#railiance-platform: installed-message" in target_after: + if inbox and inbox[-1]["displayed_at"] is not None and "From:railiance-platform/o: installed-message" in target_after: break time.sleep(0.05) assert inbox, run( @@ -206,10 +206,10 @@ def test_isolated_installed_tool_session_smoke(tmp_path): assert inbox[-1]["displayed_at"] is not None message_id = inbox[-1]["message_id"] assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == ( - f"#railiance-platform: installed-message [{message_id}]\n" + "From:railiance-platform/o: installed-message\n" ) assert target_after != target_before - assert f"#railiance-platform: installed-message [{message_id}]" in target_after + assert "From:railiance-platform/o: installed-message" in target_after assert run(str(tamq), "ack", message_id).stdout.strip() == message_id assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == "" @@ -226,7 +226,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): "send-keys", "-t", "tamq:railiance-platform", - "@activity-core filtered-message", + "To:activity-core: filtered-message", "C-m", ) deadline = time.monotonic() + 5 @@ -242,7 +242,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): str(tamq), "inbox", "--repo", "activity-core", "--filter", f"cat >> {message_log}", ) - assert "#railiance-platform: filtered-message [m-" in message_log.read_text( + assert "From:railiance-platform/o: filtered-message" in message_log.read_text( encoding="utf-8" ) assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == "" diff --git a/tests/test_line_limits.py b/tests/test_line_limits.py new file mode 100644 index 0000000..08ee51e --- /dev/null +++ b/tests/test_line_limits.py @@ -0,0 +1,65 @@ +from concurrent.futures import ThreadPoolExecutor + +from tamq.broker import BrokerIdentity, InputBroker +from tamq.store import Store + + +def test_window_counters_survive_reopen_and_isolate_repositories(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) + path = tmp_path / "queue.sqlite3" + store = Store(path) + left = InputBroker(store, BrokerIdentity("session-1", "left"), limits=(1, 10, 10)) + right = InputBroker(store, BrokerIdentity("session-1", "right"), limits=(1, 10, 10)) + assert left.inspect_operator_line("To:target: left") is not None + assert right.inspect_operator_line("To:target: right") is not None + store.close() + + reopened = Store(path) + assert reopened.line_state("session-1", "left")["messages"] == 1 + assert reopened.line_state("session-1", "right")["messages"] == 1 + blocked = [] + left = InputBroker( + reopened, + BrokerIdentity("session-1", "left"), + limits=(1, 10, 10), + notify=blocked.append, + ) + assert left.inspect_worker_line("To:target: blocked") is None + assert "lines of messages" in blocked[-1] + + +def test_atomic_message_admission_does_not_cross_limit(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) + path = tmp_path / "queue.sqlite3" + seed = Store(path) + seed.configure_window("session-1", "source", (1, 100, 100)) + seed.close() + + def send(body): + store = Store(path) + try: + broker = InputBroker( + store, + BrokerIdentity("session-1", "source"), + limits=(1, 100, 100), + ) + return broker.inspect_worker_line(f"To:target: {body}") is not None + finally: + store.close() + + with ThreadPoolExecutor(max_workers=2) as executor: + accepted = list(executor.map(send, ("one", "two"))) + assert sorted(accepted) == [False, True] + final = Store(path) + assert len(final.list()) == 1 + assert final.line_state("session-1", "source")["messages"] == 1 + + +def test_new_session_generation_starts_a_fresh_ledger(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) + store = Store(tmp_path / "queue.sqlite3") + first = InputBroker(store, BrokerIdentity("generation-1", "source"), limits=(1, 10, 10)) + second = InputBroker(store, BrokerIdentity("generation-2", "source"), limits=(1, 10, 10)) + assert first.inspect_operator_line("To:target: first") is not None + assert second.inspect_operator_line("To:target: second") is not None + assert len(store.list()) == 2 diff --git a/tests/test_manual_messaging.py b/tests/test_manual_messaging.py index 4f69ff5..06382d0 100644 --- a/tests/test_manual_messaging.py +++ b/tests/test_manual_messaging.py @@ -15,7 +15,7 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk 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 + assert main(["send", "To:audit-core:", "review", "this"]) == 0 message_id = capsys.readouterr().out.strip() assert main(["inbox", "--repo", "audit-core", "--json"]) == 0 @@ -24,7 +24,7 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk assert row["sender_repo"] == "flex-auth" assert row["target_repo"] == "audit-core" assert row["body"] == "review this" - assert row["state"] == "pending" + assert row["provenance"] == "operator_input" assert main(["ack", message_id]) == 0 capsys.readouterr() @@ -32,103 +32,6 @@ def test_manual_send_inbox_and_ack_use_window_repository_identity(tmp_path, monk assert capsys.readouterr().out == "" -def test_bare_reply_targets_latest_inbound_sender(tmp_path, monkeypatch, capsys): - monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) - monkeypatch.setenv("TAMQ_REPO", "audit-core") - monkeypatch.setattr("tamq.cli.ping", service_is_down) - monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) - store = Store(tmp_path / "state" / "tamq.sqlite3") - store.add("flex-auth", "audit-core", "older") - latest = store.add("railiance-platform", "audit-core", "latest") - store.acknowledge(latest) - store.add("audit-core", "audit-core", "self note") - store.close() - - assert main(["reply", "--", "--looks-like-an-option", "thanks"]) == 0 - message_id = capsys.readouterr().out.strip() - - store = Store(tmp_path / "state" / "tamq.sqlite3") - row = next(row for row in store.list() if row["message_id"] == message_id) - assert row["sender_repo"] == "audit-core" - assert row["target_repo"] == "railiance-platform" - assert row["body"] == "--looks-like-an-option thanks" - store.close() - - -def test_interactive_reply_shows_latest_and_cycles_session_recipients(tmp_path, monkeypatch, capsys): - monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) - monkeypatch.setenv("TAMQ_REPO", "audit-core") - monkeypatch.setenv( - "TAMQ_RECIPIENTS", '["audit-core","railiance-platform","flex-auth"]' - ) - monkeypatch.setattr("tamq.cli.ping", service_is_down) - monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) - store = Store(tmp_path / "state" / "tamq.sqlite3") - store.add("flex-auth", "audit-core", "latest inbound") - store.close() - seen = [] - - def compose(recipients): - seen.append(recipients) - return "railiance-platform", "What's up?" - - monkeypatch.setattr("tamq.cli.compose_message", compose) - assert main(["reply"]) == 0 - message_id = capsys.readouterr().out.strip() - - assert seen == [("flex-auth", "railiance-platform")] - store = Store(tmp_path / "state" / "tamq.sqlite3") - row = next(row for row in store.list() if row["message_id"] == message_id) - assert (row["sender_repo"], row["target_repo"], row["body"]) == ( - "audit-core", "railiance-platform", "What's up?" - ) - store.close() - - -def test_interactive_reply_uses_first_peer_without_history(tmp_path, monkeypatch, capsys): - monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) - monkeypatch.setenv("TAMQ_REPO", "audit-core") - monkeypatch.setenv("TAMQ_RECIPIENTS", '["audit-core","flex-auth"]') - monkeypatch.setattr("tamq.cli.ping", service_is_down) - monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) - monkeypatch.setattr( - "tamq.cli.compose_message", lambda recipients: (recipients[0], "hello") - ) - - assert main(["reply"]) == 0 - message_id = capsys.readouterr().out.strip() - store = Store(tmp_path / "state" / "tamq.sqlite3") - row = next(row for row in store.list() if row["message_id"] == message_id) - assert row["target_repo"] == "flex-auth" - store.close() - - -def test_interactive_reply_cancellation_queues_nothing(tmp_path, monkeypatch): - monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) - monkeypatch.setenv("TAMQ_REPO", "audit-core") - monkeypatch.setenv("TAMQ_RECIPIENTS", '["audit-core","flex-auth"]') - monkeypatch.setattr("tamq.cli.compose_message", lambda recipients: None) - - assert main(["reply"]) == 130 - store = Store(tmp_path / "state" / "tamq.sqlite3") - assert store.list() == [] - store.close() - - -def test_bare_reply_requires_managed_identity_and_prior_sender(tmp_path, monkeypatch, capsys): - monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) - monkeypatch.setattr("tamq.cli.ping", service_is_down) - monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) - monkeypatch.delenv("TAMQ_REPO", raising=False) - - assert main(["reply", "hello"]) == 2 - assert "managed window" in capsys.readouterr().err - - monkeypatch.setenv("TAMQ_REPO", "audit-core") - assert main(["reply", "hello"]) == 2 - assert "no counterparty" in capsys.readouterr().err - - 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) @@ -136,16 +39,19 @@ def test_inbox_requires_repository_outside_managed_window(tmp_path, monkeypatch, assert "requires --repo" in capsys.readouterr().err -def test_human_inbox_is_comment_safe_on_every_line(tmp_path, monkeypatch, capsys): +def test_human_inbox_uses_readable_origin_framing(tmp_path, monkeypatch, capsys): monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) store = Store(tmp_path / "state" / "tamq.sqlite3") - message_id = store.add("flex-auth", "audit-core", "first\nsecond\x1b[31m") + store.add( + "flex-auth", "audit-core", "first\nsecond\x1b[31m", + provenance="operator_input", + ) store.close() assert main(["inbox", "--repo", "audit-core"]) == 0 assert capsys.readouterr().out == ( - f"#flex-auth: first\n# second\\x1b[31m [{message_id}]\n" + "From:flex-auth/o: first\\x0asecond\\x1b[31m\n" ) @@ -153,8 +59,8 @@ def test_inbox_filter_logs_and_acknowledges_successful_messages(tmp_path, monkey monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) monkeypatch.setattr("tamq.cli.validate_targets", lambda repos: None) store = Store(tmp_path / "state" / "tamq.sqlite3") - first_id = store.add("flex-auth", "audit-core", "review 'quoted' $value") - second_id = store.add("railiance-platform", "audit-core", "second") + store.add("flex-auth", "audit-core", "review 'quoted' $value", provenance="operator_input") + store.add("railiance-platform", "audit-core", "second", provenance="worker_output") store.close() log = tmp_path / "messages.log" @@ -163,8 +69,8 @@ def test_inbox_filter_logs_and_acknowledges_successful_messages(tmp_path, monkey f"cat >> {shlex.quote(str(log))}", ]) == 0 assert log.read_text(encoding="utf-8") == ( - f"#flex-auth: review 'quoted' $value [{first_id}]\n" - f"#railiance-platform: second [{second_id}]\n" + "From:flex-auth/o: review 'quoted' $value\n" + "From:railiance-platform: second\n" ) assert main(["inbox", "--repo", "audit-core"]) == 0 assert capsys.readouterr().out == "" @@ -183,7 +89,6 @@ def test_failed_inbox_filter_leaves_current_and_later_messages_pending(tmp_path, assert main(["inbox", "--repo", "audit-core", "--json"]) == 0 rows = [json.loads(line) for line in capsys.readouterr().out.splitlines()] assert [row["message_id"] for row in rows] == [first_id, second_id] - assert all(row["state"] == "pending" for row in rows) def test_inbox_filter_rejects_non_pending_and_json_modes(tmp_path, monkeypatch, capsys): diff --git a/tests/test_ptytap.py b/tests/test_ptytap.py index e1100d9..5ab732b 100644 --- a/tests/test_ptytap.py +++ b/tests/test_ptytap.py @@ -4,7 +4,13 @@ import pty import struct import termios -from tamq.ptytap import PtyTap, copy_winsize, observed_line, write_all +from tamq.ptytap import ( + PtyTap, + TerminalOutputObserver, + copy_winsize, + observed_line, + write_all, +) class RecordingBroker: @@ -13,7 +19,7 @@ class RecordingBroker: def inspect_line(self, line): self.lines.append(line) - return line.startswith(("@", "#")) + return line.startswith("To:") def test_input_observer_accepts_raw_terminal_carriage_returns(): @@ -22,11 +28,11 @@ def test_input_observer_accepts_raw_terminal_carriage_returns(): tap = PtyTap(["true"], broker, on_line=observed.append) buffer = bytearray() - tap._observe_input(buffer, b"#activity-core: hel") + tap._observe_input(buffer, b"To:activity-core: hel") tap._observe_input(buffer, b"lo\rplain line\n") - assert broker.lines == ["#activity-core: hello", "plain line"] - assert observed == ["#activity-core: hello"] + assert broker.lines == ["To:activity-core: hello", "plain line"] + assert observed == ["To:activity-core: hello"] def test_input_observer_accepts_enhanced_terminal_enter_sequences(): @@ -35,17 +41,17 @@ def test_input_observer_accepts_enhanced_terminal_enter_sequences(): tap = PtyTap(["true"], broker, on_line=observed.append) buffer = bytearray() - tap._observe_input(buffer, b"#activity-core: CSI-u\x1b[13;1") + tap._observe_input(buffer, b"To:activity-core: CSI-u\x1b[13;1") assert broker.lines == [] - tap._observe_input(buffer, b":1u#other: modify\x1b[27;1;13~") + tap._observe_input(buffer, b":1uTo:other: modify\x1b[27;1;13~") tap._observe_input(buffer, b"plain\x1bOM") assert broker.lines == [ - "#activity-core: CSI-u", - "#other: modify", + "To:activity-core: CSI-u", + "To:other: modify", "plain", ] - assert observed == ["#activity-core: CSI-u", "#other: modify"] + assert observed == ["To:activity-core: CSI-u", "To:other: modify"] def test_input_observer_ignores_terminal_replies_before_typed_line(): @@ -55,20 +61,36 @@ def test_input_observer_ignores_terminal_replies_before_typed_line(): buffer = bytearray() tap._observe_input(buffer, b"\x1b[9;1R\x1b[?1;2;4c") - tap._observe_input(buffer, b"#activity-core: Helol\x7f\x7flo\r") + tap._observe_input(buffer, b"To:activity-core: Helol\x7f\x7flo\r") - assert broker.lines == ["#activity-core: Hello"] - assert observed == ["#activity-core: Hello"] + assert broker.lines == ["To:activity-core: Hello"] + assert observed == ["To:activity-core: Hello"] def test_observed_line_removes_common_terminal_protocol_wrappers(): raw = ( - b"\x1b[200~#target: pasted\x1b[201~" + b"\x1b[200~To:target: pasted\x1b[201~" b"\x1b]10;rgb:ffff/ffff/ffff\x07" b"\x1bP1$r0m\x1b\\" ) - assert observed_line(raw) == "#target: pasted" + assert observed_line(raw) == "To:target: pasted" + + +def test_output_observer_normalizes_ansi_and_deduplicates_redraws(): + lines = [] + observer = TerminalOutputObserver(lines.append) + observer.feed(b"plain\r\n\x1b[31mTo:target: worker\x1b[0m\r") + observer.feed(b"To:target: worker\r") + assert lines == ["plain", "To:target: worker"] + + +def test_output_observer_fails_closed_on_recent_operator_echo(): + lines = [] + observer = TerminalOutputObserver(lines.append) + observer.note_operator_line("To:target: operator") + observer.feed(b"To:target: operator\r\nTo:target: worker\r\n") + assert lines == ["To:target: worker"] def test_copy_winsize_preserves_rows_columns_and_pixels(): diff --git a/tests/test_ptytap_integration.py b/tests/test_ptytap_integration.py index bd87603..da8dcd3 100644 --- a/tests/test_ptytap_integration.py +++ b/tests/test_ptytap_integration.py @@ -8,6 +8,7 @@ from uuid import uuid4 import pytest from tamq.tmux import LaunchPlan, TmuxManager +from tamq.store import Store @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") @@ -98,3 +99,82 @@ def test_tap_propagates_terminal_size_resize_and_raw_mouse_input(tmp_path, monke assert f"RESIZE {resized}" in output finally: manager._run("kill-server", check=False) + + +@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") +def test_real_pty_attributes_worker_output_and_suppresses_operator_echo( + tmp_path, monkeypatch +): + source = tmp_path / "source" + source.mkdir() + fixture = tmp_path / "duplex_fixture.py" + fixture.write_text( + "import sys\n" + "print('READY', flush=True)\n" + "for line in sys.stdin:\n" + " print(line.rstrip('\\r\\n'), flush=True)\n" + " print('To:target: worker reply', flush=True)\n", + encoding="utf-8", + ) + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + gita = bin_dir / "gita" + gita.write_text( + "#!/bin/sh\n" + "if [ \"$1\" = ls ]; then printf 'source target\\n'; exit 0; fi\n" + "exit 2\n", + encoding="utf-8", + ) + gita.chmod(0o755) + project_src = str(Path(__file__).resolve().parents[1] / "src") + monkeypatch.setenv("PYTHONPATH", project_src) + monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}") + state = tmp_path / "state" + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + socket_name = f"tamq-duplex-{uuid4().hex[:8]}" + session = f"tamq-duplex-{uuid4().hex[:8]}" + manager = TmuxManager( + session, + tmux_command=("tmux", "-L", socket_name), + tamq_command=(sys.executable, "-m", "tamq.cli"), + command_dir=tmp_path / "commands", + ) + plan = LaunchPlan( + ("source",), {"source": str(source)}, (sys.executable, str(fixture)) + ) + store = None + try: + endpoint = manager.ensure_plan(plan, tap=True, tap_limits=(8, 1024, 32768)) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if "READY" in manager._run( + "capture-pane", "-p", "-t", f"{session}:source" + ): + break + time.sleep(0.05) + manager._run( + "send-keys", + "-t", + f"{session}:source", + "-l", + "--", + "To:target: operator request", + ) + manager._run("send-keys", "-t", f"{session}:source", "Enter") + store = Store(state / "tamq.sqlite3") + deadline = time.monotonic() + 5 + rows = [] + while time.monotonic() < deadline: + rows = store.list() + if len(rows) >= 2: + break + time.sleep(0.05) + assert [(row["body"], row["provenance"]) for row in rows] == [ + ("operator request", "operator_input"), + ("worker reply", "worker_output"), + ] + assert all(row["endpoint_id"] == endpoint.instance_id for row in rows) + finally: + if store is not None: + store.close() + manager._run("kill-server", check=False) diff --git a/tests/test_routing.py b/tests/test_routing.py index 2d59355..6111d4c 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -1,17 +1,30 @@ import pytest -from tamq.routing import parse_address_line +from tamq.routing import parse_address_line, parse_command_line -@pytest.mark.parametrize("prefix", ["@", "#"]) -def test_direct_address(prefix): - routed = parse_address_line(f"{prefix}railiance-platform: do something!") +def test_direct_address_uses_readable_case_sensitive_grammar(): + routed = parse_address_line("To:railiance-platform: do something!") assert routed.body == "do something!" assert routed.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 - assert parse_address_line("# from repo: inbound envelope") is None - assert parse_address_line("##repo: not an address") is None +@pytest.mark.parametrize( + "line", + [ + "@repo: removed", + "#repo: removed", + "to:repo: wrong case", + "hello To:repo: not at start", + "To:repo:", + "From:repo: inbound envelope", + ], +) +def test_non_protocol_and_legacy_input_is_not_routed(line): + assert parse_address_line(line) is None + + +def test_operator_command_grammar(): + assert parse_command_line("Cmd: mode=trigger") == "mode=trigger" + assert parse_command_line("Cmd: reset-limits") == "reset-limits" + assert parse_command_line("cmd: mode=trigger") is None diff --git a/tests/test_send_endpoint.py b/tests/test_send_endpoint.py index 84a39f2..6ab8043 100644 --- a/tests/test_send_endpoint.py +++ b/tests/test_send_endpoint.py @@ -4,5 +4,5 @@ from tamq.cli import build_parser def test_send_accepts_endpoint_id(): - args = build_parser().parse_args(["send", "--endpoint-id", "ep", "@repo:", "hello"]) + args = build_parser().parse_args(["send", "--endpoint-id", "ep", "To:repo:", "hello"]) assert args.endpoint_id == "ep" diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py index 593393e..accfaaf 100644 --- a/tests/test_service_delivery.py +++ b/tests/test_service_delivery.py @@ -9,6 +9,7 @@ class FakeControl: def __init__(self, session): self.session = session self.injected = [] + self.placed = [] self.submitted = [] self.__class__.instances.append(self) @@ -24,6 +25,9 @@ class FakeControl: def submit(self, window, text): self.submitted.append((window, text)) + def place(self, window, text): + self.placed.append((window, text)) + def pane_display(self, window): return PaneDisplay( tty_path=f"/dev/pts/{window.rsplit(':', 1)[-1]}", @@ -43,7 +47,7 @@ def test_service_delivery_injects_pending_messages(tmp_path, monkeypatch): 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 FakeControl.instances[-1].injected == [("tamq:repo-b", "From:repo-a: continue")] assert store.list()[0]["message_id"] == message_id assert store.list()[0]["state"] == "injected" @@ -60,7 +64,7 @@ def test_service_never_injects_manual_endpoint_messages(tmp_path, monkeypatch): assert store.list()[0]["state"] == "pending" -def test_service_pushy_mode_submits_once_and_marks_injected(tmp_path, monkeypatch): +def test_service_pushy_mode_places_once_without_enter(tmp_path, monkeypatch): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "pushy") message_id = store.add("repo-a", "repo-b", "first\nsecond") @@ -71,22 +75,36 @@ def test_service_pushy_mode_submits_once_and_marks_injected(tmp_path, monkeypatc control = FakeControl.instances[-1] service._deliver_once() - assert control.submitted == [ + assert control.placed == [ ( "tamq:repo-b", - f"# from repo-a: first\\x0asecond [{message_id}]", + "From:repo-a: first\\x0asecond", ) ] + assert control.submitted == [] assert store.list()[0]["state"] == "injected" +def test_service_trigger_mode_submits_exactly_once(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep", 9, "tamq", ["repo-b"], "trigger") + store.add("repo-a", "repo-b", "go", provenance="operator_input") + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + service = Service(store=store) + service._deliver_once() + control = FakeControl.instances[-1] + service._deliver_once() + assert control.submitted == [("tamq:repo-b", "From:repo-a/o: go")] + assert control.placed == [] + + def test_failed_pushy_submission_remains_pending(tmp_path, monkeypatch): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "pushy") store.add("repo-a", "repo-b", "continue") class FailingControl(FakeControl): - def submit(self, window, text): + def place(self, window, text): raise RuntimeError("tmux rejected input") monkeypatch.setattr("tamq.service.ControlModeClient", FailingControl) @@ -113,7 +131,7 @@ def test_service_writes_output_once_and_retains_pending_ack(tmp_path, monkeypatc assert writes == [ ( "/dev/pts/repo-b", - f"\r\n#repo-a: continue [{message_id}]\r\n", + "\r\nFrom:repo-a: continue\r\n", ) ] row = store.list()[0] diff --git a/tests/test_store.py b/tests/test_store.py index 01dc193..6bcf9a3 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -41,7 +41,7 @@ def test_store_migrates_legacy_endpoint_rows_to_manual_delivery(tmp_path): assert store.endpoints()[0]["delivery_mode"] == "manual" assert store.db.execute( "SELECT value FROM metadata WHERE key='schema_version'" - ).fetchone()[0] == "3" + ).fetchone()[0] == "4" assert "displayed_at" in { row["name"] for row in store.db.execute("PRAGMA table_info(messages)") } @@ -59,15 +59,3 @@ def test_mark_displayed_releases_lease_without_acknowledging(tmp_path): assert row["displayed_at"] is not None assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0 store.close() - - -def test_latest_counterparty_uses_latest_inbound_message_regardless_of_state(tmp_path): - store = Store(tmp_path / "queue.sqlite3") - store.add("audit-core", "audit-core", "self note") - older = store.add("flex-auth", "audit-core", "first") - store.acknowledge(older) - store.add("railiance-platform", "audit-core", "latest") - - assert store.latest_counterparty("audit-core") == "railiance-platform" - assert store.latest_counterparty("unknown") is None - store.close() diff --git a/tests/test_terminal_output.py b/tests/test_terminal_output.py index 4d7d763..0074d64 100644 --- a/tests/test_terminal_output.py +++ b/tests/test_terminal_output.py @@ -6,55 +6,48 @@ import pytest from tamq.terminal import ( TerminalOutputError, - format_comment, - format_pushy_input, + format_delivery, terminal_frame, write_terminal_output, ) -def test_comment_format_escapes_controls_and_prefixes_every_line(): - assert format_comment("repo-a", "first\nsecond\x1b[31m", "m-1") == ( - "#repo-a: first\n# second\\x1b[31m [m-1]" +def test_delivery_format_escapes_controls_and_marks_operator_origin(): + assert format_delivery("repo-a", "first\nsecond\x1b[31m", "operator_input") == ( + "From:repo-a/o: first\\x0asecond\\x1b[31m" ) - - -def test_pushy_input_is_one_sanitized_shell_comment(): - assert format_pushy_input("repo-a", "first\nsecond\x1b[31m", "m-1") == ( - "# from repo-a: first\\x0asecond\\x1b[31m [m-1]" + assert format_delivery("repo-a", "worker", "worker_output") == ( + "From:repo-a: worker" ) def test_terminal_frame_scrolls_only_rows_above_the_cursor(): assert terminal_frame( - "repo-a", "first\nsecond", "m-1", cursor_y=8, pane_height=24 + "repo-a", "hello", "operator_input", cursor_y=8, pane_height=24 ) == ( "\x1b7\x1b[1;8r\x1b[8;1H" - "\n\r#repo-a: first\n\r# second [m-1]" + "\n\rFrom:repo-a/o: hello" "\x1b[r\x1b8" ) def test_terminal_frame_falls_back_when_stable_region_is_not_safe(): - expected = "\r\n#repo-a: hello [m-1]\r\n" - assert terminal_frame("repo-a", "hello", "m-1", cursor_y=0, pane_height=24) == expected + expected = "\r\nFrom:repo-a: hello\r\n" + assert terminal_frame("repo-a", "hello", "worker_output", cursor_y=0, pane_height=24) == expected assert terminal_frame( - "repo-a", "hello", "m-1", cursor_y=8, pane_height=24, alternate_on=True + "repo-a", "hello", "worker_output", cursor_y=8, pane_height=24, alternate_on=True ) == expected - assert terminal_frame("repo-a", "hello", "m-1") == ( - "\r\n#repo-a: hello [m-1]\r\n" - ) def test_terminal_output_reaches_pty_output_but_not_input(): master, slave = pty.openpty() try: tty_path = os.ttyname(slave) - write_terminal_output(tty_path, terminal_frame("repo-a", "hello", "m-1")) + write_terminal_output(tty_path, terminal_frame("repo-a", "hello", "worker_output")) readable, _, _ = select.select([master], [], [], 1) assert readable == [master] - assert b"#repo-a: hello [m-1]" in os.read(master, 4096) + assert b"From:repo-a: hello" in os.read(master, 4096) readable_input, _, _ = select.select([slave], [], [], 0) assert readable_input == [] finally: diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index b68d5bc..95fab71 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -105,10 +105,10 @@ def test_real_tmux_starts_two_repo_windows_and_reuses_them(tmp_path, monkeypatch capture = manager._run( "capture-pane", "-p", "-t", f"{session}:activity-core" ) - if "#local: integration-message" in capture: + if "From:local: integration-message" in capture: break time.sleep(0.05) - assert "#local: integration-message" in capture + assert "From:local: integration-message" in capture finally: control.close() store.close() @@ -173,7 +173,7 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path): terminal_frame( "flex-auth", "stable-message", - "m-stable", + "worker_output", cursor_y=before.cursor_y, pane_height=before.pane_height, alternate_on=before.alternate_on, @@ -183,11 +183,11 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path): deadline = time.monotonic() + 5 while time.monotonic() < deadline: after_capture = manager._run("capture-pane", "-p", "-t", target) - if "#flex-auth: stable-message [m-stable]" in after_capture: + if "From:flex-auth: stable-message" in after_capture: break time.sleep(0.05) after = control.pane_display(target) - assert "#flex-auth: stable-message [m-stable]" in after_capture + assert "From:flex-auth: stable-message" in after_capture before_input = next(line for line in before_capture.splitlines() if "PARTIAL-INPUT" in line) after_lines = after_capture.splitlines() assert any("PARTIAL-INPUT" in line for line in after_lines), repr(after_capture) @@ -195,14 +195,14 @@ def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path): index for index, line in enumerate(after_lines) if "PARTIAL-INPUT" in line ) assert after_lines[after_input_index] == before_input - assert after_lines[after_input_index - 1] == "#flex-auth: stable-message [m-stable]" + assert after_lines[after_input_index - 1] == "From:flex-auth: stable-message" assert (after.cursor_x, after.cursor_y) == (before.cursor_x, before.cursor_y) finally: manager._run("kill-server", check=False) @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") -def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch): +def test_real_tmux_pushy_mode_places_one_readable_input_without_enter(tmp_path, monkeypatch): repo = tmp_path / "audit-core" repo.mkdir() socket_name = f"tamq-pushy-{os.getpid()}-{uuid4().hex[:8]}" @@ -238,11 +238,11 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch endpoint.repos, "pushy", ) - message_id = store.add("flex-auth", "audit-core", "What's next?\nsecond") + store.add("flex-auth", "audit-core", "What's next?\nsecond", provenance="operator_input") service = Service(store=store) service._deliver_once() - expected = f"# from flex-auth: What's next?\\x0asecond [{message_id}]" + expected = "From:flex-auth/o: What's next?\\x0asecond" deadline = time.monotonic() + 5 while time.monotonic() < deadline: capture = manager._run("capture-pane", "-p", "-J", "-t", target) @@ -262,7 +262,7 @@ def test_real_tmux_pushy_mode_submits_one_shell_safe_input(tmp_path, monkeypatch @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") -def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch): +def test_real_tmux_to_route_triggers_once_without_feedback(tmp_path, monkeypatch): repo_a = tmp_path / "railiance-platform" repo_b = tmp_path / "activity-core" bin_dir = tmp_path / "bin" @@ -330,7 +330,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch endpoint.pid, endpoint.session, endpoint.repos, - "pushy", + "trigger", ) manager._run( "send-keys", @@ -338,7 +338,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch f"{session}:railiance-platform", "-l", "--", - "#activity-core: Hello!", + "To:activity-core: Hello!", ) manager._run("send-keys", "-t", f"{session}:railiance-platform", "Enter") @@ -357,7 +357,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch service = Service(store=store) service._deliver_once() - expected = f"# from railiance-platform: Hello! [{row['message_id']}]" + expected = "From:railiance-platform/o: Hello!" deadline = time.monotonic() + 5 while time.monotonic() < deadline: target_capture = manager._run( @@ -369,9 +369,7 @@ def test_real_tmux_hash_route_pushes_once_without_feedback(tmp_path, monkeypatch assert f"AGENT:{expected}" in target_capture assert store.list()[0]["state"] == "injected" - legacy_envelope = ( - f"#railiance-platform: Hello! [{row['message_id']}]" - ) + legacy_envelope = "#railiance-platform: Hello!" manager._run( "send-keys", "-t", diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py index 9ed92a2..5b9c9ac 100644 --- a/tests/test_tmux_manager.py +++ b/tests/test_tmux_manager.py @@ -98,6 +98,15 @@ def test_preflight_rejects_foreign_session(tmp_path, monkeypatch): manager.preflight(["a"]) +def test_tap_refuses_managed_session_without_duplex_protocol(tmp_path, monkeypatch): + manager = tmux.TmuxManager("tamq-test", command_dir=tmp_path / "commands") + plan = tmux.LaunchPlan(("a",), {"a": str(tmp_path)}, ("sh",)) + monkeypatch.setattr(manager, "_existing_session_identity", lambda: (42, "tmux-amq-42-old")) + monkeypatch.setattr(manager, "_run", lambda *args, **kwargs: "") + with pytest.raises(tmux.TmuxError, match="predates readable duplex"): + manager.ensure_plan(plan, tap=True) + + def test_tmux_only_plan_starts_agent_without_tap(tmp_path, monkeypatch): manager = tmux.TmuxManager("tamq-test", command_dir=tmp_path / "commands") calls = [] @@ -152,39 +161,27 @@ def test_neutral_plan_starts_shell_without_sending_keystrokes(tmp_path, monkeypa assert "TAMQ_REPO=b" in new_window assert f"PATH={command_dir}{os.pathsep}" in " ".join(new_session) assert f"PATH={command_dir}{os.pathsep}" in " ".join(new_window) - assert (command_dir / "@a").is_file() - assert (command_dir / "@a:").is_file() - assert (command_dir / "@b").is_file() - assert (command_dir / "@b:").is_file() - assert (command_dir / "@").is_file() + assert (command_dir / "To:a:").is_file() + assert (command_dir / "To:b:").is_file() + assert (command_dir / "Cmd:").is_file() + assert not any(path.name.startswith("@") for path in command_dir.iterdir()) -def test_address_commands_preserve_message_arguments(tmp_path): +def test_protocol_commands_absorb_forwarded_shell_lines(tmp_path): command_dir = tmp_path / "commands" manager = tmux.TmuxManager( "tamq-test", tamq_command=("echo",), command_dir=command_dir, ) manager._install_address_commands(["audit-core"]) - for name in ("@audit-core", "@audit-core:"): + for name in ("To:audit-core:", "Cmd:"): result = subprocess.run( - [str(command_dir / name), "Some message!", "$value", "; literal", "--from"], + [str(command_dir / name), "Some message!", "$value"], text=True, capture_output=True, check=True, ) - assert result.stdout == "send -- @audit-core: Some message! $value ; literal --from\n" - - result = subprocess.run( - [str(command_dir / "@"), "Some reply!", "$value", "--literal"], - text=True, - capture_output=True, - check=True, - ) - assert result.stdout == "reply -- Some reply! $value --literal\n" - generic = (command_dir / "@").read_text(encoding="utf-8") - assert 'TAMQ_RECIPIENTS=' in generic - assert '["audit-core"]' in generic + assert result.stdout == "" def test_address_commands_reject_unsafe_repository_names(tmp_path): @@ -196,7 +193,7 @@ def test_address_commands_reject_unsafe_repository_names(tmp_path): def test_address_commands_do_not_replace_unowned_commands(tmp_path): command_dir = tmp_path / "commands" command_dir.mkdir() - existing = command_dir / "@audit-core" + existing = command_dir / "To:audit-core:" existing.write_text("#!/bin/sh\necho mine\n", encoding="utf-8") manager = tmux.TmuxManager("tamq-test", command_dir=command_dir) @@ -205,10 +202,10 @@ def test_address_commands_do_not_replace_unowned_commands(tmp_path): assert existing.read_text(encoding="utf-8") == "#!/bin/sh\necho mine\n" -def test_address_commands_do_not_replace_unowned_bare_reply(tmp_path): +def test_address_commands_do_not_replace_unowned_command_lane(tmp_path): command_dir = tmp_path / "commands" command_dir.mkdir() - existing = command_dir / "@" + existing = command_dir / "Cmd:" existing.write_text("#!/bin/sh\necho mine\n", encoding="utf-8") manager = tmux.TmuxManager("tamq-test", command_dir=command_dir) diff --git a/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md b/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md index c6293ce..eb74891 100644 --- a/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md +++ b/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md @@ -4,7 +4,7 @@ type: workplan title: "Readable duplex messaging, runtime commands, and line limits" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social planning_priority: P0 @@ -101,7 +101,7 @@ atomically resets all three counters for the current source window only. ```task id: TAMQ-WP-0014-T01 -status: todo +status: done priority: critical state_hub_task_id: "66ac5b8c-6435-5d62-8b3c-4516bddae75f" ``` @@ -124,7 +124,7 @@ line/echo boundary and fixtures before routing worker output. ```task id: TAMQ-WP-0014-T02 -status: wait +status: done priority: critical state_hub_task_id: "95cf3b20-8b23-5155-8b6d-3431baf36486" ``` @@ -145,7 +145,7 @@ deduplication, echo suppression, and feedback-chain containment. ```task id: TAMQ-WP-0014-T03 -status: wait +status: done priority: high state_hub_task_id: "3ae47806-9808-5b38-b0cd-1f63536ebce9" ``` @@ -165,7 +165,7 @@ neither mode repeats after accepted delivery, and no mode turns delivered ```task id: TAMQ-WP-0014-T04 -status: wait +status: done priority: high state_hub_task_id: "67c3aa0d-ad5e-5fcd-bd7a-fe4bdcdebbcb" ``` @@ -190,7 +190,7 @@ delay normal input, or implement an agent-specific erasure trick. ```task id: TAMQ-WP-0014-T05 -status: wait +status: done priority: critical state_hub_task_id: "5101a264-5fde-5e94-b924-bdaa114d18fd" ``` @@ -214,7 +214,7 @@ sends, and notice feedback suppression. ```task id: TAMQ-WP-0014-T06 -status: wait +status: done priority: high state_hub_task_id: "5b630cab-e431-5f8f-83b0-cc1bd56c262c" ``` @@ -236,7 +236,7 @@ new protocol until this workplan's local contract is proven. ```task id: TAMQ-WP-0014-T07 -status: wait +status: done priority: critical state_hub_task_id: "bae85317-60d0-5e12-8ffd-d08a70f9d9e7" ``` @@ -251,3 +251,36 @@ unrelated panes or durable history, and leave runaway-safe cleanup available. Do not mark the workplan finished while T01 attribution is only simulated or while a worker output replay can duplicate an operator message. + +## Completion evidence + +- The readable, case-sensitive `To:` parser is shared by attributed operator + input and normalized worker output. Durable provenance renders exactly as + `From:sender/o:` or `From:sender:`; `From:` and worker-output `Cmd:` lines are + inert. Legacy `@/#` routing, reply/composer code, and generated v1 shims were + removed from the active contract. +- A real tmux PTY fixture echoed an operator `To:` line and then emitted a + distinct worker `To:` line. Exactly two records resulted, with operator and + worker provenance respectively; the echo was suppressed. ANSI normalization, + redraw deduplication, resize, raw mouse input, and byte forwarding are + covered. +- `pushy` now places one sanitized `From:` line without Enter. `trigger` places + the same line and submits exactly once. Startup requires readable framing, + trigger, and line-limit capabilities, and managed sessions without the new + tap marker fail with an explicit recreate instruction. +- `Cmd:` supports the allowlisted modes, three positive per-window limits, and + current-window `reset-limits`. It is operator-only. The selected transparent + contract leaves the command visible to the foreground program; shell + absorbers prevent ordinary-shell command-not-found noise. +- SQLite schema v4 holds atomic counters keyed by session generation and repo. + Tests prove equality blocking, deterministic precedence, overlapping + accounting, per-window isolation, persistence across reopen, fresh session + generations, atomic concurrent admission, reset, and the exact local block + notice. `tamq status` exposes the ledger. +- `make check`: 133 tests passed, including real tmux and the isolated installed + package workflow. `make install` refreshed `tmux-amq==0.1.0`. Installed + `tamq cleanup` dry-run verified the live broker identity, preserved all 26 + durable records, and reported legacy owned shims and stale sockets without + mutation. +- Retry and acknowledgement residuals remain owned by `TAMQ-WP-0003`; + coordination-engine interoperability remains owned by `TAMQ-WP-0002`.