diff --git a/README.md b/README.md index ff9d307..e424cdc 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,12 @@ 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. + ## Exchange messages manually Each managed shell exports its own repository slug as `TAMQ_REPO`. From the diff --git a/SCOPE.md b/SCOPE.md index 1a0b40e..88d235c 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -57,7 +57,7 @@ transport. | 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. | | 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. It marks successful delivery `injected`, but cannot identify pane occupants or protect input already being edited. | +| 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. | | 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. | @@ -98,7 +98,7 @@ Not yet suitable: and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 123 passing tests and 77% statement coverage. Coverage +The suite currently has 126 passing tests and 77% statement coverage. Coverage is strongest in durable storage and registry handling, and weakest in the PTY tap and CLI orchestration; PTY statement coverage increased from 23% to 33%, while subprocess behavior is primarily proven by the real-tmux test. The diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 3a2f554..888807f 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -21,6 +21,7 @@ | workplan | TAMQ-WP-0009 | finished | — | workplans/TAMQ-WP-0009-interactive-recipient-composer.md | | workplan | TAMQ-WP-0010 | finished | — | workplans/TAMQ-WP-0010-experimental-pushy-delivery.md | | 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 | | 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 | @@ -65,3 +66,6 @@ | task | TAMQ-WP-0011-T01 | done | — | workplans/TAMQ-WP-0011-hash-routing-for-pushy-agents.md | | task | TAMQ-WP-0011-T02 | done | — | workplans/TAMQ-WP-0011-hash-routing-for-pushy-agents.md | | task | TAMQ-WP-0011-T03 | done | — | workplans/TAMQ-WP-0011-hash-routing-for-pushy-agents.md | +| task | TAMQ-WP-0012-T01 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | +| task | TAMQ-WP-0012-T02 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | +| task | TAMQ-WP-0012-T03 | done | — | workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md | diff --git a/src/tamq/broker.py b/src/tamq/broker.py index c7ddc5a..11778bf 100644 --- a/src/tamq/broker.py +++ b/src/tamq/broker.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re from dataclasses import dataclass from .routing import RoutedMessage, parse_address_line @@ -7,6 +8,12 @@ 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, +) + + @dataclass(frozen=True) class BrokerIdentity: endpoint_id: str @@ -24,6 +31,15 @@ class InputBroker: 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: diff --git a/src/tamq/cli.py b/src/tamq/cli.py index b6f41a5..256dc87 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -18,7 +18,7 @@ from pathlib import Path from . import __version__ from .config import db_path, pid_path, lock_path, config_path, socket_path, state_dir, setting -from .service import Service, ping, request +from .service import PUSHY_FRAMING_CAPABILITY, Service, ping, request from .store import Store from .tmux import TmuxError, TmuxManager from .broker import BrokerIdentity, InputBroker @@ -130,7 +130,9 @@ def ensure_output_service() -> bool: def ensure_pushy_service() -> bool: - return ensure_service_capabilities({"manual_delivery", "pushy_input"}) + return ensure_service_capabilities( + {"manual_delivery", "pushy_input", PUSHY_FRAMING_CAPABILITY} + ) def preflight_runtime_paths() -> None: diff --git a/src/tamq/service.py b/src/tamq/service.py index 1ebba9f..bcbb323 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -16,6 +16,16 @@ from .control import ControlModeClient from .terminal import format_pushy_input, terminal_frame, write_terminal_output PROTOCOL_VERSION = "0.1" +PUSHY_FRAMING_CAPABILITY = "pushy_input_non_routable_v1" +SERVICE_CAPABILITIES = [ + "register", + "send", + "history", + "manual_delivery", + "terminal_output", + "pushy_input", + PUSHY_FRAMING_CAPABILITY, +] class Service: @@ -144,7 +154,7 @@ class Service: if protocol.split(".")[0] != PROTOCOL_VERSION.split(".")[0]: response = {"ok": False, "error": f"incompatible protocol: {protocol}", "protocol": PROTOCOL_VERSION} elif op == "ping": - response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": ["register", "send", "history", "manual_delivery", "terminal_output", "pushy_input"]} + response = {"ok": True, "protocol": PROTOCOL_VERSION, "capabilities": SERVICE_CAPABILITIES} elif op == "register": required = ("endpoint_id", "pid", "session", "repos") if any(key not in request for key in required): diff --git a/src/tamq/store.py b/src/tamq/store.py index 4f2b096..c403555 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -153,6 +153,11 @@ class Store: where = f" WHERE {' AND '.join(clauses)}" if clauses else "" return list(self.db.execute(f"SELECT * FROM messages{where} ORDER BY created_at", values)) + def message(self, message_id: str) -> sqlite3.Row | None: + return self.db.execute( + "SELECT * FROM messages WHERE message_id=?", (message_id,) + ).fetchone() + def latest_counterparty(self, target: str) -> str | None: row = self.db.execute( "SELECT sender_repo FROM messages " diff --git a/tests/test_broker.py b/tests/test_broker.py index b815d9a..ae955b3 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -19,3 +19,36 @@ def test_broker_accepts_hash_address_alias(tmp_path): assert row["sender_repo"] == "net-kingdom" assert row["target_repo"] == "railiance-platform" assert row["endpoint_id"] == "tmux-amq-42-boot" + + +def test_broker_rejects_legacy_injected_envelope_feedback(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add( + "flex-auth", + "audit-core", + "Hello worker agent!", + endpoint="tmux-amq-42-boot", + ) + broker = InputBroker( + store, BrokerIdentity("tmux-amq-42-boot", "audit-core") + ) + + assert ( + broker.inspect_line( + f"#flex-auth: Hello worker agent! [{message_id}]" + ) + is None + ) + assert len(store.list()) == 1 + + +def test_broker_does_not_block_unknown_receipt_like_user_text(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + broker = InputBroker( + store, BrokerIdentity("tmux-amq-42-boot", "audit-core") + ) + + assert broker.inspect_line( + "#flex-auth: Please inspect [m-00000000-0000-0000-0000-000000000000]" + ) + assert len(store.list()) == 1 diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index 831e986..bed0b09 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -4,6 +4,7 @@ from tamq.cli import ( ensure_pushy_service, parse_size, ) +from tamq.service import PUSHY_FRAMING_CAPABILITY def test_parse_size(): @@ -51,8 +52,13 @@ def test_output_service_requires_terminal_output_capability(monkeypatch): def test_pushy_service_requires_pushy_input_capability(monkeypatch): capabilities = iter([ - ["manual_delivery", "terminal_output"], ["manual_delivery", "terminal_output", "pushy_input"], + [ + "manual_delivery", + "terminal_output", + "pushy_input", + PUSHY_FRAMING_CAPABILITY, + ], ]) starts = [] stops = [] diff --git a/tests/test_protocol.py b/tests/test_protocol.py index ec89606..c46b83a 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,7 +1,7 @@ import asyncio import json -from tamq.service import Service +from tamq.service import PUSHY_FRAMING_CAPABILITY, Service from tamq.store import Store @@ -19,3 +19,24 @@ def test_incompatible_protocol_rejected(tmp_path): finally: server.close(); await server.wait_closed(); service.close() asyncio.run(run()) + + +def test_ping_advertises_non_routable_pushy_framing(tmp_path): + async def run(): + path = tmp_path / "tamq.sock" + service = Service(path, Store(tmp_path / "queue.sqlite3")) + server = await asyncio.start_unix_server(service.handle, path=str(path)) + try: + reader, writer = await asyncio.open_unix_connection(str(path)) + writer.write(b'{"op":"ping"}\n') + await writer.drain() + response = json.loads(await reader.readline()) + assert PUSHY_FRAMING_CAPABILITY in response["capabilities"] + writer.close() + await writer.wait_closed() + finally: + server.close() + await server.wait_closed() + service.close() + + asyncio.run(run()) diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index dee87b2..b68d5bc 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -369,6 +369,23 @@ 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']}]" + ) + manager._run( + "send-keys", + "-t", + f"{session}:activity-core", + "-l", + "--", + legacy_envelope, + ) + manager._run( + "send-keys", "-t", f"{session}:activity-core", "Enter" + ) + time.sleep(0.1) + assert len(store.list()) == 1 + service._deliver_once() time.sleep(0.1) assert len(store.list()) == 1 diff --git a/workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md b/workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md new file mode 100644 index 0000000..d8e73f8 --- /dev/null +++ b/workplans/TAMQ-WP-0012-pushy-feedback-circuit-breaker.md @@ -0,0 +1,76 @@ +--- +id: TAMQ-WP-0012 +type: workplan +title: "Pushy feedback circuit breaker and upgrade gate" +domain: communication +repo: tmux-amq +status: finished +owner: codex +topic_slug: coulomb-social +planning_priority: P0 +planning_order: 17 +created: "2026-08-25" +updated: "2026-08-25" +--- + +# Pushy feedback circuit breaker and upgrade gate + +Stop a live pushy feedback incident, make stale delivery framing detectable, +and reject any legacy injected envelope that re-enters a tapped pane. + +## Halt and characterize the live incident + +```task +id: TAMQ-WP-0012-T01 +status: done +priority: critical +``` + +Stop the broker without changing panes or history and preserve evidence of the +alternating sender/target chain and recursively appended message identifiers. + +## Enforce compatible pushy framing + +```task +id: TAMQ-WP-0012-T02 +status: done +priority: critical +``` + +Advertise and require a capability specific to non-routable pushy envelopes so +startup restarts a broker that still emits the legacy routable `#sender:` form. + +## Add a tap-side circuit breaker and prove containment + +```task +id: TAMQ-WP-0012-T03 +status: done +priority: critical +``` + +Recognize legacy injected envelopes by their durable message identity and +direction, refuse to enqueue them, and cover stale-service restart, false +positive boundaries, one-pass delivery, and isolated real-tmux containment. +Install the corrected build but leave the operator broker stopped and preserve +incident history unless cleanup is separately authorized. + +## Completion evidence + +- Stopped live broker PID 2656594 before diagnosis; panes and durable history + were preserved. Message count stabilized at 105 with 14 pre-existing pending + records and no active endpoint. +- The incident records alternate `flex-auth` and `audit-core`, append the prior + message identifier on every hop, and prove an older `#sender:` pushy envelope + was being routed by newer taps. +- Pushy startup now requires `pushy_input_non_routable_v1`, forcing a one-time + restart of brokers that predate the `# from sender:` framing. +- The broker independently rejects a legacy hash envelope only when its final + message identifier resolves to a durable delivery whose sender and target + are the reverse of the observing tap. Unknown receipt-like user text remains + routable. +- `make check`: 126 tests passed, including real tmux injection of a legacy + envelope followed by proof that no second durable record appears. +- `make install`: refreshed `tmux-amq==0.1.0`. The operator broker remains + stopped; the 82 incident-generated records were not deleted or acknowledged. +- Retry and acknowledgement residuals remain owned by `TAMQ-WP-0003`; this + incident adds no untracked residual.