From cd73b954e92bd50d816c1ab9a3e6580e3ca47b84 Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 25 Aug 2026 22:13:32 +0200 Subject: [PATCH] fix: route case-insensitive worker messages Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- README.md | 11 +++++- SCOPE.md | 8 ++-- WORK-RECORDS.md | 1 + src/tamq/broker.py | 29 ++++++++++---- src/tamq/cli.py | 2 +- src/tamq/routing.py | 23 ++++++++++- src/tamq/tmux.py | 8 ++-- tests/test_broker.py | 32 ++++++++++++++- tests/test_ptytap.py | 11 ++++++ tests/test_ptytap_integration.py | 2 +- tests/test_routing.py | 39 +++++++++++++++---- tests/test_tmux_manager.py | 19 +++++++-- workplans/ADHOC-2026-08-25.md | 24 ++++++++++++ ...14-readable-duplex-messaging-and-limits.md | 11 ++++-- 14 files changed, 183 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 61190f9..02ec9ba 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,12 @@ An operator or a worker can emit the same readable line: To:audit-core: Please review the authentication change. ``` +Protocol keywords are case-insensitive, so `to:`, `TO:`, `cmd:`, and their +mixed-case forms work as well. Repository slugs remain exact. For full-screen +terminal programs, tamq also recognizes `To:` immediately after a conservative +worker-output gutter such as the `•` used to frame assistant output; operator +prompt gutters are not worker output. + Operator input is forwarded unchanged and is delivered with `/o`; a line originating in worker output has no suffix: @@ -105,6 +111,8 @@ To:audit-core: please review the auth boundary ``` The same line at the start of worker output queues a worker-originated message. +The keyword may use any letter case, and a recognized full-screen output gutter +may precede it. The `@`, `#`, reply shorthand, and interactive recipient composer from earlier alphas have been removed. @@ -262,7 +270,8 @@ later phase. 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. + normalizes worker output for case-insensitive start-of-content `To:` routing, + including conservative full-screen worker-output gutters. This keeps tmux-specific topology concerns separate from reusable terminal I/O observation and message identity. diff --git a/SCOPE.md b/SCOPE.md index ab61c1b..3427240 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -25,8 +25,8 @@ tamq does not choose or infer them. queue-capable interactive programs. - Explicit opt-in control-mode pane delivery and the full-duplex `tamq tap` PTY broker for integration experiments. -- Exact `gita` validation and direct start-of-line `To:repo:` routing from - operator input or normalized worker output. +- Exact `gita` validation and direct, case-insensitive `To:repo:` keyword + 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 @@ -57,14 +57,14 @@ transport. | Intent capability | State | Evidence and remaining gap | | --- | --- | --- | -| 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. | +| Direct repository addressing | Implemented for local alpha | Exact `gita` validation and case-insensitive `To:repo: body` keyword 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 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. | +| 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, recognizes conservative full-screen worker-output gutters, 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. | diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 5f9c850..1a5a07d 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -27,6 +27,7 @@ | 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-ADHOC-2026-08-25-T02 | done | — | workplans/ADHOC-2026-08-25.md | +| task | TAMQ-WP-ADHOC-2026-08-25-T03 | done | — | workplans/ADHOC-2026-08-25.md | | task | TAMQ-WP-0001-T01 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | | task | TAMQ-WP-0001-T02 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | | task | TAMQ-WP-0001-T03 | done | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | diff --git a/src/tamq/broker.py b/src/tamq/broker.py index 5979c6e..a723bf0 100644 --- a/src/tamq/broker.py +++ b/src/tamq/broker.py @@ -4,7 +4,13 @@ from collections.abc import Callable from dataclasses import dataclass from .registry import RegistryError, validate_targets -from .routing import RoutedMessage, parse_address_line, parse_command_line +from .routing import ( + RoutedMessage, + is_delivery_line, + parse_address_line, + parse_command_line, + parse_worker_address_line, +) from .store import LimitBlock, Store from .terminal import format_delivery @@ -42,8 +48,15 @@ class InputBroker: "in this terminal. Use 'Cmd: reset-limits' to unblock." ) - def _route(self, line: str, provenance: str) -> RoutedMessage | None: - routed = parse_address_line(line) + def _route( + self, + line: str, + provenance: str, + *, + worker_output: bool = False, + ) -> RoutedMessage | None: + parser = parse_worker_address_line if worker_output else parse_address_line + routed = parser(line) if routed is None: return None try: @@ -65,7 +78,7 @@ class InputBroker: def inspect_operator_line(self, line: str) -> RoutedMessage | None: """Observe one submitted operator line; it is still forwarded unchanged.""" - if line.startswith("From:"): + if is_delivery_line(line): return None self.store.count_line( self.identity.endpoint_id, self.identity.source_repo, "input" @@ -78,19 +91,19 @@ class InputBroker: def inspect_worker_line(self, line: str) -> RoutedMessage | None: """Observe worker output; Cmd is intentionally inert on this path.""" - if line.startswith("From:"): + if is_delivery_line(line): return None self.store.count_line( self.identity.endpoint_id, self.identity.source_repo, "output" ) - return self._route(line, "worker_output") + return self._route(line, "worker_output", worker_output=True) 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": + if command.casefold() == "reset-limits": self.store.reset_limits( self.identity.endpoint_id, self.identity.source_repo ) @@ -100,7 +113,9 @@ class InputBroker: self._notice(f"Unknown command: {command}") return name, value = (part.strip() for part in command.split("=", 1)) + name = name.casefold() if name == "mode": + value = value.casefold() if value not in {"inbox", "output", "pushy", "trigger"}: self._notice(f"Invalid mode: {value}") return diff --git a/src/tamq/cli.py b/src/tamq/cli.py index 31a8227..d49113e 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -196,7 +196,7 @@ def build_parser() -> argparse.ArgumentParser: Line-limit defaults are 8 messages, 1024 input, and 32768 output lines. duplex messaging from a managed terminal: - To:TARGET: MESSAGE... operator input or worker output + To:TARGET: MESSAGE... operator input or worker output (case-insensitive) Cmd: mode=trigger operator-only runtime command Cmd: reset-limits reset this window's running counters tamq inbox [--filter COMMAND] diff --git a/src/tamq/routing.py b/src/tamq/routing.py index aed9867..3ea6c00 100644 --- a/src/tamq/routing.py +++ b/src/tamq/routing.py @@ -3,8 +3,15 @@ from __future__ import annotations import re from dataclasses import dataclass -ADDRESS = re.compile(r"^To:([a-z0-9][a-z0-9._-]*):[ \t]*(.+)$") -COMMAND = re.compile(r"^Cmd:[ \t]*(\S(?:.*\S)?)?[ \t]*$") +ADDRESS = re.compile(r"^(?i:to):([a-z0-9][a-z0-9._-]*):[ \t]*(.+)$") +COMMAND = re.compile(r"^(?i:cmd):[ \t]*(\S(?:.*\S)?)?[ \t]*$") +DELIVERY = re.compile(r"^(?i:from):") + +# Full-screen terminal programs commonly reserve one glyph as a presentation +# gutter for program output. These are deliberately narrower than shell and +# operator prompt markers (such as ``$``, ``>``, and ``›``), so an echoed input +# line cannot become worker output merely because the terminal decorated it. +WORKER_OUTPUT_GUTTER = re.compile(r"^[ \t]*[•●▪◆◇◦][ \t]+") @dataclass(frozen=True) @@ -21,9 +28,21 @@ def parse_address_line(line: str) -> RoutedMessage | None: return RoutedMessage(target_repo=match.group(1), body=match.group(2)) +def parse_worker_address_line(line: str) -> RoutedMessage | None: + """Parse an address after a conservative full-screen output gutter.""" + candidate = line.rstrip("\r\n") + candidate = WORKER_OUTPUT_GUTTER.sub("", candidate, count=1) + return parse_address_line(candidate) + + 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) + + +def is_delivery_line(line: str) -> bool: + """Return whether a line is an already-delivered From envelope.""" + return DELIVERY.match(line.rstrip("\r\n")) is not None diff --git a/src/tamq/tmux.py b/src/tamq/tmux.py index 2bd5cf5..4995568 100644 --- a/src/tamq/tmux.py +++ b/src/tamq/tmux.py @@ -92,17 +92,17 @@ class TmuxManager: continue if "# tamq-address-command v1" in content: existing.unlink() + script = "#!/bin/sh\n# tamq-protocol-command v2\nexit 0\n" scripts: dict[str, str] = { - "Cmd:": "#!/bin/sh\n# tamq-protocol-command v2\nexit 0\n", + f"{keyword}:": script for keyword in ("Cmd", "cmd", "CMD", "cMd") } 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}" ) - scripts[f"To:{repo}:"] = ( - "#!/bin/sh\n# tamq-protocol-command v2\nexit 0\n" - ) + for keyword in ("To", "to", "TO", "tO"): + scripts[f"{keyword}:{repo}:"] = script for name in scripts: destination = self.command_dir / name if destination.exists() or destination.is_symlink(): diff --git a/tests/test_broker.py b/tests/test_broker.py index 5f2f08e..dc58db3 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -21,6 +21,23 @@ def test_broker_accepts_worker_output_with_distinct_provenance(tmp_path, monkeyp assert store.list()[0]["provenance"] == "worker_output" +def test_broker_accepts_case_insensitive_framed_worker_output(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None) + store = Store(tmp_path / "queue.sqlite3") + broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom")) + assert broker.inspect_worker_line("• tO:railiance-platform: hello") is not None + row = store.list()[0] + assert row["body"] == "hello" + assert row["provenance"] == "worker_output" + + +def test_case_insensitive_delivery_envelopes_are_not_rescanned(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom")) + assert broker.inspect_worker_line("fRoM:railiance-platform: hello") is None + assert store.line_state("ep", "net-kingdom")["output"] == 0 + + def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("ep", 9, "tamq", ["repo"], "output") @@ -30,11 +47,24 @@ def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path): ) broker.inspect_worker_line("Cmd: mode=trigger") assert store.endpoint("ep")["delivery_mode"] == "output" - broker.inspect_operator_line("Cmd: mode=trigger") + 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_operator_command_names_are_case_insensitive(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep", 9, "tamq", ["repo"], "output") + notices = [] + broker = InputBroker( + store, BrokerIdentity("ep", "repo"), notify=notices.append + ) + broker.inspect_operator_line("CMD: MAXMSG=12") + assert store.line_state("ep", "repo")["maxmsg"] == 12 + broker.inspect_operator_line("cmd: RESET-LIMITS") + assert notices[-1] == "From:tamq: Limits reset for this terminal." + + 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") diff --git a/tests/test_ptytap.py b/tests/test_ptytap.py index 5ab732b..a1bcebb 100644 --- a/tests/test_ptytap.py +++ b/tests/test_ptytap.py @@ -93,6 +93,17 @@ def test_output_observer_fails_closed_on_recent_operator_echo(): assert lines == ["To:target: worker"] +def test_output_observer_preserves_full_screen_program_gutters(): + lines = [] + observer = TerminalOutputObserver(lines.append) + observer.feed( + b"\x1b[4;1H\xe2\x80\xba To:target: operator echo" + b"\x1b[5;1H\xe2\x80\xa2 tO:target: worker reply" + b"\x1b[6;1H" + ) + assert lines == ["› To:target: operator echo", "• tO:target: worker reply"] + + def test_copy_winsize_preserves_rows_columns_and_pixels(): source_master, source_slave = pty.openpty() target_master, target_slave = pty.openpty() diff --git a/tests/test_ptytap_integration.py b/tests/test_ptytap_integration.py index da8dcd3..6560845 100644 --- a/tests/test_ptytap_integration.py +++ b/tests/test_ptytap_integration.py @@ -113,7 +113,7 @@ def test_real_pty_attributes_worker_output_and_suppresses_operator_echo( "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", + " print('\\x1b[5;1H• tO:target: worker reply\\x1b[6;1H', flush=True)\n", encoding="utf-8", ) bin_dir = tmp_path / "bin" diff --git a/tests/test_routing.py b/tests/test_routing.py index 6111d4c..624871a 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -1,10 +1,18 @@ import pytest -from tamq.routing import parse_address_line, parse_command_line +from tamq.routing import ( + is_delivery_line, + parse_address_line, + parse_command_line, + parse_worker_address_line, +) -def test_direct_address_uses_readable_case_sensitive_grammar(): - routed = parse_address_line("To:railiance-platform: do something!") +@pytest.mark.parametrize("keyword", ["to", "To", "TO", "tO"]) +def test_direct_address_keyword_is_case_insensitive(keyword): + routed = parse_address_line( + f"{keyword}:railiance-platform: do something!" + ) assert routed.body == "do something!" assert routed.target_repo == "railiance-platform" @@ -14,7 +22,6 @@ def test_direct_address_uses_readable_case_sensitive_grammar(): [ "@repo: removed", "#repo: removed", - "to:repo: wrong case", "hello To:repo: not at start", "To:repo:", "From:repo: inbound envelope", @@ -24,7 +31,23 @@ 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 +@pytest.mark.parametrize("keyword", ["cmd", "Cmd", "CMD", "cMd"]) +def test_operator_command_keyword_is_case_insensitive(keyword): + assert parse_command_line(f"{keyword}: mode=trigger") == "mode=trigger" + assert parse_command_line(f"{keyword}: reset-limits") == "reset-limits" + + +@pytest.mark.parametrize("keyword", ["from", "From", "FROM", "fRoM"]) +def test_delivery_keyword_is_case_insensitive(keyword): + assert is_delivery_line(f"{keyword}:repo: inbound envelope") is True + + +@pytest.mark.parametrize("gutter", ["• ", " ● ", "▪ ", "◆ "]) +def test_worker_address_accepts_terminal_output_gutter(gutter): + routed = parse_worker_address_line(f"{gutter}tO:target: worker reply") + assert routed == parse_address_line("To:target: worker reply") + + +@pytest.mark.parametrize("gutter", ["› ", "> ", "$ ", "# ", "* ", "- "]) +def test_worker_address_rejects_operator_and_markup_gutters(gutter): + assert parse_worker_address_line(f"{gutter}To:target: operator echo") is None diff --git a/tests/test_tmux_manager.py b/tests/test_tmux_manager.py index 8ae5f10..0971277 100644 --- a/tests/test_tmux_manager.py +++ b/tests/test_tmux_manager.py @@ -161,9 +161,11 @@ 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 / "To:a:").is_file() - assert (command_dir / "To:b:").is_file() - assert (command_dir / "Cmd:").is_file() + for keyword in ("To", "to", "TO", "tO"): + assert (command_dir / f"{keyword}:a:").is_file() + assert (command_dir / f"{keyword}:b:").is_file() + for keyword in ("Cmd", "cmd", "CMD", "cMd"): + assert (command_dir / f"{keyword}:").is_file() assert not any(path.name.startswith("@") for path in command_dir.iterdir()) @@ -174,7 +176,16 @@ def test_protocol_commands_absorb_forwarded_shell_lines(tmp_path): ) manager._install_address_commands(["audit-core"]) - for name in ("To:audit-core:", "Cmd:"): + for name in ( + "To:audit-core:", + "to:audit-core:", + "TO:audit-core:", + "tO:audit-core:", + "Cmd:", + "cmd:", + "CMD:", + "cMd:", + ): result = subprocess.run( [str(command_dir / name), "Some message!", "$value"], text=True, diff --git a/workplans/ADHOC-2026-08-25.md b/workplans/ADHOC-2026-08-25.md index 0afd179..e591db1 100644 --- a/workplans/ADHOC-2026-08-25.md +++ b/workplans/ADHOC-2026-08-25.md @@ -66,3 +66,27 @@ rather than a traceback. - Plain `make` lists `help`, `install`, `uninstall`, `test`, and `check` with descriptions. `make check` passes 136 tests and `make install` refreshes the installed `tmux-amq==0.1.0` tool. + +## Case-insensitive protocol and full-screen worker output + +```task +id: TAMQ-WP-ADHOC-2026-08-25-T03 +status: done +priority: high +``` + +Make the `To:`, `From:`, and `Cmd:` protocol keywords case-insensitive and +restore agent-authored routing when a full-screen terminal interface renders +worker content behind a presentation gutter. + +### Evidence + +- All four representative letter-case variants route for `To:` and parse for + `Cmd:`; delivered `From:` envelopes remain inert regardless of case. Plain + shell panes install matching harmless absorbers for every keyword variant. +- Worker output accepts a conservative full-screen output gutter (including + Codex's `•`) but rejects operator, shell, and markup gutters such as `›`, + `>`, `$`, `#`, `*`, and `-`. +- A real tmux/PTY integration fixture emits a cursor-positioned, mixed-case, + bullet-framed worker message. It is durably attributed as `worker_output` + while the operator echo is suppressed. 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 eb74891..4a3ec21 100644 --- a/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md +++ b/workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md @@ -27,8 +27,9 @@ old address aliases and changes `pushy` so that submission requires the new ## Required protocol contract -The normalized line grammar is case-sensitive and recognized only at the start -of a logical line: +The normalized line keywords are case-insensitive and recognized only at the +start of logical content. A terminal program's conservative worker-output +gutter may precede worker content: | Observed source | Line | Effect | | --- | --- | --- | @@ -254,8 +255,10 @@ 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 +- The readable, case-insensitive `To:` parser is shared by attributed operator + input and normalized worker output. Conservative full-screen worker-output + gutters are removed before worker parsing, while operator prompt gutters stay + inert. 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.