feat(issue-5): persistent MQTT publisher prototype and benchmark harness

Splits publish_event.main into three pieces with no behaviour change:

  prepare_event   load the job, consume the seq, build the signed payload,
                  decide retain. The seq is burned here, before any transport
                  runs, so a failed publish can never leave a number a replayed
                  packet could reuse.
  record_outcome  audit record, registry event log and status sync -- all three
                  best effort, all three regardless of the publish result. That
                  is exactly the contract G-1/G-2/G-4 pin down, now in one place.
  main            parse, prepare, publish with retry, record.

publisher.py holds the two transports behind one interface. OneShotPublisher is
the current behaviour. PersistentPublisher keeps a single connection and lets
paho reconnect underneath, raising on an unacknowledged publish so the caller's
retry policy stays the only thing deciding attempt counts. It refuses an event
for a different broker rather than publishing to the wrong place.

Nothing in production uses the persistent path: the CLI and shell helpers exit
after one publish. The consumer is the phase 2a engine, which pays the CONNACK
cost on every event today -- this job's own started event took ~12.5s because
of one CONNACK timeout plus reconnect.

bench.py measures both arms. The fake transport is deterministic and is what
the tests assert on; the real-broker arm is informational, since the PoC
default is a public broker whose latency is not ours to control.

Verified: N publishes cost N connections one-shot and 1 persistent; the two
transports leave field-for-field identical registry and audit records including
the HMAC signature and retain flag; an unacknowledged publish retries on the
same connection and records published=false; a second broker is refused. No
test opens a socket.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

tests/test_issue5_phase1_delegation.py

@@ -720,3 +720,163 @@ def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry)
     published = [e for e in events if e.get("event") == "published"]
     assert published and published[-1]["published"] is False
     assert published[-1]["publish_error"]
