feat: complete reliable coordination adapter

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc

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):

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

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 = []

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())

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

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 == (

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")

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:

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

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

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"])

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):