fix(issue-5): record an accepted intent, not a refused one

The subscriber marked a notification intent as seen before the monotonic
sequence guard ran. A late packet was then dropped by that guard with its
intent already burned, so the genuine retry of a notification this
subscriber had never delivered was mistaken for a replay and dropped too.

Identity comparison stays ahead of the sequence guard -- a duplicate must
not consume a sequence number -- but the bookkeeping write moves behind
it, so "seen" means "accepted" rather than "arrived".

An intent that cannot yield a stable identity (not an object, or holding
an unhashable value) is now refused instead of raising out of the
callback: a paho callback that raises takes the network loop down for
every job that subscriber is watching.

Six regressions; three of them fail on the unfixed tree, and one drives
a real notifier retry through the real consumer with no hand-built
payload. Full suite: 732 passed, no skips.

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

tests/test_issue5_phase2b_autocomplete.py

@@ -954,3 +954,132 @@ def test_r4_events_without_an_intent_are_unaffected():
     assert watcher.events.qsize() == 2
     deliver(2)             # non-increasing: still dropped by the sequence guard
     assert watcher.events.qsize() == 2
+
+
+_MISSING = object()
+
+
+def _intent_watcher(last_seq=0):
+    from lib_py.delegation import job_subscriber
+    return job_subscriber._Watcher({"j1"}, {"j1": None}, {"j1": last_seq})
+
+
+def _deliver_intent(watcher, wire_seq, intent, event="completed"):
+    """Hand one schema-v1 event to the real callback. `intent` is placed in
+    data.notification_intent verbatim, so a malformed one can be modelled."""
+    import types
+    payload = {"schema_version": 1, "seq": wire_seq, "job_id": "j1",
+               "event": event, "timestamp": "t", "detail": "d", "data": {}}
+    if intent is not _MISSING:
+        payload["data"]["notification_intent"] = intent
+    watcher.on_message(None, None, types.SimpleNamespace(
+        topic="python/mqtt/jobs/j1/events",
+        payload=json.dumps(payload).encode()))
+
+
+def test_r4b_a_stale_packet_does_not_burn_the_intent_for_its_own_retry():
+    """The defect: the identity was recorded before the sequence guard ran, so a
+    late packet marked its intent as seen and the legitimate retry that followed
+    was mistaken for a replay of an event the subscriber never delivered."""
+    watcher = _intent_watcher(last_seq=10)
+    intent = {"seq": 7, "event": "completed", "timestamp": "T"}
+
+    _deliver_intent(watcher, 9, intent)     # late/out-of-order: refused by the seq guard
+    assert watcher.events.qsize() == 0
+    assert watcher.seen_intents.get("j1", set()) == set(), (
+        "a refused event must not leave its intent marked as accepted"
+    )
+
+    _deliver_intent(watcher, 11, intent)    # the genuine retry of that notification
+    assert watcher.events.qsize() == 1, "the retry of a never-delivered intent must be accepted"
+
+    _deliver_intent(watcher, 12, intent)    # now it really is a replay
+    assert watcher.events.qsize() == 1
+
+
+def test_r4b_a_duplicate_intent_does_not_advance_the_sequence_floor():
+    """Dropping a replay must not consume sequence numbers: the wire seq of an
+    event that was never delivered has no claim on the monotonic floor."""
+    watcher = _intent_watcher()
+    intent = {"seq": 7, "event": "completed", "timestamp": "T"}
+
+    _deliver_intent(watcher, 1, intent)
+    assert watcher.events.qsize() == 1 and watcher.last_seq["j1"] == 1
+    _deliver_intent(watcher, 9, intent)     # replay, refused
+    assert watcher.events.qsize() == 1
+    assert watcher.last_seq["j1"] == 1, "a refused replay must not raise the sequence floor"
+
+    _deliver_intent(watcher, 2, {"seq": 8, "event": "completed", "timestamp": "T"})
+    assert watcher.events.qsize() == 2, "an ordinary event after a refused replay still lands"
+
+
+def test_r4b_an_unhashable_intent_is_refused_instead_of_raising():
+    """A callback that raises takes down the network loop for every watched job,
+    so a malformed remote payload must be refused inside defensive parsing."""
+    watcher = _intent_watcher()
+    _deliver_intent(watcher, 1, {"seq": [1, 2], "event": "completed", "timestamp": "T"})
+    assert watcher.events.qsize() == 0
+    assert watcher.last_seq["j1"] == 0, "a refused event must not advance the floor"
+
+    # A well-formed intent on the same job is still accepted afterwards.
+    _deliver_intent(watcher, 2, {"seq": 7, "event": "completed", "timestamp": "T"})
+    assert watcher.events.qsize() == 1
+
+
+def test_r4b_a_non_object_intent_is_refused():
+    """Present but not an object: no stable identity can be derived, so the
+    replay guard could not protect it."""
+    watcher = _intent_watcher()
+    _deliver_intent(watcher, 1, "completed:7")
+    assert watcher.events.qsize() == 0
+    _deliver_intent(watcher, 2, ["completed", 7])
+    assert watcher.events.qsize() == 0
+
+
+def test_r4b_an_absent_or_null_intent_is_treated_as_no_intent():
+    """The ordinary path is unchanged: events that carry no intent are governed
+    by the sequence guard alone."""
+    watcher = _intent_watcher()
+    _deliver_intent(watcher, 1, _MISSING, event="progress")
+    _deliver_intent(watcher, 2, None, event="progress")
+    assert watcher.events.qsize() == 2
+    assert watcher.seen_intents.get("j1", set()) == set()
+    _deliver_intent(watcher, 2, None, event="progress")   # non-increasing
+    assert watcher.events.qsize() == 2
+
+
+def test_r4b_a_real_retry_is_deduplicated_by_the_real_consumer(mam_sandbox, tmp_path, monkeypatch):
+    """End to end, no hand-built payload: the notifier's own retry of an unsent
+    notification is handed to the real subscriber callback and must land once."""
+    import types
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import engine, job_subscriber, mqtt_common, notifier
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    record = _load(reg, job_id)
+    record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
+    Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8")
+    engine.transition(job_id, "claim", registry_dir=reg)
+    engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
+
+    # The broker is unreachable, so the intent stays owed and flush retries it.
+    notifier.flush(job_id, reg, attempts=1)
+    assert notifier.pending(job_id, reg), "an unsent notification stays owed"
+    notifier.flush(job_id, reg, attempts=1)
+
+    payloads = [e["payload"] for e
+                in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())
+                if e.get("event") == "published"]
+    assert len(payloads) == 2, "two send attempts for one notification"
+    assert payloads[0]["seq"] != payloads[1]["seq"], (
+        "the retry gets a fresh wire sequence: the monotonic guard cannot see it"
+    )
+
+    watcher = job_subscriber._Watcher({job_id}, {job_id: None},
+                                      {job_id: payloads[0]["seq"] - 1})
+    for payload in payloads:
+        watcher.on_message(None, None, types.SimpleNamespace(
+            topic="python/mqtt/jobs/%s/events" % job_id,
+            payload=json.dumps(payload).encode()))
+    assert watcher.events.qsize() == 1, "the same notification must be acted on once"