+
+
+# ==============================================================================
+# 6. P1-03 — transports, parity and the benchmark harness (no network at all)
+# ==============================================================================
+
+def _prepared(topic="python/mqtt/jobs/j/events", host="broker.example", port=1883):
+    from lib_py.delegation import mqtt_common, publish_event
+
+    config = mqtt_common.broker_config_from_env({"host": host, "port": port})
+    return publish_event.PreparedEvent(
+        job_id="j", event="progress", detail="d", seq=1, topic=topic, retain=False,
+        payload={"seq": 1}, body=b'{"seq": 1}', config=config,
+    )
+
+
+def _fake_factory(counters, publishes_ok=True, connect_rc=0):
+    from lib_py.delegation.bench import FakeClient, FakeInfo
+
+    def factory(_role, _config):
+        client = FakeClient(connect_latency=0.0, counters=counters)
+        if not publishes_ok:
+            def unacked(*_a, **_k):
+                counters["publish"] = counters.get("publish", 0) + 1
+                return FakeInfo(published=False)
+            client.publish = unacked
+        if connect_rc:
+            original = client.connect
+
+            def connect(host, port, keepalive=60):
+                original(host, port, keepalive)
+                client.on_connect(client, None, None, connect_rc, None)
+            client.connect = connect
+        return client
+    return factory
+
+
+def test_persistent_publisher_reuses_one_connection():
+    """The whole point of the prototype: N publishes, 1 connection."""
+    from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher
+
+    prepared = _prepared()
+    one_counters, persist_counters = {}, {}
+    one = OneShotPublisher(client_factory=_fake_factory(one_counters))
+    persist = PersistentPublisher(client_factory=_fake_factory(persist_counters))
+    try:
+        for _ in range(5):
+            assert one.publish_with_retry(prepared, attempts=1) == (True, None)
+            assert persist.publish_with_retry(prepared, attempts=1) == (True, None)
+    finally:
+        persist.close()
+    assert one_counters["connect"] == 5
+    assert persist_counters["connect"] == 1
+    assert one_counters["publish"] == persist_counters["publish"] == 5
+
+
+def test_transport_parity_of_registry_and_audit_records(mam_sandbox, monkeypatch):
+    """Same input, two transports: registry and audit records must match field for field."""
+    from lib_py.delegation import mqtt_common, publish_event, registry
+    from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher
+
+    monkeypatch.chdir(mam_sandbox)
+    reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
+    os.makedirs(reg_dir, exist_ok=True)
+
+    results = {}
+    for label, publisher in (
+        ("one_shot", OneShotPublisher(client_factory=_fake_factory({}))),
+        ("persistent", PersistentPublisher(client_factory=_fake_factory({}))),
+    ):
+        job_id = registry.register_job("parity " + label, registry_dir=reg_dir)
+        prepared = publish_event.prepare_event(
+            job_id=job_id, event="completed", detail="done", registry_dir=reg_dir,
+        )
+        ok, err = publisher.publish_with_retry(prepared, attempts=1)
+        publish_event.record_outcome(prepared, reg_dir, ok, err)
+        if hasattr(publisher, "close"):
+            publisher.close()
+        record = mqtt_common.load_job(job_id, reg_dir)
+        events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
+        audit = [e for e in events if e.get("event") == "published"][-1]
+        results[label] = {
+            "ok": ok,
+            "status": record["status"],
+            "last_seq": record.get("last_seq"),
+            "retain": audit["retain"],
+            "published": audit["published"],
+            "seq": audit["seq"],
+            "has_sig": "hmac_sig" in audit["payload"].get("data", {}),
+            "source_event": audit["source_event"],
+        }
+
+    assert results["one_shot"] == results["persistent"], results
+    assert results["one_shot"]["retain"] is True, "terminal events must retain"
+    assert results["one_shot"]["status"] == "completed"
+
+
+def test_persistent_publisher_unacked_publish_retries_then_records_failure(mam_sandbox, monkeypatch):
+    """No PUBACK -> exception -> with_retry -> record_outcome(published=False)."""
+    from lib_py.delegation import mqtt_common, publish_event, registry
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    monkeypatch.chdir(mam_sandbox)
+    reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
+    os.makedirs(reg_dir, exist_ok=True)
+    job_id = registry.register_job("unacked", registry_dir=reg_dir)
+    prepared = publish_event.prepare_event(
+        job_id=job_id, event="completed", detail="d", registry_dir=reg_dir,
+    )
+    counters = {}
+    publisher = PersistentPublisher(client_factory=_fake_factory(counters, publishes_ok=False))
+    try:
+        ok, err = publisher.publish_with_retry(prepared, attempts=3)
+    finally:
+        publisher.close()
+    assert ok is False and "not acknowledged" in err
+    assert counters["publish"] == 3, "every attempt must reach the transport"
+    assert counters["connect"] == 1, "retries reuse the connection"
+
+    publish_event.record_outcome(prepared, reg_dir, ok, err)
+    record = mqtt_common.load_job(job_id, reg_dir)
+    assert record["status"] == "completed", "status syncs even when the publish failed"
+    assert record.get("last_seq") == prepared.seq
+    audit = [e for e in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())
+             if e.get("event") == "published"][-1]
+    assert audit["published"] is False and audit["publish_error"]
+
+
+def test_persistent_publisher_refuses_a_second_broker():
+    from lib_py.delegation.publisher import PersistentPublisher
+
+    publisher = PersistentPublisher(client_factory=_fake_factory({}))
+    try:
+        assert publisher.publish_with_retry(_prepared(host="a.example"), attempts=1)[0] is True
+        ok, err = publisher.publish_with_retry(_prepared(host="b.example"), attempts=1)
+        assert ok is False
+        assert "bound to a.example" in err, err
+    finally:
+        publisher.close()
+
+
+def test_bench_fake_transport_schema_and_counts():
+    from lib_py.delegation import bench
+
+    result = bench.run_benchmark(n=7, transport="fake", connect_latency_ms=1.0)
+    assert result["transport"] == "fake"
+    for arm in ("one_shot", "persistent"):
+        assert result[arm]["publishes"] == 7
+        assert result[arm]["failures"] == 0
+        assert set(result[arm]["latency_ms"]) == {"p50", "p95", "max"}
+    assert result["one_shot"]["client_connects"] == 7
+    assert result["persistent"]["client_connects"] == 1
+    assert result["persistent"]["latency_ms"]["p50"] < result["one_shot"]["latency_ms"]["p50"]
+
+
+def test_bench_broker_transport_requires_a_job():
+    from lib_py.delegation import bench
+
+    with pytest.raises(ValueError):
+        bench.run_benchmark(n=1, transport="broker")