From 6d2ccc7760dec13a793296ac5b837a79226fc49b Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 26 Aug 2026 08:11:09 +0200 Subject: [PATCH] feat: complete reliable coordination adapter Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc --- .forgejo/workflows/ci-smoke.yaml | 3 +- README.md | 112 ++- SCOPE.md | 37 +- TamqMessagingIntroduction.md | 46 +- WORK-RECORDS.md | 21 +- spec/coordination-engine-adapter-v0.1.md | 110 +++ src/tamq/broker.py | 149 +++- src/tamq/capture.py | 163 ++++ src/tamq/cli.py | 142 +++- src/tamq/client.py | 223 +++++ src/tamq/policy.py | 13 +- src/tamq/protocol.py | 29 + src/tamq/ptytap.py | 2 +- src/tamq/service.py | 172 +++- src/tamq/store.py | 778 +++++++++++++++++- tests/test_broker.py | 13 + tests/test_capture.py | 56 ++ tests/test_cli_helpers.py | 10 +- tests/test_client_adapter.py | 202 +++++ tests/test_config_command.py | 3 + tests/test_install_target.py | 11 +- tests/test_policy.py | 14 +- tests/test_protocol.py | 11 +- tests/test_replay.py | 9 + tests/test_retry.py | 27 + tests/test_service_delivery.py | 132 ++- tests/test_store.py | 50 +- ...AMQ-WP-0002-coordination-engine-adapter.md | 38 +- .../TAMQ-WP-0003-delivery-reliability.md | 35 +- ...AMQ-WP-0016-structured-protocol-capture.md | 86 ++ 30 files changed, 2553 insertions(+), 144 deletions(-) create mode 100644 spec/coordination-engine-adapter-v0.1.md create mode 100644 src/tamq/capture.py create mode 100644 src/tamq/client.py create mode 100644 src/tamq/protocol.py create mode 100644 tests/test_capture.py create mode 100644 tests/test_client_adapter.py create mode 100644 tests/test_retry.py create mode 100644 workplans/TAMQ-WP-0016-structured-protocol-capture.md diff --git a/.forgejo/workflows/ci-smoke.yaml b/.forgejo/workflows/ci-smoke.yaml index bef92ff..5329487 100644 --- a/.forgejo/workflows/ci-smoke.yaml +++ b/.forgejo/workflows/ci-smoke.yaml @@ -16,6 +16,7 @@ jobs: - run: sudo apt-get install -y tmux - run: python -m pip install uv - run: python -m pip install -e '.[dev]' - - run: pytest -q + - run: make check - run: python -m tamq.cli --help - run: python -m tamq.cli --version + - run: python -m tamq.cli capture --help diff --git a/README.md b/README.md index 8817040..d774126 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,8 @@ 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, readable `To:` routing, JSONL export/replay, and a Unix-socket protocol -for a later coordination-engine adapter. +leases, readable `To:` routing, JSONL export/replay, bounded delivery state, and +a versioned Unix-socket coordination-engine adapter. Agents and operators should start with the concise [TAMQ messaging introduction](TamqMessagingIntroduction.md). @@ -182,6 +182,34 @@ the durable id needed by `ack`: From:flex-auth/o: please review the auth boundary ``` +### Reviewing the communication protocol + +TAMQ automatically appends structured protocol events to the local database. +The ledger includes addressed-message acceptance, operator/worker provenance, +worker block boundaries and termination reasons, allowlisted command outcomes, +line-limit blocks, endpoint lifecycle, and delivery attempts or failures. It +does not capture unrelated pane output or ordinary shell input. + +Create a review artifact with: + +```bash +tamq capture --output tamq-protocol.md +tamq capture --repo audit-core --output audit-core-protocol.md +tamq capture --message-id m-... --format jsonl +tamq capture --event delivery.failed +``` + +Markdown is the human-oriented default and starts with aggregate counts useful +for spotting protocol friction. JSONL is intended for scripts. The default is +the newest 500 matching events; `--limit` changes that bound. Repository filters +match either side of an exchange. A message body is included only on its +acceptance event, not repeated for every delivery event. Captures are durable +communication evidence, so never put credentials or secrets in TAMQ messages. +When upgrading an already-running alpha, recreate the managed session once so +its long-lived pane observers use the capture-aware build; existing durable +message lifecycle is backfilled, but old observers cannot reconstruct block or +command events retroactively. + To explicitly consume pending messages through a command, use an inbox filter: ```bash @@ -195,12 +223,26 @@ it and all later messages pending. Filters never run in the background and 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. -Pushy places input without Enter. Trigger waits for terminal paste detection to -settle and then adds exactly one Enter. A short endpoint-startup grace protects -the first input delivery while the foreground program initializes. Both input -modes record accepted placement as `injected`, not recipient acknowledgement. +`--json`. Neither `output` nor `inbox` mode injects terminal keystrokes. Pushy +places input without Enter. Trigger waits for terminal paste detection to settle +and then adds exactly one Enter. A short endpoint-startup grace protects the +first input delivery while the foreground program initializes. + +The policy profile controls completion. The default `injected` policy completes +after a successful terminal write. `acknowledged` keeps the message in +`awaiting_ack`, visible in the inbox, and redelivers the same ID after the ack +deadline. Delivery failures and expired leases use bounded backoff; exhaustion +becomes `failed`. Inspect the attempt count and reason, then explicitly reset a +terminal message if retry is safe: + +```bash +tamq inspect m-... +tamq retry m-... +``` + +Acknowledged-mode redelivery can create duplicate terminal presentation. A late +ack wins even after exhaustion. Neither state proves that an agent completed the +requested work. Every session window has independent, session-lifetime running counters for accepted outbound messages, operator input lines, and normalized worker output @@ -296,17 +338,16 @@ later phase. This keeps tmux-specific topology concerns separate from reusable terminal I/O 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` places a sanitized `From:` block +Normal endpoints use terminal-output delivery. `--mode inbox` selects +inbox-only manual mode. Neither becomes pane input. Experimental `--mode pushy` +places a sanitized `From:` block without Enter; `--mode trigger` performs the same placement, lets paste detection settle, 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`. +`tamq start --tap --command ...` opt-in. Both input paths use the same bounded +retry and acknowledgement state machine. The visible endpoint label is `tmux-amq-`; each boot also receives an instance nonce so PID reuse cannot collide with prior leases or receipts. @@ -324,6 +365,9 @@ purge_before = "365d" purge_max_size = "100MB" history_max_size = "100MB" delivery_poll_interval = "0.5" +delivery_lease_seconds = "30" +delivery_retry_backoff_seconds = "5,15,60,300" +delivery_ack_timeout_seconds = "30" maxmsg = 8 maxin = 1024 maxout = 32768 @@ -331,15 +375,43 @@ maxout = 32768 [policy.profiles.diagnostics] safety_gated_max_attempts = 2 delivery_ack_mode = "acknowledged" +delivery_max_attempts = 4 ``` -Policy profiles accept a safety-gated retry cap of at most nine attempts, but -the alpha delivery path does not enforce attempt counting yet. Explicit -acknowledgement exists, while `delivery_ack_mode` enforcement and bounded retry -state are tracked in `TAMQ-WP-0003`. Use `--policy-profile` to select a profile; -`--orwell` enables explicitly unsafe local diagnostics. +Policy profiles accept one to nine delivery attempts. Lease expiry and write +failure consume an attempt; retry delay uses the configured sequence and then +repeats its last value. `delivery_ack_mode` is either `injected` or +`acknowledged`. Use `--policy-profile` to select a profile; `--orwell` enables +explicitly unsafe local diagnostics. The Unix socket service supports structured `ping`, `register`, `send`, -`history`, `ack`, `endpoints`, and `disconnect` operations. Endpoint +`message`, `history`, `ack`, `retry`, `endpoints`, and `disconnect` operations. Endpoint registrations and messages are persisted in the same local SQLite database; delivery remains delegated to the control-mode adapter. + +## Coordination-engine adapter + +`tamq.client.CoordinationEngineAdapter` is a transport-only async adapter. It +does not import tmux/control-mode code and opens a fresh authenticated +Unix-socket connection per operation. A wake resolves exactly one live endpoint, +validates the gita target, and uses the coordination lease ID as its idempotency +key: + +```python +from tamq.client import CoordinationEngineAdapter, WakeRequest + +receipt = await CoordinationEngineAdapter().wake( + WakeRequest( + lease_id="lease-123", + trigger_id="task-456:r7", + target_repo="audit-core", + prompt="Resume the actionable task and publish a checkpoint.", + ) +) +``` + +Repeating that wake returns the same message ID. Reusing its lease ID for a +different payload is rejected. Receipt recovery, explicit acknowledgement, and +terminal retry use `receipt`, `acknowledge`, and `retry_failed`. The normative +state and ownership contract is +[`spec/coordination-engine-adapter-v0.1.md`](spec/coordination-engine-adapter-v0.1.md). diff --git a/SCOPE.md b/SCOPE.md index af94ae6..81acce4 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -14,7 +14,8 @@ tamq does not choose or infer them. ## In Scope - Local SQLite message history, leases, endpoint registrations, delivery state, - acknowledgements, replay, export, and bounded purging. + acknowledgements, replay, export, bounded purging, and a structured + communication-protocol ledger for review. - Managed neutral-shell tmux lifecycle and explicit initial commands. - Durable send/inbox/acknowledgement with per-window repository identity, readable `To:`/`From:` framing, explicit operator/worker provenance, and @@ -30,8 +31,8 @@ tamq does not choose or infer them. an empty line. - 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. +- Versioned Unix-socket operations and a transport-only coordination-engine + client adapter with correlation and idempotent wake admission. - Policy profiles, safety-gated retries, local diagnostics, tests, packaging, shell completion, and operator documentation. - Dry-run emergency cleanup for verified services, tamq-marked sessions, @@ -63,13 +64,14 @@ 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. | | 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, non-routable `From:` block above a stable shell input row without injecting stdin; `/o` marks operator origin. Messages remain pending until acknowledgement. Inbox-only mode is explicit. | +| Safe output messaging | Implemented for local alpha | Normal endpoints write one sanitized, non-routable `From:` block above a stable shell input row without injecting stdin; `/o` marks operator origin. Inbox-only mode is explicit. The selected acknowledgement policy determines completion. | | Experimental pushy and trigger delivery | Explicit opt-in | Pushy places a non-routable `From:` block without Enter; trigger waits past paste detection and adds exactly one Enter. A startup grace protects first delivery. 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; collects worker `To:` blocks through an empty row, recognizes conservative full-screen worker-output gutters and row gaps, 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`. | +| Protocol review capture | Implemented for local alpha | An append-only SQLite ledger records addressed-message acceptance, provenance, worker block boundaries, allowlisted command outcomes, endpoint changes, limit blocks, and delivery outcomes. `tamq capture` renders filtered Markdown or JSONL without recording unrelated pane output or shell input. | +| Bounded retry behavior | Implemented for local alpha | Lease acquisition increments a persistent attempt count. Failures and lease expiry schedule bounded backoff; cap exhaustion becomes inspectable `failed`, and `tamq retry` explicitly resets a terminal message. | +| Acknowledgement policy | Implemented for local alpha | `injected` completes on successful terminal write. `acknowledged` enters `awaiting_ack`, redelivers the same message ID after a deadline, exhausts with `ack_timeout`, accepts late acknowledgements, and exposes duplicate-delivery semantics. | +| Coordination-engine interoperability | Implemented transport boundary | The versioned same-user Unix-socket client negotiates capabilities, resolves an exact live endpoint, admits idempotent correlated wakes, recovers receipts after restart, and exposes ack/retry without importing tmux control code. Coordination-engine still owns its orchestration runtime. | ## Practical Usability @@ -89,22 +91,24 @@ Suitable today: gita-registered repositories. - Durable message exchange between managed repository windows, including explicit pull-time loggers and filters. +- Reviewing repository-to-repository exchanges and onboarding friction through + structured Markdown or JSONL protocol captures. - 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. +- Integrating coordination-engine through the implemented transport adapter and + durable receipt boundary. Not yet suitable: -- Unattended pane injection where bounded retries and positive recipient - acknowledgement are required. +- Treating terminal delivery or acknowledgement as proof of completed work; + workflow completion remains coordination-engine/State Hub state. - Pushy delivery to arbitrary shells, editors, or panes with unknown input state. - Production-style operation without longer-running crash/terminal soak tests and stronger process-supervision evidence. - Cross-host messaging or use as a general-purpose broker. -The suite currently has 166 passing tests. It includes atomic counter races, +The suite currently has 182 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 @@ -112,10 +116,11 @@ checks on Python 3.11. ## Next Usability Gates -- `TAMQ-WP-0003-T01` and `T02` still own bounded retry state and acknowledgement - enforcement. Its real tmux/PTY lifecycle gate (`T03`) is complete. -- `TAMQ-WP-0002` owns the coordination-engine adapter after the local delivery - contract is sufficiently reliable. +- Operational soak evidence is still needed before treating input delivery as + production-grade or unattended across arbitrary terminal applications. +- Coordination-engine owns implementing its trigger observer, coordination + leases, safety gates, checkpoints, and State Hub projections on top of this + completed local transport boundary. ## Getting Oriented diff --git a/TamqMessagingIntroduction.md b/TamqMessagingIntroduction.md index 7f34466..2e9e3ae 100644 --- a/TamqMessagingIntroduction.md +++ b/TamqMessagingIntroduction.md @@ -75,8 +75,9 @@ The endpoint mode controls what delivery means: New pushy/trigger endpoints have a short startup grace so their foreground program can initialize before the first delivery. These modes still cannot prove that an arbitrary program accepted or understood the input. A durable -state of `injected` records successful terminal placement, not agent -acknowledgement. +state of `injected` records completion under the default terminal-write policy, +not agent comprehension or task completion. An `acknowledged` policy instead +retains the message in `awaiting_ack` until explicit acknowledgement. An operator can change the live mode: @@ -112,6 +113,38 @@ tamq inbox --repo audit-core tamq inspect tamq ack tamq send --from audit-core 'To:flex-auth: Direct CLI message.' +tamq capture --repo audit-core --output tamq-protocol.md +``` + +TAMQ automatically keeps a structured protocol ledger for review. It records +addressed-message acceptance, operator/worker provenance, worker block start +and close reasons, allowlisted `Cmd:` outcomes, limit blocks, endpoint changes, +and delivery attempts or failures. It does **not** record unrelated shell input +or arbitrary pane output. Message bodies appear once, at their durable +acceptance event. + +Use Markdown for a human review or JSONL for analysis: + +```bash +tamq capture --output tamq-protocol.md +tamq capture --repo audit-core --format jsonl --output audit-core-protocol.jsonl +tamq capture --message-id m-... --output one-message.md +tamq capture --event delivery.failed +``` + +The default report contains the newest 500 matching events and a summary of +message provenance, attempts, failures, blocks, and outcomes. Filters match a +repository as sender or recipient. Do not put secrets in TAMQ messages: durable +history and captures intentionally retain addressed message content. + +After first installing capture support over an older running alpha, recreate +the managed session once when convenient. Existing durable message lifecycle +is backfilled, but old already-running pane observers cannot emit worker-block +and command events retroactively: + +```bash +make cleanup +tamq --command codex --mode trigger flex-auth audit-core ``` If no reply arrives, distinguish these states before resending: @@ -119,9 +152,12 @@ If no reply arrives, distinguish these states before resending: - no durable history row: the source line or completed worker block was not observed, the slug was invalid, or a line budget blocked it; - `pending`: no eligible live endpoint has completed the selected delivery; -- `displayed`: output mode showed it, but did not submit it to the worker; -- `injected`: pushy/trigger placement succeeded, but the worker has not - necessarily processed it; +- `awaiting_ack`: terminal delivery succeeded, but explicit acknowledgement is + outstanding and same-ID redelivery may occur after the deadline; +- `injected`: the selected terminal-write policy completed, but the worker has + not necessarily processed it; +- `failed`: bounded delivery or acknowledgement attempts were exhausted; inspect + the reason and use `tamq retry ` only when redelivery is safe; - explicit reply or acknowledgement: the recipient confirmed receipt. Use a correlation token when duplicate requests would be harmful. TAMQ is an diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 37d592b..c462dc8 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -11,8 +11,8 @@ | workplan | TAMQ-WP-ADHOC-2026-08-24 | finished | — | workplans/ADHOC-2026-08-24.md | | workplan | TAMQ-WP-ADHOC-2026-08-25 | finished | — | workplans/ADHOC-2026-08-25.md | | workplan | TAMQ-WP-0001 | finished | — | workplans/TAMQ-WP-0001-statehub-bootstrap.md | -| workplan | TAMQ-WP-0002 | active | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | -| workplan | TAMQ-WP-0003 | active | — | workplans/TAMQ-WP-0003-delivery-reliability.md | +| workplan | TAMQ-WP-0002 | finished | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | +| workplan | TAMQ-WP-0003 | finished | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | workplan | TAMQ-WP-0004 | finished | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | workplan | TAMQ-WP-0005 | finished | — | workplans/TAMQ-WP-0005-terminal-neutral-manual-messaging.md | | workplan | TAMQ-WP-0006 | finished | — | workplans/TAMQ-WP-0006-shell-native-message-routing.md | @@ -25,6 +25,7 @@ | workplan | TAMQ-WP-0013 | finished | — | workplans/TAMQ-WP-0013-emergency-cleanup.md | | workplan | TAMQ-WP-0014 | finished | — | workplans/TAMQ-WP-0014-readable-duplex-messaging-and-limits.md | | workplan | TAMQ-WP-0015 | finished | — | workplans/TAMQ-WP-0015-reliable-trigger-and-worker-blocks.md | +| workplan | TAMQ-WP-0016 | finished | — | workplans/TAMQ-WP-0016-structured-protocol-capture.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-ADHOC-2026-08-25-T02 | done | — | workplans/ADHOC-2026-08-25.md | @@ -33,13 +34,13 @@ | 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 | -| task | TAMQ-WP-0002-T01 | todo | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | -| task | TAMQ-WP-0002-T02 | wait | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | -| task | TAMQ-WP-0002-T03 | wait | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | -| task | TAMQ-WP-0003-T01 | todo | — | workplans/TAMQ-WP-0003-delivery-reliability.md | -| task | TAMQ-WP-0003-T02 | wait | — | workplans/TAMQ-WP-0003-delivery-reliability.md | +| task | TAMQ-WP-0002-T01 | done | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | +| task | TAMQ-WP-0002-T02 | done | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | +| task | TAMQ-WP-0002-T03 | done | — | workplans/TAMQ-WP-0002-coordination-engine-adapter.md | +| task | TAMQ-WP-0003-T01 | done | — | workplans/TAMQ-WP-0003-delivery-reliability.md | +| task | TAMQ-WP-0003-T02 | done | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | task | TAMQ-WP-0003-T03 | done | — | workplans/TAMQ-WP-0003-delivery-reliability.md | -| task | TAMQ-WP-0003-T04 | wait | — | workplans/TAMQ-WP-0003-delivery-reliability.md | +| task | TAMQ-WP-0003-T04 | done | — | workplans/TAMQ-WP-0003-delivery-reliability.md | | task | TAMQ-WP-0004-T01 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | task | TAMQ-WP-0004-T02 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | | task | TAMQ-WP-0004-T03 | done | — | workplans/TAMQ-WP-0004-operator-installable-local-alpha.md | @@ -89,3 +90,7 @@ | task | TAMQ-WP-0015-T02 | done | — | workplans/TAMQ-WP-0015-reliable-trigger-and-worker-blocks.md | | task | TAMQ-WP-0015-T03 | done | — | workplans/TAMQ-WP-0015-reliable-trigger-and-worker-blocks.md | | task | TAMQ-WP-0015-T04 | done | — | workplans/TAMQ-WP-0015-reliable-trigger-and-worker-blocks.md | +| task | TAMQ-WP-0016-T01 | done | — | workplans/TAMQ-WP-0016-structured-protocol-capture.md | +| task | TAMQ-WP-0016-T02 | done | — | workplans/TAMQ-WP-0016-structured-protocol-capture.md | +| task | TAMQ-WP-0016-T03 | done | — | workplans/TAMQ-WP-0016-structured-protocol-capture.md | +| task | TAMQ-WP-0016-T04 | done | — | workplans/TAMQ-WP-0016-structured-protocol-capture.md | diff --git a/spec/coordination-engine-adapter-v0.1.md b/spec/coordination-engine-adapter-v0.1.md new file mode 100644 index 0000000..50764ea --- /dev/null +++ b/spec/coordination-engine-adapter-v0.1.md @@ -0,0 +1,110 @@ +# Coordination-engine ↔ TAMQ adapter contract v0.1 + +Status: implemented local-alpha contract. TAMQ owns this transport contract; +coordination-engine owns trigger policy, coordination leases, checkpoints, and +workflow state. + +## Boundary and authentication + +The adapter connects to TAMQ's configured Unix-domain socket. The socket is +mode `0600`, and TAMQ accepts only the same Unix user through `SO_PEERCRED` when +the platform exposes it. There is no TCP listener or command-line bearer token. +The client opens a new connection for every operation, so restarting TAMQ needs +no reconnect handshake or client-side session state. + +The client first sends `ping` with protocol `0.1`. Major versions must match and +the service must advertise `bounded_delivery_ack_v1` and `idempotent_send_v1`. +Unknown minor capabilities are ignored. An incompatible major version or a +missing required capability stops the wake before message admission. + +## Wake mapping + +`CoordinationEngineAdapter.wake(WakeRequest)` resolves exactly one live endpoint +whose registered repository list includes `target_repo`. Zero matches returns +unavailable. Multiple matches require an explicit endpoint ID; TAMQ never +chooses an arbitrary worker session. + +The wake becomes a `send` request: + +```json +{ + "op": "send", + "protocol": "0.1", + "client_id": "coordination-engine", + "idempotency_key": "", + "correlation_id": "", + "sender_repo": "coordination-engine", + "target_repo": "", + "endpoint_id": "", + "provenance": "coordination_engine", + "metadata": {"lease_id": "...", "trigger_id": "..."}, + "body": "" +} +``` + +Targets and non-local senders are validated against the current `gita` +registry. Bodies are UTF-8 text up to 8 KiB; metadata is a JSON object up to 4 +KiB. TAMQ transports the prompt but does not interpret it as task state or +grant authority. + +`(client_id, idempotency_key)` is unique. Repeating an identical request returns +the original message ID with `deduplicated: true`. Reusing the key with a +different sender, target, endpoint, or body is an error. The durable identity +used by coordination-engine is therefore the pair of TAMQ endpoint instance +and local message ID, correlated to its own lease and trigger IDs. + +## Message and delivery states + +| State | Meaning | Coordination interpretation | +| --- | --- | --- | +| `pending` | Accepted and eligible now or after `next_attempt_at` | Transport owns retry timing. | +| `awaiting_ack` | Written successfully under explicit-ack policy | Recipient comprehension is still unproven. | +| `injected` | Written successfully under injected policy | Transport delivery is complete, not task completion. | +| `acknowledged` | Explicit acknowledgement received | Strongest TAMQ receipt; still not workflow completion. | +| `failed` | Delivery or acknowledgement attempts exhausted | Terminal until an operator/client explicitly retries or a late ack arrives. | + +Every delivery lease acquisition increments the persistent `attempt_count`. +Failure releases the lease and sets `last_failure_reason` plus a bounded +`next_attempt_at`. Lease expiry is itself a failed attempt. Retry delays default +to 5, 15, 60, and 300 seconds and use the last value thereafter. The selected +endpoint policy supplies `delivery_max_attempts` in the range 1–9. + +With `delivery_ack_mode=injected`, a successful terminal write completes the +message immediately. With `acknowledged`, success enters `awaiting_ack`; the +default 30-second deadline then redelivers the same message ID. Duplicate visual +or input delivery is therefore possible and recipients must correlate by +message ID when the operation is not naturally idempotent. Exhausting the cap +while waiting produces `failed` with reason `ack_timeout`. A late explicit ack +wins even after terminal failure and cancels any outstanding retry. `retry` +resets only a `failed` message's attempts and returns it to `pending`. + +## Operations and failure semantics + +- `endpoints`: discover live repository attachment and disambiguate a wake. +- `send`: admit an idempotent durable message. +- `message`: retrieve one receipt including state, attempts, deadlines, and + correlation metadata. +- `history`: recover receipts after either process restarts. +- `ack`: record explicit or late acknowledgement and cancel pending delivery. +- `retry`: operator/policy-authorized reset of terminal delivery failure. + +Endpoint disappearance before admission is `unavailable` and creates no +message. Disappearance after admission leaves the message durable. Repeating the +identical wake after a replacement endpoint appears rebinds that same message +ID, but only when its previous endpoint is no longer live; TAMQ will not move a +message between two live sessions implicitly. A process crash after claiming a +message is recovered through lease expiry and bounded backoff. + +JSONL replay uses the original message ID (or a deterministic content digest) +as a `tamq-replay` idempotency key. Repeating the same replay reports it as +deduplicated rather than creating another message. Replay still uses normal +gita validation and never changes coordination-engine workflow state. + +## Ownership + +TAMQ owns local queue rows, endpoint registration, delivery leases, retry and +ack state, terminal interaction, and protocol capture. Coordination-engine owns +trigger deduplication, coordination leases, actionability and safety policy, +checkpoints, State Hub projection, and the decision to issue a wake or request +an explicit retry. Neither system treats terminal injection as proof that an +agent completed work. diff --git a/src/tamq/broker.py b/src/tamq/broker.py index e04b1d1..ffe3383 100644 --- a/src/tamq/broker.py +++ b/src/tamq/broker.py @@ -2,6 +2,7 @@ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass +import json from .registry import RegistryError, validate_targets from .routing import ( @@ -38,8 +39,30 @@ class InputBroker: self.notify = notify self._worker_target: str | None = None self._worker_lines: list[str] = [] + self._last_admitted_message_id: str | None = None self.store.configure_window(identity.endpoint_id, identity.source_repo, limits) + def _event( + self, + event_type: str, + *, + peer_repo: str | None = None, + message_id: str | None = None, + provenance: str | None = None, + outcome: str = "observed", + detail: dict | None = None, + ) -> None: + self.store.record_protocol_event( + event_type, + endpoint_id=self.identity.endpoint_id, + repo=self.identity.source_repo, + peer_repo=peer_repo, + message_id=message_id, + provenance=provenance, + outcome=outcome, + detail=detail, + ) + def _notice(self, text: str) -> None: if self.notify is not None: self.notify(f"From:tamq: {text}") @@ -52,9 +75,16 @@ class InputBroker: ) def _admit(self, routed: RoutedMessage, provenance: str) -> RoutedMessage | None: + self._last_admitted_message_id = None try: validate_targets([routed.target_repo]) except RegistryError: + self._event( + "message.rejected", + peer_repo=routed.target_repo, + provenance=provenance, + outcome="invalid_target", + ) return None result = self.store.admit_message( self.identity.source_repo, @@ -67,6 +97,7 @@ class InputBroker: if isinstance(result, LimitBlock): self._blocked(result) return None + self._last_admitted_message_id = result return routed def _route(self, line: str, provenance: str) -> RoutedMessage | None: @@ -77,21 +108,44 @@ class InputBroker: try: validate_targets([routed.target_repo]) except RegistryError: + self._event( + "worker.block_rejected", + peer_repo=routed.target_repo, + provenance="worker_output", + outcome="invalid_target", + ) return self._worker_target = routed.target_repo self._worker_lines = [routed.body] + self._event( + "worker.block_started", + peer_repo=routed.target_repo, + provenance="worker_output", + outcome="collecting", + ) - def flush_worker_message(self) -> RoutedMessage | None: + def flush_worker_message(self, reason: str = "explicit_flush") -> RoutedMessage | None: """Admit the worker block accumulated through its empty terminator.""" if self._worker_target is None: return None + target = self._worker_target + line_count = len(self._worker_lines) routed = RoutedMessage( - target_repo=self._worker_target, + target_repo=target, body="\n".join(self._worker_lines), ) self._worker_target = None self._worker_lines = [] - return self._admit(routed, "worker_output") + admitted = self._admit(routed, "worker_output") + self._event( + "worker.block_closed", + peer_repo=target, + message_id=self._last_admitted_message_id, + provenance="worker_output", + outcome="accepted" if admitted is not None else "not_accepted", + detail={"reason": reason, "line_count": line_count}, + ) + return admitted def inspect_operator_line(self, line: str) -> RoutedMessage | None: """Observe one submitted operator line; it is still forwarded unchanged.""" @@ -117,14 +171,14 @@ class InputBroker: ) routed = parse_worker_address_line(line) if routed is not None: - previous = self.flush_worker_message() + previous = self.flush_worker_message("new_address") self._start_worker_message(routed) return previous if self._worker_target is None: return None content = worker_content_line(line).strip() if not content: - return self.flush_worker_message() + return self.flush_worker_message("empty_line") self._worker_lines.append(content) return None @@ -137,9 +191,20 @@ class InputBroker: self.store.reset_limits( self.identity.endpoint_id, self.identity.source_repo ) + self._event( + "command.applied", + provenance="operator_input", + outcome="applied", + detail={"command": "reset-limits"}, + ) self._notice("Limits reset for this terminal.") return if "=" not in command: + self._event( + "command.rejected", + provenance="operator_input", + outcome="unknown_command", + ) self._notice(f"Unknown command: {command}") return name, value = (part.strip() for part in command.split("=", 1)) @@ -147,11 +212,32 @@ class InputBroker: if name == "mode": value = value.casefold() if value not in {"inbox", "output", "pushy", "trigger"}: + self._event( + "command.rejected", + provenance="operator_input", + outcome="invalid_value", + detail={"command": "mode"}, + ) self._notice(f"Invalid mode: {value}") return if not self.store.set_endpoint_mode(self.identity.endpoint_id, value): + self._event( + "command.rejected", + provenance="operator_input", + outcome="endpoint_missing", + detail={"command": "mode"}, + ) self._notice("Cannot change mode before endpoint registration.") return + self.store.record_protocol_event( + "command.applied", + endpoint_id=self.identity.endpoint_id, + repo=self.identity.source_repo, + provenance="operator_input", + delivery_mode=value, + outcome="applied", + detail={"command": "mode", "value": value}, + ) self._notice(f"Mode set to {value}.") return if name in {"maxmsg", "maxin", "maxout"}: @@ -164,21 +250,50 @@ class InputBroker: parsed, ) except ValueError as exc: + self._event( + "command.rejected", + provenance="operator_input", + outcome="invalid_value", + detail={"command": name}, + ) self._notice(str(exc)) return + self._event( + "command.applied", + provenance="operator_input", + outcome="applied", + detail={"command": name, "value": parsed}, + ) self._notice(f"{name} set to {parsed} for this terminal.") return + self._event( + "command.rejected", + provenance="operator_input", + outcome="unknown_command", + ) self._notice(f"Unknown command: {name}") def deliver_pending(self, control, *, window_for_repo): """Place pending messages through the authoritative control client.""" delivered = 0 - for row in self.store.list(state="pending"): + endpoint = self.store.endpoint(self.identity.endpoint_id) + repos = ( + json.loads(endpoint["repos"]) + if endpoint is not None + else [row["target_repo"] for row in self.store.list()] + ) + for row in self.store.deliverable(self.identity.endpoint_id, repos): if row["endpoint_id"] not in (None, self.identity.endpoint_id): continue target = window_for_repo(row["target_repo"]) - lease_id = self.store.claim(row["message_id"], self.identity.endpoint_id) - if lease_id is None: + claim = self.store.claim_delivery( + row["message_id"], + self.identity.endpoint_id, + attempt_limit=(int(endpoint["delivery_max_attempts"]) if endpoint is not None else 4), + retry_delay=5.0, + delivery_mode="pane", + ) + if claim is None: continue try: control.inject( @@ -187,8 +302,20 @@ class InputBroker: row["sender_repo"], row["body"], row["provenance"] ), ) - except Exception: + except Exception as exc: + self.store.fail_delivery( + row["message_id"], + claim.lease_id, + type(exc).__name__, + delivery_mode="pane", + ) continue - self.store.release(row["message_id"], lease_id, "injected") - delivered += 1 + if self.store.complete_delivery( + row["message_id"], + claim.lease_id, + delivery_mode="pane", + ack_mode=(endpoint["delivery_ack_mode"] if endpoint is not None else "injected"), + ack_timeout=(float(endpoint["ack_timeout_seconds"]) if endpoint is not None else 30.0), + ): + delivered += 1 return delivered diff --git a/src/tamq/capture.py b/src/tamq/capture.py new file mode 100644 index 0000000..0d8ca3a --- /dev/null +++ b/src/tamq/capture.py @@ -0,0 +1,163 @@ +"""Render the structured TAMQ protocol ledger for review.""" + +from __future__ import annotations + +import json +from collections import Counter +from datetime import datetime, timezone +from typing import Iterable, Mapping + + +def _timestamp(value: float) -> str: + return datetime.fromtimestamp(value, timezone.utc).isoformat(timespec="milliseconds") + + +def _safe_text(value: str) -> str: + """Keep report structure readable without treating terminal controls as evidence.""" + return "".join( + character + for character in value + if character in "\n\t" or ord(character) >= 32 + ) + + +def event_record(row: Mapping) -> dict: + """Build a stable JSON record and include a body only at queue acceptance.""" + record = { + "sequence": row["sequence"], + "event_id": row["event_id"], + "occurred_at": _timestamp(float(row["occurred_at"])), + "event_type": row["event_type"], + "endpoint_id": row["endpoint_id"], + "repo": row["repo"], + "peer_repo": row["peer_repo"], + "message_id": row["message_id"], + "provenance": row["provenance"], + "delivery_mode": row["delivery_mode"], + "outcome": row["outcome"], + "detail": json.loads(row["detail"]), + } + if row["event_type"] == "message.accepted" and row["message_body"] is not None: + record["message_body"] = _safe_text(row["message_body"]) + return record + + +def render_jsonl(rows: Iterable[Mapping]) -> str: + return "".join( + json.dumps(event_record(row), sort_keys=True) + "\n" for row in rows + ) + + +def render_markdown(rows: Iterable[Mapping], *, scope: str = "all repositories") -> str: + events = list(rows) + types = Counter(row["event_type"] for row in events) + outcomes = Counter(row["outcome"] for row in events) + provenances = Counter( + row["provenance"] for row in events if row["provenance"] is not None + ) + generated = datetime.now(timezone.utc).isoformat(timespec="seconds") + lines = [ + "# TAMQ protocol capture", + "", + f"Generated: `{generated}` ", + f"Scope: {scope} ", + f"Events: {len(events)}", + "", + "This report contains addressed TAMQ messages and structured lifecycle", + "metadata. It does not contain unrelated pane output or shell input.", + "", + "## Summary", + "", + f"- Accepted messages: {types['message.accepted']}", + f"- Delivery attempts: {types['delivery.attempted']}", + f"- Delivery failures: {types['delivery.failed']}", + f"- Limit blocks: {types['message.blocked']}", + ] + if provenances: + lines.append( + "- Message provenance: " + + ", ".join(f"{name}={count}" for name, count in sorted(provenances.items())) + ) + if outcomes: + lines.append( + "- Outcomes: " + + ", ".join(f"{name}={count}" for name, count in sorted(outcomes.items())) + ) + close_reasons = Counter() + for row in events: + if row["event_type"] != "worker.block_closed": + continue + reason = json.loads(row["detail"]).get("reason") + if reason: + close_reasons[reason] += 1 + lines.extend(["", "## Review cues", ""]) + cues: list[str] = [] + open_blocks = types["worker.block_started"] - types["worker.block_closed"] + if open_blocks > 0: + cues.append( + f"{open_blocks} worker block(s) started without a captured close; review " + "blank-line termination and worker shutdown handling." + ) + non_blank_closes = close_reasons["new_address"] + close_reasons["worker_exit"] + if non_blank_closes: + cues.append( + f"{non_blank_closes} worker block(s) closed on a new address or worker exit; " + "an explicit blank-line example may improve onboarding." + ) + if types["command.rejected"]: + cues.append( + f"{types['command.rejected']} operator command(s) were rejected; review command discoverability." + ) + if types["message.rejected"] + types["worker.block_rejected"]: + cues.append( + f"{types['message.rejected'] + types['worker.block_rejected']} address(es) were rejected; " + "review repository-slug discovery." + ) + if types["message.blocked"]: + cues.append( + f"{types['message.blocked']} message(s) hit a line budget; review limits and reset guidance." + ) + if types["delivery.failed"]: + cues.append( + f"{types['delivery.failed']} delivery attempt(s) failed; group JSONL by mode and failure reason." + ) + if not cues: + cues.append( + "No rejected commands, invalid addresses, limit blocks, unclosed worker blocks, " + "or delivery failures are visible in this selection." + ) + lines.extend(f"- {cue}" for cue in cues) + lines.extend(["", "## Timeline", ""]) + if not events: + lines.extend(["No matching protocol events.", ""]) + return "\n".join(lines) + + for row in events: + lines.append( + f"### {_timestamp(float(row['occurred_at']))} — `{row['event_type']}`" + ) + lines.append("") + fields = ( + ("Outcome", row["outcome"]), + ("Repository", row["repo"]), + ("Counterparty", row["peer_repo"]), + ("Provenance", row["provenance"]), + ("Mode", row["delivery_mode"]), + ("Endpoint", row["endpoint_id"]), + ("Message", row["message_id"]), + ("Event", row["event_id"]), + ) + for label, value in fields: + if value is not None: + lines.append(f"- {label}: `{_safe_text(str(value))}`") + detail = json.loads(row["detail"]) + if detail: + lines.append( + "- Detail: `" + _safe_text(json.dumps(detail, sort_keys=True)) + "`" + ) + if row["event_type"] == "message.accepted" and row["message_body"] is not None: + lines.extend(["", "Message body:", ""]) + body_lines = _safe_text(row["message_body"]).splitlines() or [""] + lines.extend(f" {line}" for line in body_lines) + lines.append("") + return "\n".join(lines) diff --git a/src/tamq/cli.py b/src/tamq/cli.py index 9617b81..a9a3502 100644 --- a/src/tamq/cli.py +++ b/src/tamq/cli.py @@ -9,6 +9,7 @@ import os import signal import sqlite3 import fcntl +import hashlib from uuid import uuid4 import subprocess import sys @@ -20,6 +21,9 @@ from . import __version__ from .config import db_path, pid_path, lock_path, config_path, socket_path, state_dir, setting from .service import ( LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, PUSHY_FRAMING_CAPABILITY, TRIGGER_CAPABILITY, Service, @@ -37,10 +41,11 @@ from .policy import load_profile from .registry import RegistryError, validate_targets from .routing import parse_address_line from .terminal import format_delivery +from .capture import render_jsonl, render_markdown SUBCOMMANDS = frozenset( - "start attach serve stop cleanup status ping inbox history inspect ack send export replay purge completion db-version config tap".split() + "start attach serve stop cleanup status ping inbox history capture inspect ack retry send export replay purge completion db-version config tap".split() ) START_OPTIONS = frozenset({"--command", "--cmd", "--tap", "--detach", "--no-service", "--no-display", "--mode", "--maxmsg", "--maxin", "--maxout"}) GLOBAL_FLAGS = frozenset({"--orwell", "--verbose"}) @@ -131,13 +136,28 @@ def ensure_service_capabilities(required: set[str]) -> bool: def ensure_manual_service() -> bool: return ensure_service_capabilities( - {"manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY} + { + "manual_delivery", + PUSHY_FRAMING_CAPABILITY, + LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, + } ) def ensure_output_service() -> bool: return ensure_service_capabilities( - {"manual_delivery", "terminal_output", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY} + { + "manual_delivery", + "terminal_output", + PUSHY_FRAMING_CAPABILITY, + LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, + } ) @@ -149,6 +169,9 @@ def ensure_pushy_service() -> bool: PUSHY_FRAMING_CAPABILITY, TRIGGER_CAPABILITY, LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, } ) @@ -202,6 +225,7 @@ duplex messaging from a managed terminal: Cmd: mode=trigger operator-only runtime command Cmd: reset-limits reset this window's running counters tamq inbox [--filter COMMAND] + tamq capture --output tamq-protocol.md """, ) parser.add_argument("--version", "-V", action="version", version=__version__) @@ -241,6 +265,27 @@ duplex messaging from a managed terminal: history = subparsers.add_parser("history", help="inspect local message history") history.add_argument("--repo", dest="target_repo") history.add_argument("--state") + capture = subparsers.add_parser( + "capture", + help="render the structured communication protocol ledger for review", + ) + capture.add_argument("--repo", help="include events where this repository is either party") + capture.add_argument("--endpoint-id", help="include only one endpoint instance") + capture.add_argument("--message-id", help="include only one durable message lifecycle") + capture.add_argument("--event", dest="event_type", help="include only one event type") + capture.add_argument( + "--format", + choices=("markdown", "jsonl"), + default="markdown", + help="review format (default: markdown)", + ) + capture.add_argument( + "--limit", + type=int, + default=500, + help="newest matching events to include (default: 500)", + ) + capture.add_argument("--output", type=Path, help="write to a file instead of stdout") inbox = subparsers.add_parser("inbox", help="show durable messages for a repository without injecting terminal input") inbox.add_argument("--repo", dest="target_repo", help="target repository (default: TAMQ_REPO in a managed window)") inbox.add_argument("--all", action="store_true", help="include non-pending messages") @@ -254,6 +299,8 @@ duplex messaging from a managed terminal: inspect.add_argument("message_id") ack = subparsers.add_parser("ack", help="acknowledge one durable message") ack.add_argument("message_id") + retry = subparsers.add_parser("retry", help="reset one terminally failed message for bounded redelivery") + retry.add_argument("message_id") send = subparsers.add_parser("send", help="queue a direct message") send.add_argument("address", help="To:repo: message") send.add_argument("body", nargs="*", help="message body when address is a repo slug") @@ -366,7 +413,7 @@ def main(argv: list[str] | None = None) -> int: 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)) + 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, "delivery_max_attempts": profile.delivery_max_attempts, "ack_timeout_seconds": float(setting("delivery_ack_timeout_seconds", "30")), "delivery_retry_backoff_seconds": setting("delivery_retry_backoff_seconds", "5,15,60,300"), **configured_limits}, sort_keys=True)) return 0 if args.command == "tap": command = list(args.wrapped_command) @@ -465,7 +512,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, "maxmsg": limits[0], "maxin": limits[1], "maxout": limits[2]})) + registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos, "delivery_mode": delivery_mode, "maxmsg": limits[0], "maxin": limits[1], "maxout": limits[2], "policy_profile": profile.name, "delivery_ack_mode": profile.delivery_ack_mode, "delivery_max_attempts": profile.delivery_max_attempts, "ack_timeout_seconds": float(setting("delivery_ack_timeout_seconds", "30"))})) if not registration.get("ok"): raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}") registered = True @@ -553,7 +600,7 @@ def main(argv: list[str] | None = None) -> int: if args.filter_command and (args.all or args.json): print("tamq: --filter cannot be combined with --all or --json", file=sys.stderr) return 2 - rows = store.list(target, None if args.all else "pending") + rows = store.inbox(target, include_terminal=args.all) for row in rows: if args.json: print(json.dumps(dict(row), sort_keys=True)) @@ -587,6 +634,43 @@ def main(argv: list[str] | None = None) -> int: if args.command == "history": for row in store.list(args.target_repo, args.state): print(json.dumps(dict(row), sort_keys=True)) return 0 + if args.command == "capture": + try: + rows = store.protocol_events( + repo=args.repo, + endpoint_id=args.endpoint_id, + message_id=args.message_id, + event_type=args.event_type, + limit=args.limit, + ) + except ValueError as exc: + print(f"tamq: {exc}", file=sys.stderr) + return 2 + scope_parts = [] + if args.repo: + scope_parts.append(f"repository `{args.repo}`") + if args.endpoint_id: + scope_parts.append(f"endpoint `{args.endpoint_id}`") + if args.message_id: + scope_parts.append(f"message `{args.message_id}`") + if args.event_type: + scope_parts.append(f"event `{args.event_type}`") + rendered = ( + render_jsonl(rows) + if args.format == "jsonl" + else render_markdown(rows, scope=", ".join(scope_parts) or "all repositories") + ) + if args.output is None: + print(rendered, end="" if rendered.endswith("\n") else "\n") + return 0 + try: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered, encoding="utf-8") + except OSError as exc: + print(f"tamq: cannot write capture: {exc}", file=sys.stderr) + return 1 + print(f"Wrote {len(rows)} protocol event(s) to {args.output}") + return 0 if args.command == "inspect": rows = [row for row in store.list() if row["message_id"] == args.message_id] if not rows: @@ -597,11 +681,16 @@ def main(argv: list[str] | None = None) -> int: if not store.acknowledge(args.message_id): print("tamq: message not found", file=sys.stderr); return 1 print(args.message_id); return 0 + if args.command == "retry": + if not store.retry_message(args.message_id): + print("tamq: message is not failed or was not found", file=sys.stderr); return 1 + print(args.message_id); return 0 if args.command == "export": store.export(store.list(args.target_repo, args.state), Path(args.output)); return 0 if args.command == "replay": batch_id = f"replay-{uuid4()}" count = 0 + deduplicated = 0 if args.endpoint_id and store.endpoint(args.endpoint_id) is None: print(f"tamq: endpoint is not registered: {args.endpoint_id}", file=sys.stderr) return 1 @@ -611,14 +700,46 @@ def main(argv: list[str] | None = None) -> int: if not line.strip(): continue item = json.loads(line) + validate_targets([item["target_repo"]]) + if item.get("sender_repo", "replay") != "local": + validate_targets([item.get("sender_repo", "replay")]) provenance = json.dumps({"batch_id": batch_id, "original_message_id": item.get("message_id")}, sort_keys=True) endpoint = args.endpoint_id or item.get("endpoint_id") - store.add(item.get("sender_repo", "replay"), item["target_repo"], item["body"], endpoint=endpoint, provenance=provenance) + replay_key = str( + item.get("message_id") + or hashlib.sha256( + json.dumps( + { + "sender_repo": item.get("sender_repo", "replay"), + "target_repo": item["target_repo"], + "body": item["body"], + }, + sort_keys=True, + ).encode() + ).hexdigest() + ) + if store.message_by_idempotency("tamq-replay", replay_key): + deduplicated += 1 + continue + store.add( + item.get("sender_repo", "replay"), + item["target_repo"], + item["body"], + endpoint=endpoint, + provenance=provenance, + client_id="tamq-replay", + idempotency_key=replay_key, + correlation_id=item.get("correlation_id"), + metadata={ + "batch_id": batch_id, + "original_message_id": item.get("message_id"), + }, + ) count += 1 - except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + except (OSError, KeyError, TypeError, ValueError, RegistryError, json.JSONDecodeError) as exc: print(f"tamq: cannot replay {args.file}: {exc}", file=sys.stderr) return 2 - print(json.dumps({"batch_id": batch_id, "count": count})); return 0 + print(json.dumps({"batch_id": batch_id, "count": count, "deduplicated": deduplicated})); return 0 if args.command == "purge": if args.feedback_chain: if args.before or args.max_size: @@ -653,7 +774,8 @@ 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", []), "window_counters": [dict(row) for row in store.counters()]}, sort_keys=True)); return 0 + states = {state: count for state, count in store.db.execute("SELECT state,COUNT(*) FROM messages GROUP BY state")} + print(json.dumps({"version": __version__, "db": str(db_path()), "messages": len(store.list()), "pending": states.get("pending", 0), "awaiting_ack": states.get("awaiting_ack", 0), "failed": states.get("failed", 0), "message_states": states, "leases": store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0], "service": live, "policy_profile": profile.name, "delivery_ack_mode": profile.delivery_ack_mode, "delivery_max_attempts": profile.delivery_max_attempts, "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/client.py b/src/tamq/client.py new file mode 100644 index 0000000..2015099 --- /dev/null +++ b/src/tamq/client.py @@ -0,0 +1,223 @@ +"""Transport-only Unix-socket client for TAMQ integrations.""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .config import socket_path +from .protocol import ( + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, + PROTOCOL_VERSION, +) + + +class TamqClientError(RuntimeError): + """The local TAMQ service rejected or could not complete an operation.""" + + +class TamqProtocolError(TamqClientError): + """The service and client do not share a compatible protocol contract.""" + + +class TamqTargetUnavailable(TamqClientError): + """No unambiguous live endpoint owns the requested repository.""" + + +@dataclass(frozen=True) +class WakeRequest: + lease_id: str + target_repo: str + prompt: str + trigger_id: str + source_repo: str = "coordination-engine" + endpoint_id: str | None = None + target_agent: str | None = None + + +@dataclass(frozen=True) +class WakeReceipt: + lease_id: str + trigger_id: str + endpoint_id: str + message_id: str + state: str + deduplicated: bool + + +class TamqClient: + """Small async client with no tmux/control-mode dependency.""" + + def __init__( + self, + path: Path | None = None, + *, + client_id: str = "coordination-engine", + protocol: str = PROTOCOL_VERSION, + timeout: float = 5.0, + ): + self.path = path or socket_path() + self.client_id = client_id + self.protocol = protocol + self.timeout = timeout + + async def request(self, operation: str, **fields: Any) -> dict: + payload = {"op": operation, "protocol": self.protocol, **fields} + try: + reader, writer = await asyncio.wait_for( + asyncio.open_unix_connection(str(self.path), limit=1024 * 1024), + timeout=self.timeout, + ) + try: + writer.write((json.dumps(payload, sort_keys=True) + "\n").encode()) + await asyncio.wait_for(writer.drain(), timeout=self.timeout) + line = await asyncio.wait_for(reader.readline(), timeout=self.timeout) + finally: + writer.close() + await writer.wait_closed() + except (OSError, asyncio.TimeoutError) as exc: + raise TamqClientError(f"TAMQ service unavailable: {exc}") from exc + try: + response = json.loads(line) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise TamqProtocolError("TAMQ returned an invalid response") from exc + if not response.get("ok"): + error = str(response.get("error") or "operation failed") + if error.startswith("incompatible protocol"): + raise TamqProtocolError(error) + raise TamqClientError(error) + return response + + async def negotiate(self) -> dict: + response = await self.request("ping") + server_protocol = str(response.get("protocol", "")) + if server_protocol.split(".")[0] != self.protocol.split(".")[0]: + raise TamqProtocolError( + f"incompatible protocol: client {self.protocol}, server {server_protocol}" + ) + required = {DELIVERY_RELIABILITY_CAPABILITY, IDEMPOTENT_SEND_CAPABILITY} + missing = required.difference(response.get("capabilities", [])) + if missing: + raise TamqProtocolError( + "TAMQ lacks required capabilities: " + ", ".join(sorted(missing)) + ) + return response + + async def send( + self, + *, + sender_repo: str, + target_repo: str, + body: str, + idempotency_key: str, + endpoint_id: str | None = None, + correlation_id: str | None = None, + metadata: dict | None = None, + provenance: str = "coordination_engine", + ) -> dict: + fields: dict[str, Any] = { + "sender_repo": sender_repo, + "target_repo": target_repo, + "body": body, + "client_id": self.client_id, + "idempotency_key": idempotency_key, + "provenance": provenance, + } + if endpoint_id is not None: + fields["endpoint_id"] = endpoint_id + if correlation_id is not None: + fields["correlation_id"] = correlation_id + if metadata is not None: + fields["metadata"] = metadata + return await self.request("send", **fields) + + async def message(self, message_id: str) -> dict: + return (await self.request("message", message_id=message_id))["message"] + + async def history( + self, *, target_repo: str | None = None, state: str | None = None + ) -> list[dict]: + fields = { + key: value + for key, value in {"target_repo": target_repo, "state": state}.items() + if value is not None + } + return (await self.request("history", **fields))["messages"] + + async def endpoints(self) -> list[dict]: + return (await self.request("endpoints"))["endpoints"] + + async def acknowledge(self, message_id: str) -> dict: + return await self.request("ack", message_id=message_id) + + async def retry(self, message_id: str) -> dict: + return await self.request("retry", message_id=message_id) + + +class CoordinationEngineAdapter: + """Map coordination-engine wake requests onto durable TAMQ messages.""" + + def __init__(self, client: TamqClient | None = None): + self.client = client or TamqClient() + + async def _resolve_endpoint(self, target_repo: str, requested: str | None) -> str: + endpoints = await self.client.endpoints() + matching = [ + endpoint + for endpoint in endpoints + if target_repo in json.loads(endpoint["repos"]) + and (requested is None or endpoint["endpoint_id"] == requested) + ] + if not matching: + raise TamqTargetUnavailable( + f"no live TAMQ endpoint is attached to {target_repo}" + ) + if len(matching) > 1: + raise TamqTargetUnavailable( + f"multiple TAMQ endpoints are attached to {target_repo}; endpoint_id is required" + ) + return str(matching[0]["endpoint_id"]) + + async def wake(self, request: WakeRequest) -> WakeReceipt: + await self.client.negotiate() + endpoint_id = await self._resolve_endpoint( + request.target_repo, request.endpoint_id + ) + response = await self.client.send( + sender_repo=request.source_repo, + target_repo=request.target_repo, + body=request.prompt, + idempotency_key=request.lease_id, + endpoint_id=endpoint_id, + correlation_id=request.trigger_id, + metadata={ + "lease_id": request.lease_id, + "trigger_id": request.trigger_id, + **( + {"target_agent": request.target_agent} + if request.target_agent is not None + else {} + ), + }, + ) + return WakeReceipt( + lease_id=request.lease_id, + trigger_id=request.trigger_id, + endpoint_id=endpoint_id, + message_id=response["message_id"], + state=response["state"], + deduplicated=bool(response.get("deduplicated")), + ) + + async def receipt(self, message_id: str) -> dict: + return await self.client.message(message_id) + + async def acknowledge(self, message_id: str) -> dict: + return await self.client.acknowledge(message_id) + + async def retry_failed(self, message_id: str) -> dict: + return await self.client.retry(message_id) diff --git a/src/tamq/policy.py b/src/tamq/policy.py index cd0e79b..e685708 100644 --- a/src/tamq/policy.py +++ b/src/tamq/policy.py @@ -14,6 +14,7 @@ class PolicyProfile: require_human: frozenset[str] safety_gated_max_attempts: int = 0 delivery_ack_mode: str = "injected" + delivery_max_attempts: int = 4 DEFAULT = PolicyProfile( @@ -40,4 +41,14 @@ def load_profile(path: Path | None = None, selected: str = "default") -> PolicyP ack_mode = profile.get("delivery_ack_mode", "injected") if ack_mode not in {"injected", "acknowledged"}: raise ValueError("delivery_ack_mode must be injected or acknowledged") - return PolicyProfile(selected, frozenset(profile.get("allow", [])), frozenset(profile.get("require_human", [])), attempts, ack_mode) + delivery_attempts = int(profile.get("delivery_max_attempts", 4)) + if not 1 <= delivery_attempts <= 9: + raise ValueError("delivery_max_attempts must be between 1 and 9") + return PolicyProfile( + selected, + frozenset(profile.get("allow", [])), + frozenset(profile.get("require_human", [])), + attempts, + ack_mode, + delivery_attempts, + ) diff --git a/src/tamq/protocol.py b/src/tamq/protocol.py new file mode 100644 index 0000000..3ec49a8 --- /dev/null +++ b/src/tamq/protocol.py @@ -0,0 +1,29 @@ +"""Stable local socket protocol identifiers shared by service and clients.""" + +PROTOCOL_VERSION = "0.1" +PUSHY_FRAMING_CAPABILITY = "readable_duplex_framing_v2" +TRIGGER_CAPABILITY = "trigger_input_v2" +LINE_LIMITS_CAPABILITY = "session_window_line_limits_v1" +PROTOCOL_CAPTURE_CAPABILITY = "structured_protocol_capture_v1" +DELIVERY_RELIABILITY_CAPABILITY = "bounded_delivery_ack_v1" +IDEMPOTENT_SEND_CAPABILITY = "idempotent_send_v1" + +SERVICE_CAPABILITIES = [ + "register", + "send", + "history", + "message", + "ack", + "retry", + "endpoints", + "disconnect", + "manual_delivery", + "terminal_output", + "pushy_input", + PUSHY_FRAMING_CAPABILITY, + TRIGGER_CAPABILITY, + LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, +] diff --git a/src/tamq/ptytap.py b/src/tamq/ptytap.py index 6dd9e87..c3aae3d 100644 --- a/src/tamq/ptytap.py +++ b/src/tamq/ptytap.py @@ -300,7 +300,7 @@ class PtyTap: self.output_observer.flush() flush_worker = getattr(self.broker, "flush_worker_message", None) if flush_worker is not None: - flush_worker() + flush_worker("worker_exit") for signum, handler in saved_handlers.items(): signal.signal(signum, handler) if saved_terminal is not None: diff --git a/src/tamq/service.py b/src/tamq/service.py index 132d57e..dc602d9 100644 --- a/src/tamq/service.py +++ b/src/tamq/service.py @@ -15,22 +15,16 @@ from .store import LimitBlock, Store from .registry import RegistryError, validate_targets from .control import ControlModeClient from .terminal import format_delivery, terminal_frame, write_terminal_output - -PROTOCOL_VERSION = "0.1" -PUSHY_FRAMING_CAPABILITY = "readable_duplex_framing_v2" -TRIGGER_CAPABILITY = "trigger_input_v2" -LINE_LIMITS_CAPABILITY = "session_window_line_limits_v1" -SERVICE_CAPABILITIES = [ - "register", - "send", - "history", - "manual_delivery", - "terminal_output", - "pushy_input", - PUSHY_FRAMING_CAPABILITY, - TRIGGER_CAPABILITY, +from .protocol import ( + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, LINE_LIMITS_CAPABILITY, -] + PROTOCOL_CAPTURE_CAPABILITY, + PROTOCOL_VERSION, + PUSHY_FRAMING_CAPABILITY, + SERVICE_CAPABILITIES, + TRIGGER_CAPABILITY, +) class Service: @@ -40,6 +34,8 @@ class Service: store: Store | None = None, poll_interval: float | None = None, input_grace: float | None = None, + retry_backoff: tuple[float, ...] | None = None, + lease_ttl: float | None = None, ): self.path = path or socket_path() self.store = store or Store(db_path()) @@ -59,6 +55,22 @@ class Service: ) except ValueError: self.input_grace = max(0.0, input_grace or 1.0) + configured_backoff = setting("delivery_retry_backoff_seconds", "5,15,60,300") + try: + parsed_backoff = tuple(float(item.strip()) for item in configured_backoff.split(",")) + if not parsed_backoff or any(value < 0 for value in parsed_backoff): + raise ValueError + self.retry_backoff = retry_backoff or parsed_backoff + except ValueError: + self.retry_backoff = retry_backoff or (5.0, 15.0, 60.0, 300.0) + try: + self.lease_ttl = ( + max(0.05, lease_ttl) + if lease_ttl is not None + else max(0.05, float(setting("delivery_lease_seconds", "30"))) + ) + except ValueError: + self.lease_ttl = max(0.05, lease_ttl or 30.0) async def run(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) @@ -103,6 +115,7 @@ class Service: continue def _deliver_once(self) -> None: + self.store.expire_delivery_leases() for endpoint in self.store.endpoints(): repos = json.loads(endpoint["repos"]) control = ControlModeClient(endpoint["session"]) @@ -117,20 +130,26 @@ class Service: and time.time() - float(endpoint["connected_at"]) < self.input_grace ): continue - pending = [ - row for row in self.store.list(state="pending") - if row["endpoint_id"] in (None, endpoint["endpoint_id"]) - and row["target_repo"] in repos - and (delivery_mode == "pane" or row["displayed_at"] is None) - ] + pending = self.store.deliverable(endpoint["endpoint_id"], repos) if not pending: continue try: if delivery_mode == "pane": control.start() for row in pending: - lease = self.store.claim(row["message_id"], endpoint["endpoint_id"]) - if lease is None: + attempt = int(row["attempt_count"]) + 1 + retry_delay = self.retry_backoff[ + min(attempt - 1, len(self.retry_backoff) - 1) + ] + claim = self.store.claim_delivery( + row["message_id"], + endpoint["endpoint_id"], + attempt_limit=int(endpoint["delivery_max_attempts"]), + retry_delay=retry_delay, + delivery_mode=delivery_mode, + ttl=self.lease_ttl, + ) + if claim is None: continue try: target = f'{endpoint["session"]}:{row["target_repo"]}' @@ -172,12 +191,21 @@ class Service: row["sender_repo"], row["body"], row["provenance"] ), ) - except Exception: + except Exception as exc: + self.store.fail_delivery( + row["message_id"], + claim.lease_id, + type(exc).__name__, + delivery_mode=delivery_mode, + ) continue - if delivery_mode == "output": - self.store.mark_displayed(row["message_id"], lease) - else: - self.store.release(row["message_id"], lease, "injected") + self.store.complete_delivery( + row["message_id"], + claim.lease_id, + delivery_mode=delivery_mode, + ack_mode=endpoint["delivery_ack_mode"], + ack_timeout=float(endpoint["ack_timeout_seconds"]), + ) finally: control.close() @@ -219,6 +247,10 @@ class Service: int(request.get("maxin", 1024)), int(request.get("maxout", 32768)), ), + policy_profile=str(request.get("policy_profile", "default")), + delivery_ack_mode=str(request.get("delivery_ack_mode", "injected")), + delivery_max_attempts=int(request.get("delivery_max_attempts", 4)), + ack_timeout_seconds=float(request.get("ack_timeout_seconds", 30)), ) response = {"ok": True, "endpoint_id": request["endpoint_id"], "instance_id": endpoint_id, "protocol": PROTOCOL_VERSION} except (RegistryError, ValueError) as exc: @@ -242,7 +274,60 @@ class Service: raise RegistryError("target is not attached to endpoint") endpoint_id = endpoint["endpoint_id"] provenance = request.get("provenance", "operator_input") - if endpoint_id and request["sender_repo"] != "local": + client_id = request.get("client_id") + idempotency_key = request.get("idempotency_key") + correlation_id = request.get("correlation_id") + metadata = request.get("metadata") + if bool(client_id) != bool(idempotency_key): + raise ValueError( + "client_id and idempotency_key must be provided together" + ) + if client_id and ( + len(str(client_id)) > 128 or len(str(idempotency_key)) > 256 + ): + raise ValueError("client or idempotency identity is too long") + if metadata is not None and not isinstance(metadata, dict): + raise ValueError("metadata must be a JSON object") + existing = ( + self.store.message_by_idempotency( + str(client_id), str(idempotency_key) + ) + if client_id + else None + ) + if existing is not None: + if ( + existing["sender_repo"] != request["sender_repo"] + or existing["target_repo"] != request["target_repo"] + or existing["body"] != request["body"] + ): + raise ValueError( + "idempotency key was already used for a different message" + ) + if existing["endpoint_id"] != endpoint_id and existing["state"] in { + "pending", "awaiting_ack", "failed" + }: + old_endpoint = ( + self.store.endpoint(existing["endpoint_id"]) + if existing["endpoint_id"] + else None + ) + if old_endpoint is not None: + raise ValueError( + "idempotent message is still bound to another live endpoint" + ) + self.store.rebind_message_endpoint( + existing["message_id"], endpoint_id + ) + existing = self.store.message(existing["message_id"]) + assert existing is not None + response = { + "ok": True, + "message_id": existing["message_id"], + "state": existing["state"], + "deduplicated": True, + } + elif endpoint_id and request["sender_repo"] != "local": result = self.store.admit_message( request["sender_repo"], request["target_repo"], @@ -250,6 +335,10 @@ class Service: endpoint=endpoint_id, source_repo=request["sender_repo"], provenance=provenance, + client_id=str(client_id) if client_id else None, + idempotency_key=str(idempotency_key) if idempotency_key else None, + correlation_id=str(correlation_id) if correlation_id else None, + metadata=metadata, ) if isinstance(result, LimitBlock): response = { @@ -260,7 +349,7 @@ class Service: "blocked": result.dimension, } else: - response = {"ok": True, "message_id": result, "state": "pending"} + response = {"ok": True, "message_id": result, "state": "pending", "deduplicated": False} else: message_id = self.store.add( request["sender_repo"], @@ -268,18 +357,39 @@ class Service: request["body"], endpoint=endpoint_id, provenance=provenance, + client_id=str(client_id) if client_id else None, + idempotency_key=str(idempotency_key) if idempotency_key else None, + correlation_id=str(correlation_id) if correlation_id else None, + metadata=metadata, ) - response = {"ok": True, "message_id": message_id, "state": "pending"} + response = {"ok": True, "message_id": message_id, "state": "pending", "deduplicated": False} except (RegistryError, ValueError) as exc: response = {"ok": False, "error": str(exc)} elif op == "history": response = {"ok": True, "messages": [dict(row) for row in self.store.list(request.get("target_repo"), request.get("state"))]} + elif op == "message": + row = self.store.message(str(request.get("message_id", ""))) + response = { + "ok": row is not None, + "message": dict(row) if row is not None else None, + "error": None if row is not None else "message not found", + } elif op == "ack": if not request.get("message_id"): response = {"ok": False, "error": "ack requires message_id"} else: ok = self.store.acknowledge(request["message_id"]) response = {"ok": ok, "message_id": request["message_id"], "state": "acknowledged" if ok else "missing"} + elif op == "retry": + if not request.get("message_id"): + response = {"ok": False, "error": "retry requires message_id"} + else: + ok = self.store.retry_message(request["message_id"]) + response = { + "ok": ok, + "message_id": request["message_id"], + "state": "pending" if ok else "not_failed_or_missing", + } elif op == "endpoints": response = {"ok": True, "endpoints": [dict(row) for row in self.store.endpoints()]} elif op == "disconnect": diff --git a/src/tamq/store.py b/src/tamq/store.py index fe7a17c..bf2007a 100644 --- a/src/tamq/store.py +++ b/src/tamq/store.py @@ -9,7 +9,7 @@ from pathlib import Path from typing import Iterable from uuid import uuid4 -SCHEMA_VERSION = 4 +SCHEMA_VERSION = 6 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, @@ -23,6 +23,12 @@ class LimitBlock: limit: int +@dataclass(frozen=True) +class DeliveryClaim: + lease_id: str + attempt: int + + class Store: def __init__(self, path: Path): path.parent.mkdir(parents=True, exist_ok=True) @@ -44,7 +50,16 @@ class Store: provenance TEXT, injected_at REAL, acknowledged_at REAL, - displayed_at REAL + displayed_at REAL, + delivered_at REAL, + attempt_count INTEGER NOT NULL DEFAULT 0, + last_attempt_at REAL, + last_failure_reason TEXT, + next_attempt_at REAL, + client_id TEXT, + idempotency_key TEXT, + correlation_id TEXT, + envelope_metadata TEXT NOT NULL DEFAULT '{}' ); CREATE INDEX IF NOT EXISTS messages_target_state ON messages(target_repo, state); CREATE TABLE IF NOT EXISTS endpoints ( @@ -53,6 +68,10 @@ class Store: session TEXT NOT NULL, repos TEXT NOT NULL, delivery_mode TEXT NOT NULL DEFAULT 'manual', + policy_profile TEXT NOT NULL DEFAULT 'default', + delivery_ack_mode TEXT NOT NULL DEFAULT 'injected', + delivery_max_attempts INTEGER NOT NULL DEFAULT 4, + ack_timeout_seconds REAL NOT NULL DEFAULT 30, connected_at REAL NOT NULL, disconnected_at REAL ); @@ -73,8 +92,30 @@ class Store: lease_id TEXT NOT NULL, endpoint_id TEXT NOT NULL, acquired_at REAL NOT NULL, - expires_at REAL NOT NULL + expires_at REAL NOT NULL, + attempt_limit INTEGER NOT NULL DEFAULT 1, + retry_delay REAL NOT NULL DEFAULT 5 ); + CREATE TABLE IF NOT EXISTS protocol_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + occurred_at REAL NOT NULL, + event_type TEXT NOT NULL, + endpoint_id TEXT, + repo TEXT, + peer_repo TEXT, + message_id TEXT, + provenance TEXT, + delivery_mode TEXT, + outcome TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '{}' + ); + CREATE INDEX IF NOT EXISTS protocol_events_time + ON protocol_events(occurred_at, sequence); + CREATE INDEX IF NOT EXISTS protocol_events_message + ON protocol_events(message_id, occurred_at); + CREATE INDEX IF NOT EXISTS protocol_events_endpoint + ON protocol_events(endpoint_id, occurred_at); """) endpoint_columns = { row["name"] for row in self.db.execute("PRAGMA table_info(endpoints)") @@ -83,11 +124,47 @@ class Store: self.db.execute( "ALTER TABLE endpoints ADD COLUMN delivery_mode TEXT NOT NULL DEFAULT 'manual'" ) + for name, definition in ( + ("policy_profile", "TEXT NOT NULL DEFAULT 'default'"), + ("delivery_ack_mode", "TEXT NOT NULL DEFAULT 'injected'"), + ("delivery_max_attempts", "INTEGER NOT NULL DEFAULT 4"), + ("ack_timeout_seconds", "REAL NOT NULL DEFAULT 30"), + ): + if name not in endpoint_columns: + self.db.execute(f"ALTER TABLE endpoints ADD COLUMN {name} {definition}") message_columns = { row["name"] for row in self.db.execute("PRAGMA table_info(messages)") } if "displayed_at" not in message_columns: self.db.execute("ALTER TABLE messages ADD COLUMN displayed_at REAL") + for name, definition in ( + ("delivered_at", "REAL"), + ("attempt_count", "INTEGER NOT NULL DEFAULT 0"), + ("last_attempt_at", "REAL"), + ("last_failure_reason", "TEXT"), + ("next_attempt_at", "REAL"), + ("client_id", "TEXT"), + ("idempotency_key", "TEXT"), + ("correlation_id", "TEXT"), + ("envelope_metadata", "TEXT NOT NULL DEFAULT '{}'"), + ): + if name not in message_columns: + self.db.execute(f"ALTER TABLE messages ADD COLUMN {name} {definition}") + lease_columns = { + row["name"] for row in self.db.execute("PRAGMA table_info(leases)") + } + for name, definition in ( + ("attempt_limit", "INTEGER NOT NULL DEFAULT 1"), + ("retry_delay", "REAL NOT NULL DEFAULT 5"), + ): + if name not in lease_columns: + self.db.execute(f"ALTER TABLE leases ADD COLUMN {name} {definition}") + self.db.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS messages_client_idempotency " + "ON messages(client_id,idempotency_key) " + "WHERE client_id IS NOT NULL AND idempotency_key IS NOT NULL" + ) + self._backfill_protocol_events() self.db.execute( "INSERT INTO metadata(key,value) VALUES('schema_version',?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", @@ -95,6 +172,131 @@ class Store: ) self.db.commit() + def _backfill_protocol_events(self) -> None: + """Represent pre-v5 message history without inventing terminal activity.""" + self.db.execute( + "INSERT OR IGNORE INTO protocol_events(" + "event_id,occurred_at,event_type,endpoint_id,repo,peer_repo,message_id," + "provenance,delivery_mode,outcome,detail) " + "SELECT 'pe-accepted-' || message_id,created_at,'message.accepted'," + "endpoint_id,sender_repo,target_repo,message_id,provenance,NULL,'accepted'," + "'{\"backfilled\":true}' FROM messages m WHERE NOT EXISTS (" + "SELECT 1 FROM protocol_events e WHERE e.message_id=m.message_id " + "AND e.event_type='message.accepted')" + ) + self.db.execute( + "INSERT OR IGNORE INTO protocol_events(" + "event_id,occurred_at,event_type,endpoint_id,repo,peer_repo,message_id," + "provenance,delivery_mode,outcome,detail) " + "SELECT 'pe-displayed-' || message_id,displayed_at,'delivery.displayed'," + "endpoint_id,target_repo,sender_repo,message_id,provenance,'output','displayed'," + "'{\"backfilled\":true}' FROM messages m WHERE displayed_at IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM protocol_events e WHERE " + "e.message_id=m.message_id AND e.event_type='delivery.displayed')" + ) + self.db.execute( + "INSERT OR IGNORE INTO protocol_events(" + "event_id,occurred_at,event_type,endpoint_id,repo,peer_repo,message_id," + "provenance,delivery_mode,outcome,detail) " + "SELECT 'pe-injected-' || message_id,injected_at,'delivery.injected'," + "endpoint_id,target_repo,sender_repo,message_id,provenance,NULL,'injected'," + "'{\"backfilled\":true}' FROM messages m WHERE injected_at IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM protocol_events e WHERE " + "e.message_id=m.message_id AND e.event_type='delivery.injected')" + ) + self.db.execute( + "INSERT OR IGNORE INTO protocol_events(" + "event_id,occurred_at,event_type,endpoint_id,repo,peer_repo,message_id," + "provenance,delivery_mode,outcome,detail) " + "SELECT 'pe-acknowledged-' || message_id,acknowledged_at,'message.acknowledged'," + "endpoint_id,target_repo,sender_repo,message_id,provenance,NULL,'acknowledged'," + "'{\"backfilled\":true}' FROM messages m WHERE acknowledged_at IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM protocol_events e WHERE " + "e.message_id=m.message_id AND e.event_type='message.acknowledged')" + ) + + def record_protocol_event( + self, + event_type: str, + *, + endpoint_id: str | None = None, + repo: str | None = None, + peer_repo: str | None = None, + message_id: str | None = None, + provenance: str | None = None, + delivery_mode: str | None = None, + outcome: str = "observed", + detail: dict | None = None, + occurred_at: float | None = None, + commit: bool = True, + ) -> str: + """Append one structured TAMQ event; never record arbitrary pane content.""" + event_id = f"pe-{uuid4()}" + self.db.execute( + "INSERT INTO protocol_events(" + "event_id,occurred_at,event_type,endpoint_id,repo,peer_repo,message_id," + "provenance,delivery_mode,outcome,detail) VALUES(?,?,?,?,?,?,?,?,?,?,?)", + ( + event_id, + occurred_at if occurred_at is not None else time.time(), + event_type, + endpoint_id, + repo, + peer_repo, + message_id, + provenance, + delivery_mode, + outcome, + json.dumps(detail or {}, sort_keys=True, separators=(",", ":")), + ), + ) + if commit: + self.db.commit() + return event_id + + def protocol_events( + self, + *, + repo: str | None = None, + endpoint_id: str | None = None, + message_id: str | None = None, + event_type: str | None = None, + limit: int = 500, + ) -> list[sqlite3.Row]: + """Return the newest matching protocol events in chronological order.""" + if isinstance(limit, bool) or limit <= 0: + raise ValueError("capture limit must be a positive integer") + clauses: list[str] = [] + values: list[str | int] = [] + if repo: + clauses.append( + "(e.repo=? OR e.peer_repo=? OR m.sender_repo=? OR m.target_repo=?)" + ) + values.extend([repo, repo, repo, repo]) + if endpoint_id: + clauses.append("e.endpoint_id=?") + values.append(endpoint_id) + if message_id: + clauses.append("e.message_id=?") + values.append(message_id) + if event_type: + clauses.append("e.event_type=?") + values.append(event_type) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + values.append(limit) + return list( + self.db.execute( + "SELECT * FROM (SELECT e.*,m.body AS message_body," + "m.sender_repo AS message_sender_repo," + "m.target_repo AS message_target_repo,m.state AS message_state " + "FROM protocol_events e LEFT JOIN messages m " + f"ON m.message_id=e.message_id {where} " + "ORDER BY e.occurred_at DESC,e.sequence DESC LIMIT ?) " + "ORDER BY occurred_at,sequence", + values, + ) + ) + def close(self) -> None: self.db.close() @@ -106,23 +308,83 @@ class Store: repos: list[str], delivery_mode: str = "manual", limits: tuple[int, int, int] = (8, 1024, 32768), + *, + policy_profile: str = "default", + delivery_ack_mode: str = "injected", + delivery_max_attempts: int = 4, + ack_timeout_seconds: float = 30.0, ) -> None: import json if delivery_mode not in {"manual", "output", "pane", "pushy", "trigger"}: raise ValueError(f"invalid delivery mode: {delivery_mode}") + if delivery_ack_mode not in {"injected", "acknowledged"}: + raise ValueError("delivery_ack_mode must be injected or acknowledged") + if isinstance(delivery_max_attempts, bool) or not 1 <= delivery_max_attempts <= 9: + raise ValueError("delivery_max_attempts must be between 1 and 9") + if ack_timeout_seconds <= 0: + raise ValueError("ack_timeout_seconds must be positive") self._validate_limits(limits) + replaced = [ + row["endpoint_id"] + for row in self.db.execute( + "SELECT endpoint_id FROM endpoints WHERE pid=? AND session=? " + "AND endpoint_id<>? AND disconnected_at IS NULL", + (pid, session, endpoint_id), + ) + ] self.db.execute( "UPDATE endpoints SET disconnected_at=strftime('%s','now') " "WHERE pid=? AND session=? AND endpoint_id<>? AND disconnected_at IS NULL", (pid, session, endpoint_id), ) self.db.execute( - "INSERT INTO endpoints(endpoint_id,pid,session,repos,delivery_mode,connected_at,disconnected_at) VALUES(?,?,?,?,?,?,NULL) " - "ON CONFLICT(endpoint_id) DO UPDATE SET pid=excluded.pid, session=excluded.session, repos=excluded.repos, delivery_mode=excluded.delivery_mode, connected_at=excluded.connected_at, disconnected_at=NULL", - (endpoint_id, pid, session, json.dumps(repos), delivery_mode, time.time()), + "INSERT INTO endpoints(endpoint_id,pid,session,repos,delivery_mode," + "policy_profile,delivery_ack_mode,delivery_max_attempts,ack_timeout_seconds," + "connected_at,disconnected_at) VALUES(?,?,?,?,?,?,?,?,?,?,NULL) " + "ON CONFLICT(endpoint_id) DO UPDATE SET pid=excluded.pid, " + "session=excluded.session, repos=excluded.repos, " + "delivery_mode=excluded.delivery_mode, policy_profile=excluded.policy_profile, " + "delivery_ack_mode=excluded.delivery_ack_mode, " + "delivery_max_attempts=excluded.delivery_max_attempts, " + "ack_timeout_seconds=excluded.ack_timeout_seconds, " + "connected_at=excluded.connected_at, disconnected_at=NULL", + ( + endpoint_id, + pid, + session, + json.dumps(repos), + delivery_mode, + policy_profile, + delivery_ack_mode, + delivery_max_attempts, + ack_timeout_seconds, + time.time(), + ), ) for repo in repos: self.configure_window(endpoint_id, repo, limits, commit=False) + for replaced_id in replaced: + self.record_protocol_event( + "endpoint.disconnected", + endpoint_id=replaced_id, + outcome="replaced", + commit=False, + ) + self.record_protocol_event( + "endpoint.registered", + endpoint_id=endpoint_id, + delivery_mode=delivery_mode, + outcome="registered", + detail={ + "repos": repos, + "limits": {"messages": limits[0], "input": limits[1], "output": limits[2]}, + "policy_profile": policy_profile, + "delivery_ack_mode": delivery_ack_mode, + "delivery_max_attempts": delivery_max_attempts, + "ack_timeout_seconds": ack_timeout_seconds, + }, + commit=False, + ) self.db.commit() @staticmethod @@ -229,11 +491,30 @@ class Store: 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,)) + result = self.db.execute( + "UPDATE endpoints SET disconnected_at=strftime('%s','now') " + "WHERE endpoint_id=? AND disconnected_at IS NULL", + (endpoint_id,), + ) + if result.rowcount: + self.record_protocol_event( + "endpoint.disconnected", + endpoint_id=endpoint_id, + outcome="disconnected", + commit=False, + ) self.db.commit() def disconnect_all(self) -> None: + endpoint_ids = [row["endpoint_id"] for row in self.endpoints()] self.db.execute("UPDATE endpoints SET disconnected_at=strftime('%s','now') WHERE disconnected_at IS NULL") + for endpoint_id in endpoint_ids: + self.record_protocol_event( + "endpoint.disconnected", + endpoint_id=endpoint_id, + outcome="service_stopped", + commit=False, + ) self.db.commit() def clear_runtime_state(self) -> tuple[int, int]: @@ -244,6 +525,7 @@ class Store: "SELECT COUNT(*) FROM endpoints WHERE disconnected_at IS NULL" ).fetchone()[0] ) + endpoint_ids = [row["endpoint_id"] for row in self.endpoints()] with self.db: self.db.execute("DELETE FROM leases") self.db.execute("DELETE FROM window_counters") @@ -251,6 +533,13 @@ class Store: "UPDATE endpoints SET disconnected_at=strftime('%s','now') " "WHERE disconnected_at IS NULL" ) + for endpoint_id in endpoint_ids: + self.record_protocol_event( + "endpoint.disconnected", + endpoint_id=endpoint_id, + outcome="runtime_cleanup", + commit=False, + ) return endpoints, leases def endpoints(self) -> list[sqlite3.Row]: @@ -281,14 +570,59 @@ class Store: oldest = self.db.execute("SELECT MIN(created_at) FROM messages").fetchone()[0] return size, oldest - def add(self, sender: str, target: str, body: str, *, endpoint: str | None = None, provenance: str | None = None) -> str: + @staticmethod + def _encode_envelope_metadata(metadata: dict | None) -> str: + encoded = json.dumps(metadata or {}, sort_keys=True, separators=(",", ":")) + if len(encoded.encode("utf-8")) > 4096: + raise ValueError("message metadata exceeds 4 KiB limit") + return encoded + + def add( + self, + sender: str, + target: str, + body: str, + *, + endpoint: str | None = None, + provenance: str | None = None, + client_id: str | None = None, + idempotency_key: str | None = None, + correlation_id: str | None = None, + metadata: dict | None = None, + ) -> str: if len(body.encode("utf-8")) > 8192: raise ValueError("message body exceeds 8 KiB limit") + if bool(client_id) != bool(idempotency_key): + raise ValueError("client_id and idempotency_key must be provided together") + envelope_metadata = self._encode_envelope_metadata(metadata) message_id = f"m-{uuid4()}" + created_at = time.time() 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), + "INSERT INTO messages(message_id,sender_repo,target_repo,body,created_at," + "state,endpoint_id,provenance,injected_at,acknowledged_at,displayed_at," + "client_id,idempotency_key,correlation_id,envelope_metadata) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + ( + message_id, sender, target, body, created_at, "pending", endpoint, + provenance, None, None, None, client_id, idempotency_key, + correlation_id, envelope_metadata, + ), + ) + self.record_protocol_event( + "message.accepted", + endpoint_id=endpoint, + repo=sender, + peer_repo=target, + message_id=message_id, + provenance=provenance, + outcome="accepted", + detail={ + "line_count": max(1, len(body.splitlines())), + **({"client_id": client_id} if client_id else {}), + **({"correlation_id": correlation_id} if correlation_id else {}), + }, + occurred_at=created_at, + commit=False, ) self.db.commit() return message_id @@ -302,12 +636,20 @@ class Store: endpoint: str, source_repo: str, provenance: str, + client_id: str | None = None, + idempotency_key: str | None = None, + correlation_id: str | None = None, + metadata: dict | None = None, ) -> 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") + if bool(client_id) != bool(idempotency_key): + raise ValueError("client_id and idempotency_key must be provided together") + envelope_metadata = self._encode_envelope_metadata(metadata) self.db.execute("BEGIN IMMEDIATE") try: + created_at = time.time() self.db.execute( "INSERT INTO window_counters(endpoint_id,repo,updated_at) VALUES(?,?,?) " "ON CONFLICT(endpoint_id,repo) DO NOTHING", @@ -324,18 +666,56 @@ class Store: ("output", "maxout", "output"), ): if int(state[count_name]) >= int(state[limit_name]): + blocked = LimitBlock( + label, int(state[count_name]), int(state[limit_name]) + ) self.db.rollback() - return LimitBlock(label, int(state[count_name]), int(state[limit_name])) + self.record_protocol_event( + "message.blocked", + endpoint_id=endpoint, + repo=sender, + peer_repo=target, + provenance=provenance, + outcome="blocked", + detail={ + "dimension": blocked.dimension, + "count": blocked.count, + "limit": blocked.limit, + }, + ) + return blocked 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), + "INSERT INTO messages(message_id,sender_repo,target_repo,body,created_at," + "state,endpoint_id,provenance,injected_at,acknowledged_at,displayed_at," + "client_id,idempotency_key,correlation_id,envelope_metadata) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + ( + message_id, sender, target, body, created_at, "pending", endpoint, + provenance, None, None, None, client_id, idempotency_key, + correlation_id, envelope_metadata, + ), ) self.db.execute( "UPDATE window_counters SET messages=messages+1,updated_at=? WHERE endpoint_id=? AND repo=?", (time.time(), endpoint, source_repo), ) + self.record_protocol_event( + "message.accepted", + endpoint_id=endpoint, + repo=sender, + peer_repo=target, + message_id=message_id, + provenance=provenance, + outcome="accepted", + detail={ + "line_count": max(1, len(body.splitlines())), + **({"client_id": client_id} if client_id else {}), + **({"correlation_id": correlation_id} if correlation_id else {}), + }, + occurred_at=created_at, + commit=False, + ) self.db.commit() return message_id except BaseException: @@ -351,11 +731,54 @@ 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 inbox(self, target: str, *, include_terminal: bool = False) -> list[sqlite3.Row]: + if include_terminal: + return self.list(target) + return list( + self.db.execute( + "SELECT * FROM messages WHERE target_repo=? " + "AND state IN ('pending','awaiting_ack') ORDER BY created_at", + (target,), + ) + ) + def message(self, message_id: str) -> sqlite3.Row | None: return self.db.execute( "SELECT * FROM messages WHERE message_id=?", (message_id,) ).fetchone() + def message_by_idempotency( + self, client_id: str, idempotency_key: str + ) -> sqlite3.Row | None: + return self.db.execute( + "SELECT * FROM messages WHERE client_id=? AND idempotency_key=?", + (client_id, idempotency_key), + ).fetchone() + + def rebind_message_endpoint( + self, message_id: str, endpoint_id: str + ) -> bool: + row = self.message(message_id) + if row is None or row["state"] not in {"pending", "awaiting_ack", "failed"}: + return False + with self.db: + self.db.execute( + "UPDATE messages SET endpoint_id=? WHERE message_id=?", + (endpoint_id, message_id), + ) + self.record_protocol_event( + "message.rebound", + endpoint_id=endpoint_id, + repo=row["sender_repo"], + peer_repo=row["target_repo"], + message_id=message_id, + provenance=row["provenance"], + outcome="rebound", + detail={"previous_endpoint_id": row["endpoint_id"]}, + commit=False, + ) + return True + def feedback_chain(self, root_id: str) -> list[sqlite3.Row]: """Resolve a reflected-delivery chain using receipt and direction evidence.""" rows = self.list() @@ -405,10 +828,323 @@ class Store: self.db.commit() def acknowledge(self, message_id: str) -> bool: - row = self.db.execute("SELECT 1 FROM messages WHERE message_id=?", (message_id,)).fetchone() + row = self.db.execute("SELECT * FROM messages WHERE message_id=?", (message_id,)).fetchone() if row is None: return False - self.set_state(message_id, "acknowledged") + with self.db: + self.db.execute("DELETE FROM leases WHERE message_id=?", (message_id,)) + self.db.execute( + "UPDATE messages SET state='acknowledged',acknowledged_at=?," + "next_attempt_at=NULL,last_failure_reason=NULL WHERE message_id=?", + (time.time(), message_id), + ) + self.record_protocol_event( + "message.acknowledged", + endpoint_id=row["endpoint_id"], + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + outcome="late_acknowledged" if row["state"] == "failed" else "acknowledged", + commit=False, + ) + return True + + def deliverable( + self, endpoint_id: str, repos: list[str], *, now: float | None = None + ) -> list[sqlite3.Row]: + if not repos: + return [] + when = time.time() if now is None else now + placeholders = ",".join("?" for _ in repos) + return list( + self.db.execute( + "SELECT * FROM messages WHERE state IN ('pending','awaiting_ack') " + "AND (endpoint_id IS NULL OR endpoint_id=?) " + f"AND target_repo IN ({placeholders}) " + "AND (next_attempt_at IS NULL OR next_attempt_at<=?) " + "ORDER BY created_at", + [endpoint_id, *repos, when], + ) + ) + + def expire_delivery_leases(self, *, now: float | None = None) -> int: + """Turn expired delivery ownership into bounded retry or terminal failure.""" + when = time.time() if now is None else now + rows = list( + self.db.execute( + "SELECT l.*,m.sender_repo,m.target_repo,m.provenance,m.attempt_count " + "FROM leases l JOIN messages m USING(message_id) WHERE l.expires_at= int(row["attempt_limit"]) + self.db.execute( + "DELETE FROM leases WHERE message_id=? AND lease_id=?", + (row["message_id"], row["lease_id"]), + ) + self.db.execute( + "UPDATE messages SET state=?,last_failure_reason='lease_expired'," + "next_attempt_at=? WHERE message_id=? AND state!='acknowledged'", + ( + "failed" if terminal else "pending", + None if terminal else when + float(row["retry_delay"]), + row["message_id"], + ), + ) + self.record_protocol_event( + "delivery.failed", + endpoint_id=row["endpoint_id"], + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=row["message_id"], + provenance=row["provenance"], + outcome="exhausted" if terminal else "retry_scheduled", + detail={ + "reason": "lease_expired", + "attempt": int(row["attempt_count"]), + "attempt_limit": int(row["attempt_limit"]), + }, + commit=False, + ) + return len(rows) + + def claim_delivery( + self, + message_id: str, + endpoint_id: str, + *, + attempt_limit: int, + retry_delay: float, + delivery_mode: str, + ttl: float = 30.0, + now: float | None = None, + ) -> DeliveryClaim | None: + if not 1 <= attempt_limit <= 9: + raise ValueError("delivery attempt limit must be between 1 and 9") + if retry_delay < 0 or ttl <= 0: + raise ValueError("delivery retry delay must be non-negative and lease TTL positive") + when = time.time() if now is None else now + self.expire_delivery_leases(now=when) + lease_id = f"lease-{uuid4()}" + with self.db: + row = self.db.execute( + "SELECT * FROM messages WHERE message_id=? AND " + "state IN ('pending','awaiting_ack') AND " + "(next_attempt_at IS NULL OR next_attempt_at<=?)", + (message_id, when), + ).fetchone() + if row is None: + return None + if int(row["attempt_count"]) >= attempt_limit: + reason = ( + "ack_timeout" + if row["state"] == "awaiting_ack" + else "attempt_limit_exhausted" + ) + self.db.execute( + "UPDATE messages SET state='failed',last_failure_reason=?," + "next_attempt_at=NULL WHERE message_id=?", + (reason, message_id), + ) + self.record_protocol_event( + "delivery.failed", + endpoint_id=endpoint_id, + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + delivery_mode=delivery_mode, + outcome="exhausted", + detail={ + "reason": reason, + "attempt": int(row["attempt_count"]), + "attempt_limit": attempt_limit, + }, + commit=False, + ) + return None + if row["state"] == "awaiting_ack": + self.record_protocol_event( + "delivery.failed", + endpoint_id=endpoint_id, + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + delivery_mode=delivery_mode, + outcome="retrying", + detail={ + "reason": "ack_timeout", + "attempt": int(row["attempt_count"]), + "attempt_limit": attempt_limit, + }, + commit=False, + ) + try: + self.db.execute( + "INSERT INTO leases(message_id,lease_id,endpoint_id,acquired_at," + "expires_at,attempt_limit,retry_delay) VALUES(?,?,?,?,?,?,?)", + ( + message_id, lease_id, endpoint_id, when, when + ttl, + attempt_limit, retry_delay, + ), + ) + except sqlite3.IntegrityError: + return None + attempt = int(row["attempt_count"]) + 1 + self.db.execute( + "UPDATE messages SET state='pending',attempt_count=?,last_attempt_at=?," + "last_failure_reason=NULL,next_attempt_at=NULL WHERE message_id=?", + (attempt, when, message_id), + ) + self.record_protocol_event( + "delivery.attempted", + endpoint_id=endpoint_id, + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + delivery_mode=delivery_mode, + outcome="attempted", + detail={"attempt": attempt, "attempt_limit": attempt_limit}, + commit=False, + ) + return DeliveryClaim(lease_id, attempt) + + def fail_delivery( + self, + message_id: str, + lease_id: str, + reason: str, + *, + delivery_mode: str, + now: float | None = None, + ) -> bool: + when = time.time() if now is None else now + row = self.db.execute( + "SELECT l.*,m.sender_repo,m.target_repo,m.provenance,m.attempt_count " + "FROM leases l JOIN messages m USING(message_id) " + "WHERE l.message_id=? AND l.lease_id=?", + (message_id, lease_id), + ).fetchone() + if row is None: + return False + terminal = int(row["attempt_count"]) >= int(row["attempt_limit"]) + with self.db: + self.db.execute( + "DELETE FROM leases WHERE message_id=? AND lease_id=?", + (message_id, lease_id), + ) + self.db.execute( + "UPDATE messages SET state=?,last_failure_reason=?,next_attempt_at=? " + "WHERE message_id=? AND state!='acknowledged'", + ( + "failed" if terminal else "pending", + reason, + None if terminal else when + float(row["retry_delay"]), + message_id, + ), + ) + self.record_protocol_event( + "delivery.failed", + endpoint_id=row["endpoint_id"], + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + delivery_mode=delivery_mode, + outcome="exhausted" if terminal else "retry_scheduled", + detail={ + "reason": reason, + "attempt": int(row["attempt_count"]), + "attempt_limit": int(row["attempt_limit"]), + "retry_at": None if terminal else when + float(row["retry_delay"]), + }, + commit=False, + ) + return True + + def complete_delivery( + self, + message_id: str, + lease_id: str, + *, + delivery_mode: str, + ack_mode: str, + ack_timeout: float, + now: float | None = None, + ) -> bool: + if ack_mode not in {"injected", "acknowledged"}: + raise ValueError("invalid acknowledgement mode") + if ack_timeout <= 0: + raise ValueError("acknowledgement timeout must be positive") + when = time.time() if now is None else now + row = self.db.execute( + "SELECT l.endpoint_id,m.* FROM leases l JOIN messages m USING(message_id) " + "WHERE l.message_id=? AND l.lease_id=?", + (message_id, lease_id), + ).fetchone() + if row is None: + return False + state = "injected" if ack_mode == "injected" else "awaiting_ack" + event_type = "delivery.displayed" if delivery_mode == "output" else "delivery.injected" + with self.db: + self.db.execute( + "DELETE FROM leases WHERE message_id=? AND lease_id=?", + (message_id, lease_id), + ) + self.db.execute( + "UPDATE messages SET state=?,delivered_at=?,displayed_at=CASE " + "WHEN ?='output' THEN ? ELSE displayed_at END,injected_at=CASE " + "WHEN ?!='output' THEN ? ELSE injected_at END,next_attempt_at=?," + "last_failure_reason=NULL WHERE message_id=?", + ( + state, when, delivery_mode, when, delivery_mode, when, + None if ack_mode == "injected" else when + ack_timeout, + message_id, + ), + ) + self.record_protocol_event( + event_type, + endpoint_id=row["endpoint_id"], + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + delivery_mode=delivery_mode, + outcome=state, + detail={ + "attempt": int(row["attempt_count"]), + "ack_mode": ack_mode, + "ack_deadline": None if ack_mode == "injected" else when + ack_timeout, + }, + commit=False, + ) + return True + + def retry_message(self, message_id: str) -> bool: + row = self.message(message_id) + if row is None or row["state"] != "failed": + return False + with self.db: + self.db.execute( + "UPDATE messages SET state='pending',attempt_count=0,last_attempt_at=NULL," + "last_failure_reason=NULL,next_attempt_at=NULL WHERE message_id=?", + (message_id,), + ) + self.record_protocol_event( + "delivery.retry_reset", + endpoint_id=row["endpoint_id"], + repo=row["target_repo"], + peer_repo=row["sender_repo"], + message_id=message_id, + provenance=row["provenance"], + outcome="pending", + commit=False, + ) return True def mark_displayed(self, message_id: str, lease_id: str) -> bool: @@ -427,10 +1163,14 @@ class Store: def claim(self, message_id: str, endpoint_id: str, ttl: float = 30.0) -> str | None: now = time.time() lease_id = f"lease-{uuid4()}" + self.expire_delivery_leases(now=now) with self.db: - self.db.execute("DELETE FROM leases WHERE expires_at < ?", (now,)) try: - self.db.execute("INSERT INTO leases VALUES(?,?,?,?,?)", (message_id, lease_id, endpoint_id, now, now + ttl)) + self.db.execute( + "INSERT INTO leases(message_id,lease_id,endpoint_id,acquired_at,expires_at) " + "VALUES(?,?,?,?,?)", + (message_id, lease_id, endpoint_id, now, now + ttl), + ) except sqlite3.IntegrityError: return None return lease_id diff --git a/tests/test_broker.py b/tests/test_broker.py index 768d6c3..5bad361 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -47,6 +47,15 @@ def test_broker_collects_worker_followup_lines_until_empty(tmp_path, monkeypatch assert row["body"] == ( "Review this\nContext: AUTH-WP-4.\nReply with accepted or blocked." ) + events = store.protocol_events() + assert [event["event_type"] for event in events] == [ + "worker.block_started", + "message.accepted", + "worker.block_closed", + ] + assert events[-1]["message_id"] == row["message_id"] + assert events[-1]["outcome"] == "accepted" + assert events[-1]["detail"] == '{"line_count":3,"reason":"empty_line"}' def test_new_worker_address_flushes_previous_block(tmp_path, monkeypatch): @@ -81,6 +90,10 @@ def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path): broker.inspect_operator_line("cMD: MODE=TRIGGER") assert store.endpoint("ep")["delivery_mode"] == "trigger" assert notices[-1] == "From:tamq: Mode set to trigger." + command_events = store.protocol_events(event_type="command.applied") + assert len(command_events) == 1 + assert command_events[0]["delivery_mode"] == "trigger" + assert command_events[0]["provenance"] == "operator_input" def test_operator_command_names_are_case_insensitive(tmp_path): diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..a6deb5f --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,56 @@ +import json + +from tamq.capture import render_jsonl, render_markdown +from tamq.cli import main +from tamq.store import Store + + +def test_capture_renderers_expose_protocol_but_not_duplicate_message_bodies(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add( + "source", "target", "Please review.\nReply accepted.", provenance="worker_output" + ) + store.record_protocol_event( + "delivery.attempted", + repo="target", + peer_repo="source", + message_id=message_id, + provenance="worker_output", + delivery_mode="trigger", + outcome="attempted", + ) + rows = store.protocol_events(repo="source") + + markdown = render_markdown(rows, scope="repository `source`") + assert "# TAMQ protocol capture" in markdown + assert "Accepted messages: 1" in markdown + assert "## Review cues" in markdown + assert markdown.count("Please review.") == 1 + assert "does not contain unrelated pane output" in markdown + + records = [json.loads(line) for line in render_jsonl(rows).splitlines()] + assert records[0]["message_body"] == "Please review.\nReply accepted." + assert "message_body" not in records[1] + store.close() + + +def test_capture_cli_filters_and_writes_review_file(tmp_path, monkeypatch, capsys): + database = tmp_path / "queue.sqlite3" + monkeypatch.setattr("tamq.cli.db_path", lambda: database) + store = Store(database) + store.add("one", "two", "hello", provenance="operator_input") + store.add("three", "four", "ignore", provenance="worker_output") + store.close() + output = tmp_path / "review" / "protocol.md" + + assert main(["capture", "--repo", "one", "--output", str(output)]) == 0 + assert "Wrote 1 protocol event" in capsys.readouterr().out + report = output.read_text() + assert "hello" in report + assert "ignore" not in report + + +def test_capture_cli_rejects_non_positive_limit(tmp_path, monkeypatch, capsys): + monkeypatch.setattr("tamq.cli.db_path", lambda: tmp_path / "queue.sqlite3") + assert main(["capture", "--limit", "0"]) == 2 + assert "capture limit must be a positive integer" in capsys.readouterr().err diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py index 6860182..4ea54ea 100644 --- a/tests/test_cli_helpers.py +++ b/tests/test_cli_helpers.py @@ -5,7 +5,10 @@ from tamq.cli import ( parse_size, ) from tamq.service import ( + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, PUSHY_FRAMING_CAPABILITY, TRIGGER_CAPABILITY, ) @@ -17,7 +20,7 @@ def test_parse_size(): def test_manual_service_restarts_legacy_broker(monkeypatch): - capabilities = iter([[], ["manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY]]) + capabilities = iter([[], ["manual_delivery", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY, PROTOCOL_CAPTURE_CAPABILITY, DELIVERY_RELIABILITY_CAPABILITY, IDEMPOTENT_SEND_CAPABILITY]]) starts = [] stops = [] @@ -37,7 +40,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", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY], + ["manual_delivery", "terminal_output", PUSHY_FRAMING_CAPABILITY, LINE_LIMITS_CAPABILITY, PROTOCOL_CAPTURE_CAPABILITY, DELIVERY_RELIABILITY_CAPABILITY, IDEMPOTENT_SEND_CAPABILITY], ]) starts = [] stops = [] @@ -64,6 +67,9 @@ def test_pushy_service_requires_pushy_input_capability(monkeypatch): PUSHY_FRAMING_CAPABILITY, TRIGGER_CAPABILITY, LINE_LIMITS_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, ], ]) starts = [] diff --git a/tests/test_client_adapter.py b/tests/test_client_adapter.py new file mode 100644 index 0000000..0bbb994 --- /dev/null +++ b/tests/test_client_adapter.py @@ -0,0 +1,202 @@ +import asyncio +import json + +import pytest + +from tamq.client import ( + CoordinationEngineAdapter, + TamqClient, + TamqClientError, + TamqProtocolError, + TamqTargetUnavailable, + WakeRequest, +) +from tamq.service import Service +from tamq.store import Store + + +async def _start_service(path, store): + service = Service(path, store) + server = await asyncio.start_unix_server(service.handle, path=str(path)) + return service, server + + +def test_coordination_adapter_negotiates_and_deduplicates_wake(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.service.validate_targets", lambda repos: None) + + async def run(): + path = tmp_path / "tamq.sock" + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint( + "ep-1", 42, "tamq", ["target"], "manual", + delivery_ack_mode="acknowledged", + ) + service, server = await _start_service(path, store) + adapter = CoordinationEngineAdapter(TamqClient(path)) + wake = WakeRequest( + lease_id="lease-7", + trigger_id="trigger-9", + target_repo="target", + prompt="Continue task T1.", + ) + try: + first = await adapter.wake(wake) + repeated = await adapter.wake(wake) + assert repeated.message_id == first.message_id + assert first.deduplicated is False + assert repeated.deduplicated is True + row = await adapter.receipt(first.message_id) + assert row["client_id"] == "coordination-engine" + assert row["idempotency_key"] == "lease-7" + assert row["correlation_id"] == "trigger-9" + assert json.loads(row["envelope_metadata"])["trigger_id"] == "trigger-9" + assert len(store.list()) == 1 + assert store.line_state("ep-1", "coordination-engine")["messages"] == 1 + await adapter.acknowledge(first.message_id) + assert (await adapter.receipt(first.message_id))["state"] == "acknowledged" + finally: + server.close() + await server.wait_closed() + service.close() + + asyncio.run(run()) + + +def test_client_reconnects_and_recovers_receipt_from_durable_store(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.service.validate_targets", lambda repos: None) + + async def run(): + path = tmp_path / "tamq.sock" + database = tmp_path / "queue.sqlite3" + store = Store(database) + store.register_endpoint("ep", 42, "tamq", ["target"], "manual") + first_service, first_server = await _start_service(path, store) + client = TamqClient(path) + sent = await client.send( + sender_repo="coordination-engine", + target_repo="target", + body="wake", + endpoint_id="ep", + idempotency_key="lease-restart", + ) + first_server.close() + await first_server.wait_closed() + first_service.close() + path.unlink(missing_ok=True) + + reopened = Store(database) + second_service, second_server = await _start_service(path, reopened) + try: + receipt = await client.message(sent["message_id"]) + assert receipt["state"] == "pending" + repeated = await client.send( + sender_repo="coordination-engine", + target_repo="target", + body="wake", + endpoint_id="ep", + idempotency_key="lease-restart", + ) + assert repeated["deduplicated"] is True + finally: + second_server.close() + await second_server.wait_closed() + second_service.close() + + asyncio.run(run()) + + +def test_adapter_rejects_incompatible_protocol_and_disappeared_endpoint( + tmp_path, monkeypatch +): + monkeypatch.setattr("tamq.service.validate_targets", lambda repos: None) + + async def run(): + path = tmp_path / "tamq.sock" + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("ep", 42, "tamq", ["target"], "manual") + service, server = await _start_service(path, store) + try: + with pytest.raises(TamqProtocolError): + await TamqClient(path, protocol="2.0").negotiate() + store.disconnect_endpoint("ep") + adapter = CoordinationEngineAdapter(TamqClient(path)) + with pytest.raises(TamqTargetUnavailable): + await adapter.wake( + WakeRequest("lease", "target", "wake", "trigger") + ) + finally: + server.close() + await server.wait_closed() + service.close() + + asyncio.run(run()) + + +def test_repeated_wake_rebinds_same_message_after_endpoint_replacement( + tmp_path, monkeypatch +): + monkeypatch.setattr("tamq.service.validate_targets", lambda repos: None) + + async def run(): + path = tmp_path / "tamq.sock" + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint("old", 41, "tamq-old", ["target"], "manual") + service, server = await _start_service(path, store) + adapter = CoordinationEngineAdapter(TamqClient(path)) + wake = WakeRequest("lease-rebind", "target", "wake", "trigger") + try: + first = await adapter.wake(wake) + store.disconnect_endpoint("old") + store.register_endpoint("new", 42, "tamq-new", ["target"], "manual") + repeated = await adapter.wake(wake) + assert repeated.message_id == first.message_id + assert repeated.endpoint_id == "new" + assert repeated.deduplicated is True + assert store.message(first.message_id)["endpoint_id"] == "new" + rebound = store.protocol_events( + message_id=first.message_id, event_type="message.rebound" + ) + assert len(rebound) == 1 + finally: + server.close() + await server.wait_closed() + service.close() + + asyncio.run(run()) + + +def test_idempotency_key_conflict_is_rejected(tmp_path, monkeypatch): + monkeypatch.setattr("tamq.service.validate_targets", lambda repos: None) + + async def run(): + path = tmp_path / "tamq.sock" + store = Store(tmp_path / "queue.sqlite3") + service, server = await _start_service(path, store) + client = TamqClient(path) + try: + first = await client.send( + sender_repo="coordination-engine", + target_repo="one", + body="first", + idempotency_key="same", + ) + with pytest.raises(TamqClientError, match="different message"): + await client.send( + sender_repo="coordination-engine", + target_repo="two", + body="second", + idempotency_key="same", + ) + store.db.execute( + "UPDATE messages SET state='failed',attempt_count=4 WHERE message_id=?", + (first["message_id"],), + ) + store.db.commit() + retried = await client.retry(first["message_id"]) + assert retried["state"] == "pending" + finally: + server.close() + await server.wait_closed() + service.close() + + asyncio.run(run()) diff --git a/tests/test_config_command.py b/tests/test_config_command.py index 5a3265c..cf72279 100644 --- a/tests/test_config_command.py +++ b/tests/test_config_command.py @@ -12,3 +12,6 @@ def test_config_command(monkeypatch, capsys, tmp_path): assert result["maxout"] == 32768 assert result["database"].endswith("tamq.sqlite3") assert result["policy_profile"] == "default" + assert result["delivery_ack_mode"] == "injected" + assert result["delivery_max_attempts"] == 4 + assert result["ack_timeout_seconds"] == 30 diff --git a/tests/test_install_target.py b/tests/test_install_target.py index daec056..7e923b5 100644 --- a/tests/test_install_target.py +++ b/tests/test_install_target.py @@ -27,6 +27,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): bin_dir = tmp_path / "bin" runtime_dir = tmp_path / "runtime" state_dir = tmp_path / "state" + config = tmp_path / "config.toml" repo_a = tmp_path / "railiance-platform" repo_b = tmp_path / "activity-core" for path in (bin_dir, runtime_dir, repo_a, repo_b): @@ -52,6 +53,12 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ) gita.chmod(0o755) agent.chmod(0o755) + config.write_text( + "[policy.profiles.integration]\n" + "delivery_ack_mode='acknowledged'\n" + "delivery_max_attempts=3\n", + encoding="utf-8", + ) socket_name = f"tamq-installed-{os.getpid()}" env = os.environ.copy() @@ -63,6 +70,8 @@ def test_isolated_installed_tool_session_smoke(tmp_path): "UV_CACHE_DIR": str(tmp_path / "uv-cache"), "XDG_RUNTIME_DIR": str(runtime_dir), "TAMQ_STATE_DIR": str(state_dir), + "TAMQ_CONFIG": str(config), + "TAMQ_POLICY_PROFILE": "integration", "TAMQ_TMUX_SOCKET": socket_name, } ) @@ -202,7 +211,7 @@ def test_isolated_installed_tool_session_smoke(tmp_path): ).stdout assert inbox[-1]["sender_repo"] == "railiance-platform" assert inbox[-1]["body"] == "installed-message" - assert inbox[-1]["state"] == "pending" + assert inbox[-1]["state"] == "awaiting_ack" assert inbox[-1]["displayed_at"] is not None message_id = inbox[-1]["message_id"] assert run(str(tamq), "inbox", "--repo", "activity-core").stdout == ( diff --git a/tests/test_policy.py b/tests/test_policy.py index f68e466..46fb07e 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -9,6 +9,18 @@ def test_default_policy(tmp_path): def test_configured_policy(tmp_path): path = tmp_path / "config.toml" - path.write_text("[policy.profiles.dev]\nallow=['repo_inspect']\nrequire_human=['destructive']\nsafety_gated_max_attempts=2\n") + path.write_text("[policy.profiles.dev]\nallow=['repo_inspect']\nrequire_human=['destructive']\nsafety_gated_max_attempts=2\ndelivery_max_attempts=3\n") profile = load_profile(path, "dev") assert profile.safety_gated_max_attempts == 2 + assert profile.delivery_max_attempts == 3 + + +def test_delivery_attempt_cap_is_bounded(tmp_path): + path = tmp_path / "config.toml" + path.write_text("[policy.profiles.bad]\ndelivery_max_attempts=10\n") + try: + load_profile(path, "bad") + except ValueError as exc: + assert str(exc) == "delivery_max_attempts must be between 1 and 9" + else: + raise AssertionError("invalid delivery cap was accepted") diff --git a/tests/test_protocol.py b/tests/test_protocol.py index c46b83a..038c515 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,7 +1,13 @@ import asyncio import json -from tamq.service import PUSHY_FRAMING_CAPABILITY, Service +from tamq.service import ( + DELIVERY_RELIABILITY_CAPABILITY, + IDEMPOTENT_SEND_CAPABILITY, + PROTOCOL_CAPTURE_CAPABILITY, + PUSHY_FRAMING_CAPABILITY, + Service, +) from tamq.store import Store @@ -32,6 +38,9 @@ def test_ping_advertises_non_routable_pushy_framing(tmp_path): await writer.drain() response = json.loads(await reader.readline()) assert PUSHY_FRAMING_CAPABILITY in response["capabilities"] + assert PROTOCOL_CAPTURE_CAPABILITY in response["capabilities"] + assert DELIVERY_RELIABILITY_CAPABILITY in response["capabilities"] + assert IDEMPOTENT_SEND_CAPABILITY in response["capabilities"] writer.close() await writer.wait_closed() finally: diff --git a/tests/test_replay.py b/tests/test_replay.py index 8afbf95..54b3bff 100644 --- a/tests/test_replay.py +++ b/tests/test_replay.py @@ -8,16 +8,24 @@ def test_replay_reports_batch(tmp_path, monkeypatch, capsys): source = tmp_path / "messages.jsonl" source.write_text(json.dumps({"message_id": "old-1", "sender_repo": "a", "target_repo": "b", "body": "hello"}) + "\n") monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setattr("tamq.cli.validate_targets", lambda targets: None) assert main(["replay", str(source)]) == 0 output = json.loads(capsys.readouterr().out) assert output["batch_id"].startswith("replay-") assert output["count"] == 1 + assert output["deduplicated"] == 0 + assert main(["replay", str(source)]) == 0 + repeated = json.loads(capsys.readouterr().out) + assert repeated["count"] == 0 + assert repeated["deduplicated"] == 1 + assert len(Store(tmp_path / "state" / "tamq.sqlite3").list()) == 1 def test_replay_preserves_endpoint(tmp_path, monkeypatch, capsys): source = tmp_path / "messages.jsonl" source.write_text(json.dumps({"sender_repo": "a", "target_repo": "b", "body": "hello", "endpoint_id": "tmux-amq-42"}) + "\n") monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setattr("tamq.cli.validate_targets", lambda targets: None) assert main(["replay", str(source)]) == 0 row = Store(tmp_path / "state" / "tamq.sqlite3").list()[0] assert row["endpoint_id"] == "tmux-amq-42" @@ -27,5 +35,6 @@ def test_replay_rejects_bad_json(tmp_path, monkeypatch, capsys): source = tmp_path / "bad.jsonl" source.write_text("not-json\n") monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setattr("tamq.cli.validate_targets", lambda targets: None) assert main(["replay", str(source)]) == 2 assert "cannot replay" in capsys.readouterr().err diff --git a/tests/test_retry.py b/tests/test_retry.py new file mode 100644 index 0000000..cca17bd --- /dev/null +++ b/tests/test_retry.py @@ -0,0 +1,27 @@ +from tamq.cli import main +from tamq.store import Store + + +def test_retry_cli_resets_only_terminal_failures(tmp_path, monkeypatch, capsys): + state = tmp_path / "state" + monkeypatch.setenv("TAMQ_STATE_DIR", str(state)) + store = Store(state / "tamq.sqlite3") + failed = store.add("a", "b", "failed") + pending = store.add("a", "b", "pending") + store.db.execute( + "UPDATE messages SET state='failed',attempt_count=4,last_failure_reason='OSError' " + "WHERE message_id=?", + (failed,), + ) + store.db.commit() + store.close() + + assert main(["retry", failed]) == 0 + assert capsys.readouterr().out.strip() == failed + reopened = Store(state / "tamq.sqlite3") + assert reopened.message(failed)["state"] == "pending" + assert reopened.message(failed)["attempt_count"] == 0 + reopened.close() + + assert main(["retry", pending]) == 1 + assert "not failed" in capsys.readouterr().err diff --git a/tests/test_service_delivery.py b/tests/test_service_delivery.py index 74fd0fd..3b9e39c 100644 --- a/tests/test_service_delivery.py +++ b/tests/test_service_delivery.py @@ -1,3 +1,5 @@ +import json + from tamq.control import PaneDisplay from tamq.service import Service from tamq.store import Store @@ -96,6 +98,13 @@ def test_service_trigger_mode_submits_exactly_once(tmp_path, monkeypatch): service._deliver_once() assert control.submitted == [("tamq:repo-b", "From:repo-a/o: go")] assert control.placed == [] + lifecycle = store.protocol_events(message_id=store.list()[0]["message_id"]) + assert [event["event_type"] for event in lifecycle] == [ + "message.accepted", + "delivery.attempted", + "delivery.injected", + ] + assert lifecycle[-1]["delivery_mode"] == "trigger" def test_failed_pushy_submission_remains_pending(tmp_path, monkeypatch): @@ -111,6 +120,12 @@ def test_failed_pushy_submission_remains_pending(tmp_path, monkeypatch): Service(store=store, input_grace=0)._deliver_once() assert store.list()[0]["state"] == "pending" + failure = store.protocol_events(event_type="delivery.failed")[0] + assert failure["delivery_mode"] == "pushy" + detail = json.loads(failure["detail"]) + assert detail["reason"] == "RuntimeError" + assert detail["attempt"] == 1 + assert detail["attempt_limit"] == 4 def test_input_delivery_waits_for_new_endpoint_grace(tmp_path, monkeypatch): @@ -126,7 +141,7 @@ def test_input_delivery_waits_for_new_endpoint_grace(tmp_path, monkeypatch): assert store.list()[0]["state"] == "pending" -def test_service_writes_output_once_and_retains_pending_ack(tmp_path, monkeypatch): +def test_default_injected_policy_completes_output_delivery_once(tmp_path, monkeypatch): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"], "output") message_id = store.add("repo-a", "repo-b", "continue") @@ -148,7 +163,7 @@ def test_service_writes_output_once_and_retains_pending_ack(tmp_path, monkeypatc ) ] row = store.list()[0] - assert row["state"] == "pending" + assert row["state"] == "injected" assert row["displayed_at"] is not None assert store.db.execute("SELECT COUNT(*) FROM leases").fetchone()[0] == 0 @@ -166,18 +181,125 @@ def test_failed_terminal_output_remains_undisplayed_and_retryable(tmp_path, monk raise OSError("temporary failure") monkeypatch.setattr("tamq.service.write_terminal_output", fail_once) - service = Service(store=store) + service = Service(store=store, retry_backoff=(0,)) service._deliver_once() assert store.list()[0]["displayed_at"] is None - store.db.execute("UPDATE leases SET expires_at=0") - store.db.commit() service._deliver_once() assert len(attempts) == 2 assert store.list()[0]["displayed_at"] is not None +def test_acknowledged_policy_waits_then_redelivers_same_message(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint( + "ep", 9, "tamq", ["repo-b"], "trigger", + delivery_ack_mode="acknowledged", + delivery_max_attempts=3, + ack_timeout_seconds=10, + ) + message_id = store.add("repo-a", "repo-b", "confirm") + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + service = Service(store=store, input_grace=0, retry_backoff=(0,)) + + service._deliver_once() + first_control = FakeControl.instances[-1] + assert store.message(message_id)["state"] == "awaiting_ack" + assert store.message(message_id)["attempt_count"] == 1 + assert len(first_control.submitted) == 1 + service._deliver_once() + assert FakeControl.instances[-1].submitted == [] + + store.db.execute( + "UPDATE messages SET next_attempt_at=0 WHERE message_id=?", (message_id,) + ) + store.db.commit() + service._deliver_once() + second_control = FakeControl.instances[-1] + assert store.message(message_id)["attempt_count"] == 2 + assert len(second_control.submitted) == 1 + assert store.acknowledge(message_id) is True + service._deliver_once() + assert FakeControl.instances[-1].submitted == [] + + +def test_delivery_attempt_cap_is_terminal_and_operator_retry_resets(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint( + "ep", 9, "tamq", ["repo-b"], "pushy", delivery_max_attempts=2 + ) + message_id = store.add("repo-a", "repo-b", "fail") + + class FailingControl(FakeControl): + def place(self, window, text): + raise OSError("no pane") + + monkeypatch.setattr("tamq.service.ControlModeClient", FailingControl) + service = Service(store=store, input_grace=0, retry_backoff=(0,)) + service._deliver_once() + service._deliver_once() + service._deliver_once() + + row = store.message(message_id) + assert row["state"] == "failed" + assert row["attempt_count"] == 2 + assert row["last_failure_reason"] == "OSError" + assert store.retry_message(message_id) is True + assert store.message(message_id)["state"] == "pending" + assert store.message(message_id)["attempt_count"] == 0 + + +def test_ack_timeout_exhaustion_is_terminal_but_late_ack_wins(tmp_path, monkeypatch): + store = Store(tmp_path / "queue.sqlite3") + store.register_endpoint( + "ep", 9, "tamq", ["repo-b"], "trigger", + delivery_ack_mode="acknowledged", + delivery_max_attempts=2, + ack_timeout_seconds=10, + ) + message_id = store.add("repo-a", "repo-b", "confirm") + monkeypatch.setattr("tamq.service.ControlModeClient", FakeControl) + service = Service(store=store, input_grace=0, retry_backoff=(0,)) + + service._deliver_once() + for _ in range(2): + store.db.execute( + "UPDATE messages SET next_attempt_at=0 WHERE message_id=?", (message_id,) + ) + store.db.commit() + service._deliver_once() + + row = store.message(message_id) + assert row["state"] == "failed" + assert row["attempt_count"] == 2 + assert row["last_failure_reason"] == "ack_timeout" + assert store.acknowledge(message_id) is True + assert store.message(message_id)["state"] == "acknowledged" + assert store.protocol_events(message_id=message_id)[-1]["outcome"] == "late_acknowledged" + + +def test_expired_delivery_lease_consumes_attempt_and_schedules_retry(tmp_path): + store = Store(tmp_path / "queue.sqlite3") + message_id = store.add("repo-a", "repo-b", "lease") + claim = store.claim_delivery( + message_id, + "ep", + attempt_limit=2, + retry_delay=7, + delivery_mode="output", + ttl=1, + now=10, + ) + assert claim is not None + assert store.expire_delivery_leases(now=12) == 1 + row = store.message(message_id) + assert row["state"] == "pending" + assert row["attempt_count"] == 1 + assert row["last_failure_reason"] == "lease_expired" + assert row["next_attempt_at"] == 19 + + def test_service_disconnects_disappeared_tmux_endpoint(tmp_path, monkeypatch): store = Store(tmp_path / "queue.sqlite3") store.register_endpoint("tmux-amq-9-boot", 9, "tamq", ["repo-b"]) diff --git a/tests/test_store.py b/tests/test_store.py index 6bcf9a3..86aeb28 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -41,10 +41,58 @@ 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] == "4" + ).fetchone()[0] == "6" assert "displayed_at" in { row["name"] for row in store.db.execute("PRAGMA table_info(messages)") } + assert store.db.execute( + "SELECT COUNT(*) FROM protocol_events" + ).fetchone()[0] == 0 + + +def test_protocol_ledger_is_atomic_and_not_duplicated_on_reopen(tmp_path): + path = tmp_path / "queue.sqlite3" + store = Store(path) + message_id = store.add( + "source", "target", "first\nsecond", provenance="worker_output" + ) + event = store.protocol_events(message_id=message_id)[0] + assert event["event_type"] == "message.accepted" + assert event["repo"] == "source" + assert event["peer_repo"] == "target" + assert json.loads(event["detail"]) == {"line_count": 2} + store.close() + + reopened = Store(path) + assert len(reopened.protocol_events(message_id=message_id)) == 1 + reopened.close() + + +def test_schema_v5_backfills_existing_message_lifecycle(tmp_path): + path = tmp_path / "legacy-messages.sqlite3" + store = Store(path) + message_id = store.add("a", "b", "old") + store.db.execute("DELETE FROM protocol_events") + store.db.execute( + "UPDATE messages SET displayed_at=2,injected_at=3,acknowledged_at=4 " + "WHERE message_id=?", + (message_id,), + ) + store.db.commit() + store.close() + + migrated = Store(path) + assert [row["event_type"] for row in migrated.protocol_events()] == [ + "delivery.displayed", + "delivery.injected", + "message.acknowledged", + "message.accepted", + ] + assert all( + json.loads(row["detail"])["backfilled"] is True + for row in migrated.protocol_events() + ) + migrated.close() def test_mark_displayed_releases_lease_without_acknowledging(tmp_path): diff --git a/workplans/TAMQ-WP-0002-coordination-engine-adapter.md b/workplans/TAMQ-WP-0002-coordination-engine-adapter.md index d8ac374..56efc4e 100644 --- a/workplans/TAMQ-WP-0002-coordination-engine-adapter.md +++ b/workplans/TAMQ-WP-0002-coordination-engine-adapter.md @@ -4,11 +4,11 @@ type: workplan title: "Coordination-engine adapter boundary" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social created: "2026-08-24" -updated: "2026-08-24" +updated: "2026-08-26" state_hub_workstream_id: "48da5fa7-b7bf-5d13-8edc-b90f266e241c" --- @@ -22,7 +22,7 @@ workflow/orchestration policy in coordination-engine. ```task id: TAMQ-WP-0002-T01 -status: todo +status: done priority: high state_hub_task_id: "a222dab7-edea-58c6-9dab-d684eb65c412" ``` @@ -31,11 +31,17 @@ Define request, response, identity, delivery, acknowledgement, retry, and failure semantics for a coordination-engine client. Resolve protocol-version negotiation and document which system owns each state transition. +The implemented contract is `spec/coordination-engine-adapter-v0.1.md`. It +defines same-user socket authentication, major-version and capability +negotiation, exact endpoint resolution, message/correlation identity, +idempotency conflicts, bounded delivery states, ack timeouts, late +acknowledgement, recovery, and the TAMQ/coordination ownership boundary. + ## Implement the coordination-engine client adapter ```task id: TAMQ-WP-0002-T02 -status: wait +status: done priority: high state_hub_task_id: "195db613-05a3-58d8-9e94-2db34920576f" ``` @@ -43,11 +49,18 @@ state_hub_task_id: "195db613-05a3-58d8-9e94-2db34920576f" Implement the adapter against the stable socket contract without importing tmux/control-mode concerns into coordination-engine. This task waits for T01. +`tamq.client` provides a transport-only async client and +`CoordinationEngineAdapter`. It opens a fresh socket connection per operation, +resolves one live repository endpoint, maps the coordination lease ID to a +durable idempotency key, preserves trigger correlation metadata, and exposes +receipt, acknowledgement, and terminal retry operations. Protocol constants +are separate from the service and terminal implementation. + ## Prove interoperability and recovery ```task id: TAMQ-WP-0002-T03 -status: wait +status: done priority: high state_hub_task_id: "23c7f6c7-be48-5b6c-b85c-ae05d82cc277" ``` @@ -55,3 +68,18 @@ state_hub_task_id: "23c7f6c7-be48-5b6c-b85c-ae05d82cc277" Add end-to-end coverage for delivery, acknowledgement, reconnect, replay, deduplication, incompatible protocol versions, and endpoint disappearance. Update both repositories' operator documentation with the verified workflow. + +Unix-socket integration tests cover negotiation, identical-wake deduplication, +idempotency conflict, durable receipt recovery after service restart, +incompatible protocol, and disappeared endpoints. Reliability tests cover +delivery, explicit and late acknowledgement, same-ID ack-timeout redelivery, +lease expiry, exhaustion, and retry reset. JSONL replay now uses deterministic +idempotency while retaining normal gita validation. TAMQ documentation and the +coordination-engine worker-service specification link the verified contract. + +## Residuals + +This workplan does not implement coordination-engine's trigger observer, +coordination leases, actionability decisions, checkpoints, or State Hub +projection. Those remain owned by `COORDINATION-WP-0003` and consume this +adapter rather than expanding TAMQ's scope. diff --git a/workplans/TAMQ-WP-0003-delivery-reliability.md b/workplans/TAMQ-WP-0003-delivery-reliability.md index 13cc3ec..cd24c17 100644 --- a/workplans/TAMQ-WP-0003-delivery-reliability.md +++ b/workplans/TAMQ-WP-0003-delivery-reliability.md @@ -4,11 +4,11 @@ type: workplan title: "Delivery reliability and practical integration evidence" domain: communication repo: tmux-amq -status: active +status: finished owner: codex topic_slug: coulomb-social created: "2026-08-24" -updated: "2026-08-24" +updated: "2026-08-26" state_hub_workstream_id: "7961275b-8f1f-5827-b9fc-46b3ac35fb73" --- @@ -22,7 +22,7 @@ stable coordination-engine dependency. ```task id: TAMQ-WP-0003-T01 -status: todo +status: done priority: high state_hub_task_id: "52c43745-5c28-599e-b74c-0386d5b51e68" ``` @@ -31,11 +31,18 @@ Persist attempt counts and failure reasons, apply the selected policy profile's retry cap, define lease-expiry behavior, and introduce an inspectable terminal failure state. Prove restart-safe behavior and avoid tight retry loops. +Completed with schema-v6 persistent attempt, deadline, and failure fields. +Lease acquisition consumes an attempt; write failure or lease expiry releases +ownership into configured bounded backoff. The selected endpoint policy caps +attempts from one to nine, exhaustion becomes `failed`, and `tamq retry` is the +explicit recovery action. Protocol events retain attempt, cap, reason, and +retry timing without raw terminal output. + ## Enforce acknowledgement semantics ```task id: TAMQ-WP-0003-T02 -status: wait +status: done priority: high state_hub_task_id: "ff88d6c9-f4d7-52c7-9645-7c3a7a34ccb4" ``` @@ -44,6 +51,13 @@ Make `delivery_ack_mode` control whether injection completes delivery or waits for explicit recipient acknowledgement. Specify timeout, redelivery, duplicate, and late-acknowledgement behavior. This task follows the state model from T01. +Completed for every automatic delivery mode. `injected` completes after a +successful terminal write. `acknowledged` enters `awaiting_ack`, schedules +same-message-ID redelivery after the configured deadline, records `ack_timeout` +on retries and exhaustion, and accepts a late acknowledgement even after +`failed`. The contract warns that redelivery can duplicate terminal +presentation and never represents task completion. + ## Add real tmux and PTY lifecycle coverage ```task @@ -70,7 +84,7 @@ entirely as documented. ```task id: TAMQ-WP-0003-T04 -status: wait +status: done priority: medium state_hub_task_id: "44939829-a52e-582a-b06b-60337b36080e" ``` @@ -78,3 +92,14 @@ state_hub_task_id: "44939829-a52e-582a-b06b-60337b36080e" Run the integration suite in Forgejo CI with explicit tmux and gita setup. Update README claims about retries, acknowledgement, prerequisites, and maturity from verified behavior. This task follows T01-T03. + +Forgejo CI now runs `make check`, including generated-gita and isolated tmux +installation fixtures, compile/diff checks, and CLI/capture help smokes on +Python 3.11. README, SCOPE, configuration examples, and the agent introduction +describe the implemented state machine and terminal-comprehension boundary. + +## Residuals + +Long-duration and arbitrary-terminal soak coverage remains a maturity gate in +SCOPE rather than a missing part of this bounded reliability contract. The +coordination runtime above TAMQ remains owned by coordination-engine. diff --git a/workplans/TAMQ-WP-0016-structured-protocol-capture.md b/workplans/TAMQ-WP-0016-structured-protocol-capture.md new file mode 100644 index 0000000..f30559b --- /dev/null +++ b/workplans/TAMQ-WP-0016-structured-protocol-capture.md @@ -0,0 +1,86 @@ +--- +id: TAMQ-WP-0016 +type: workplan +title: "Structured communication protocol capture" +domain: communication +repo: tmux-amq +status: finished +owner: codex +topic_slug: structured-protocol-capture +created: "2026-08-25" +updated: "2026-08-25" +--- + +# Structured communication protocol capture + +Retain reviewable evidence of TAMQ exchanges so agent onboarding and protocol +friction can be assessed without recording arbitrary terminal sessions. + +## Define the evidence and privacy boundary + +```task +id: TAMQ-WP-0016-T01 +status: done +priority: high +``` + +Capture only TAMQ protocol facts: addressed messages, provenance, worker block +boundaries, allowlisted commands, line-limit decisions, endpoint lifecycle, and +delivery outcomes. Do not capture unrelated pane output or ordinary shell +input. Retain a message body once at acceptance rather than duplicating it at +each lifecycle event. + +## Add an append-only protocol ledger + +```task +id: TAMQ-WP-0016-T02 +status: done +priority: high +``` + +Add structured protocol events to SQLite schema version 5. Make acceptance +events atomic with message admission, represent existing message lifecycle +history during migration, and retain stable message/endpoint/repository links +for filtering. + +## Instrument onboarding-sensitive transitions + +```task +id: TAMQ-WP-0016-T03 +status: done +priority: high +``` + +Record operator versus worker origin, worker block start and close reason, +invalid targets, limit blocks, operator-only command results, delivery mode, +attempt, success, and failure class. Never store unrecognized command text or +raw terminal output in event details. + +## Provide a review and analysis path + +```task +id: TAMQ-WP-0016-T04 +status: done +priority: medium +``` + +Add `tamq capture` with repository, endpoint, message, event-type, and bounded +event-count filters. Render a summarized Markdown review by default and stable +JSONL for analysis. Document the workflow and retention/privacy boundary in the +agent introduction, README, CLI help, and scope assessment. + +## Evidence + +- Schema migration backfills existing acceptance/display/injection/ack events + without duplicating newly captured events on later opens. +- Focused store, broker, service, renderer, and CLI tests cover atomic capture, + worker termination evidence, provenance, failure classes, filtering, and + Markdown/JSONL output. +- `make check` passes the complete suite. + +## Residuals + +The capture deliberately describes transport behavior, not recipient +comprehension or authorization. Long-term event retention and automated +protocol-quality recommendations can be added after real captures show which +summaries are useful.